Stream a model reply and thoughts separately

Run one A11 action that calls a real model and consume two outputs independently: the visible reply and the model's thoughts. The example uses the public A11 demo service, so it needs no account or API key. The action remains ordinary asynchronous Python around a replaceable model call.

Install A11#

Use Python 3.11 or later. In an existing virtual environment, installation is one command:

pip install a11-kit

For an empty directory, create an environment first:

python3 -m venv .venv
source .venv/bin/activate
pip install a11-kit

The example requires Python 3.11 or later.

Run both streams#

Download multiple_streams.py, save it in your working directory, then run:

python multiple_streams.py

The terminal labels values by the output they came from:

[a1b2c3#thoughts] The user wants a concise explanation…
[a1b2c3#reply] Streaming improves an AI application by showing useful…

The prefix contains the invocation ID and output name; your ID will differ, and both streams will vary. Reply and thought fragments belong to the same invocation and arrive while it is running. The landing demo uses Studio’s stream visualiser to show either stream as joined text or individual chunks.

Keep connection details separate from application behavior#

model_call is an ordinary async context manager. It opens the public demo connection, configures the provider-neutral interact_with_llm action and supplies the prompt. The provider, model and demo credential remain configuration on that call:

@asynccontextmanager
async def model_call(prompt: str):
    async with open_gateway(
        "wss://a11.to/ws/demoserver", allow_embedded=False
    ) as connection:
        interact = connection.action(
            "interact_with_llm", INTERACT_WITH_LLM_SCHEMA
        )
        interact.set_header(LlmHeaders.PROVIDER.value, "ollama")
        interact.set_header(LlmHeaders.MODEL.value, "glm-5.3-flash:cloud")
        interact.set_header(LlmHeaders.BASE_URL.value, "https://ollama.com")
        interact.set_header(
            LlmHeaders.API_KEY.value, "use-a11-demo-resources"
        )
        await interact.call()
        # Supply the prompt and yield interact to the action below.

The complete helper includes the portable Interaction input, drains outputs that this example does not display and closes the connection. An existing provider SDK call can occupy the same helper instead. The registered action still owns the behavior its callers observe.

Forward thoughts and reply separately#

The action adds the interface the application needs around that model call. Its outputs are real A11 AsyncNode instances, and Annotated attaches an OutputPort declaration to each one. Both declarations state directly that the output contains plain text:

@REGISTRY.action
async def answer(
    prompt: str,
    reply: Annotated[AsyncNode, OutputPort(mimetype="text/plain")],
    thoughts: Annotated[AsyncNode, OutputPort(mimetype="text/plain")],
):
    async with model_call(prompt) as interact:
        async for thought in interact["thoughts"]:
            await thoughts.put(str(thought))
        async for text in interact["text_output"]:
            await reply.put(str(text))
        await interact.wait()

The action contains the meaningful routing and lifecycle logic, while provider connection details stay out of its interface. A11 exposes thoughts and visible text as separate streams, and interact.wait() checks the model call's final status. A11 closes both action outputs when the handler finishes.

Start one call and read both streams#

The caller supplies the prompt and starts the action with run():

action = REGISTRY.make_action("answer").run()
await action["prompt"].finalize(
    "Explain why streaming helps an AI application in one sentence."
)

Each reader handles one named output:

async def _display(node: a11.AsyncNode):
    node_id = node.get_id()
    async for value in node:
        print(f"[{node_id}] {value}", flush=True)

async with asyncio.TaskGroup() as readers:
    readers.create_task(_display(action["reply"]))
    readers.create_task(_display(action["thoughts"]))
await action.wait()

Two TaskGroup readers consume the outputs concurrently, so a UI can present the visible answer separately from the model's thought stream. wait() checks the final status after both streams close.

Stop both streams together#

Press Stop on the landing page, or Ctrl+C in the terminal, to cancel the action. Already received text can remain visible, but the remote model call and both streams stop with the parent action. Running again starts a fresh invocation with empty output.

Run independent functions concurrently#

Multiple outputs describe one function’s interface. Running two functions at once is a separate capability. When two independent requests arrive, the second need not wait for the first stream to finish. Download the single-output action and its concurrent caller:

curl -fLO https://a11.to/examples/first_stream.py
curl -fLO https://a11.to/examples/parallel_streams.py
python parallel_streams.py

The caller imports the registry from first_stream.py. Each call gets its own invocation and input:

from first_stream import REGISTRY


async def read_one(text: str):
    action = REGISTRY.make_action("read").run()
    await action["text"].finalize(text)
    async for word in action["words"]:
        print(f"{text.split()[0]}: {word.strip()}", flush=True)
    await action.wait()

Ordinary Python tasks overlap the two invocations:

async def main():
    async with asyncio.TaskGroup() as tasks:
        tasks.create_task(read_one("One stream starts here"))
        tasks.create_task(read_one("Another stream runs alongside"))

The terminal interleaves lines prefixed One: and Another:. Each invocation retains its own output stream and final status. The research walkthrough applies concurrency to model investigations.

Start with an ordinary return value#

A function that returns one value does not need a streaming parameter. Here is the complete handler and registration from first_function.py:

import a11

REGISTRY = a11.ActionRegistry()


@REGISTRY.action(output="greeting")
async def greet(name: str) -> str:
    return f"Hello, {name}"

The caller supplies name, starts the action and consumes its greeting output:

async def main():
    action = REGISTRY.make_action("greet").run()
    await action["name"].finalize("Ada")
    print(await action["greeting"].consume(str))
    await action.wait()

Run the complete file, which includes asyncio and the entry point:

curl -fLO https://a11.to/examples/first_function.py
python first_function.py

It prints Hello, Ada.

Call the same function through MCP#

When an MCP-compatible model application needs to invoke a function you already registered, expose the same registry through A11’s MCP server. This exercise uses an MCP client directly so you can verify tool discovery and invocation without a model or an API key.

In the same environment, install the MCP extra and download the complete caller:

pip install 'a11-kit[mcp]==0.6.2'
curl -fLO https://a11.to/examples/first_function.py
curl -fLO https://a11.to/examples/mcp_client.py
python mcp_client.py

The client uses your current Python interpreter to start A11’s stdio server with the downloaded registry. This is the actual process configuration in mcp_client.py:

parameters = mcp.StdioServerParameters(
    command=sys.executable,
    args=[
        "-m",
        "a11.cli",
        "serve",
        str(Path(__file__).with_name("first_function.py")),
        "--mcp-stdio",
    ],
)

The client imports Status and StatusCode from a11.status for error reporting. It discovers the tool, calls greet with a real argument, checks the result and prints its structured output:

async with mcp.Client(parameters) as client:
    listed = await client.list_tools()
    print("Tools:", ", ".join(tool.name for tool in listed.tools))
    result = await client.call_tool("greet", {"name": "Ada"})
    if result.is_error:
        raise Status(
            code=StatusCode.INTERNAL,
            message="MCP tool invocation failed.",
            details=[result.content],
        ).to_exception()
    print(result.structured_content["greeting"])

Expect Tools: greet followed by Hello, Ada. The client context owns the server process and closes it on exit. The original greet handler stays unchanged while MCP provides another invocation path. The exchange verifies tool discovery and invocation; the model application still decides when to call the tool.

Add A11 to one operation#

Add A11 around one operation whose caller benefits from streaming, structured progress or remote execution:

Each downloadable example uses the public SDK directly and shows the integration point with the surrounding application.

Keep going#

• Understand the mental model.
• Practice a remote call with the simpler word-stream function.
• See what streaming changes about an interface.

If installation fails, check python --version and that python -m pip points to the active environment. The SDK package is a11-kit, imported as a11. Installing the Exchange service is unnecessary for this local example.

Complete source#

Download multiple_streams.py. It combines the snippets above with the imports and executable entry point; run it with python multiple_streams.py.

import asyncio
from contextlib import asynccontextmanager
from typing import Annotated

import a11
from a11 import AsyncNode, OutputPort
from a11.client.connection import open_gateway
from a11.sdk.interact_with_llm_schema import INTERACT_WITH_LLM_SCHEMA
from a11.sdk.llm import Interaction, LlmHeaders, Role

REGISTRY = a11.ActionRegistry()


async def _drain(node: a11.AsyncNode):
    async for _ in node:
        pass


@asynccontextmanager
async def model_call(prompt: str):
    async with open_gateway(
        "wss://a11.to/ws/demoserver", allow_embedded=False
    ) as connection:
        interact = connection.action(
            "interact_with_llm", INTERACT_WITH_LLM_SCHEMA
        )
        interact.set_header(LlmHeaders.PROVIDER.value, "ollama")
        interact.set_header(LlmHeaders.MODEL.value, "glm-5.3-flash:cloud")
        interact.set_header(LlmHeaders.BASE_URL.value, "https://ollama.com")
        interact.set_header(
            LlmHeaders.API_KEY.value, "use-a11-demo-resources"
        )
        await interact.call()

        readers = [
            asyncio.create_task(_drain(interact[name]))
            for name in ("event_stream", "new_interactions")
        ]
        try:
            turn = Interaction(
                role=Role.USER,
                content=[a11.to_chunk({
                    "role": "user",
                    "content": [{"type": "text", "text": prompt}],
                })],
            )
            await interact["interactions"].finalize(turn)
            await interact["config"].finalize()
            await interact["tools"].finalize()

            yield interact
            await asyncio.gather(*readers)
        finally:
            for reader in readers:
                reader.cancel()
            await asyncio.gather(
                *readers,
                return_exceptions=True,
            )
            interact.cancel()


@REGISTRY.action
async def answer(
    prompt: str,
    reply: Annotated[AsyncNode, OutputPort(mimetype="text/plain")],
    thoughts: Annotated[AsyncNode, OutputPort(mimetype="text/plain")],
):
    async with model_call(prompt) as interact:
        async for thought in interact["thoughts"]:
            await thoughts.put(str(thought))
        async for text in interact["text_output"]:
            await reply.put(str(text))
        await interact.wait()


async def _display(node: a11.AsyncNode):
    node_id = node.get_id()

    async for value in node:
        print(f"[{node_id}] {value}", flush=True)


async def main():
    action = REGISTRY.make_action("answer").run()
    await action["prompt"].finalize(
        "Explain why streaming helps an AI application in one sentence."
    )

    async with asyncio.TaskGroup() as readers:
        readers.create_task(_display(action["reply"]))
        readers.create_task(_display(action["thoughts"]))

    await action.wait()


if __name__ == "__main__":
    asyncio.run(main())

import asyncio

import a11

REGISTRY = a11.ActionRegistry()


@REGISTRY.action(output="greeting")
async def greet(name: str) -> str:
    return f"Hello, {name}"


async def main():
    action = REGISTRY.make_action("greet").run()
    await action["name"].finalize("Ada")
    print(await action["greeting"].consume(str))
    await action.wait()


if __name__ == "__main__":
    asyncio.run(main())

import asyncio

from first_stream import REGISTRY


async def read_one(text: str):
    action = REGISTRY.make_action("read").run()
    await action["text"].finalize(text)
    async for word in action["words"]:
        print(f"{text.split()[0]}: {word.strip()}", flush=True)
    await action.wait()


async def main():
    async with asyncio.TaskGroup() as tasks:
        tasks.create_task(read_one("One stream starts here"))
        tasks.create_task(read_one("Another stream runs alongside"))


if __name__ == "__main__":
    asyncio.run(main())

"""Call the actual greeting action through an MCP server over stdio."""

import asyncio
import sys
from pathlib import Path

import mcp
from a11.status import Status, StatusCode


async def main():
    parameters = mcp.StdioServerParameters(
        command=sys.executable,
        args=[
            "-m",
            "a11.cli",
            "serve",
            str(Path(__file__).with_name("first_function.py")),
            "--mcp-stdio",
        ],
    )
    async with mcp.Client(parameters) as client:
        listed = await client.list_tools()
        print("Tools:", ", ".join(tool.name for tool in listed.tools))
        result = await client.call_tool("greet", {"name": "Ada"})
        if result.is_error:
            raise Status(
                code=StatusCode.INTERNAL,
                message="MCP tool invocation failed.",
                details=[result.content],
            ).to_exception()
        print(result.structured_content["greeting"])


if __name__ == "__main__":
    asyncio.run(main())