Run parallel model research and stream the report to Python

Run this guide to receive a real investigation plan and a streamed report from the hosted Deep Research action. The Python client owns the question, deadline, cancellation and presentation; the backend owns its model composition. You can replace either side without changing the two named outputs between them.

Run a real investigation#

The Research in your browser button opens the existing hosted demo. Keep its default provider settings, enter Compare asyncio.TaskGroup and gather. Keep the final report under 200 words., then select Research. The plan appears beside the report. This is a live model run, so wording and latency vary.

To get the same result in your terminal, copy the commands in Run this guide above into an active Python 3.11+ virtual environment.

Those commands install the tested SDK version, download the complete client and run it. The shared backend supplies the model; your computer does not download model weights. The question is sent to that backend. Shared demo capacity is limited. An unavailable provider returns an error.

Watch for output in this order:

[plan] Research asyncio.TaskGroup's API and semantics…
[plan] Research asyncio.gather's behavior…
[plan] Find authoritative sources…
…a streamed comparison…
[completed]

Those are abbreviated observations from a real run. There can be a pause between the plan and report while the investigations run. Completion is a separate runtime result; receiving the first text does not mean the investigation succeeded.

Ask a second question with the same client#

After the TaskGroup comparison, try a service-design question. Pass it as the next command-line argument; the caller does not need a new schema for each topic:

python research_client.py "Compare queues and direct calls for a small Python service. Keep the report under 200 words."

The client first connects to the endpoint and discovers its action interface:

connection = await GatewayConnection.connect(
    args.endpoint, timeout=timing.Duration.seconds(15)
)
registry = a11.ActionRegistry()
await install_peer_actions(connection, registry)

The argument becomes the action’s topic input. Set the hosted demo’s public resource header before dispatching the call:

action = connection.action(
    "deep-research", registry.get_schema("deep-research")
)
action.set_header("x-a11-llm-api-key", "use-a11-demo-resources")
await action.call()
await action["topic"].finalize(args.topic)

connection reaches the hosted service. Discovery supplies its action schema. finalize sends the question and tells the server there is no more input to wait for. Your application does not need to know how many investigations the planner will choose.

Show the investigation plan while the report is still empty#

A research request can spend time investigating before it has a sentence of final prose. If the UI only reads the report, that period looks like nothing is happening. Showing the plan tells the reader which questions are being investigated.

The plan and report have different jobs. The plan explains what is about to happen; the report is the final synthesis. The client gives each a reader:

async def read_plan():
    async for brief in action["plan"]:
        print(f"\n[plan] {brief}", flush=True)

async def read_report():
    async for text in action["report"]:
        print(text, end="", flush=True)

Start both readers before waiting for either to finish:

async with asyncio.timeout(args.timeout):
    async with asyncio.TaskGroup() as readers:
        readers.create_task(read_plan())
        readers.create_task(read_report())
    await action.wait()
print("\n[completed]", flush=True)

An asyncio.TaskGroup starts both readers. The client can consume plan entries while the report reader waits for its first text. Waiting for the entire report before starting the plan reader would hide the plan during the period when it is most useful. In a UI these readers update separate views instead of sharing stdout.

Run independent investigations together, then combine their findings#

The TaskGroup comparison has independent questions about API shape, failure behavior and cancellation. Investigating each in sequence would make one wait for the others. The backend runs several investigations concurrently, then waits for their findings before starting synthesis.

The hosted implementation lives in deep_research.flow. Its public deep-research action coordinates three stages:

The server’s Flow composition contains the actual concurrency limit below. This is an excerpt from the linked server implementation, separate from the downloadable Python client:

for brief in planned.briefs parallel 3 {
  one = run investigate(topic: topic, brief: brief)
  one.report -> findings
  for line in one.narration {
    log info line
  }
}

written = run synthesise-findings(
  topic: topic,
  brief: planned.synthesis,
  findings: findings,
)
written.report -> report

parallel 3 bounds concurrent investigations; findings carries their reports into synthesis. The synthesis stage collects those findings before making its model call, then forwards the response pieces to the caller’s report output.

