Skip to content

Recipe: Real-Time Streaming Output

Goal

Use Agent.stream() to display model output as it generates.

Basic streaming

import asyncio
from doc_processing_toolkit import Agent

async def main():
    agent = Agent(
        model="claude-sonnet-4-5",
        instructions="You are concise.",
    )

    async for event in agent.stream("Explain quantum computing"):
        if event.event_type == "text_delta":
            print(event.content, end="", flush=True)
        elif event.event_type == "complete":
            print("\n\nDone!")
            result = event.metadata  # Full AgentResult as dict

asyncio.run(main())

Provider streams often arrive as many small chunks. Grouping consecutive text_delta and thinking_delta events makes console and log output much easier to read without changing the underlying streamed events.

class GroupedStreamPrinter:
    def __init__(self):
        self.current_type = None
        self.parts = []

    def push(self, event_type, content):
        if content is None:
            return

        if self.current_type is None:
            self.current_type = event_type
        elif event_type != self.current_type:
            self.flush()
            self.current_type = event_type

        self.parts.append(content)

    def flush(self):
        if not self.parts or self.current_type is None:
            self.current_type = None
            self.parts = []
            return

        aggregated = "".join(self.parts)
        print(f"\n[{self.current_type}]\n{aggregated}\n---", flush=True)
        self.current_type = None
        self.parts = []

Monitoring tool calls

When the agent uses tools, tool_call_start and tool_call_result events fire between text deltas:

async for event in agent.stream("What's the weather in Tokyo?"):
    if event.event_type == "tool_call_start":
        print(f"\n[Calling tool: {event.metadata['tool_name']}]")
    elif event.event_type == "tool_call_result":
        status = "✓" if event.metadata['success'] else "✗"
        print(f"[{status} Result ({event.metadata['duration_ms']}ms): "
              f"{event.metadata['result'][:100]}]")
    elif event.event_type == "text_delta":
        print(event.content, end="", flush=True)
    elif event.event_type == "error":
        print(f"\nError: {event.content}")

result is a preview: the first 500 characters of the result's str(), and event.metadata["result_truncated"] says whether that is all of it. When the client needs the whole result (a file path deep in a long reply, say), build the agent with tool_result_preview_chars=0: the event then carries all of it, plus result_data, the result as JSON data. The model is sent exactly what it was before; only the event's copy grows.

agent = Agent(model="claude-sonnet-5", tools=[fetch_report], tool_result_preview_chars=0)

async for event in agent.stream("Summarise the Q3 report."):
    if event.event_type == "tool_call_result" and event.metadata["success"]:
        data = event.metadata["result_data"]   # the tool's dict, whole
        print(data["path"], data["pages"])

See Tool results in events.

Monitoring thinking output

When the agent uses a thinking/reasoning model, thinking_delta events stream the model's reasoning process. Group the deltas before printing so reasoning and answer text stay readable:

printer = GroupedStreamPrinter()

async for event in agent.stream("What is the GCD of 1071 and 462?"):
    if event.event_type in {"thinking_delta", "text_delta"}:
        printer.push(event.event_type, event.content)
    elif event.event_type == "tool_call_start":
        printer.flush()
        print(f"\n[Calling: {event.metadata['tool_name']}]")
    elif event.event_type == "tool_call_result":
        printer.flush()
        status = "✓" if event.metadata['success'] else "✗"
        print(f"[{status} {event.metadata['tool_name']}: {event.metadata['duration_ms']}ms]")
    elif event.event_type == "iteration_complete":
        printer.flush()
        print(f"\n[Iteration {event.metadata['iteration']} — "
              f"tokens: {event.metadata['input_tokens']}+{event.metadata['output_tokens']}, "
              f"cost: ${event.metadata['cost']:.4f}]")
    elif event.event_type in {"error", "complete"}:
        printer.flush()

Debug grouped stream logs

For local debugging, set DEBUG_AGENT_STREAMING=true before starting your process.

  • Consecutive text_delta and thinking_delta chunks are grouped into readable blocks in the logger and console.
  • Tool and iteration events are still printed individually with formatted metadata.
  • The async iterator output is unchanged, so you can enable this without changing your application code.

Streaming in sessions

AgentSession.stream() delegates to Agent.stream() with full session history and cost tracking:

from doc_processing_toolkit.agent import AgentSession

async def chat():
    agent = Agent(
        model="gpt-5.4-mini",
        instructions="You are a helpful assistant.",
    )
    session = AgentSession(agent, session_id="my-chat")

    async for event in session.stream("Hello!"):
        if event.event_type == "text_delta":
            print(event.content, end="", flush=True)

    # Session costs are tracked automatically
    print(f"\nSession cost: ${session.metadata['total_cost']:.4f}")

asyncio.run(chat())

Streaming with structured output

Agent.stream() accepts the same output_schema as Agent.run(). During the stream, text_delta events carry the raw JSON as it is generated; the parsed and validated Pydantic instance is attached to the final complete event:

from pydantic import BaseModel
from doc_processing_toolkit import Agent

class Summary(BaseModel):
    topic: str
    bullets: list[str]

agent = Agent(model="claude-sonnet-4-5", instructions="Be concise.")

async for event in agent.stream(
    "Summarize the benefits of async I/O in three bullets.",
    output_schema=Summary,
):
    if event.event_type == "text_delta":
        print(event.content, end="", flush=True)  # streaming raw JSON
    elif event.event_type == "complete":
        result = event.metadata
        parsed = result["output"]  # already validated against Summary
        print("\n", parsed)

Early termination

Breaking out of the async for loop cancels the underlying execution cleanly:

max_chars = 1000
chars = 0

async for event in agent.stream("Write a long essay about AI"):
    if event.event_type == "text_delta":
        chars += len(event.content)
        print(event.content, end="", flush=True)
        if chars > max_chars:
            print("\n\n[Truncated]")
            break  # Cancels the stream

Provider billing on cancellation

When a stream is cancelled, you are still billed for tokens generated before cancellation. The savings come from tokens that would have been generated after the break.

What to validate

  • event.event_type is one of the documented types (text_delta, thinking_delta, tool_call_start, tool_call_result, iteration_complete, error, complete)
  • The complete event's metadata contains token and cost fields
  • Breaking out of the loop does not raise exceptions