Getting Started with PydanticAI

Overview

PydanticAI is a type-safe AI agent framework built on the Pydantic ecosystem. This tutorial walks you through building your first AI agent with PydanticAI, from basic setup to advanced patterns.

Prerequisites

• Python 3.10+
• OpenAI API key (or other LLM provider)
• Basic Python knowledge

Installation

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install PydanticAI
pip install pydantic-ai

# Or with extras
pip install pydantic-ai[openai,anthropic,google]

# Create .env file
echo "OPENAI_API_KEY=sk-..." > .env

Your First Agent

Basic Example

from pydantic_ai import Agent

# Create a simple agent
agent = Agent('openai:gpt-4o')

# Run a query
result = agent.run_sync('What is the capital of France?')
print(result.data)

With Structured Output

from pydantic_ai import Agent
from pydantic import BaseModel, Field

class WeatherInfo(BaseModel):
    """Weather information for a location."""
    location: str = Field(description="The location name")
    temperature: float = Field(description="Temperature in Celsius")
    conditions: str = Field(description="Weather conditions")
    humidity: int = Field(description="Humidity percentage")

agent = Agent(
    'openai:gpt-4o',
    result_type=WeatherInfo,  # Type-safe result
)

result = agent.run_sync('What is the weather like in London?')
print(result.data.location)
print(result.data.temperature)
print(result.data.conditions)

Understanding Agents

Agent Components

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

# Model configuration
model = OpenAIModel(
    'gpt-4o',
    api_key='sk-...',
    base_url='https://api.openai.com/v1',
)

# Agent with model
agent = Agent(
    model,
    system_prompt='You are a helpful assistant.',
    result_type=str,
)

System Prompt

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-4o',
    system_prompt='''
    You are an expert data analyst.
    Your job is to analyze data and provide insights.
    Always be concise and focus on actionable findings.
    ''',
)

Dynamic System Prompt

from pydantic_ai import Agent, RunContext

class AgentDeps(BaseModel):
    context: str
    tone: str

agent = Agent(
    'openai:gpt-4o',
    deps_type=AgentDeps,
)

@agent.system_prompt
def add_context(ctx: RunContext[AgentDeps]) -> str:
    return f'''
    You are a helpful assistant.
    
    Context: {ctx.deps.context}
    Tone: {ctx.deps.tone}
    '''

result = agent.run_sync(
    'Summarize this data',
    deps=AgentDeps(
        context='Q4 sales data shows 15% growth',
        tone='professional',
    ),
)

Tools

Defining Tools

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str

agent = Agent('openai:gpt-4o')

@agent.tool
async def search_web(ctx: RunContext, query: str) -> list[SearchResult]:
    """Search the web for information."""
    # Implement search logic
    results = await perform_search(query)
    return [SearchResult(**r) for r in results]

@agent.tool
async def fetch_page(ctx: RunContext, url: str) -> str:
    """Fetch and return the content of a webpage."""
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

# Use in conversation
result = agent.run_sync('Find information about AI agents')

Tool with Dependencies

from pydantic_ai import Agent, RunContext

class ToolDeps(BaseModel):
    api_key: str
    base_url: str

agent = Agent(
    'openai:gpt-4o',
    deps_type=ToolDeps,
)

@agent.tool
async def search_api(ctx: RunContext[ToolDeps], query: str) -> str:
    """Search using the configured API."""
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{ctx.deps.base_url}/search",
            params={"q": query},
            headers={"Authorization": f"Bearer {ctx.deps.api_key}"},
        )
        return response.text

result = agent.run_sync(
    'Search for AI frameworks',
    deps=ToolDeps(
        api_key='sk-...',
        base_url='https://api.example.com',
    ),
)

Multi-Agent Systems

Agent Dependencies

from pydantic_ai import Agent
from pydantic import BaseModel

class ResearchResult(BaseModel):
    findings: list[str]
    sources: list[str]

class WritingResult(BaseModel):
    article: str
    word_count: int

# Research agent
research_agent = Agent(
    'openai:gpt-4o',
    result_type=ResearchResult,
)

# Writer agent depends on research result
writer_agent = Agent(
    'openai:gpt-4o',
    result_type=WritingResult,
    deps_type=ResearchResult,
)

@writer_agent.system_prompt
def add_research_context(ctx: RunContext[ResearchResult]) -> str:
    return f"Based on research findings: {', '.join(ctx.deps.findings[:3])}"

# Chain agents
research = research_agent.run_sync('Research quantum computing')
writing = writer_agent.run_sync(
    'Write an article',
    deps=research.data,
)

Agent Handoffs

from pydantic_ai import Agent
from pydantic import BaseModel

class TaskResult(BaseModel):
    completed: bool
    output: str

researcher = Agent('openai:gpt-4o', result_type=TaskResult)
writer = Agent('openai:gpt-4o', result_type=TaskResult)
editor = Agent('openai:gpt-4o', result_type=TaskResult)

def run_workflow(topic: str) -> TaskResult:
    # Research
    research_result = researcher.run_sync(f'Research {topic}')
    
    # Write
    writer_result = writer.run_sync(
        f'Write about {topic}',
        deps=research_result.data,
    )
    
    # Edit
    editor_result = editor.run_sync(
        'Edit the content',
        deps=writer_result.data,
    )
    
    return editor_result.data

Streaming

Stream Text

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

async def stream_response():
    async for chunk in agent.run_stream('Explain quantum computing'):
        print(chunk.data, end='', flush=True)

import asyncio
asyncio.run(stream_response())

Stream Events

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