That is one planning call, one call per brief and one final call. Intermediate findings stay on the backend. The service registers the model actions and composition in web_demos_server.py.

Flow is how this particular hosted demo stores its composition. It is optional. The Python implementation in the research guide expresses the same plan/investigate/synthesize structure with coroutines, a semaphore and asyncio.gather. Both expose the same two output ports. A caller depends on that contract, not on the implementation language.

The hosted workload uses model knowledge and has no web-search action. For evidence-backed research, add a search or fetch action, retain the returned documents and pass those records into synthesis. Keep retrieved evidence separate from generated prose.

Cancel a question the user has replaced#

Suppose someone starts a research request, then changes the topic before the report arrives. The first request no longer belongs in the active result view. Cancel its action and keep any partial text associated with that original request, so it cannot appear as the answer to the replacement question.

Press Ctrl+C in the terminal to cancel the action and close its connection. A deadline covers a different case: the user still wants the answer, but the application has reached its waiting limit. Exercise that path:

python research_client.py --timeout 1

The asyncio.timeout block above cancels the reader tasks when its deadline expires. Cleanup belongs in finally, so an exception or Ctrl+C also closes the remote call:

finally:
    if action is not None:
        action.cancel()
    await connection.aclose()

The complete client sets action = None before entering try, so cleanup also works if discovery or dispatch fails. The timeout covers the running investigation; the connection has its own 15-second timeout. The client never prints [completed] after a timeout. A reader may already have seen a plan or part of the report, so mark that material as incomplete and offer a retry rather than presenting it as a finished investigation. Cancellation propagates through the A11 action. A provider request that has already been accepted follows that provider's cancellation semantics.

If the shared backend is busy, wait and retry. To use your own server, --endpoint ws://127.0.0.1:9010/a11-demos changes the connection; you must also configure its provider credentials. The public demo resource token in the source selects the hosted service’s configured credentials. Your own server needs its own provider configuration.

Complete source#

Download research_client.py. This is the file used by the launch commands, including discovery, concurrent readers, timeout and cleanup.

"""Stream a real plan and report from the hosted A11 research demo."""

import argparse
import asyncio

import a11
from a11 import timing
from a11.client.connection import GatewayConnection
from a11.client.discovery import install_peer_actions


async def main(args):
    connection = await GatewayConnection.connect(
        args.endpoint, timeout=timing.Duration.seconds(15)
    )
    action = None
    try:
        registry = a11.ActionRegistry()
        await install_peer_actions(connection, registry)
        action = connection.action(
            "deep-research", registry.get_schema("deep-research")
        )
        action.set_header("x-a11-llm-api-key", "use-a11-demo-resources")
        await action.call()
        await action["topic"].finalize(args.topic)

        async def read_plan():
            async for brief in action["plan"]:
                print(f"\n[plan] {brief}", flush=True)

        async def read_report():
            async for text in action["report"]:
                print(text, end="", flush=True)

        # Both ports stay moving, even while one is waiting for its first value.
        async with asyncio.timeout(args.timeout):
            async with asyncio.TaskGroup() as readers:
                readers.create_task(read_plan())
                readers.create_task(read_report())
            await action.wait()
        print("\n[completed]", flush=True)
    finally:
        if action is not None:
            action.cancel()
        await connection.aclose()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "topic",
        nargs="?",
        default=(
            "Compare asyncio.TaskGroup and gather. Keep the final report "
            "under 200 words."
        ),
    )
    parser.add_argument("--endpoint", default="wss://a11.to/ws/demoserver")
    parser.add_argument("--timeout", type=float, default=180)
    args = parser.parse_args()
    try:
        asyncio.run(main(args))
    except KeyboardInterrupt:
        print("\nStopped. The connection is closed.")
    except TimeoutError:
        raise SystemExit(
            "Timed out; the action was cancelled. Retry later or increase "
            "--timeout."
        )

After running the two-port interface, the debugging walkthrough gives you a small, predictable nested workflow with success, failure and cancellation cases.