Streaming Output

Why stream

run_sync waits for the entire answer before returning anything, fine for short answers, frustrating for a long one where a user is staring at a blank screen. run_stream gives you the output incrementally, as the model produces it, the same motivation as .stream() on a LangChain runnable.

run_stream is an async context manager, there's no _sync version, streaming is inherently async, so it needs an async def main() and asyncio.run(main()), same as the async lessons in the MCP course.

Streaming structured output, not just text

The genuinely different part, compared to LangChain streaming, is that this works for structured output_types too, not just plain text.

import asyncio

from dotenv import load_dotenv
from pydantic import BaseModel
from pydantic_ai import Agent

load_dotenv()


class CityFact(BaseModel):
    city: str
    country: str


text_agent = Agent("google:gemini-3.5-flash-lite")
structured_agent = Agent("google:gemini-3.5-flash-lite", output_type=CityFact)


async def main() -> None:
    print("Streaming plain text:")
    async with text_agent.run_stream("Count from 1 to 5, one number per line.") as result:
        async for chunk in result.stream_output():
            print("  ", repr(chunk))

    print("\nStreaming structured output:")
    async with structured_agent.run_stream("Tell me about Tokyo.") as result:
        async for partial in result.stream_output():
            print("  ", partial)


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

Each item yielded by stream_output() is a fully re-validated instance of your output type, built from however much of the response has arrived so far, fields may be missing or partial early on, then complete by the last chunk. This lets a UI render a structured card that fills itself in, instead of a wall of streaming JSON text you'd have to parse yourself.