Streaming Responses in Production with Pydantic AI: Complete Guide

Introduction to Streaming Responses with Pydantic AI

Streaming responses have become a cornerstone of modern AI-powered applications. When you're building production-grade systems with large language models, waiting for an entire response to generate before showing anything to the user creates a poor experience. Pydantic AI, the agent framework from the team behind Pydantic, offers first-class support for streaming structured and unstructured responses in a way that's both type-safe and production-ready.

In this guide, we'll explore what streaming means in the context of Pydantic AI, why it matters for production workloads, and how to implement it correctly with robust error handling, validation, and observability.

What Is Response Streaming in Pydantic AI?

Pydantic AI is an agent framework that brings Pydantic's type validation philosophy to LLM applications. Instead of treating LLM outputs as opaque strings, you define typed result models and the framework ensures outputs conform to your schema. Streaming in this context means receiving the model's output incrementally — token by token or chunk by chunk — rather than waiting for the full completion.

Pydantic AI supports two main streaming modes:

• Text streaming: Yields partial string chunks as the model generates them. Ideal for chat interfaces and conversational agents.
• Structured streaming: Yields validated partial objects as the model produces JSON. Useful when you need typed outputs but still want incremental delivery.

Under the hood, Pydantic AI leverages the streaming capabilities of underlying providers (OpenAI, Anthropic, Gemini, Groq, Ollama, and others) and wraps them in a consistent, typed interface.

Why Streaming Matters in Production

Streaming isn't just a nice-to-have feature. In production systems, it directly impacts several critical metrics:

• Time to first token (TTFT): Users perceive systems as faster when they see partial output quickly, even if total generation time is unchanged.
• Timeout mitigation: Long generations can exceed gateway or load balancer timeouts. Streaming keeps the connection active with continuous data flow.
• Cancellation: Streaming lets you abort generation mid-way if the user navigates away, saving token costs.
• Progressive rendering: For structured outputs like reports or code, you can render sections as they arrive instead of blocking the UI.
• Backpressure handling: Slow consumers can apply backpressure rather than buffering entire responses in memory.

In short, streaming transforms a brittle "wait then display" pattern into a responsive, cost-aware, and resilient flow.

Setting Up Your Environment

Before diving into code, install Pydantic AI and any provider packages you need. The examples below use OpenAI, but the patterns apply across providers.

pip install pydantic-ai
pip install "pydantic-ai-slim[openai]"
pip install uvicorn fastapi

Set your API keys as environment variables:

export OPENAI_API_KEY="sk-..."

Basic Text Streaming

The simplest streaming use case is yielding text chunks from an agent. Pydantic AI exposes this through the run_stream method on an Agent instance.

from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o",
    system_prompt="You are a concise, helpful assistant.",
)

async def main():
    async with agent.run_stream("Explain how HTTP/2 multiplexing works.") as result:
        async for chunk in result.stream():
            print(chunk, end="", flush=True)
        print()

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

The run_stream method returns an async context manager. Inside it, result.stream() is an async iterator yielding string chunks. Using end="" and flush=True ensures chunks print inline without newlines or buffering delays.

Streaming with Validation and Final Output

Often you want both incremental display and a final validated value. Pydantic AI lets you access the consolidated output after the stream completes.

async def stream_and_collect(prompt: str) -> str:
    async with agent.run_stream(prompt) as result:
        async for chunk in result.stream_text():
            print(chunk, end="", flush=True)
        print()
        # After the stream ends, get the full validated output
        final = await result.get_output()
        return final


The stream_text() method is an alias that explicitly yields text chunks, while get_output() returns the final consolidated response once iteration finishes.

Streaming Structured Outputs

Where Pydantic AI shines is structured streaming. You define a Pydantic model as the result type, and the agent streams partial validated objects as the JSON is generated.

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class CodeReview(BaseModel):
    summary: str = Field(description="One-sentence summary of the code")
    issues: list[str] = Field(default_factory=list, description="List of issues found")
    severity: str = Field(description="low, medium, or high")
    suggested_fix: str = Field(description="Proposed fix in code")

