PydanticAI

Type-safe, structured agent framework built on Pydantic v2. Designed for production reliability.

Philosophy

Validation first, execution second. Every input/output validated by Pydantic models.

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel, Field
from typing import Literal

# Define structured output
class SupportResponse(BaseModel):
    response: str = Field(description="Customer-facing response")
    needs_escalation: bool = Field(description="Whether human needed")
    category: Literal["billing", "technical", "general"] = Field(
        description="Issue category"
    )

# Agent with structured output
support_agent = Agent(
    "openai:gpt-4o",
    result_type=SupportResponse,
    system_prompt="Classify and respond to customer inquiries."
)

# Run with validation
result = await support_agent.run("I was charged twice!")
print(result.data.response)
print(result.data.needs_escalation)
print(result.data.category)
# All validated automatically!

Core Architecture

Agent Definition

from pydantic_ai import Agent
from pydantic_ai.models import KnownModelName

agent = Agent(
    model: KnownModelName | Model,  # "openai:gpt-4o", "anthropic:claude-3-5-sonnet"
    result_type: type[ResultT] | None = None,  # Structured output
    system_prompt: str | None = None,
    deps_type: type[DepsT] = None,  # Dependency injection
    retries: int = 3,
    model_settings: ModelSettings = ModelSettings(),
    tools: list[Tool] = [],
    result_validator: Callable[[ResultT], ResultT] | None = None,
    result_retries: int = 3,
)

Dependency Injection (Type-Safe)

from dataclasses import dataclass
from pydantic_ai import RunContext

@dataclass
class SupportDeps:
    customer_id: str
    db: Database
    settings: Settings

support_agent = Agent(
    "openai:gpt-4o",
    result_type=SupportResponse,
    deps_type=SupportDeps,
    system_prompt="You have access to customer database."
)

@support_agent.tool
async def get_customer(ctx: RunContext[SupportDeps], email: str) -> Customer:
    return await ctx.deps.db.get_customer(email)

# Run with dependencies
deps = SupportDeps(customer_id="123", db=db, settings=settings)
result = await support_agent.run("Help me!", deps=deps)

Structured Outputs

Result Types

# Simple
class Answer(BaseModel):
    answer: str

# Complex nested
class Analysis(BaseModel):
    summary: str
    findings: list[Finding]
    confidence: float = Field(ge=0, le=1)
    metadata: dict[str, str] = Field(default_factory=dict)

# Union types
class Result(BaseModel):
    type: Literal["success", "error"]
    data: SuccessData | ErrorData

# Streaming structured output
async for partial in agent.run_stream("Query", result_type=Analysis):
    print(partial)  # Partial validated models

Validation

# Automatic: Pydantic validates on every run
# Custom validator
def validate_analysis(result: Analysis) -> Analysis:
    if result.confidence < 0.5:
        raise ValueError("Confidence too low")
    return result

agent = Agent(..., result_validator=validate_analysis)

Tools

Function Tools

from pydantic_ai import RunContext

@agent.tool
async def search_web(ctx: RunContext[Deps], query: str) -> list[SearchResult]:
    """Search the web for information."""
    return await ctx.deps.search_client.search(query)

@agent.tool
async def calculate(ctx: RunContext[Deps], expression: str) -> float:
    """Evaluate mathematical expression."""
    return safe_eval(expression)

# Tool with complex input
class SearchInput(BaseModel):
    query: str
    max_results: int = 10
    recency_days: int | None = None

@agent.tool
async def advanced_search(ctx: RunContext[Deps], input: SearchInput) -> list[Result]:
    ...

Tool Retries & Errors

@agent.tool(retries=2)
async def flaky_api(ctx: RunContext[Deps], param: str) -> str:
    # Auto-retries on exception
    return await external_api.call(param)

# Custom error handling
@agent.tool
async def safe_tool(ctx: RunContext[Deps], param: str) -> str:
    try:
        return await risky_operation(param)
    except ExternalError as e:
        raise ToolError(f"Failed: {e}") from e  # Converted to model error

Multi-Agent Patterns

Agent as Tool

research_agent = Agent("openai:gpt-4o", result_type=ResearchResult)

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

@main_agent.tool
async def research(ctx: RunContext[Deps], topic: str) -> ResearchResult:
    return await research_agent.run(topic, deps=ctx.deps)

Handoff Pattern

class Handoff(BaseModel):
    to: Literal["billing", "technical", "sales"]
    reason: str
    context: str

triage_agent = Agent(
    "openai:gpt-4o",
    result_type=Handoff,
    system_prompt="Route to specialist."
)

# In main loop
handoff = await triage_agent.run(user_query)
specialist = agents[handoff.data.to]
result = await specialist.run(handoff.data.context)

Streaming

Text Streaming