async def stream_events():
    async for event in agent.iter_events('Analyze this data'):
        if event.type == 'text':
            print(event.data, end='', flush=True)
        elif event.type == 'tool_call':
            print(f"\nCalling tool: {event.data.name}")
        elif event.type == 'retries':
            print(f"\nRetries: {event.data}")

asyncio.run(stream_events())

Error Handling

Result Validation

from pydantic_ai import Agent, ModelRetry
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    findings: list[str]
    confidence: float

agent = Agent('openai:gpt-4o', result_type=AnalysisResult)

@agent.result_validator
def validate_result(result: AnalysisResult) -> AnalysisResult:
    if not result.findings:
        raise ModelRetry('No findings returned, please try again')
    if result.confidence < 0.5:
        raise ModelRetry('Low confidence, please re-analyze')
    return result

result = agent.run_sync('Analyze this dataset')

Tool Error Handling

from pydantic_ai import Agent, RunContext

agent = Agent('openai:gpt-4o')

@agent.tool
async def reliable_tool(ctx: RunContext, query: str) -> str:
    """A tool with error handling."""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            return await perform_search(query)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)
    return ""

result = agent.run_sync('Search for information')

Testing

Mock Model

from pydantic_ai import Agent
from pydantic_ai.test import MockModel

agent = Agent('openai:gpt-4o')

# Create mock model
mock = MockModel()
mock.add_chat_completion('Hello', 'Hi there!')

# Replace model for testing
agent._model = mock

result = agent.run_sync('Say hello')
assert result.data == 'Hi there!'

Test with Real API

import pytest
from pydantic_ai import Agent

@pytest.mark.asyncio
async def test_agent_response():
    agent = Agent('openai:gpt-4o')
    result = await agent.run('What is 2+2?')
    assert '4' in result.data.lower()

Advanced Patterns

Conversation State

from pydantic_ai import Agent
from pydantic import BaseModel

class ConversationState(BaseModel):
    messages: list[dict]
    context: dict

agent = Agent(
    'openai:gpt-4o',
    result_type=str,
)

# State is managed automatically
result1 = agent.run_sync('What is the project about?')
result2 = agent.run_sync('Tell me more about the timeline')
# Both calls share context

print(agent.state)

Retry Logic

from pydantic_ai import Agent
import asyncio

agent = Agent('openai:gpt-4o')

async def run_with_retry(query: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            result = await agent.run(query)
            return result
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            print(f"Attempt {attempt + 1} failed: {e}")
            await asyncio.sleep(2 ** attempt)

result = asyncio.run(run_with_retry('Analyze this data'))

Batch Processing

from pydantic_ai import Agent
import asyncio

agent = Agent('openai:gpt-4o')

async def process_batch(queries: list[str]):
    tasks = [agent.run(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return [r.data for r in results]

queries = ['Query 1', 'Query 2', 'Query 3']
results = asyncio.run(process_batch(queries))

Multi-Provider Support

Switching Providers

from pydantic_ai import Agent

# OpenAI
agent = Agent('openai:gpt-4o')

# Anthropic
agent = Agent('anthropic:claude-3-5-sonnet-20241022')

# Google Gemini
agent = Agent('google:gemini-1.5-pro')

# Groq
agent = Agent('groq:llama-3.1-405b')

# DeepSeek
agent = Agent('deepseek:deepseek-chat')

# Ollama (local)
agent = Agent('ollama:llama3.1')

Provider Configuration

from pydantic_ai import Agent
from pydantic_ai.providers.openai import OpenAIProvider

provider = OpenAIProvider(
    api_key="sk-...",
    base_url="https://api.openai.com/v1",
)
agent = Agent(provider, model="gpt-4o")

Best Practices

1. Use Structured Outputs

# Good: Type-safe result
class AnalysisResult(BaseModel):
    summary: str
    findings: list[str]
    recommendations: list[str]

agent = Agent('openai:gpt-4o', result_type=AnalysisResult)

# Bad: Unstructured string
agent = Agent('openai:gpt-4o', result_type=str)

2. Be Specific with System Prompts

# Good: Specific role and constraints
agent = Agent(
    'openai:gpt-4o',
    system_prompt='You are a senior data analyst. Provide concise, actionable insights.',
)

# Bad: Vague
agent = Agent('openai:gpt-4o', system_prompt='You are helpful.')

3. Use Dependencies for Context

# Good: Pass context via dependencies
class AgentDeps(BaseModel):
    user_context: str
    preferences: dict

agent = Agent('openai:gpt-4o', deps_type=AgentDeps)

# Bad: Hardcoded context
agent = Agent('openai:gpt-4o', system_prompt='User likes Python.')

4. Handle Errors Gracefully

# Good: Validate results
@agent.result_validator
def validate(result):
    if not result.findings:
        raise ModelRetry('No findings')
    return result

# Bad: No validation
result = agent.run_sync('Analyze')

Troubleshooting

Common Issues

API rate limits:

import asyncio
from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

async def rate_limited_run(query: str):
    await asyncio.sleep(1)  # Rate limit delay
    return await agent.run(query)

Model not found:

# Check available models
from pydantic_ai.models.openai import OpenAIModel

model = OpenAIModel('gpt-4o')  # Must be a valid model name

Tool not called:

# Ensure tool description is clear
@agent.tool
async def search_web(query: str) -> str:
    """Search the web for information about the query."""  # Clear description
    ...

Resources

• PydanticAI Documentation: https://ai.pydantic.dev/
• PydanticAI GitHub: https://github.com/pydantic/pydantic-ai
• Pydantic Documentation: https://docs.pydantic.dev/
• OpenAI API Reference: https://platform.openai.com/docs/api-reference

---

Last updated: May 2026