Pydantic AI (pydantic/pydantic-ai)

Pydantic AI

https://github.com/pydantic/pydantic-ai
Admin
Pydantic AI is a Python agent framework designed to simplify building production-grade applications...

Tokens:307,660
Snippets:1,856
Trust Score:9.6
Update:6 days ago
Show doc for...
Context Summary (auto-generated)
Raw
# Pydantic AI

Pydantic AI is a Python agent framework designed to help build production-grade applications with Generative AI. Built by the creators of Pydantic, it brings FastAPI-style ergonomic design to GenAI development, offering type-safe interactions with LLMs, model-agnostic architecture supporting virtually every major provider (OpenAI, Anthropic, Gemini, Bedrock, Groq, Mistral, and many more), and seamless integration with Pydantic Logfire for observability. The framework emphasizes structured outputs, dependency injection for tools and system prompts, and comprehensive streaming support.

The core abstraction is the `Agent` class, which orchestrates interactions with LLMs by combining instructions, function tools, structured output types, and dependency injection. Agents can run synchronously or asynchronously, support streaming responses, and maintain conversation history across multiple runs. The framework also includes Pydantic Graph for complex state-machine workflows, Pydantic Evals for systematic testing of AI systems, and MCP (Model Context Protocol) integration for connecting to external tools and services.

## Creating a Basic Agent

The Agent class is the primary interface for interacting with LLMs. Agents are configured with a model, optional instructions, and can be run synchronously or asynchronously to generate responses.

```python
from pydantic_ai import Agent

# Create a simple agent with static instructions
agent = Agent(
    'openai:gpt-5.2',
    instructions='Be concise, reply with one sentence.',
)

# Run synchronously
result = agent.run_sync('Where does "hello world" come from?')
print(result.output)
# Output: The first known use of "hello, world" was in a 1974 textbook about the C programming language.

# Run asynchronously
async def main():
    result = await agent.run('What is the capital of France?')
    print(result.output)
    # Output: The capital of France is Paris.
```

## Structured Output with Pydantic Models

Agents can be configured to return structured data using Pydantic models as the output_type. The LLM is forced to return data matching the specified schema, with automatic validation.

```python
from pydantic import BaseModel, Field
from pydantic_ai import Agent


class CityLocation(BaseModel):
    city: str
    country: str


class SupportOutput(BaseModel):
    support_advice: str = Field(description='Advice returned to the customer')
    block_card: bool = Field(description="Whether to block the customer's card")
    risk: int = Field(description='Risk level of query', ge=0, le=10)


# Agent with structured output
agent = Agent('google-gla:gemini-3-flash-preview', output_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.output)
# Output: city='London' country='United Kingdom'
print(result.usage())
# Output: RunUsage(input_tokens=57, output_tokens=8, requests=1)

# Multiple output types (union)
agent = Agent(
    'openai:gpt-5-mini',
    output_type=[CityLocation, str],
    instructions='Extract city location if provided, otherwise respond with text.',
)
result = agent.run_sync('Tell me about Paris')
print(result.output)
# Output: city='Paris' country='France'
```

## Function Tools with @agent.tool Decorator

Tools provide a mechanism for models to perform actions and retrieve information. Use @agent.tool for tools that need context access, or @agent.tool_plain for standalone tools.

```python
import random
from pydantic_ai import Agent, RunContext


agent = Agent(
    'google-gla:gemini-3-flash-preview',
    deps_type=str,  # Player's name as dependency
    instructions=(
        "You're a dice game, roll the die and see if the number "
        "matches the user's guess. Use the player's name in the response."
    ),
)


@agent.tool_plain
def roll_dice() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))


@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps


# Run the agent with dependencies
result = agent.run_sync('My guess is 4', deps='Anne')
print(result.output)
# Output: Congratulations Anne, you guessed correctly! You're a winner!
```

## Dependency Injection System

Dependencies allow passing data, connections, and services to system prompts, tools, and validators. Define dependencies using dataclasses and access them via RunContext.

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


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


agent = Agent(
    'openai:gpt-5.2',
    deps_type=MyDeps,
)


@agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await ctx.deps.http_client.get(
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    response.raise_for_status()
    return f'System context: {response.text}'


@agent.tool
async def fetch_data(ctx: RunContext[MyDeps], query: str) -> str:
    """Fetch data from external API."""
    response = await ctx.deps.http_client.get(
        'https://api.example.com/data',
        params={'q': query},
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    return response.text


async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('api-key-here', client)
        result = await agent.run('Search for Python tutorials', deps=deps)
        print(result.output)
```

## Streaming Responses

Agents support streaming text and structured output for real-time response handling. Use run_stream() for streaming with context manager.

```python
from pydantic_ai import Agent
from typing_extensions import NotRequired, TypedDict


agent = Agent('google-gla:gemini-3-flash-preview')


# Stream text output
async def stream_text():
    async with agent.run_stream('Tell me a story about a cat') as result:
        async for text in result.stream_text():
            print(text)
            # Output (incremental):
            # Once upon a
            # Once upon a time, there was
            # Once upon a time, there was a curious cat...


# Stream structured output
class UserProfile(TypedDict):
    name: str
    dob: NotRequired[str]
    bio: NotRequired[str]


profile_agent = Agent(
    'openai:gpt-5.2',
    output_type=UserProfile,
    instructions='Extract a user profile from the input',
)


async def stream_structured():
    user_input = 'My name is Ben, born January 28th 1990, I like hiking.'
    async with profile_agent.run_stream(user_input) as result:
        async for profile in result.stream_output():
            print(profile)
            # Output (incremental):
            # {'name': 'Ben'}
            # {'name': 'Ben', 'dob': '1990-01-28'}
            # {'name': 'Ben', 'dob': '1990-01-28', 'bio': 'Likes hiking'}
```

## Message History and Conversations

Maintain conversation context across multiple agent runs by passing message history. Messages are model-agnostic and can be serialized for persistence.

```python
from pydantic_ai import Agent, ModelMessagesTypeAdapter
from pydantic_core import to_jsonable_python

agent = Agent('openai:gpt-5.2', instructions='Be a helpful assistant.')

# First run
result1 = agent.run_sync('Who was Albert Einstein?')
print(result1.output)
# Output: Albert Einstein was a German-born theoretical physicist.

# Continue conversation with message history
result2 = agent.run_sync(
    'What was his most famous equation?',
    message_history=result1.new_messages(),
)
print(result2.output)
# Output: Albert Einstein's most famous equation is E = mc^2.

# Serialize messages for storage
history = result2.all_messages()
json_history = to_jsonable_python(history)

# Restore messages from storage
restored_history = ModelMessagesTypeAdapter.validate_python(json_history)
result3 = agent.run_sync('Tell me more about it', message_history=restored_history)
```

## Dynamic Instructions with Decorators

Use the @agent.instructions decorator to create dynamic instructions that can access dependencies and runtime context.

```python
from datetime import date
from pydantic_ai import Agent, RunContext


agent = Agent(
    'openai:gpt-5.2',
    deps_type=str,
    instructions="Use the customer's name while replying to them.",
)


@agent.instructions
def add_customer_name(ctx: RunContext[str]) -> str:
    return f"The customer's name is {ctx.deps}."


@agent.instructions
def add_date() -> str:
    return f'The date is {date.today()}.'


result = agent.run_sync('What is the date?', deps='Frank')
print(result.output)
# Output: Hello Frank, the date today is 2032-01-02.
```

## Model Retry and Error Handling

Use ModelRetry to ask the model to retry with different parameters. Configure retry limits at agent, tool, or output level.

```python
from pydantic_ai import Agent, RunContext, ModelRetry
from pydantic_ai.exceptions import UnexpectedModelBehavior, UsageLimitExceeded
from pydantic_ai.usage import UsageLimits


agent = Agent(
    'openai:gpt-5.2',
    retries=3,  # Global retry limit
)


@agent.tool(retries=2)  # Tool-specific retry limit
def get_user_by_name(ctx: RunContext, name: str) -> int:
    """Get a user's ID from their full name."""
    user_id = lookup_user(name)  # Your lookup function
    if user_id is None:
        raise ModelRetry(
            f'No user found with name {name!r}, provide their full name'
        )
    return user_id


# Usage limits to prevent runaway costs
try:
    result = agent.run_sync(
        'Find user John',
        usage_limits=UsageLimits(
            response_tokens_limit=100,
            request_limit=5,
            tool_calls_limit=10,
        ),
    )
except UsageLimitExceeded as e:
    print(f'Limit exceeded: {e}')
except UnexpectedModelBehavior as e:
    print(f'Model error: {e}')
```

## Output Functions

Output functions let the model call a function as its final action. The function result becomes the agent's output without being sent back to the model.

```python
from pydantic import BaseModel
from pydantic_ai import Agent, ModelRetry, RunContext


class Row(BaseModel):
    name: str
    country: str


def run_sql_query(query: str) -> list[Row]:
    """Run a SQL query on the database."""
    if 'SELECT' not in query.upper():
        raise ModelRetry('Only SELECT queries are supported')
    # Execute query and return results
    return [Row(name='Paris', country='France')]


sql_agent = Agent[None, list[Row]](
    'openai:gpt-5.2',
    output_type=run_sql_query,  # Function as output type
    instructions='You are a SQL agent that runs queries on a database.',
)

result = sql_agent.run_sync('Get all capital cities')
print(result.output)
# Output: [Row(name='Paris', country='France')]
```

## Output Validators

Add custom validation logic for agent outputs using the @agent.output_validator decorator. Validators can be async and raise ModelRetry.

```python
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext, ModelRetry


class SQLQuery(BaseModel):
    query: str


agent = Agent('openai:gpt-5.2', output_type=SQLQuery)


@agent.output_validator
async def validate_sql(ctx: RunContext, output: SQLQuery) -> SQLQuery:
    try:
        # Validate query syntax
        await execute_explain(output.query)
    except QueryError as e:
        raise ModelRetry(f'Invalid query: {e}')
    return output


result = agent.run_sync('Generate a query to get all users')
print(result.output.query)
```

## Multiple Model Providers

Pydantic AI supports multiple model providers with a consistent interface. Switch providers by changing the model string or using model classes directly.

```python
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.models.google import GoogleModel
from pydantic_ai.models.fallback import FallbackModel

# Using model strings (auto-detection)
openai_agent = Agent('openai:gpt-5.2')
anthropic_agent = Agent('anthropic:claude-sonnet-4-5')
gemini_agent = Agent('google-gla:gemini-3-flash-preview')
groq_agent = Agent('groq:llama-4-scout-17b-16e-instruct')

# Using model classes for more control
model = OpenAIChatModel('gpt-5.2', api_key='your-key')
agent = Agent(model)

# Fallback model for reliability
fallback = FallbackModel(
    OpenAIChatModel('gpt-5.2'),
    AnthropicModel('claude-sonnet-4-5'),
)
reliable_agent = Agent(fallback)
result = reliable_agent.run_sync('Hello')  # Falls back if first model fails
```

## MCP (Model Context Protocol) Integration

Connect agents to MCP servers for external tools and services. Pydantic AI can act as an MCP client or be used within MCP servers.

```python
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServer

# Connect to an MCP server
mcp_server = MCPServer('http://localhost:8080')

agent = Agent(
    'openai:gpt-5.2',
    toolsets=[mcp_server],  # Add MCP tools to agent
)

result = agent.run_sync('Use the external tool to fetch data')
print(result.output)
```

## Pydantic Evals for Testing

Pydantic Evals provides a framework for systematically testing AI systems with datasets, evaluators, and experiment reports.

```python
from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import Evaluator, EvaluatorContext, IsInstance


# Define test cases
case1 = Case(
    name='capital_question',
    inputs='What is the capital of France?',
    expected_output='Paris',
    metadata={'difficulty': 'easy'},
)


# Custom evaluator
class MyEvaluator(Evaluator[str, str]):
    def evaluate(self, ctx: EvaluatorContext[str, str]) -> float:
        if ctx.output == ctx.expected_output:
            return 1.0
        elif ctx.expected_output.lower() in ctx.output.lower():
            return 0.8
        return 0.0


# Create dataset with evaluators
dataset = Dataset(
    cases=[case1],
    evaluators=[IsInstance(type_name='str'), MyEvaluator()],
)


# Define the task to evaluate
async def answer_question(question: str) -> str:
    agent = Agent('openai:gpt-5-mini')
    result = await agent.run(question)
    return result.output


# Run evaluation
report = dataset.evaluate_sync(answer_question)
report.print(include_input=True, include_output=True)
```

## Pydantic Graph for Complex Workflows

Use pydantic-graph for state-machine workflows where nodes and edges are defined using type hints.

```python
from __future__ import annotations
from dataclasses import dataclass
from pydantic_graph import BaseNode, End, Graph, GraphRunContext


@dataclass
class DivisibleBy5(BaseNode[None, None, int]):
    foo: int

    async def run(self, ctx: GraphRunContext) -> Increment | End[int]:
        if self.foo % 5 == 0:
            return End(self.foo)
        return Increment(self.foo)


@dataclass
class Increment(BaseNode):
    foo: int

    async def run(self, ctx: GraphRunContext) -> DivisibleBy5:
        return DivisibleBy5(self.foo + 1)


# Create and run the graph
graph = Graph(nodes=[DivisibleBy5, Increment])
result = graph.run_sync(DivisibleBy5(4))
print(result.output)  # Output: 5

# Generate mermaid diagram
print(graph.mermaid_code(start_node=DivisibleBy5))
```

## Logfire Integration for Observability

Integrate with Pydantic Logfire for real-time debugging, tracing, and monitoring of agent behavior.

```python
import logfire
from pydantic_ai import Agent

# Configure Logfire
logfire.configure()
logfire.instrument_pydantic_ai()

# All agent runs are now traced
agent = Agent('openai:gpt-5.2')
result = agent.run_sync('What is 2 + 2?')

# Traces show:
# - Messages exchanged with the model
# - Tool calls with arguments and return values
# - Token usage per request
# - Latency for each operation
# - Any errors with full context
```

## Complete Bank Support Agent Example

A complete example demonstrating dynamic instructions, structured output, tools, and dependency injection.

```python
import sqlite3
from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext


@dataclass
class DatabaseConn:
    conn: sqlite3.Connection

    async def customer_name(self, id: int) -> str | None:
        cur = self.conn.cursor()
        res = cur.execute('SELECT name FROM customers WHERE id=?', (id,))
        row = res.fetchone()
        return row[0] if row else None

    async def customer_balance(self, id: int) -> float:
        cur = self.conn.cursor()
        res = cur.execute('SELECT balance FROM customers WHERE id=?', (id,))
        row = res.fetchone()
        if row:
            return row[0]
        raise ValueError('Customer not found')


@dataclass
class SupportDeps:
    customer_id: int
    db: DatabaseConn


class SupportOutput(BaseModel):
    support_advice: str
    block_card: bool
    risk: int


support_agent = Agent(
    'openai:gpt-5.2',
    deps_type=SupportDeps,
    output_type=SupportOutput,
    instructions=(
        'You are a support agent in our bank. Give the customer '
        'support and judge the risk level of their query.'
    ),
)


@support_agent.instructions
async def add_customer_name(ctx: RunContext[SupportDeps]) -> str:
    name = await ctx.deps.db.customer_name(ctx.deps.customer_id)
    return f"The customer's name is {name!r}"


@support_agent.tool
async def customer_balance(ctx: RunContext[SupportDeps]) -> str:
    """Returns the customer's current account balance."""
    balance = await ctx.deps.db.customer_balance(ctx.deps.customer_id)
    return f'${balance:.2f}'


# Usage
with sqlite3.connect(':memory:') as conn:
    conn.execute('CREATE TABLE customers(id, name, balance)')
    conn.execute("INSERT INTO customers VALUES (123, 'John', 123.45)")
    conn.commit()

    deps = SupportDeps(customer_id=123, db=DatabaseConn(conn))

    result = support_agent.run_sync('What is my balance?', deps=deps)
    print(result.output)
    # Output: support_advice='Hello John, your balance is $123.45.' block_card=False risk=1

    result = support_agent.run_sync('I just lost my card!', deps=deps)
    print(result.output)
    # Output: support_advice="I'm sorry John, blocking your card now." block_card=True risk=8
```

## Summary

Pydantic AI provides a comprehensive framework for building production-grade AI applications with strong typing, model-agnostic design, and enterprise-ready features. The core Agent class serves as the primary interface, supporting synchronous and asynchronous execution, streaming responses, and structured outputs validated by Pydantic. The dependency injection system enables clean separation of concerns, while tools and dynamic instructions allow agents to interact with external services and adapt to runtime context.

Key integration patterns include: using message history for multi-turn conversations, implementing fallback models for reliability, connecting to MCP servers for external tools, and leveraging Pydantic Evals for systematic testing. For complex workflows, pydantic-graph provides a type-safe state machine abstraction. Throughout all these patterns, Logfire integration enables comprehensive observability for debugging and monitoring agent behavior in production environments.