async for text in agent.run_stream("Query"):
    print(text, end="", flush=True)

Structured Streaming

async for partial in agent.run_stream("Query", result_type=Analysis):
    # partial is Analysis with potentially incomplete fields
    if partial.confidence:
        update_confidence_bar(partial.confidence)

Custom Stream

async with agent.run_stream("Query") as stream:
    async for node in stream:
        if node.type == "model_response":
            print(node.data)
        elif node.type == "tool_call":
            print(f"Calling {node.tool}...")
        elif node.type == "tool_result":
            print(f"Result: {node.result}")

Models

Supported Providers

# OpenAI
Agent("openai:gpt-4o")
Agent("openai:gpt-4o-mini")

# Anthropic
Agent("anthropic:claude-3-5-sonnet")
Agent("anthropic:claude-3-haiku")

# Google
Agent("google:gemini-1.5-pro")
Agent("google:gemini-1.5-flash")

# Mistral
Agent("mistral:mistral-large")

# Groq
Agent("groq:llama-3.1-70b")

# Ollama (local)
Agent("ollama:llama3.1")

# Custom
from pydantic_ai.models import Model
Agent(MyCustomModel())

Model Settings

from pydantic_ai import ModelSettings

agent = Agent(
    "openai:gpt-4o",
    model_settings=ModelSettings(
        temperature=0.7,
        max_tokens=4096,
        top_p=0.9,
        parallel_tool_calls=True,
        timeout=60.0
    )
)

Testing

Test Helpers

from pydantic_ai.testing import TestModel, FunctionModel

# TestModel - returns structured test data
agent = Agent(TestModel(), result_type=MyResult)

# FunctionModel - custom behavior
def custom_model(messages, tools):
    return {"tool_calls": [...], "text": "..."}

agent = Agent(FunctionModel(custom_model))

# CaptureToolCalls - inspect tool usage
from pydantic_ai.testing import CaptureToolCalls

with CaptureToolCalls() as capture:
    await agent.run("Query")
    print(capture.tool_calls)

Unit Testing

@pytest.mark.asyncio
async def test_support_agent():
    agent = Agent(TestModel(), result_type=SupportResponse, deps_type=SupportDeps)

    result = await agent.run("Billing issue", deps=mock_deps)

    assert result.data.category == "billing"
    assert result.data.needs_escalation is True

Observability

Logfire Integration

import logfire
logfire.configure()

agent = Agent("openai:gpt-4o", instrument=True)
# Automatic: traces, spans, token counts, errors

Custom Instrumentation

from pydantic_ai import AgentRun, AgentRunResult

def on_run_start(run: AgentRun):
    logfire.info("Agent started", agent=run.agent.name)

def on_run_end(result: AgentRunResult):
    logfire.info("Agent completed", 
                 tokens=result.usage().total_tokens,
                 cost=result.cost())

agent = Agent(..., on_run_start=on_run_start, on_run_end=on_run_end)

Model Context Protocol (MCP)

Client

from pydantic_ai.mcp import MCPClient

client = MCPClient("http://localhost:8000/mcp")
tools = await client.list_tools()

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

Server

from pydantic_ai.mcp import MCPServer, Tool

@MCPServer.tool
async def my_tool(param: str) -> str:
    return f"Result: {param}"

server = MCPServer(tools=[my_tool])
server.run(host="0.0.0.0", port=8000)

Deployment

FastAPI

from fastapi import FastAPI
from pydantic_ai import Agent

app = FastAPI()
agent = Agent("openai:gpt-4o", result_type=Response)

@app.post("/chat")
async def chat(request: ChatRequest):
    result = await agent.run(request.message, deps=get_deps(request.user))
    return {"response": result.data.response}

# Streaming
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def stream():
        async for text in agent.run_stream(request.message):
            yield f"data: {text}\n\n"
    return StreamingResponse(stream(), media_type="text/event-stream")

Docker

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Best Practices

Define result_type always — Enables validation, IDE support
Use deps for dependencies — Type-safe, testable, no globals
Keep tools small — Single responsibility, clear schemas
Validate aggressively — Custom validators for business logic
Test with TestModel — Fast, deterministic unit tests
Instrument with Logfire — Production observability
Handle ToolError — Graceful degradation

Framework Comparison — vs LangGraph, CrewAI
Core Abstractions — Type-safe abstractions
Function Calling — Function calling patterns
OpenAI Agents SDK — Alternative with handoffs
LangGraph — Alternative for graphs
Memory Systems — Memory via deps
Observability — Logfire integration
Agent Benchmarks — Testing patterns

Sources: PydanticAI docs (ai.pydantic.dev), PydanticAI GitHub, Pydantic v2 docs, Logfire docs, example repos, production case studies.

From the Agentic Engineering research vault · CC BY 4.0 · all notes · raw markdown