Agent Builder Pydantic Ai

---
name: agent-builder-pydantic-ai
description: Build conversational AI agents using Pydantic AI + OpenRouter. Use when creating type-safe Python agents with tool calling, validation, and streaming.
license: MIT
---

# 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
```bash
pip install pydantic-ai httpx pydantic python-dotenv
```

### Base Agent Pattern
```python
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
```python
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
```bash
OPENROUTER_API_KEY=sk-or-v1-...
FRONTEND_URL=http://localhost:3000
```

## Tool Definition Pattern

```python
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

```python
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

```python
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

```python
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

```python
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

1. **Type Safety**: Always define Pydantic models for inputs/outputs
2. **Dependency Injection**: Use FastAPI-style DI for tools
3. **Auto-Retry**: Configure retry logic for robustness
4. **Logging**: Add structured logging for debugging
5. **Testing**: Write pytest tests for agent behaviors
6. **Validation**: Let Pydantic handle validation automatically
7. **Context**: Pass context dict for stateful conversations

## Example: Complete Agent

```python
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](https://ai.pydantic.dev)
- [OpenRouter Docs](https://openrouter.ai/docs)
- [FastAPI Docs](https://fastapi.tiangolo.com)