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

Resources

• Pydantic AI Docs

• OpenRouter Docs

• FastAPI Docs