PydanticAI: Building Type-Safe Agent Workflows with Structured Outputs

LLM outputs are notoriously unstructured. You ask for JSON and get markdown-wrapped prose. You define a schema and the model skips fields or invents new ones. PydanticAI solves this by making type-safe structured output the core of its agent model — not a post-processing step bolted on after the fact [1].

Built by the team behind Pydantic (the most widely used data validation library in Python), PydanticAI provides an agent framework where every agent has a typed result, type-checked tool signatures, and dependency injection that composes cleanly with your existing codebase [2]. This guide walks through building a multi-step document analysis agent from scratch.

Prerequisites

• Python 3.10+ (3.11+ recommended for Self type and better dataclass support)
• pip install pydantic-ai — the full framework (includes model providers)
• An API key for at least one supported provider: OpenAI, Anthropic, Google Gemini, or any OpenAI-compatible endpoint

Step 1: Your First Structured Agent

PydanticAI agents declare their output type at construction time via the result_type parameter. The framework handles LLM response parsing, validation, and retry.

from pydantic import BaseModel
from pydantic_ai import Agent

class DocumentSummary(BaseModel):
    title: str
    word_count: int
    key_topics: list[str]
    sentiment: str  # positive, negative, neutral

agent = Agent(
    "openai:gpt-4o",
    result_type=DocumentSummary,
    system_prompt="Extract structured metadata from the given document.",
)

result = agent.run_sync(
    "The new FPGA accelerator achieves 4.2 TFLOPS at 45W thermal design power, "
    "marking a 3x improvement over the previous generation. Early benchmarks "
    "show particular strength in sparse matrix operations."
)
print(result.data)
# DocumentSummary(
#   title='FPGA Accelerator Benchmarks',
#   word_count=28,
#   key_topics=['FPGA', 'accelerator', 'benchmarks', 'sparse matrix'],
#   sentiment='positive'
# )

The result.data field is a fully validated DocumentSummary instance. If the LLM returns malformed output, PydanticAI automatically retries with the validation error message as context — no manual parsing or error handling needed [3].

Step 2: Registering Tools with Type-Safe Signatures

Tools in PydanticAI are plain Python functions with type annotations. The framework infers the JSON schema for tool calling from the function signature and docstring.

Use @agent.tool when the tool needs access to the agent’s context (dependencies). Use @agent.tool_plain for stateless tools.

from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class SearchDep:
    api_key: str
    max_results: int = 5

search_agent = Agent(
    "anthropic:claude-sonnet-4-20260514",
    deps_type=SearchDep,
    result_type=list[str],
)

@search_agent.tool
async def web_search(ctx: RunContext[SearchDep], query: str) -> list[str]:
    """Search the web for recent information on a topic.

    Args:
        ctx: Agent context with API credentials.
        query: The search query string.
    Returns:
        A list of result URLs.
    """
    # ctx.deps.api_key available here
    results = await run_search_api(query, api_key=ctx.deps.api_key)
    return [r.url for r in results[: ctx.deps.max_results]]

Key details:

• The LLM sees the function’s docstring as the tool description and type annotations as parameter schemas.
• RunContext[T] gives access to dependencies injected at runtime.
• Tools can be sync or async; the agent handles both.
• Return type annotations are validated at runtime — if a tool returns something unexpected, the agent catches it before sending to the LLM [4].

Step 3: Dependency Injection for Composable Agents

PydanticAI’s dependency system decouples agent logic from external services. Dependencies are declared via deps_type and passed at runtime through the run() / run_sync() call.

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass
import httpx

@dataclass
class DocServiceDep:
    client: httpx.AsyncClient
    base_url: str
    auth_token: str

class AnalysisResult(BaseModel):
    document_id: str
    entities: list[str]
    summary: str
    risk_score: float

doc_agent = Agent(
    "google-gla:gemini-2.5-pro",
    deps_type=DocServiceDep,
    result_type=AnalysisResult,
    system_prompt="Analyze documents and produce structured risk assessments.",
)

@doc_agent.tool
async def fetch_document(ctx: RunContext[DocServiceDep], doc_id: str) -> str:
    """Retrieve a document from the content service by ID."""
    headers = {"Authorization": f"Bearer {ctx.deps.auth_token}"}
    response = await ctx.deps.client.get(
        f"{ctx.deps.base_url}/docs/{doc_id}", headers=headers
    )
    response.raise_for_status()
    return response.text

# Runtime injection
async def analyze_document(doc_id: str) -> AnalysisResult:
    async with httpx.AsyncClient() as client:
        deps = DocServiceDep(
            client=client,
            base_url="https://api.internal.example.com",
            auth_token=os.environ["DOC_SERVICE_TOKEN"],
        )
        result = await doc_agent.run(
            f"Analyze document {doc_id} for compliance risks",
            deps=deps,
        )
    return result.data

This pattern lets you swap implementations without changing agent code — use a mock client in tests, a real client in production, and inject different credentials per environment [5].

Step 4: Multi-Step Workflows with Agent Composition

Real agent workflows involve multiple steps: fetch → classify → enrich → report. Each step is its own agent; you orchestrate them in plain Python.

from pydantic import BaseModel
from pydantic_ai import Agent

# Step 1: Classify the document
class DocumentClass(BaseModel):
    category: str
    confidence: float
    language: str

classifier = Agent(
    "openai:gpt-4o-mini",
    result_type=DocumentClass,
    system_prompt="Classify the document into a category.",
)

# Step 2: Extract structured data
class ExtractedData(BaseModel):
    dates: list[str]
    monetary_amounts: list[float]
    named_entities: dict[str, list[str]]

