Stream Every Action¶

The guarded document assistant works, but await agent.run(...) reveals only text. During a tool call, the terminal appears idle. It hides the tool name, the result, failures, and token usage.

run() is useful when another function needs only the final text. A harness needs the typed event stream.

Outcome¶

harness.py uses run_stream() and renders text, tool activity, errors, and the final session status. It also closes an interrupted stream explicitly.

Fast Track¶

Replace await agent.run(...) with agent.run_stream(...).
Match concrete event dataclasses instead of inspecting untyped dictionaries.
Send assistant text to stdout and operational details to stderr.
Close AgentStream in finally.

Download the complete example.

Hands-on delta¶

1. Preserve both levels of progress¶

A user turn starts with one run_stream(prompt, context) call. It ends when Axio emits one SessionEndEvent.

An agent iteration is one request to the transport. Each request ends with IterationEnd. A tool_use stop runs the requested tools and starts another iteration inside the same user turn. An end_turn stop finishes the user turn.

For example, one document lookup normally uses two iterations:

The model requests read_document, then the tool returns its result.
The model reads that result and writes the answer.

SessionEndEvent.total_usage adds the usage from both iterations.

2. Add a terminal renderer¶

Keep DOCUMENTS, both tools, and DocumentAccessGuard unchanged. Add this function to harness.py:

examples/tutorial/stream_every_action.py¶

async def render_turn(
    agent: Agent,
    prompt: str,
    context: ContextStore,
    *,
    stdout: TextIO = sys.stdout,
    stderr: TextIO = sys.stderr,
) -> SessionEndEvent:
    stream = agent.run_stream(prompt, context)
    session_end: SessionEndEvent | None = None

    try:
        async for event in stream:
            match event:
                case TextDelta(delta=text):
                    print(text, end="", file=stdout, flush=True)
                case ToolUseStart(name=name):
                    print(f"\n[tool] {name}", file=stderr)
                case ToolResult(name=name, content=content, is_error=is_error):
                    label = "tool error" if is_error else "tool result"
                    print(
                        f"[{label}] {name}: {len(content)} characters",
                        file=stderr,
                    )
                case Error(exception=exception):
                    print(
                        f"[error] {type(exception).__name__}",
                        file=stderr,
                    )
                case SessionEndEvent() as ending:
                    session_end = ending
                    usage = ending.total_usage
                    print(
                        f"[done] {ending.stop_reason.value}; input={usage.input_tokens}, output={usage.output_tokens}",
                        file=stderr,
                    )
    finally:
        await stream.aclose()

    if session_end is None:
        raise RuntimeError("Agent stream ended without SessionEndEvent")

    print(file=stdout)
    return session_end

ToolUseStart identifies the tool before execution. ToolResult contains the completed result and its is_error flag. A failed tool is therefore different from Error, which reports a stream or transport failure.

The renderer reports result size and exception type without printing raw tool content or exception text. Those values can contain credentials, paths, or private records. Send full details only to access-controlled logs after the application applies its redaction policy.

Axio still adds the complete tool result to conversation history. Compact Long Context bounds that stored content at its source.

AgentStream is an async iterator, but it is not an async context manager. The finally block closes its underlying async generator when iteration finishes, the user cancels the task, or rendering raises an exception.

See Stream Events for all event variants and Agent & the Agentic Loop for the complete loop contract.

Try It¶

The offline example scripts one tool request and one final answer. It captures both terminal channels, checks the typed result and usage, then forwards the rendered output:

examples/tutorial/stream_every_action.py¶

async def run_scripted_turn() -> None:
    transport = StubTransport(
        [
            make_tool_use_response(
                "read_document",
                tool_input={"path": "README.md"},
            ),
            make_text_response("README.md: Project overview."),
        ]
    )
    context = MemoryContextStore()
    stdout = StringIO()
    stderr = StringIO()

    ending = await render_turn(
        build_agent(transport),
        "Summarize README.md.",
        context,
        stdout=stdout,
        stderr=stderr,
    )

    history = await context.get_history()
    results = [
        block
        for message in history
        for block in message.content
        if isinstance(block, ToolResultBlock)
    ]
    assert ending.total_usage == Usage(input_tokens=20, output_tokens=10)
    assert results[0].content == "README.md: Project overview"
    assert stdout.getvalue() == "README.md: Project overview.\n"
    assert "[tool] read_document" in stderr.getvalue()
    assert "[tool result] read_document" in stderr.getvalue()

    sys.stdout.write(stdout.getvalue())
    sys.stderr.write(stderr.getvalue())

Run uv run python examples/tutorial/stream_every_action.py from the repository root. The tool activity appears before the final answer. No API key is required.

Done when¶

[ ] Assistant text streams without waiting for the complete turn.
[ ] ToolUseStart and ToolResult show tool activity.
[ ] Error and SessionEndEvent show failure and completion state.
[ ] One tool call can create multiple iterations within one user turn.
[ ] Every stream closes in a finally block.

Next failure¶

The harness is now observable, but its MemoryContextStore still disappears with the process. The next lesson gives each conversation durable identity.

Continue with Persist the Conversation.