review_agent = Agent(
    "openai:gpt-4o",
    output_type=CodeReview,
    system_prompt="You are a senior code reviewer. Respond with structured JSON.",
)

async def review_code(snippet: str):
    async with review_agent.run_stream(f"Review this code:\n{snippet}") as result:
        async for partial in result.stream():
            print("Partial:", partial)
        final = await result.get_output()
        print("\nFinal review:")
        print(final.model_dump_json(indent=2))


Each yielded partial is a validated CodeReview instance reflecting whatever fields the model has produced so far. Fields not yet generated will hold their defaults. This lets you render progressive UIs — for example, showing the summary first, then issues as they arrive.

Handling Partial Validation Failures

Structured streaming means partial JSON, which can temporarily be invalid. Pydantic AI handles this gracefully by skipping chunks that don't yet validate and yielding the last valid partial. You can also configure behavior explicitly:

async with review_agent.run_stream(prompt) as result:
    async for partial in result.stream(debounce_by=0.1):
        # debounce_by collapses rapid updates to reduce UI thrash
        update_ui(partial)


The debounce_by parameter is invaluable in production frontends — it prevents flooding the client with hundreds of micro-updates per second.

Building a Streaming API with FastAPI

In production, you'll typically expose streaming through an HTTP endpoint. FastAPI's StreamingResponse pairs naturally with Pydantic AI's async streams. Here's a complete, production-minded example using Server-Sent Events (SSE).

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from pydantic_ai import Agent
import json
import asyncio

app = FastAPI()

agent = Agent(
    "openai:gpt-4o",
    system_prompt="You are a helpful technical assistant.",
)

class ChatRequest(BaseModel):
    message: str

async def event_generator(prompt: str, request: Request):
    try:
        async with agent.run_stream(prompt) as result:
            async for chunk in result.stream_text():
                if await request.is_disconnected():
                    # Client went away — stop generating to save tokens
                    break
                payload = json.dumps({"type": "chunk", "data": chunk})
                yield f"data: {payload}\n\n"
        yield f"data: {json.dumps({'type': 'done'})}\n\n"
    except Exception as exc:
        err = json.dumps({"type": "error", "data": str(exc)})
        yield f"data: {err}\n\n"

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest, request: Request):
    return StreamingResponse(
        event_generator(req.message, request),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Disable Nginx buffering
            "Connection": "keep-alive",
        },
    )


Key production details in this endpoint:


  
Disconnection detection: request.is_disconnected() lets you stop generation when the client closes the connection, avoiding wasted token spend.
  
SSE framing: Each event is wrapped in data: ...\n\n per the SSE spec, making it consumable by browser EventSource APIs.
  
Buffering disabled: The X-Accel-Buffering: no header tells Nginx not to buffer the response, which would otherwise break streaming.
  
Error envelope: Errors are serialized as SSE events so the client always receives structured feedback.


Consuming the Stream from JavaScript

Here's a minimal browser client that consumes the endpoint above:

const response = await fetch("/chat/stream", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ message: "Explain async/await in Python" }),
});

const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });
  const lines = buffer.split("\n\n");
  buffer = lines.pop();
  for (const line of lines) {
    if (!line.startsWith("data: ")) continue;
    const evt = JSON.parse(line.slice(6));
    if (evt.type === "chunk") appendToUI(evt.data);
    if (evt.type === "done") finalizeUI();
    if (evt.type === "error") showError(evt.data);
  }
}


Streaming with Tools and Multi-Step Agents

Pydantic AI agents can call tools during a run. Streaming works alongside tool calls, but you need to understand the lifecycle. When the model decides to call a tool, the stream pauses, the tool executes, and the model continues generating with the tool result in context.

from pydantic_ai import Agent, RunContext

agent = Agent("openai:gpt-4o")

@agent.tool
async def get_weather(ctx: RunContext[None], city: str) -> str:
    # Simulated external call
    await asyncio.sleep(0.5)
    return f"The weather in {city} is sunny and 22C."

async def stream_with_tools(query: str):
    async with agent.run_stream(query) as result:
        async for chunk in result.stream_text():
            print(chunk, end="", flush=True)
        print()


