Streaming

A real agent run takes a while — it plans, calls tools, thinks, writes. Awaiting one agent.run() gives you the answer, but the screen sits frozen until it lands. Streaming lets you watch the work happen instead: tokens as they're generated, tool calls as they fire, thinking as it unfolds.

You stream by iterating over events as the run progresses, rather than awaiting one final result.

Python

import asyncio

from pydantic_ai import (
    AgentRunResultEvent,
    FunctionToolCallEvent,
    FunctionToolResultEvent,
    PartDeltaEvent,
)

from pydantic_deep import create_deep_agent, DeepAgentDeps, StateBackend


async def main():
    agent = create_deep_agent(
        model="anthropic:claude-sonnet-4-6",
        instructions="You are a helpful coding assistant.",
    )

    deps = DeepAgentDeps(backend=StateBackend())

    async for event in agent.run_stream_events(
        "Write a Python function that returns the nth Fibonacci number, "
        "save it to fib.py, then read it back to me.",
        deps=deps,
    ):
        if isinstance(event, PartDeltaEvent):
            if hasattr(event.delta, "content_delta"):
                print(event.delta.content_delta, end="", flush=True)

        elif isinstance(event, FunctionToolCallEvent):
            print(f"\n[tool] {event.part.tool_name}", flush=True)

        elif isinstance(event, AgentRunResultEvent):
            print(f"\n\nDone. Files: {sorted(event.result.deps.files)}")


asyncio.run(main())

Run it

Save it to main.py and run:

Bash Session

$ python main.py

Instead of one silent pause, you'll see the answer typing itself out, with the agent's tool calls interleaved as they happen:

Text Only

[tool] write_todos
[tool] write_file
[tool] read_file
Done! I wrote fib.py with a function that returns the nth Fibonacci number,
then read it back. Here's what it contains:

def fib(n): ...
[tool] (none)

Done. Files: ['fib.py']

Same agent, same prompt as Your first agent — you just watched it work instead of waiting for it to finish.

Step by step

Let's look at the streaming loop one piece at a time.

Step 1: ask for events instead of a result

Python

async for event in agent.run_stream_events(prompt, deps=deps):
    ...

run_stream_events() returns an async iterator. Where agent.run() hands back a single AgentRunResult at the end, this yields a typed event for every small thing that happens along the way — and you async for over them.

Step 2: print text as it's generated

Python

if isinstance(event, PartDeltaEvent):
    if hasattr(event.delta, "content_delta"):
        print(event.delta.content_delta, end="", flush=True)

A PartDeltaEvent carries a delta — one more chunk of a part the model is building. For text, event.delta.content_delta is the next snippet of the answer. Print it with end="" and flush=True and the reply appears token by token, exactly like a chat UI.

Step 3: see tool calls as they fire

Python

elif isinstance(event, FunctionToolCallEvent):
    print(f"\n[tool] {event.part.tool_name}")

When the agent decides to act — write a file, update its todos, search the web — you get a FunctionToolCallEvent before the tool runs. event.part.tool_name tells you which one, and event.part.args holds the arguments. There's a matching FunctionToolResultEvent once the tool returns, with event.result.content.

Step 4: grab the final result

Python

elif isinstance(event, AgentRunResultEvent):
    print(event.result.output)

The very last event is an AgentRunResultEvent. Its event.result is the same AgentRunResult you'd get from agent.run() — so event.result.output, event.result.usage(), and event.result.deps are all there when the stream ends.

Watching the agent think

Reasoning models emit their thinking as its own kind of part. It arrives as PartDeltaEvents too — just with a thinking delta instead of a content delta:

Python

if isinstance(event, PartDeltaEvent):
    if hasattr(event.delta, "content_delta"):
        print(event.delta.content_delta, end="", flush=True)
    elif hasattr(event.delta, "thinking_delta"):
        print(f"\033[2m{event.delta.thinking_delta}\033[0m", end="", flush=True)

Now the model's reasoning streams in dim text and its actual answer streams in normal text — the same split you see in a polished assistant UI.

Nodes vs. events

There are two granularities. agent.run_stream_events() (used here) gives fine-grained events — text deltas, thinking, individual tool calls — ideal for a live UI. agent.iter() gives coarser nodes, one per step of the run, which is handy for a simple progress indicator. Reach for events when you want the typing-it-out feel.

The result is only there at the end

run.result (with iter()) and AgentRunResultEvent (with events) are only available once the stream finishes. Don't try to read the output mid-stream — collect deltas as they arrive, or wait for the final event.

Handle every event type

Don't assume an order. A run interleaves thinking, text, and many tool calls, and new event types can appear. Match the ones you care about and let the rest fall through — never index into event assuming it's the kind you expect.

Recap

You turned a silent wait into a live view of the agent at work:

• agent.run_stream_events(prompt, deps=deps) yields typed events as the run unfolds — async for over them instead of awaiting one result.
• PartDeltaEvent carries incremental output: content_delta for the answer, thinking_delta for reasoning. Print with flush=True for the typing effect.
• FunctionToolCallEvent and FunctionToolResultEvent surface tool calls as they fire and return.
• AgentRunResultEvent is the last event; event.result is the same AgentRunResult you'd get from agent.run().
• Want coarse, per-step progress instead? agent.iter() yields one node per step.

Next, let's let the agent pause and check with you before it does something risky.

• Human-in-the-loop →