extractor = Agent(
    "openai:gpt-4o-mini",
    result_type=ExtractedData,
    system_prompt="Extract all dates, monetary values, and named entities.",
)

# Step 3: Generate the final report
class FinalReport(BaseModel):
    classification: DocumentClass
    extracted: ExtractedData
    recommendations: list[str]

reporter = Agent(
    "openai:gpt-4o",
    result_type=FinalReport,
    system_prompt=(
        "Combine classification and extracted data into a structured "
        "compliance report with actionable recommendations."
    ),
)

async def process_document(text: str) -> FinalReport:
    # Step 1
    cls = await classifier.run(text)
    # Step 2
    data = await extractor.run(text)
    # Step 3 — pass previous results as context
    report = await reporter.run(
        f"Classification: {cls.data.model_dump_json()}\n"
        f"Extracted data: {data.data.model_dump_json()}"
    )
    return report.data

Each agent is independently testable. You can swap models per step (use cheap models for classification, expensive ones for report generation). The pipeline is just function composition — no framework-specific DAG syntax to learn [6].

Step 5: Retry, Fallback, and Error Handling

PydanticAI’s ModelRetry exception lets tools signal the LLM to retry with additional context. Combine this with agent run settings for robust error handling.

from pydantic_ai import Agent, ModelRetry
from pydantic_ai.exceptions import ModelRetryError
import httpx

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

@agent.tool
async def fetch_weather(city: str) -> dict:
    """Get current weather for a city. Retry if the API is temporarily down."""
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"https://api.weather.example.com/v1/{city}",
                timeout=5.0,
            )
            resp.raise_for_status()
            return resp.json()
    except httpx.TimeoutException:
        raise ModelRetry("Weather API timed out, please try again")
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 503:
            raise ModelRetry("Weather service temporarily unavailable, retry later")
        raise

# Run with retry configuration
result = agent.run_sync(
    "What's the weather in Tokyo?",
    max_retries=3,          # Max retries for the whole run
    max_result_retries=2,    # Max retries for result validation
)

When ModelRetry is raised, the agent sends the error message back to the LLM and lets it decide how to proceed — try the tool again, try a different city, or explain the failure to the user. This is fundamentally different from a hard exception that crashes the agent [7].

Step 6: Streaming Structured Outputs

For long-running agents, PydanticAI supports streaming both text tokens and structured result updates.

from pydantic_ai import Agent
from pydantic import BaseModel

class StreamResult(BaseModel):
    progress: float
    current_step: str
    partial_output: str | None = None

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

async def stream_analysis():
    async with agent.run_stream(
        "Process this 500-page annual report..."
    ) as stream:
        async for partial in stream.stream_structured():
            # partial.data is a partial StreamResult
            if partial.data:
                print(
                    f"Progress: {partial.data.progress:.0%} — "
                    f"{partial.data.current_step}"
                )
        # Final validated result
        final = await stream.get_data()
        return final

Streaming structured outputs enables real-time progress indicators in UIs without waiting for the full agent run to complete [8].

Best Practices

1. Use specific models per task — Classification and extraction work well with gpt-4o-mini or claude-3-haiku. Synthesis and report generation benefit from gpt-4o or claude-sonnet-4. Set model per-agent, not globally.

2. Keep tools focused — Each tool should do one thing well. A tool that “searches and summarizes and formats” is harder for the LLM to reason about than three separate tools [9].

3. Prefer deps_type over closures — Dependency injection through RunContext is testable and composable. Closing over globals in tool functions breaks isolation and makes testing impossible.

4. Set max_result_retries — The default is low. For complex result types, set max_result_retries=3 to give the LLM room to correct malformed outputs before surfacing an error to the user.

5. Use model_dump_json() for agent-to-agent communication — When passing results between agents, serialize with model_dump_json() and deserialize with model_validate_json() at the receiving end. This preserves type safety across the pipeline.

6. Pin your pydantic-ai version — The framework is under active development. Pin to a specific version in requirements.txt and test upgrades against your agent suite before deploying.

Key Takeaways

• PydanticAI makes structured output the core contract, not an afterthought — every agent has a typed result_type that’s validated at runtime.
• Tools are plain Python functions with type annotations; the framework generates JSON schemas automatically from function signatures.
• Dependency injection via deps_type / RunContext decouples agent logic from infrastructure — swap implementations without changing agent code.
• Multi-step workflows are plain Python function composition, not a framework-specific DAG — each agent is independently testable.
• ModelRetry gives the LLM a chance to recover from tool failures instead of crashing the agent.

References

• PydanticAI Official Documentation — Framework overview and API reference [1]
• GitHub: pydantic/pydantic-ai — Source code, issues, and examples [2]
• PydanticAI: Output Validation — How structured output validation and retry work [3]
• PydanticAI: Tools — Tool registration and context access [4]
• PydanticAI: Dependencies — Dependency injection patterns [5]
• Real Python: Building Type-Safe LLM Agents With PydanticAI — Walkthrough of multi-agent composition [6]
• PydanticAI: ModelRetry — Error recovery and retry mechanism [7]
• PydanticAI: Streaming — Streaming structured and text output [8]
• Anthropic: Tool Use Best Practices — General guidance on tool design for LLMs [9]

📖 Related Reads

• NiteAgent: Multi-Agent Pipeline Resiliency Patterns — Error recovery and circuit-breaking for agent pipelines
• NiteAgent: Production Tool-Calling Architecture — Patterns for reliable tool dispatch in production
• NiteAgent: AI Agent Testing Strategies 2026 — Evaluating and testing agent frameworks in production

Cross-links automatically generated from NiteAgent.

← Back to all posts