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 PydanticAI Official Documentation.

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 GitHub: pydantic/pydantic-ai. 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 PydanticAI: Output Validation.

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 PydanticAI: Tools.

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 PydanticAI: Dependencies.

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 Real Python: Building Type-Safe LLM Agents With PydanticAI.

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 PydanticAI: ModelRetry.

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 PydanticAI: Streaming.

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 Anthropic: Tool Use Best Practices.

  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

📖 Related Reads

Cross-links automatically generated from NiteAgent.

← Back to all posts