Pydantic AI

PydanticAI — Type-Safe AI Agents

Source: pydantic/pydantic-ai (MIT) — production-grade AI agents with Pydantic validation

Why PydanticAI

• Type-safe by default: result_type enforces Pydantic schema on LLM output
• Dependency injection: clean way to pass DB, HTTP clients, config to tools
• First-class streaming: stream structured responses with partial validation
• Model-agnostic: OpenAI, Anthropic, Google, Groq, Ollama out of the box
• Testable: swap models for TestModel in unit tests

Install

pip install pydantic-ai
# model providers
pip install pydantic-ai[anthropic]
pip install pydantic-ai[openai]

Minimal Agent

from pydantic_ai import Agent

agent = Agent(
    "claude-sonnet-4-5",
    system_prompt="You are a helpful assistant.",
)

result = agent.run_sync("What is the capital of France?")
print(result.data)  # "Paris"

Structured Output

from pydantic import BaseModel
from pydantic_ai import Agent

class CityInfo(BaseModel):
    city: str
    country: str
    population: int
    fun_fact: str

agent = Agent(
    "claude-sonnet-4-5",
    result_type=CityInfo,
    system_prompt="Extract city information from user messages.",
)

result = agent.run_sync("Tell me about Tokyo")
info = result.data  # CityInfo instance — fully typed
print(info.city, info.population)

Tools (Function Tools)

from pydantic_ai import Agent, RunContext
from dataclasses import dataclass
import httpx

@dataclass
class Deps:
    http_client: httpx.AsyncClient
    api_key: str

agent = Agent(
    "claude-sonnet-4-5",
    deps_type=Deps,
    system_prompt="You can look up weather information.",
)

@agent.tool
async def get_weather(ctx: RunContext[Deps], city: str) -> str:
    """Get current weather for a city."""
    resp = await ctx.deps.http_client.get(
        f"https://api.weather.com/v1/current",
        params={"city": city, "key": ctx.deps.api_key},
    )
    data = resp.json()
    return f"{data['temp']}°C, {data['condition']}"

@agent.tool_plain  # no ctx access, pure function
def celsius_to_fahrenheit(celsius: float) -> float:
    """Convert Celsius to Fahrenheit."""
    return celsius * 9/5 + 32

async def main():
    async with httpx.AsyncClient() as client:
        deps = Deps(http_client=client, api_key="my-key")
        result = await agent.run("What's the weather in Tokyo?", deps=deps)
        print(result.data)

Dynamic System Prompt

from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class UserDeps:
    username: str
    role: str

agent = Agent("claude-sonnet-4-5", deps_type=UserDeps)

@agent.system_prompt
def build_system_prompt(ctx: RunContext[UserDeps]) -> str:
    return f"You are helping {ctx.deps.username} who is a {ctx.deps.role}."

result = agent.run_sync(
    "Help me with my task",
    deps=UserDeps(username="Alice", role="data scientist"),
)

Streaming

import asyncio
from pydantic_ai import Agent
from pydantic import BaseModel

class Summary(BaseModel):
    title: str
    points: list[str]

agent = Agent("claude-sonnet-4-5", result_type=Summary)

async def main():
    async with agent.run_stream("Summarize AI trends") as result:
        # stream text deltas
        async for text in result.stream_text():
            print(text, end="", flush=True)

    # structured result available after stream completes
    final = await result.get_data()
    print(final.title, final.points)

asyncio.run(main())

Multi-Turn Conversations

from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessagesTypeAdapter

agent = Agent("claude-sonnet-4-5", system_prompt="You are a helpful assistant.")

# First turn
result1 = agent.run_sync("My name is Alice.")
print(result1.data)

# Continue conversation — pass message history
result2 = agent.run_sync(
    "What's my name?",
    message_history=result1.new_messages(),
)
print(result2.data)  # "Your name is Alice."

# Serialize history for persistence
history_json = result2.all_messages_json()
# Restore later:
history = ModelMessagesTypeAdapter.validate_json(history_json)

Result Validators

from pydantic_ai import Agent, ModelRetry

agent = Agent("claude-sonnet-4-5", result_type=int)

@agent.result_validator
async def validate_positive(ctx, result: int) -> int:
    if result <= 0:
        raise ModelRetry("Result must be positive. Try again.")
    return result

result = agent.run_sync("Give me a positive number")

Testing with TestModel

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

agent = Agent("claude-sonnet-4-5", system_prompt="You answer math questions.")

def test_addition():
    with agent.override(model=TestModel()):
        result = agent.run_sync("What is 2+2?")
        # TestModel returns deterministic canned responses
        assert result.data is not None

# Or use FunctionModel for custom test responses
from pydantic_ai.models.function import FunctionModel, ModelContext

def my_test_model(messages, info: ModelContext):
    return "4"

with agent.override(model=FunctionModel(my_test_model)):
    result = agent.run_sync("What is 2+2?")
    assert result.data == "4"

Supported Models

from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.models.google import GoogleModel
from pydantic_ai.models.groq import GroqModel
from pydantic_ai.models.ollama import OllamaModel

# Use model string shorthand
agent = Agent("claude-sonnet-4-5")
agent = Agent("gpt-4o")
agent = Agent("gemini-2.0-flash")

# Or explicit model instance with custom config
model = AnthropicModel("claude-opus-4-5", max_tokens=8192)
agent = Agent(model)

Usage Stats

result = agent.run_sync("Hello")
print(result.usage())
# Usage(requests=1, request_tokens=25, response_tokens=10, total_tokens=35)

Anti-Fake-Pass Checks

• result.data holds the validated result, not result.output or result.text
• @agent.tool receives ctx: RunContext[DepsType] as first arg — omit for @agent.tool_plain
• deps_type must be declared on Agent for ctx.deps to be typed
• ModelRetry from pydantic_ai retries the LLM call — not a Python exception
• run_sync is a sync wrapper; use await agent.run(...) in async contexts
• result.new_messages() returns only new messages; result.all_messages() returns full history