# 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.

```python
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
```python
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)
```python
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
```python
# 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
```python
# 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
```python
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
```python
@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
```python
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
```python
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
```python
async for text in agent.run_stream("Query"):
    print(text, end="", flush=True)
```

### Structured Streaming
```python
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
```python
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
```python
# 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
```python
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
```python
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
```python
@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
```python
import logfire
logfire.configure()

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

### Custom Instrumentation
```python
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
```python
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
```python
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
```python
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
```dockerfile
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

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

---

## Related Notes

- [[Frameworks/Framework-Comparison.md]] — vs LangGraph, CrewAI
- [[Foundations/Core-Abstractions.md]] — Type-safe abstractions
- [[ToolUse/Function-Calling.md]] — Function calling patterns
- [[Frameworks/OpenAI-Agents-SDK.md]] — Alternative with handoffs
- [[Frameworks/LangGraph.md]] — Alternative for graphs
- [[ContextEngineering/Memory-Systems.md]] — Memory via deps
- [[Production/Observability.md]] — Logfire integration
- [[Evaluation/Agent-Benchmarks.md]] — Testing patterns

---

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