PydanticAI Streaming: Tokens, Text, and Structured Data in Real Time

import asyncio
from pydantic_ai import Agent
 
agent = Agent('openai:gpt-4o-mini')
 
async def main():
    async with agent.run_stream('Write a short poem about the sea.') as result:
        async for chunk in result.stream_text(delta=True):
            print(chunk, end='', flush=True)
        print()  # newline at end
 
        # Access usage stats after streaming completes
        usage = result.usage()
        print(f'Tokens: {usage.total_tokens}')
 
asyncio.run(main())
 

Pass delta=True to stream_text() to get incremental chunks. Without it, each yield includes the full text accumulated so far — useful for updating a text box that re-renders on each yield.

from pydantic_ai import Agent, RunContext
from pydantic_ai.tools import tool
 
agent = Agent('openai:gpt-4o')
 
@agent.tool_plain
def get_weather(city: str) -> str:
    return f'The weather in {city} is sunny, 22C.'
 
async def stream_with_tools(query: str):
    async with agent.run_stream(query) as result:
        async for chunk in result.stream_text(delta=True):
            print(chunk, end='', flush=True)
        # After streaming, inspect all messages including tool calls
        for msg in result.all_messages():
            print(f'  [{msg.__class__.__name__}]')
 

from pydantic import BaseModel
from pydantic_ai import Agent
from typing import Optional
 
class MovieReview(BaseModel):
    title: str
    rating: int
    summary: str
    recommended: bool
 
agent = Agent('openai:gpt-4o', result_type=MovieReview)
 
async def stream_structured():
    async with agent.run_stream(
        'Review the movie Inception in 2 sentences.'
    ) as result:
        async for partial in result.stream_structured(debounce_by=0.05):
            # partial is a PartialResponse — fields fill in as JSON arrives
            if hasattr(partial, 'title') and partial.title:
                print(f'Title so far: {partial.title}')
 
        # Final validated result
        final: MovieReview = await result.get_data()
        print(final.model_dump_json(indent=2))
 

Partial structured results are not fully validated — fields may be None or incomplete mid-stream. Only call result.get_data() for the final, fully validated object.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic_ai import Agent
import asyncio
import json
 
app = FastAPI()
agent = Agent('openai:gpt-4o-mini')
 
@app.get('/stream')
async def stream_endpoint(query: str):
    async def generate():
        async with agent.run_stream(query) as result:
            async for chunk in result.stream_text(delta=True):
                # SSE format: 'data: ...\n\n'
                yield f'data: {json.dumps({"chunk": chunk})}\n\n'
        yield 'data: [DONE]\n\n'
 
    return StreamingResponse(
        generate(),
        media_type='text/event-stream',
        headers={
            'Cache-Control': 'no-cache',
            'X-Accel-Buffering': 'no',  # disable Nginx buffering
        }
    )
 

// Browser client
const es = new EventSource('/stream?query=Tell+me+a+story');
es.onmessage = (e) => {
    if (e.data === '[DONE]') { es.close(); return; }
    const { chunk } = JSON.parse(e.data);
    document.getElementById('output').textContent += chunk;
};
 

from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessagesTypeAdapter
 
agent = Agent('openai:gpt-4o-mini')
 
history = []
 
async def chat_stream(user_message: str):
    global history
    async with agent.run_stream(
        user_message,
        message_history=history,
    ) as result:
        async for chunk in result.stream_text(delta=True):
            print(chunk, end='', flush=True)
        print()
        history = result.all_messages()  # update history for next turn