During tool execution, no chunks are emitted. If you need to surface tool activity to the user (e.g., "Looking up weather..."), hook into the agent's event stream or use result.stream_responses() to inspect intermediate model responses including tool calls.

Best Practices for Production Streaming

1. Always Use Async Context Managers

The async with agent.run_stream(...) pattern ensures resources are cleaned up even if the consumer cancels mid-stream. Never call run_stream without the context manager — you risk leaking connections and provider sessions.

2. Implement Timeouts at Every Layer

Streaming can still hang. Wrap your streams with timeouts:

async def safe_stream(prompt: str):
    try:
        async with asyncio.timeout(60):
            async with agent.run_stream(prompt) as result:
                async for chunk in result.stream_text():
                    yield chunk
    except asyncio.TimeoutError:
        yield "\n[Generation timed out]"


3. Debounce Structured Updates

For structured streaming, raw token-by-token updates can produce hundreds of near-identical partial objects per second. Use debounce_by and consider diffing on the client side to only re-render changed fields.

4. Handle Provider-Specific Quirks

Different providers stream differently. Anthropic streams content blocks, OpenAI streams deltas, and local models via Ollama may have higher latency variance. Test your streaming code against each provider you support, and abstract provider-specific retry logic behind a wrapper.

5. Log Token Usage and Costs

Streaming makes it easy to lose track of token consumption. After the stream completes, access usage metadata:

async with agent.run_stream(prompt) as result:
    async for chunk in result.stream_text():
        process(chunk)

usage = result.usage()
print(f"Input tokens: {usage.input_tokens}")
print(f"Output tokens: {usage.output_tokens}")
print(f"Total tokens: {usage.total_tokens}")


Log these to your observability stack (Datadog, Grafana, OpenTelemetry) and set alerts on anomalous consumption.

6. Graceful Degradation

Not all clients or networks support streaming. Build a fallback non-streaming endpoint that calls agent.run() instead, and route based on client capability headers.

@app.post("/chat")
async def chat(req: ChatRequest):
    accept = req.headers.get("accept", "")
    if "text/event-stream" in accept:
        return StreamingResponse(event_generator(req.message, req), ...)
    # Fallback: non-streaming
    result = await agent.run(req.message)
    return {"response": result.output}


7. Test Streaming Behavior

Use TestModel and FunctionModel from Pydantic AI's testing utilities to simulate streaming without hitting real providers:

from pydantic_ai.models.test import TestModel

test_agent = Agent(TestModel(), system_prompt="test")

async def test_stream():
    async with test_agent.run_stream("hello") as result:
        chunks = [chunk async for chunk in result.stream_text()]
        assert len(chunks) > 0


This keeps your test suite fast, deterministic, and free of API costs.

Common Pitfalls and How to Avoid Them


  
Blocking the event loop: Never use synchronous time.sleep or blocking I/O inside stream handlers. Use asyncio.sleep and async libraries.
  
Forgetting to flush: In CLI tools, forgetting flush=True causes chunks to buffer and appear all at once, defeating the purpose of streaming.
  
Ignoring backpressure: If your consumer is slow (e.g., a sluggish database write per chunk), use an asyncio.Queue with bounded size to apply backpressure rather than unbounded buffering.
  
Assuming chunk boundaries align with tokens: Chunks are arbitrary slices of text. Never parse semantic units from chunk boundaries — accumulate and parse from the full buffer.
  
Not handling partial JSON in custom parsers: If you write custom structured-streaming parsers, always handle json.JSONDecodeError on incomplete input and retry on the next chunk.


Conclusion

Streaming responses with Pydantic AI combine the responsiveness users expect from modern AI interfaces with the type safety and validation that production systems demand. By leveraging run_stream for text, structured streaming for typed outputs, and FastAPI's StreamingResponse for HTTP delivery, you can build agents that feel instantaneous while remaining robust under real-world conditions. The key to success lies in treating streaming as a first-class architectural concern: implement timeouts, detect client disconnections, debounce structured updates, monitor token usage, and always provide a non-streaming fallback. With these patterns in place, your Pydantic AI agents will be ready to handle production traffic with confidence.