guaderrama Pydantic AI Agent Builder
Purpose
Create production-ready AI agents with type safety, automatic validation, and minimal boilerplate using Pydantic AI framework.
When to Use
- Building FastAPI backend with AI capabilities
- Need strict type checking and validation
- Want auto-retry on malformed LLM responses
- Creating agents with custom tools
Architecture Pattern
Project Structure
backend/
├── agents/
│ ├── __init__.py
│ ├── base_agent.py # Base agent class
│ └── [feature]_agent.py # Feature-specific agents
├── tools/
│ ├── __init__.py
│ └── [tool_name].py # Tool definitions
└── config/
└── agent_config.py # Agent configurations
Installation
pip install pydantic-ai httpx pydantic python-dotenv
Base Agent Pattern
from pydantic_ai import Agent
from pydantic import BaseModel
import os
class AgentResponse(BaseModel):
result: str
confidence: float
agent = Agent(
model='openrouter:openai/gpt-4o',
output_type=AgentResponse,
tools=[tool1, tool2],
system_prompt="You are a helpful AI assistant."
)
# Usage
result = await agent.run("user message")
Integration with OpenRouter
Setup
import os
from pydantic_ai.models import OpenRouterModel
model = OpenRouterModel(
name='openai/gpt-4o',
api_key=os.getenv('OPENROUTER_API_KEY'),
http_referer=os.getenv('FRONTEND_URL')
)
Environment Variables
OPENROUTER_API_KEY=sk-or-v1-...
FRONTEND_URL=http://localhost:3000
Tool Definition Pattern
from pydantic import BaseModel, Field
from pydantic_ai import Agent, Tool
class GenerateImageArgs(BaseModel):
prompt: str = Field(description="Image description")
num_images: int = Field(ge=1, le=10, default=1)
async def generate_image_tool(args: GenerateImageArgs) -> dict:
# Your implementation
return {"images": [...]}
# Register tool
agent.add_tool(
Tool(
name="generate_image",
description="Generate images using AI",
parameters=GenerateImageArgs,
execute=generate_image_tool
)
)
Streaming Pattern
async def stream_response(agent, message):
async for chunk in agent.stream(message):
yield {
"type": "text" if isinstance(chunk, str) else "tool_call",
"content": chunk
}
Error Handling & Retry
from pydantic_ai import Agent, RetryConfig
agent = Agent(
model='openrouter:openai/gpt-4o',
retry_config=RetryConfig(
max_retries=3,
retry_on=[ValidationError, TimeoutError]
)
)
# Auto-retry on validation errors
try:
result = await agent.run("user message")
except ValidationError as e:
# Will retry automatically
logger.error(f"Validation failed after retries: {e}")
FastAPI Integration
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class ChatRequest(BaseModel):
message: str
history: list = []
@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
try:
result = await agent.run(
request.message,
context={"history": request.history}
)
return {"response": result.result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
Testing Pattern
import pytest
from pydantic_ai import Agent
@pytest.mark.asyncio
async def test_agent_response():
agent = Agent(
model='openrouter:openai/gpt-4o',
system_prompt="You are a test assistant"
)
result = await agent.run("Say hello")
assert "hello" in result.lower()
Best Practices
- Type Safety: Always define Pydantic models for inputs/outputs
- Dependency Injection: Use FastAPI-style DI for tools
- Auto-Retry: Configure retry logic for robustness
- Logging: Add structured logging for debugging
- Testing: Write pytest tests for agent behaviors
- Validation: Let Pydantic handle validation automatically
- Context: Pass context dict for stateful conversations
Example: Complete Agent
from pydantic_ai import Agent, Tool
from pydantic import BaseModel, Field
import os
# Output type
class ChatResponse(BaseModel):
message: str
tool_used: str | None = None
confidence: float = Field(ge=0, le=1)
# Tool definition
class WeatherArgs(BaseModel):
city: str
async def get_weather(args: WeatherArgs) -> dict:
# Your API call here
return {"temp": 72, "condition": "sunny"}
# Create agent
agent = Agent(
model='openrouter:openai/gpt-4o',
output_type=ChatResponse,
system_prompt="You are a helpful weather assistant."
)
# Register tool
agent.add_tool(
Tool(
name="get_weather",
description="Get current weather for a city",
parameters=WeatherArgs,
execute=get_weather
)
)
# Usage
if __name__ == "__main__":
result = await agent.run("What's the weather in SF?")
print(result.message)
Common Pitfalls
❌ Don't: Use any type
✅ Do: Define strict Pydantic models
❌ Don't: Handle retries manually ✅ Do: Configure RetryConfig
❌ Don't: Parse LLM output manually ✅ Do: Let Pydantic AI handle it