Pydantic Ai

Pydantic AI

Python agent framework for building production-grade GenAI applications with the "FastAPI feeling".

Quick Navigation

When to Use

• Building AI agents with structured output
• Need type-safe, IDE-friendly agent development
• Require dependency injection for tools
• Multi-model support (OpenAI, Anthropic, Gemini, etc.)
• Production observability with Logfire
• Complex workflows with graphs

Installation

See references/installation.md for full/slim install options and optional dependency groups. Requires Python 3.10+.

Release Highlights (1.96.1)

• V2 preparation: Agent(..., prepare_tools=..., prepare_output_tools=..., event_stream_handler=...) is now the deprecated path; capability-based migration is the new direction.
• Capability migration: use PrepareTools, PrepareOutputTools, and ProcessEventStream capabilities instead of wiring those behaviors through constructor sugar.
• OpenAI fixes: the latest patch line also tightens OpenAIResponsesModel system-prompt-role handling and image-generation request shaping.

Release Highlights (1.97.0 -> 1.102.0)

• MCP migration: prefer MCPToolset for new MCP integrations. FastMCPToolset and the older MCPServer* client surface are now on the deprecation path.
• Google provider split: GoogleProvider and GoogleCloudProvider are now distinct, and model ids move from google-gla: / google-vertex: to google: / google-cloud:.
• Streaming migration: move from stream_responses() to stream_response(); the newer API yields ModelResponse objects directly.
• Retry configuration: prefer Agent(retries=...) or AgentRetries(...) over older constructor-level retry knobs.
• New runtime tools: ctx.enqueue() and MCP background tasks make it easier to queue follow-up work without forcing it into the current response turn.

Release Highlights (1.105.0 -> 1.107.0)

• New models: Claude Fable 5 and Claude Mythos 5 are supported (1.107.0), alongside Grok 4.3 reasoning_effort and updated xAI model names (1.105.0).
• Deferred loading: instructions, tools, model settings, and hooks can now be loaded on demand instead of eagerly at agent construction (1.105.0).
• Model introspection: known_model_names() enumerates KnownModelName members (1.107.0).
• OpenRouter caching: CachePoint and prompt caching are implemented for OpenRouter (1.107.0).
• xAI config: XaiProvider gains api_host and timeout, plus seed parameter mapping (1.106.0).
• Security: VercelAIAdapter UploadedFile handling was hardened against a confused-deputy file-read vulnerability (GHSA-h7p7-w5gc-xj3w, 1.106.0).
• Fixes: incomplete streamed responses when event_stream_handler doesn't consume the stream, from_data_uri on non-base64 data URIs, Temporal gateway/ model construction, GoogleModelSettings.google_cached_content request shaping, Anthropic Bedrock message=None start events, and AnthropicModel.count_tokens with native tools.

Release Highlights (1.103.0 -> 1.104.0)

• MCP prompts: maintained McpServer integrations can now list and fetch prompts with list_prompts and get_prompt; still prefer MCPToolset for new client code unless legacy wrappers are required.
• UI adapters: VercelAIAdapter round-trips message timestamps through UIMessage.metadata, and UIAdapter.sanitize_messages strips client-submitted force_download from FileUrl parts.
• Provider/model updates: Claude Opus 4.8 is supported, OpenRouterModel can use anthropic_eager_input_streaming, and hybrid OpenRouter/xAI/Bedrock routes now forward thinking=False consistently.
• Bedrock/toolset fixes: Bedrock maps malformed model/tool output to FinishReason.error, recognizes adaptive thinking, preserves single-tool tool_choice cache behavior, and toolset prepare callbacks warn when they accidentally return None.

Release Highlights (1.75.0 -> 1.84.1)

• Capabilities: CapabilityOrdering adds explicit wrapping/ordering control (innermost, outermost, wraps, wrapped_by, requires) for complex capability stacks.
• Compaction: new server-side compaction capabilities for OpenAI and Anthropic; OpenAI adds stateful compaction mode.
• Models: Claude Opus 4.7 support and a native OllamaModel path with corrected Ollama capability flags for structured output.
• Tools: tool hooks now consistently receive dict-shaped validated args for single-BaseModel tools, and internal output tools skip hook execution.
• Hardening: Google FileSearchTool parsing received regex hardening in the 1.83/1.84 line.

Release Highlights (1.71.0 → 1.74.0)

• Capabilities: composable, reusable units of agent behavior that bundle tools, lifecycle hooks, instructions, and model settings into a single class. Plug into any agent for maximum reuse.
• AgentSpec: load agents from YAML/JSON files via Agent.from_file. Supports TemplateStr for templated instructions referencing deps.
• Hooks capability: define hooks using decorators (@hooks.on_model_request, etc.).
• Thinking capability: cross-provider thinking model setting for reasoning.
• Provider-adaptive tools: WebSearch, WebFetch, MCP, ImageGeneration — automatically fall back from builtin (provider) tools to local tools.
• Online evaluation: evaluation infrastructure in pydantic-evals.
• `TextContent`: user prompts with metadata not sent to model.
• CaseLifecycle hooks: hooks for Dataset.evaluate lifecycle.
• Model swapping in hooks: before_model_request / wrap hooks can swap models via ModelRequestContext.
• `ModelRetry` from hooks: hooks can raise ModelRetry for retry control flow.
• Sync tool preparation functions supported. MCP capability no longer requires explicit url=.

Release Highlights (1.69.0 → 1.70.0)

• Agents: Agent(description=...) adds a human-readable description to the run span as gen_ai.agent.description when instrumentation is enabled.
• Models: FallbackModel now supports response-based fallback handlers for semantic failures in non-streaming runs.
• Tools: multimodal tool results are passed directly to provider APIs instead of always being split into extra user-message parts.
• Bedrock: bedrock_inference_profile is available on model and embedding settings for routing requests through an inference profile ARN.
• Stability: provider fixes landed for OpenRouter Anthropic model matching, Cohere embeddings, Google image sizes, Bedrock tool-name sanitization, and malformed tool-call retry handling.

Quick Start

Basic Agent

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-4o',
    instructions='Be concise, reply with one sentence.'
)

result = agent.run_sync('Where does "hello world" come from?')
print(result.output)

With Structured Output

from pydantic import BaseModel
from pydantic_ai import Agent

class CityInfo(BaseModel):
    name: str
    country: str
    population: int

agent = Agent('openai:gpt-4o', output_type=CityInfo)
result = agent.run_sync('Tell me about Paris')
print(result.output)  # CityInfo(name='Paris', country='France', population=2161000)

With Tools and Dependencies

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
    user_id: int

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

@agent.tool
async def get_user_name(ctx: RunContext[Deps]) -> str:
    """Get the current user's name."""
    return f"User #{ctx.deps.user_id}"

result = agent.run_sync('What is my name?', deps=Deps(user_id=123))

Key Features

Supported Models

Best Practices

1. Use type hints — enables IDE support and validation 2. Define output types — guarantees structured responses 3. Use dependencies — inject context into tools 4. Add tool docstrings — LLM uses them as descriptions 5. Enable Logfire — for production observability 6. Use `run_sync` for simple cases — run for async 7. Override deps for testing — agent.override(deps=...) 8. Set usage limits — prevent infinite loops with UsageLimits

Prohibitions

• Do not expose API keys in code
• Do not skip output validation in production
• Do not ignore tool errors
• Do not use run_stream without handling partial outputs
• Do not forget to close MCP connections (async with agent)
• Do not assume capability order is arbitrary once multiple wrappers/hooks are involved; define it explicitly when composition matters.

Common Patterns

Streaming Response

async with agent.run_stream('Query') as response:
    async for text in response.stream_text():
        print(text, end='')

Fallback Models

from pydantic_ai.models.fallback import FallbackModel

fallback = FallbackModel(openai_model, anthropic_model)
agent = Agent(fallback)

MCP Integration

from pydantic_ai.mcp import MCPToolset

toolset = MCPToolset(command='python', args=['mcp_server.py'])
agent = Agent('openai:gpt-4o', toolsets=[toolset])

Testing with TestModel

from pydantic_ai.models.test import TestModel

agent = Agent(model=TestModel())
result = agent.run_sync('test')  # Deterministic output

Embeddings

from pydantic_ai import Embedder

embedder = Embedder('openai:text-embedding-3-small')

# Embed search query
result = await embedder.embed_query('What is ML?')

# Embed documents for indexing
docs = ['Doc 1', 'Doc 2', 'Doc 3']
result = await embedder.embed_documents(docs)

See embeddings.md for providers and settings.

xAI Provider

from pydantic_ai import Agent

agent = Agent('xai:grok-4-1-fast-non-reasoning')

See models.md for configuration details.

Exa Neural Search

import os
from pydantic_ai import Agent
from pydantic_ai.common_tools.exa import ExaToolset

api_key = os.getenv('EXA_API_KEY')
toolset = ExaToolset(api_key, num_results=5, include_search=True)
agent = Agent('openai:gpt-4o', toolsets=[toolset])

See tools.md for all Exa tools.

Links

• Documentation
• Releases
• GitHub
• PyPI

Agents Reference

Core interface for interacting with LLMs in Pydantic AI.

Agent Components

Capabilities (v1.71.0+)

Composable, reusable units of agent behavior that bundle tools, lifecycle hooks, instructions, and model settings into a single class:

from pydantic_ai import Agent
from pydantic_ai.capabilities import WebSearch, Thinking, MCP, Hooks

# Provider-adaptive tools — auto-fallback from builtin to local
agent = Agent('openai:gpt-4o', capabilities=[
    WebSearch(),
    Thinking(),
    MCP(url='http://localhost:3000'),
])

Built-in capabilities: WebSearch, WebFetch, MCP, ImageGeneration, Thinking, Hooks.

V2 migration note (v1.96.x)

The latest release line starts deprecating several constructor-level shortcuts on Agent:

• prepare_tools=
• prepare_output_tools=
• event_stream_handler=

Treat those as migration shims and prefer capability-based composition instead:

• PrepareTools(...)
• PrepareOutputTools(...)
• ProcessEventStream(...)

This keeps tool preparation and stream processing aligned with the general capability ordering/composition model instead of hiding wrapper behavior in constructor kwargs.

Capability ordering (v1.80.0+)

When multiple capabilities wrap the same agent flow, ordering is now part of the public design surface.

• CapabilityOrdering supports explicit placement such as innermost, outermost, wraps, wrapped_by, and requires.
• Hooks also gained ordering controls and instance references so wrapper relationships can be expressed directly.
• Use explicit ordering when capability composition changes semantics, for example when you need one capability to observe or transform requests before another wrapper runs.

Hooks Capability

Define hooks using decorators:

from pydantic_ai.capabilities import Hooks

hooks = Hooks()

@hooks.on_model_request
async def log_request(ctx):
    print(f"Sending request to {ctx.model}")

agent = Agent('openai:gpt-4o', capabilities=[hooks])

Hooks can raise ModelRetry for retry control flow. before_model_request / wrap hooks can swap models via ModelRequestContext.

Server-side compaction capabilities (v1.80.0+)

Pydantic AI now exposes provider-backed compaction capabilities for long-running conversations:

• OpenAICompaction
• AnthropicCompaction

OpenAI compaction also gained a stateful mode in the 1.84.x line. Use these capabilities when you want the provider to manage context reduction instead of layering your own summarization logic on every turn.

AgentSpec (v1.71.0+)

Load agents from YAML/JSON files:

from pydantic_ai import Agent

agent = Agent.from_file('agent.yaml')

Supports TemplateStr for templated instructions referencing deps.

Multimodal Input

Support for image, audio, video, and document input.

Image Input

from pydantic_ai import Agent, ImageUrl, BinaryContent

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

# URL
result = agent.run_sync([
    'What is this?',
    ImageUrl(url='https://example.com/image.png'),
])

# Local file
result = agent.run_sync([
    'Describe this image',
    BinaryContent(data=Path('photo.png').read_bytes(), media_type='image/png'),
])

Audio/Video/Document Input

from pydantic_ai import AudioUrl, VideoUrl, DocumentUrl

# Audio
agent.run_sync(['Transcribe this', AudioUrl(url='https://...')])

# Video
agent.run_sync(['Describe', VideoUrl(url='https://...')])

# Document (PDF)
agent.run_sync(['Summarize', DocumentUrl(url='https://...pdf')])

Force Download

If provider can't fetch URL directly:

ImageUrl(url='https://...', force_download=True)

Provider Support

Creating Agents

from pydantic_ai import Agent, RunContext

agent = Agent(
    'openai:gpt-4o',           # model identifier
    deps_type=int,              # dependency type
    output_type=bool,           # structured output type
    description='Triage GitHub issues and draft concise replies',
    system_prompt='Your instructions here',
    model_settings=ModelSettings(temperature=0.5),
    retries=2,                  # default retry count
)

Agent Description (v1.69.0)

Use description= when you want traces and observability spans to carry a stable, human-readable agent label.

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-4o',
    description='Customer-support classifier',
)

When instrumentation is enabled, Pydantic AI attaches this value to the run span as gen_ai.agent.description.

Dependencies

Dependency injection system for passing data/services to prompts, tools, validators.

Defining Dependencies

from dataclasses import dataclass
import httpx

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

agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,  # pass TYPE, not instance
)

Accessing via RunContext

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

@agent.tool
async def fetch_data(ctx: RunContext[MyDeps], query: str) -> str:
    # ctx.deps available in tools
    return await ctx.deps.http_client.get(f'/search?q={query}')

@agent.output_validator
async def validate(ctx: RunContext[MyDeps], output: str) -> str:
    # ctx.deps available in validators
    return output

Passing Dependencies at Runtime

async with httpx.AsyncClient() as client:
    deps = MyDeps(api_key='secret', http_client=client)
    result = await agent.run('Query', deps=deps)

Async vs Sync Dependencies

Both work. Non-async functions run in thread pool via run_in_executor.

# Async (preferred for IO)
@agent.tool
async def async_tool(ctx: RunContext[MyDeps]) -> str:
    return await ctx.deps.http_client.get('/data')

# Sync (also works)
@agent.tool
def sync_tool(ctx: RunContext[MyDeps]) -> str:
    return ctx.deps.sync_client.get('/data')

Overriding Dependencies (Testing)

class TestDeps(MyDeps):
    async def system_prompt_factory(self) -> str:
        return "test prompt"

async def test_app():
    test_deps = TestDeps('test_key', None)
    with agent.override(deps=test_deps):
        result = await application_code('Query')

Run Methods

Basic Run

# Synchronous
result = agent.run_sync('What is 2+2?', deps=my_deps)
print(result.output)

# Async
result = await agent.run('What is 2+2?')
print(result.output)

Streaming

async with agent.run_stream('Tell me a story') as response:
    async for text in response.stream_text():
        print(text, end='')

Stream Events

from pydantic_ai import (
    AgentStreamEvent,
    FunctionToolCallEvent,
    FunctionToolResultEvent,
    PartDeltaEvent,
    TextPartDelta,
)

async for event in agent.run_stream_events('Query'):
    if isinstance(event, PartDeltaEvent):
        if isinstance(event.delta, TextPartDelta):
            print(event.delta.content_delta)
    elif isinstance(event, FunctionToolCallEvent):
        print(f'Tool: {event.part.tool_name}')

Iterate Over Graph

from pydantic_graph import End

async with agent.iter('Query') as agent_run:
    async for node in agent_run:
        print(node)
print(agent_run.result.output)

System Prompts vs Instructions

Static System Prompt

agent = Agent(
    'openai:gpt-4o',
    system_prompt="You are a helpful assistant."
)

Dynamic System Prompt

@agent.system_prompt
def add_context(ctx: RunContext[Deps]) -> str:
    return f"User: {ctx.deps.user_name}"

Instructions

agent = Agent(
    'openai:gpt-4o',
    instructions="Be concise."
)

@agent.instructions
def add_date() -> str:
    return f"Date: {date.today()}"

# Runtime instructions
result = agent.run_sync('Query', instructions="Extra context")

Usage Limits

from pydantic_ai import UsageLimits, UsageLimitExceeded

try:
    result = agent.run_sync(
        'Query',
        usage_limits=UsageLimits(
            response_tokens_limit=100,  # max response tokens
            request_limit=5,            # max model turns
            tool_calls_limit=10,        # max tool executions
        )
    )
except UsageLimitExceeded as e:
    print(f"Limit exceeded: {e}")

Model Settings

Settings merge: model defaults → agent defaults → run overrides

from pydantic_ai import ModelSettings

# Agent-level
agent = Agent(
    'openai:gpt-4o',
    model_settings=ModelSettings(temperature=0.5, max_tokens=500)
)

# Run-level override
result = agent.run_sync(
    'Query',
    model_settings=ModelSettings(temperature=0.0)
)

Run Metadata

from dataclasses import dataclass

@dataclass
class Deps:
    tenant: str

agent = Agent[Deps](
    'openai:gpt-4o',
    deps_type=Deps,
    metadata=lambda ctx: {'tenant': ctx.deps.tenant},
)

result = agent.run_sync(
    'Query',
    deps=Deps(tenant='acme'),
    metadata={'extra': 'data'},  # merged with agent metadata
)
print(result.metadata)  # {'tenant': 'acme', 'extra': 'data'}

Run context now exposes output validation retry count for observability (v1.52.0).

Reflection and Self-Correction

from pydantic_ai import ModelRetry

@agent.tool(retries=3)
def lookup_user(ctx: RunContext[Deps], name: str) -> int:
    user = ctx.deps.db.find(name)
    if not user:
        raise ModelRetry(f"User {name} not found. Try full name.")
    return user.id

Error Handling

from pydantic_ai import UnexpectedModelBehavior, capture_run_messages

with capture_run_messages() as messages:
    try:
        result = agent.run_sync('Query')
    except UnexpectedModelBehavior as e:
        print(f"Error: {e}")
        print(f"Messages: {messages}")

Agent Constructor Parameters

---

Messages and Chat History

Accessing Messages

result = agent.run_sync('Tell me a joke')

# All messages including prior runs
all_msgs = result.all_messages()

# Only messages from current run
new_msgs = result.new_messages()

# JSON serialization
json_bytes = result.all_messages_json()

Continuing Conversations

result1 = agent.run_sync('Tell me a joke')
print(result1.output)

# Continue with message history
result2 = agent.run_sync(
    'Explain?',
    message_history=result1.new_messages()
)
print(result2.output)

Serialize/Deserialize Messages

from pydantic_core import to_jsonable_python
from pydantic_ai import ModelMessagesTypeAdapter

# Serialize
history = result.all_messages()
as_python = to_jsonable_python(history)

# Deserialize
restored = ModelMessagesTypeAdapter.validate_python(as_python)

# Use restored history
result = agent.run_sync('Continue', message_history=restored)

History Processors

Intercept and modify message history before each request:

from pydantic_ai import Agent, ModelMessage, ModelRequest

def keep_recent(messages: list[ModelMessage]) -> list[ModelMessage]:
    """Keep only last 5 messages."""
    return messages[-5:] if len(messages) > 5 else messages

def filter_responses(messages: list[ModelMessage]) -> list[ModelMessage]:
    """Remove ModelResponse, keep only requests."""
    return [m for m in messages if isinstance(m, ModelRequest)]

agent = Agent(
    'openai:gpt-4o',
    history_processors=[filter_responses, keep_recent],
)

Context-Aware Processor

def token_aware(ctx: RunContext[None], messages: list[ModelMessage]) -> list[ModelMessage]:
    if ctx.usage.total_tokens > 1000:
        return messages[-3:]  # Keep recent when high token usage
    return messages

Summarize Old Messages

summarizer = Agent('openai:gpt-4o-mini', instructions='Summarize conversation.')

async def summarize_old(messages: list[ModelMessage]) -> list[ModelMessage]:
    if len(messages) > 10:
        oldest = messages[:10]
        summary = await summarizer.run(message_history=oldest)
        return summary.new_messages() + messages[-1:]
    return messages

Warning: When slicing history, ensure tool calls and returns are paired.

---

Direct Model Requests

Low-level API for making requests without full Agent functionality.

When to Use

• Need direct control over model interactions
• Building custom abstractions
• Don't need tool execution, retrying, structured output

Basic Usage

from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_sync

response = model_request_sync(
    'anthropic:claude-haiku-4-5',
    [ModelRequest.user_text_prompt('What is the capital of France?')]
)

print(response.parts[0].content)  # Paris
print(response.usage)  # RequestUsage(input_tokens=56, output_tokens=7)

Async Request

from pydantic_ai.direct import model_request

response = await model_request(
    'openai:gpt-4o',
    [ModelRequest.user_text_prompt('Hello')]
)

With Tool Definitions

from pydantic import BaseModel
from pydantic_ai import ModelRequest, ToolDefinition
from pydantic_ai.direct import model_request
from pydantic_ai.models import ModelRequestParameters

class Divide(BaseModel):
    """Divide two numbers."""
    numerator: float
    denominator: float

response = await model_request(
    'openai:gpt-4o',
    [ModelRequest.user_text_prompt('What is 123 / 456?')],
    model_request_parameters=ModelRequestParameters(
        function_tools=[
            ToolDefinition(
                name='divide',
                description=Divide.__doc__,
                parameters_json_schema=Divide.model_json_schema(),
            )
        ],
        allow_text_output=True,
    ),
)

Available Functions

---

Multi-Agent Patterns

Five levels of complexity:

1. Single agent — Basic agent workflows 2. Agent delegation — Agent calls another via tools 3. Programmatic hand-off — App code orchestrates agents 4. Graph-based control — State machine controls agents 5. Deep agents — Autonomous with planning, files, code exec

Agent Delegation

Parent agent delegates to child agent via tool:

from pydantic_ai import Agent, RunContext

parent_agent = Agent('openai:gpt-4o', system_prompt='Use joke_factory to get jokes.')
child_agent = Agent('anthropic:claude-sonnet-4-5', output_type=list[str])

@parent_agent.tool
async def joke_factory(ctx: RunContext[None], count: int) -> list[str]:
    result = await child_agent.run(
        f'Generate {count} jokes',
        usage=ctx.usage,  # Share usage tracking
    )
    return result.output

Key points:

• Pass usage=ctx.usage to track combined usage
• Pass deps=ctx.deps if child needs same dependencies
• Different models allowed (cost calculation manual)

Programmatic Hand-off

Sequential agents with app logic between:

from pydantic_ai import Agent, ModelMessage

flight_agent = Agent('openai:gpt-4o', output_type=FlightDetails | Failed)
seat_agent = Agent('openai:gpt-4o', output_type=SeatPreference | Failed)

async def main():
    # First agent
    flight_result = await flight_agent.run('Find flight to Paris')

    if isinstance(flight_result.output, FlightDetails):
        # Second agent (independent)
        seat_result = await seat_agent.run('Window seat please')

Agent with Shared Dependencies

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

parent = Agent('openai:gpt-4o', deps_type=SharedDeps)
child = Agent('anthropic:claude-sonnet-4-5', deps_type=SharedDeps)

@parent.tool
async def delegate(ctx: RunContext[SharedDeps], task: str) -> str:
    result = await child.run(
        task,
        deps=ctx.deps,   # Share dependencies
        usage=ctx.usage, # Share usage
    )
    return result.output

Deep Agent Capabilities

---

Thinking (Reasoning)

Enable step-by-step reasoning before final answer.

Provider Configuration

OpenAI Responses Example

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIResponsesModel, OpenAIResponsesModelSettings

model = OpenAIResponsesModel('gpt-5')
settings = OpenAIResponsesModelSettings(
    openai_reasoning_effort='low',
    openai_reasoning_summary='detailed',
)
agent = Agent(model, model_settings=settings)

Anthropic Example

from pydantic_ai import Agent
from pydantic_ai.models.anthropic import AnthropicModel, AnthropicModelSettings

model = AnthropicModel('claude-sonnet-4-0')
settings = AnthropicModelSettings(
    anthropic_thinking={'type': 'enabled', 'budget_tokens': 1024},
)
agent = Agent(model, model_settings=settings)

Google Example

from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel, GoogleModelSettings

model = GoogleModel('gemini-2.5-pro')
settings = GoogleModelSettings(google_thinking_config={'include_thoughts': True})
agent = Agent(model, model_settings=settings)

Bedrock Examples

from pydantic_ai import Agent
from pydantic_ai.models.bedrock import BedrockConverseModel, BedrockModelSettings

# Anthropic on Bedrock
model = BedrockConverseModel('us.anthropic.claude-sonnet-4-5-20250929-v1:0')
settings = BedrockModelSettings(
    bedrock_additional_model_requests_fields={
        'thinking': {'type': 'enabled', 'budget_tokens': 1024}
    }
)

# OpenAI on Bedrock
model = BedrockConverseModel('openai.gpt-oss-120b-1:0')
settings = BedrockModelSettings(
    bedrock_additional_model_requests_fields={'reasoning_effort': 'low'}
)

# Deepseek on Bedrock (always enabled)
model = BedrockConverseModel('us.deepseek.r1-v1:0')
agent = Agent(model=model)  # No settings needed

Thinking Output

Thinking parts are returned as ThinkingPart objects in the message history:

• OpenAI Chat: <think> tags converted to ThinkingPart
• OpenAI Responses: Native thinking parts
• Groq parsed: Structured thinking parts
• Local models: <think> tags auto-converted

---

Troubleshooting

Jupyter Notebook: Event Loop Error

# Error: RuntimeError: This event loop is already running
# Fix: Install and apply nest-asyncio BEFORE any agent runs
import nest_asyncio
nest_asyncio.apply()

Note: Works in Google Colab and Marimo too.

API Key Missing

UserError: API key must be provided or set in the [MODEL]_API_KEY environment variable

Solutions:

1. Set environment variable: export OPENAI_API_KEY=sk-... 2. Pass directly: OpenAIModel('gpt-4o', api_key='sk-...')

Monitoring HTTPX Requests

Use custom httpx clients for request/response inspection:

import httpx
import logfire

# Install logfire httpx integration for monitoring
logfire.instrument_httpx()

client = httpx.AsyncClient()
model = OpenAIModel('gpt-4o', http_client=client)

Community Support

• Slack: Join #pydantic-ai in Pydantic Slack
• GitHub Issues: https://github.com/pydantic/pydantic-ai/issues
• Logfire Pro: Private collaboration channel available

Embeddings Reference

Generate vector embeddings for semantic search, RAG, and similarity detection.

Quick Start

from pydantic_ai import Embedder

embedder = Embedder('openai:text-embedding-3-small')

async def main():
    # Embed a search query
    result = await embedder.embed_query('What is machine learning?')
    print(f'Dimensions: {len(result.embeddings[0])}')

    # Embed multiple documents
    docs = ['ML is AI subset.', 'Deep learning uses neural nets.']
    result = await embedder.embed_documents(docs)
    print(f'Embedded {len(result.embeddings)} documents')

Providers

OpenAI

pip install "pydantic-ai-slim[openai]"
export OPENAI_API_KEY='your-key'

from pydantic_ai import Embedder

embedder = Embedder('openai:text-embedding-3-small')
result = await embedder.embed_query('Hello world')
# 1536 dimensions

Google

pip install "pydantic-ai-slim[google]"
export GOOGLE_API_KEY='your-key'

from pydantic_ai import Embedder

# Gemini API
embedder = Embedder('google:gemini-embedding-001')

# Vertex AI
embedder = Embedder('google-cloud:gemini-embedding-001')

Cohere

pip install "pydantic-ai-slim[cohere]"
export CO_API_KEY='your-key'

from pydantic_ai import Embedder

embedder = Embedder('cohere:embed-v4.0')

VoyageAI

Optimized for retrieval with specialized models for code, finance, legal.

pip install "pydantic-ai-slim[voyageai]"
export VOYAGE_API_KEY='your-key'

from pydantic_ai import Embedder

embedder = Embedder('voyageai:voyage-3.5')

Bedrock (AWS)

AWS Bedrock embeddings for Nova, Cohere, and Titan models.

pip install "pydantic-ai-slim[bedrock]"
# Uses AWS credentials from environment or ~/.aws/credentials

from pydantic_ai import Embedder

# Amazon Titan
embedder = Embedder('bedrock:amazon.titan-embed-text-v1')

# Cohere via Bedrock
embedder = Embedder('bedrock:cohere.embed-english-v3')

Sentence Transformers (Local)

Run embeddings locally without API calls.

pip install "pydantic-ai-slim[sentence-transformers]"

from pydantic_ai import Embedder

# Downloaded from Hugging Face on first use
embedder = Embedder('sentence-transformers:all-MiniLM-L6-v2')

Embedding Result

result = await embedder.embed_query('Hello world')

# Access embeddings
embedding = result.embeddings[0]  # By index
embedding = result[0]  # Shorthand
embedding = result['Hello world']  # By input text

# Usage info
print(result.usage.input_tokens)

# Cost calculation (requires genai-prices)
cost = result.cost()
print(f'Cost: ${cost.total_price:.6f}')

Settings

from pydantic_ai import Embedder
from pydantic_ai.embeddings import EmbeddingSettings

# Reduce dimensions (OpenAI, Google, Cohere, VoyageAI)
embedder = Embedder(
    'openai:text-embedding-3-small',
    settings=EmbeddingSettings(dimensions=256),
)

# Per-call override
result = await embedder.embed_query(
    'Hello world',
    settings=EmbeddingSettings(dimensions=512),
)

Provider-Specific Settings

Google:

from pydantic_ai.embeddings.google import GoogleEmbeddingSettings

embedder = Embedder(
    'google:gemini-embedding-001',
    settings=GoogleEmbeddingSettings(
        dimensions=768,
        google_task_type='SEMANTIC_SIMILARITY',
    ),
)

Cohere:

from pydantic_ai.embeddings.cohere import CohereEmbeddingSettings

embedder = Embedder(
    'cohere:embed-v4.0',
    settings=CohereEmbeddingSettings(
        dimensions=512,
        cohere_truncate='END',
        cohere_max_tokens=256,
    ),
)

VoyageAI:

from pydantic_ai.embeddings.voyageai import VoyageAIEmbeddingSettings

embedder = Embedder(
    'voyageai:voyage-3.5',
    settings=VoyageAIEmbeddingSettings(
        dimensions=512,
        voyageai_input_type='document',
    ),
)

Sentence Transformers:

from pydantic_ai.embeddings.sentence_transformers import (
    SentenceTransformersEmbeddingSettings,
)

embedder = Embedder(
    'sentence-transformers:all-MiniLM-L6-v2',
    settings=SentenceTransformersEmbeddingSettings(
        sentence_transformers_device='cuda',
        sentence_transformers_normalize_embeddings=True,
    ),
)

Token Counting

embedder = Embedder('openai:text-embedding-3-small')

# Count tokens
token_count = await embedder.count_tokens('Hello world, this is a test.')
print(f'Tokens: {token_count}')

# Check max input
max_tokens = await embedder.max_input_tokens()
print(f'Max: {max_tokens}')

Query vs Documents

Use appropriate method based on input type:

• embed_query() — for search queries
• embed_documents() — for content being indexed

Some models optimize differently for queries vs documents.

Testing

from pydantic_ai import Embedder
from pydantic_ai.embeddings import TestEmbeddingModel

async def test_rag():
    embedder = Embedder('openai:text-embedding-3-small')
    test_model = TestEmbeddingModel()

    with embedder.override(model=test_model):
        result = await embedder.embed_query('test')
        # Returns deterministic [1.0] * 8
        assert result.embeddings[0] == [1.0] * 8

Instrumentation

import logfire
from pydantic_ai import Embedder

logfire.configure()

# Instrument specific embedder
embedder = Embedder('openai:text-embedding-3-small', instrument=True)

# Or instrument all globally
Embedder.instrument_all()

OpenAI-Compatible Providers

from pydantic_ai import Embedder
from pydantic_ai.embeddings.openai import OpenAIEmbeddingModel
from pydantic_ai.providers.openai import OpenAIProvider

# Azure OpenAI
from openai import AsyncAzureOpenAI

azure_client = AsyncAzureOpenAI(
    azure_endpoint='https://your-resource.openai.azure.com',
    api_version='2024-02-01',
    api_key='your-azure-key',
)
model = OpenAIEmbeddingModel(
    'text-embedding-3-small',
    provider=OpenAIProvider(openai_client=azure_client),
)
embedder = Embedder(model)

# Any OpenAI-compatible API
model = OpenAIEmbeddingModel(
    'your-model-name',
    provider=OpenAIProvider(
        base_url='https://your-provider.com/v1',
        api_key='your-api-key',
    ),
)
embedder = Embedder(model)

# Shorthand for known providers
embedder = Embedder('azure:text-embedding-3-small')
embedder = Embedder('ollama:nomic-embed-text')

Custom Embedding Model

from collections.abc import Sequence
from pydantic_ai.embeddings import EmbeddingModel, EmbeddingResult, EmbeddingSettings
from pydantic_ai.embeddings.result import EmbedInputType

class MyEmbeddingModel(EmbeddingModel):
    @property
    def model_name(self) -> str:
        return 'my-model'

    @property
    def system(self) -> str:
        return 'my-provider'

    async def embed(
        self,
        inputs: str | Sequence[str],
        *,
        input_type: EmbedInputType,
        settings: EmbeddingSettings | None = None,
    ) -> EmbeddingResult:
        inputs, settings = self.prepare_embed(inputs, settings)

        # Call your API here
        embeddings = [[0.1, 0.2, 0.3] for _ in inputs]

        return EmbeddingResult(
            embeddings=embeddings,
            inputs=inputs,
            input_type=input_type,
            model_name=self.model_name,
            provider_name=self.system,
        )

Pydantic Evals Reference

Evaluation framework for testing AI systems.

Installation

pip install pydantic-evals

# With Logfire integration
pip install 'pydantic-evals[logfire]'

Data Model

Dataset (1) ──── (Many) Case
│                │
└── (Many) Experiment ──┴── (Many) Case results
    │
    └── (1) Task
    │
    └── (Many) Evaluator

Basic Usage

Define Cases

from pydantic_evals import Case, Dataset

case = Case(
    name='capital_france',
    inputs='What is the capital of France?',
    expected_output='Paris',
    metadata={'difficulty': 'easy'},
)

dataset = Dataset(cases=[case])

Define Task

from pydantic_ai import Agent

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

async def my_task(question: str) -> str:
    result = await agent.run(question)
    return result.output

Run Evaluation

report = dataset.evaluate_sync(my_task)
report.print(include_input=True, include_output=True)

Built-in Evaluators

Contains supports pydantic.BaseModel outputs (v1.59.0).

LinePlot analysis type and additional evaluators were expanded in v1.62.0 for richer experiment reporting.

from pydantic_evals.evaluators import IsInstance, ExactMatch

dataset = Dataset(
    cases=[case],
    evaluators=[
        IsInstance(type_name='str'),
        ExactMatch(),
    ],
)

Custom Evaluators

from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext

@dataclass
class MyEvaluator(Evaluator[str, str]):
    async def evaluate(self, ctx: EvaluatorContext[str, str]) -> float:
        output = ctx.output
        expected = ctx.expected_output

        if output == expected:
            return 1.0
        elif expected.lower() in output.lower():
            return 0.8
        return 0.0

LLM Judge

Use LLM to evaluate subjective qualities:

from pydantic_evals.evaluators import LLMJudge

judge = LLMJudge(
    model='openai:gpt-4o',
    criteria='Is the response accurate, helpful, and well-formatted?',
)

dataset = Dataset(cases=[case], evaluators=[judge])

Case-Specific Evaluators

case = Case(
    name='math_case',
    inputs='What is 2+2?',
    expected_output='4',
    evaluators=[ExactMatch()],  # Only for this case
)

Dataset from YAML

# dataset.yaml
cases:
  - name: capital_france
    inputs: "What is the capital of France?"
    expected_output: "Paris"
  - name: capital_japan
    inputs: "What is the capital of Japan?"
    expected_output: "Tokyo"

from pydantic_evals import Dataset

dataset = Dataset.from_yaml('dataset.yaml')

Report Output

report = dataset.evaluate_sync(my_task)

# Print to console
report.print()

# Access results
for case_result in report.case_results:
    print(f"{case_result.name}: {case_result.scores}")

# Export
report.to_json('results.json')

Logfire Integration

import logfire

logfire.configure()

# Results auto-appear in Logfire dashboard
report = dataset.evaluate_sync(my_task)

Span-Based Evaluation

Evaluate internal agent behavior (tool calls, execution flow):

from pydantic_evals.evaluators import SpanEvaluator

class ToolCallEvaluator(SpanEvaluator):
    async def evaluate_spans(self, spans: list[Span]) -> float:
        tool_calls = [s for s in spans if s.name.startswith('tool:')]
        return 1.0 if len(tool_calls) <= 3 else 0.5

Concurrency

# Control parallel execution
report = dataset.evaluate_sync(
    my_task,
    max_concurrency=5,
)

Retry Strategies

from pydantic_evals import RetryConfig

report = dataset.evaluate_sync(
    my_task,
    retry_config=RetryConfig(
        max_retries=3,
        backoff_multiplier=2.0,
    ),
)

Best Practices

1. Start simple — Begin with exact match, add complex evaluators as needed 2. Use metadata — Tag cases with difficulty, category for analysis 3. Combine evaluators — Use deterministic + LLM-based together 4. Version datasets — Track dataset changes over time 5. Integrate Logfire — Visualize results and compare runs

Pydantic Graph Reference

pydantic-graph — async graph and state machine library for Python where nodes and edges are defined using type hints.

Installation

pip install pydantic-graph
# Or included with pydantic-ai
pip install pydantic-ai

Note: pydantic-graph is required dependency of pydantic-ai, optional for pydantic-ai-slim.

When to Use Graphs

Graphs are powerful but NOT for every job:

• Consider simpler multi-agent approaches first
• If not confident graph-based is needed — it's probably unnecessary
• Graphs are for advanced users with heavy generics/type hints usage

Core Components

GraphRunContext

Context for graph run, holds state and dependencies:

from pydantic_graph import GraphRunContext

# Generic in StateT
async def run(self, ctx: GraphRunContext[MyState]) -> NextNode:
    ctx.state.counter += 1  # Access/mutate state
    return NextNode()

End

Return value indicating graph run should end:

from pydantic_graph import End

# Generic in RunEndT (return type)
return End(result_value)

BaseNode

Subclass to define nodes. Nodes are dataclasses with:

1. Fields for parameters 2. Business logic in run method 3. Return annotations for outgoing edges

from dataclasses import dataclass
from pydantic_graph import BaseNode, End, GraphRunContext

@dataclass
class MyNode(BaseNode[StateT, DepsT, RunEndT]):
    """
    Generic parameters:
    - StateT: state type (default None)
    - DepsT: dependencies type (default None)
    - RunEndT: return type if returns End (default Never)
    """
    foo: int  # Node parameter

    async def run(
        self,
        ctx: GraphRunContext[StateT, DepsT],
    ) -> NextNode | End[RunEndT]:  # Return type = outgoing edges
        if self.foo % 5 == 0:
            return End(self.foo)
        return NextNode()

Graph

Execution graph made of node classes:

from pydantic_graph import Graph

# Generic in StateT, DepsT, RunEndT
my_graph = Graph(nodes=[NodeA, NodeB, NodeC])

# Run synchronously
result = my_graph.run_sync(StartNode(value=4))

# Run asynchronously
result = await my_graph.run(StartNode(), state=state, deps=deps)

Complete Example

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)

fives_graph = Graph(nodes=[DivisibleBy5, Increment])
result = fives_graph.run_sync(DivisibleBy5(4))
print(result.output)  # 5

Stateful Graphs

State = object passed along and mutated by nodes:

from dataclasses import dataclass
from pydantic_graph import BaseNode, End, Graph, GraphRunContext

@dataclass
class MachineState:
    user_balance: float = 0.0
    product: str | None = None

@dataclass
class InsertCoin(BaseNode[MachineState]):
    async def run(self, ctx: GraphRunContext[MachineState]) -> CoinsInserted:
        amount = float(input('Insert coins: '))
        return CoinsInserted(amount)

@dataclass
class CoinsInserted(BaseNode[MachineState]):
    amount: float

    async def run(self, ctx: GraphRunContext[MachineState]) -> SelectProduct | Purchase:
        ctx.state.user_balance += self.amount  # Mutate state
        if ctx.state.product is not None:
            return Purchase(ctx.state.product)
        return SelectProduct()

# ... more nodes ...

async def main():
    state = MachineState()
    await vending_machine_graph.run(InsertCoin(), state=state)

Iterating Over Graph

Using Graph.iter with async for

async with my_graph.iter(StartNode(), state=state) as run:
    async for node in run:
        print('Node:', node)
        # Node: StartNode()
        # Node: NextNode()
        # Node: End(data=result)
print('Final:', run.result.output)

Manual iteration with next()

async with my_graph.iter(StartNode(), state=state) as run:
    node = run.next_node
    while not isinstance(node, End):
        print('Node:', node)
        if some_condition:
            break  # Early exit possible
        node = await run.next(node)

State Persistence

Allows interruption and resumption of graph runs.

Built-in Implementations

Using FileStatePersistence

from pathlib import Path
from pydantic_graph import End
from pydantic_graph.persistence.file import FileStatePersistence

async def main():
    persistence = FileStatePersistence(Path('graph_state.json'))

    # Initialize graph
    await my_graph.initialize(StartNode(), state=state, persistence=persistence)

    # Run from persistence (can be in separate process)
    async with my_graph.iter_from_persistence(persistence) as run:
        node_or_end = await run.next()
        if isinstance(node_or_end, End):
            print('Complete:', node_or_end.data)

Human in the Loop Pattern

# Run 1: Generate question
persistence = FileStatePersistence(Path('qa.json'))
if snapshot := await persistence.load_next():
    state = snapshot.state
    node = EvaluateAnswer(user_answer)
else:
    state = QuestionState()
    node = AskQuestion()

async with qa_graph.iter(node, state=state, persistence=persistence) as run:
    while True:
        node = await run.next()
        if isinstance(node, End):
            print('Correct!')
            break
        elif isinstance(node, WaitForAnswer):
            print(node.question)  # Wait for user input
            break

Dependency Injection

from dataclasses import dataclass
from concurrent.futures import ProcessPoolExecutor
from pydantic_graph import BaseNode, GraphRunContext

@dataclass
class GraphDeps:
    executor: ProcessPoolExecutor

@dataclass
class MyNode(BaseNode[None, GraphDeps, int]):
    async def run(self, ctx: GraphRunContext[None, GraphDeps]) -> NextNode:
        # Use dependency
        result = await loop.run_in_executor(ctx.deps.executor, self.compute)
        return NextNode(result)

# Run with deps
with ProcessPoolExecutor() as executor:
    deps = GraphDeps(executor)
    result = await my_graph.run(StartNode(), deps=deps)

Mermaid Diagrams

Generate Diagram Code

code = my_graph.mermaid_code(start_node=StartNode)

Generate Image

# Get image bytes
image = my_graph.mermaid_image(start_node=StartNode)

# Save to file
my_graph.mermaid_save(start_node=StartNode, path='diagram.png')

Jupyter Display

from IPython.display import Image, display
display(Image(my_graph.mermaid_image(start_node=StartNode)))

Diagram Customization

from typing import Annotated
from pydantic_graph import Edge, BaseNode

@dataclass
class Ask(BaseNode[State]):
    """Generate question using AI."""  # Note from docstring
    docstring_notes = True  # Enable docstring as note

    async def run(self, ctx: GraphRunContext[State]) -> Annotated[Answer, Edge(label='Ask the question')]:
        return Answer(question)

# Direction: 'TB' (top-bottom), 'LR' (left-right), 'RL', 'BT'
code = my_graph.mermaid_code(
    start_node=StartNode,
    direction='LR',
    highlighted_nodes=[CurrentNode]  # Highlight specific nodes
)

GenAI Integration Example

format_as_xml now handles non-primitive BaseModel values (v1.52.0).

from pydantic_ai import Agent, format_as_xml
from pydantic_graph import BaseNode, End, Graph, GraphRunContext

email_writer = Agent('openai:gpt-4o', output_type=Email)
feedback_agent = Agent('openai:gpt-4o', output_type=FeedbackResult)

@dataclass
class State:
    user: User
    messages: list[ModelMessage] = field(default_factory=list)

@dataclass
class WriteEmail(BaseNode[State]):
    feedback: str | None = None

    async def run(self, ctx: GraphRunContext[State]) -> ReviewEmail:
        result = await email_writer.run(
            f'Write email for: {format_as_xml(ctx.state.user)}',
            message_history=ctx.state.messages,
        )
        ctx.state.messages += result.new_messages()
        return ReviewEmail(result.output)

@dataclass
class ReviewEmail(BaseNode[State, None, Email]):
    email: Email

    async def run(self, ctx: GraphRunContext[State]) -> WriteEmail | End[Email]:
        result = await feedback_agent.run(format_as_xml(self.email))
        if result.output.needs_revision:
            return WriteEmail(feedback=result.output.feedback)
        return End(self.email)

email_graph = Graph(nodes=[WriteEmail, ReviewEmail])
result = await email_graph.run(WriteEmail(), state=State(user=user))

Key Patterns

Node Return Types Define Edges

# Single edge
async def run(self, ctx) -> NextNode:
    return NextNode()

# Multiple possible edges (union)
async def run(self, ctx) -> NodeA | NodeB | End[int]:
    if condition_a:
        return NodeA()
    elif condition_b:
        return NodeB()
    return End(42)

State vs Dependencies

Prohibitions

• ❌ Using graphs when simpler multi-agent approach works
• ❌ Forgetting generic parameters for End-returning nodes
• ❌ Returning node types not in return annotation
• ❌ Using state without passing to graph.run()

Installation

Requires Python 3.10+ (Python 3.14 supported in v1.61+)

# Full install (all model dependencies)
pip install pydantic-ai

# With examples
pip install "pydantic-ai[examples]"

Slim Install

Use pydantic-ai-slim for minimal dependencies:

# Single model
pip install "pydantic-ai-slim[openai]"

# Multiple models
pip install "pydantic-ai-slim[openai,anthropic,logfire]"

Optional Groups

Integrations Reference

Pydantic AI integrates with MCP, Logfire, A2A, and durable execution platforms.

Vercel AI SDK Compatibility (v1.52.0)

Compatibility with Vercel AI SDK v5 is restored by passing the SDK version parameter in requests.

Vercel tool approvals (v1.62.0)

Vercel AI adapter integrates tool approval flows, enabling safer gated execution patterns in UI-driven chats.

---

MCP (Model Context Protocol)

Connect agents to external tools and services via standardized protocol.

Recent migration note (1.97.0+): prefer MCPToolset for new client integrations. FastMCPToolset and the older MCPServer* wrappers are now legacy migration surfaces.

Patch note (1.103.0+): maintained McpServer integrations can call list_prompts and get_prompt. Use this for legacy MCP server wrappers that expose prompt catalogs, but keep new client code on MCPToolset unless a migration constraint requires direct McpServer access.

Installation

pip install "pydantic-ai-slim[mcp]"

Legacy MCP server wrappers

Use these only when you are maintaining older code. For new code, start with MCPToolset.

Recommended client API (MCPToolset)

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset

remote = MCPToolset(url='http://localhost:8000/mcp')
local = MCPToolset(command='python', args=['mcp_server.py'], timeout=10)

agent = Agent('openai:gpt-4o', toolsets=[remote, local])

HTTP Client (Streamable)

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

server = MCPServerStreamableHTTP('http://localhost:8000/mcp')
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    async with agent:  # Opens MCP connection
        result = await agent.run('What is 7 + 5?')

Stdio Client (Subprocess)

from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio(
    'python',
    args=['mcp_server.py'],
    timeout=10,
)
agent = Agent('openai:gpt-4o', toolsets=[server])

Load from Config

{
  "mcpServers": {
    "calculator": {
      "url": "http://localhost:8000/mcp"
    },
    "weather": {
      "command": "python",
      "args": ["weather_server.py"]
    }
  }
}

from pydantic_ai.mcp import load_mcp_servers

servers = load_mcp_servers('mcp_config.json')
agent = Agent('openai:gpt-4o', toolsets=servers)

Tool Prefixes (Avoid Conflicts)

weather = MCPToolset(url='http://localhost:3001/mcp', tool_prefix='weather')
calc = MCPToolset(url='http://localhost:3002/mcp', tool_prefix='calc')

# Tools: weather_get_data, calc_get_data
agent = Agent('openai:gpt-4o', toolsets=[weather, calc])

MCP Resources

async with server:
    resources = await server.list_resources()
    content = await server.read_resource('resource://data.txt')

MCP Sampling

Allow MCP server to make LLM calls through client:

server = MCPServerStdio('python', args=['server.py'])
agent = Agent('openai:gpt-4o', toolsets=[server])

agent.set_mcp_sampling_model()  # Enable sampling

---

Building MCP Servers

With FastMCP + Pydantic AI Agent

from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent

server = FastMCP('My AI Server')
agent = Agent('anthropic:claude-haiku-4-5', system_prompt='Reply in rhyme')

@server.tool()
async def poet(theme: str) -> str:
    """Generate a poem about the theme."""
    result = await agent.run(f'Write a poem about {theme}')
    return result.output

if __name__ == '__main__':
    server.run()  # stdio transport by default

With MCP Sampling

Server uses client's LLM via MCPSamplingModel:

from mcp.server.fastmcp import Context, FastMCP
from pydantic_ai import Agent
from pydantic_ai.models.mcp_sampling import MCPSamplingModel

server = FastMCP('Sampling Server')
agent = Agent(system_prompt='Reply in rhyme')

@server.tool()
async def poet(ctx: Context, theme: str) -> str:
    """Generate poem using client's LLM."""
    result = await agent.run(
        f'Write poem about {theme}',
        model=MCPSamplingModel(session=ctx.session),
    )
    return result.output

---

Background MCP work (1.101.0+)

• MCP integrations can now run background tasks. Use this when a server needs to continue work after the main model turn has already returned.
• Pair background work with explicit lifecycle/logging so queued tasks are observable instead of silently detached.

---

Logfire Integration

Built-in observability for agent runs.

OTel alignment (v1.60.0)

Instrumentation version 4 aligns with OTel GenAI semantic conventions, including multimodal request traces.

Setup

pip install pydantic-ai  # Logfire included
logfire configure

import logfire

logfire.configure()
logfire.instrument_pydantic_ai()

View in Dashboard

• Agent runs with timing
• Tool calls and results
• Token usage
• Model responses

---

Agent-to-Agent (A2A)

Protocol for agents to communicate with each other.

pip install "pydantic-ai-slim[a2a]"

from pydantic_ai.a2a import A2AServer, A2AClient

# Server side
server = A2AServer(agent)

# Client side
client = A2AClient('http://agent-server.com')
result = await client.run('Query for remote agent')

---

Durable Execution

Persist agent state across failures/restarts.

Temporal

pip install "pydantic-ai-slim[temporal]"

DBOS

pip install "pydantic-ai-slim[dbos]"

Prefect

pip install "pydantic-ai-slim[prefect]"

---

Temporal (Durable Execution)

Overview

Temporal provides durable execution via workflows (deterministic) and activities (non-deterministic I/O).

            +---------------------+
            |   Temporal Server   |      (Stores workflow state,
            +---------------------+       schedules activities)
                     ^
                     |
+------------------------------------------------------+
|                      Worker                          |
|   +----------------------------------------------+   |
|   |              Workflow Code                   |   |
|   |       (Agent Run Loop - deterministic)       |   |
|   +----------------------------------------------+   |
|          |          |                |               |
|   +-----------+ +------------+ +-------------+       |
|   | Activity  | | Activity   | |  Activity   |       |
|   | (Tool)    | | (MCP Tool) | | (Model API) |       |
|   +-----------+ +------------+ +-------------+       |
+------------------------------------------------------+

Installation

pip install "pydantic-ai[temporal]"
# Start local Temporal server
brew install temporal
temporal server start-dev

TemporalAgent

Wrap any agent for durable execution:

import uuid
from temporalio import workflow
from temporalio.client import Client
from temporalio.worker import Worker
from pydantic_ai import Agent
from pydantic_ai.durable_exec.temporal import (
    PydanticAIPlugin,
    PydanticAIWorkflow,
    TemporalAgent,
)

# Define agent (name required for Temporal!)
agent = Agent(
    'openai:gpt-4o',
    instructions="You're an expert in geography.",
    name='geography',  # Required for stable activity names
)

# Wrap for durable execution
temporal_agent = TemporalAgent(agent)

# Define workflow
@workflow.defn
class GeographyWorkflow(PydanticAIWorkflow):
    __pydantic_ai_agents__ = [temporal_agent]

    @workflow.run
    async def run(self, prompt: str) -> str:
        result = await temporal_agent.run(prompt)
        return result.output

# Run workflow
async def main():
    client = await Client.connect(
        'localhost:7233',
        plugins=[PydanticAIPlugin()],
    )

    async with Worker(
        client,
        task_queue='geography',
        workflows=[GeographyWorkflow],
    ):
        output = await client.execute_workflow(
            GeographyWorkflow.run,
            args=['What is the capital of Mexico?'],
            id=f'geography-{uuid.uuid4()}',
            task_queue='geography',
        )
        print(output)  # Mexico City

Key Requirements

Model Selection at Runtime

from pydantic_ai.models.openai import OpenAIResponsesModel
from pydantic_ai.models.anthropic import AnthropicModel

# Pre-register models for TemporalAgent
default_model = OpenAIResponsesModel('gpt-4o')
fast_model = AnthropicModel('claude-sonnet-4-5')

temporal_agent = TemporalAgent(
    agent,
    models={
        'fast': fast_model,
        'reasoning': reasoning_model,
    },
    provider_factory=my_provider_factory,  # Optional for dynamic config
)

# In workflow: select by name or instance
result = await temporal_agent.run(prompt, model='fast')
result = await temporal_agent.run(prompt, model=fast_model)
result = await temporal_agent.run(prompt, model='openai:gpt-4.1-mini')  # model string

Activity Configuration

from temporalio.common import RetryPolicy
from temporalio.workflow import ActivityConfig

temporal_agent = TemporalAgent(
    agent,
    activity_config=ActivityConfig(start_to_close_timeout=120),  # Base config
    model_activity_config=ActivityConfig(start_to_close_timeout=300),  # Model requests
    toolset_activity_config={'my_toolset': ActivityConfig(...)},  # Per toolset
    tool_activity_config={
        ('my_toolset', 'fast_tool'): False,  # Disable activity for sync tools
    },
)

RunContext in Activities

Limited fields available in activities:

• ✅ deps, run_id, metadata, retries, tool_call_id, tool_name
• ✅ tool_call_approved, retry, max_retries, run_step, usage, partial_output
• ❌ model, prompt, messages, tracer — raise error

Custom serialization:

from pydantic_ai.durable_exec.temporal import TemporalRunContext

class MyRunContext(TemporalRunContext):
    @classmethod
    def serialize_run_context(cls, ctx): ...
    @classmethod
    def deserialize_run_context(cls, data): ...

temporal_agent = TemporalAgent(agent, run_context_type=MyRunContext)

Logfire Integration

from pydantic_ai.durable_exec.temporal import LogfirePlugin, PydanticAIPlugin

client = await Client.connect(
    'localhost:7233',
    plugins=[PydanticAIPlugin(), LogfirePlugin()],
)

Prohibitions

• ❌ Streaming (run_stream(), run_stream_events(), iter())
• ❌ HTTP retries (disable in provider: max_retries=0)
• ❌ Changing agent name/toolset id after deployment
• ❌ Non-serializable dependencies
• ❌ Non-async tools outside activities

---

Building MCP Servers

With FastMCP

from mcp.server.fastmcp import FastMCP

app = FastMCP('My Server')

@app.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

if __name__ == '__main__':
    app.run(transport='streamable-http')  # or 'stdio', 'sse'

Expose Resources

@app.resource('resource://data.txt', mime_type='text/plain')
async def get_data() -> str:
    return "Resource content"

With Pydantic AI Agent

Use agents inside MCP servers:

from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent

app = FastMCP('AI Server')
agent = Agent('openai:gpt-4o')

@app.tool()
async def ask_ai(question: str) -> str:
    """Ask AI a question."""
    result = await agent.run(question)
    return result.output

Models Reference

Pydantic AI is model-agnostic with 30+ providers.

Built-in Providers

Recent model-surface updates (1.80.0+)

• Anthropic support now includes Claude Opus 4.8.
• Pydantic AI added a native OllamaModel subclass and corrected Ollama capability flags, which is especially relevant for structured output on Ollama Cloud.
• OpenAI compaction gained a stateful mode in the 1.84.x line.

Provider reliability updates (1.103.0 -> 1.104.0)

• OpenRouterModel supports anthropic_eager_input_streaming; enable it only when the target route expects Anthropic-style eager streaming semantics.
• Hybrid OpenRouter/xAI/Bedrock routes now forward thinking=False consistently, matching direct-route behavior instead of silently dropping the setting.
• Bedrock maps malformed_model_output and malformed_tool_use to FinishReason.error, recognizes type='adaptive' in thinking detection, and preserves cache behavior for single-tool tool_choice requests.

Model-surface updates (1.105.0 -> 1.107.0)

• Anthropic support adds Claude Fable 5 and Claude Mythos 5 (1.107.0).
• xAI adds Grok 4.3 reasoning_effort support with updated model names (1.105.0), plus seed mapping and api_host/timeout on XaiProvider (1.106.0).
• OpenRouterModel implements CachePoint and prompt caching (1.107.0).
• known_model_names() enumerates the KnownModelName members at runtime (1.107.0).
• Anthropic Bedrock stream handling tolerates message=None start events, and AnthropicModel.count_tokens is corrected when native tools are present (1.107.0).

xAI (Grok)

Native xAI SDK provider (replaces deprecated GrokProvider):

pip install "pydantic-ai-slim[xai]"
export XAI_API_KEY='your-api-key'

from pydantic_ai import Agent

# Via model string
agent = Agent('xai:grok-4-1-fast-non-reasoning')

# Explicit model
from pydantic_ai.models.xai import XaiModel
model = XaiModel('grok-4-1-fast-non-reasoning')
agent = Agent(model)

# Custom provider
from pydantic_ai.providers.xai import XaiProvider
provider = XaiProvider(api_key='your-api-key')
model = XaiModel('grok-4-1-fast-non-reasoning', provider=provider)

# With xai_sdk client
from xai_sdk import AsyncClient
xai_client = AsyncClient(api_key='your-api-key')
provider = XaiProvider(xai_client=xai_client)
model = XaiModel('grok-4-1-fast-non-reasoning', provider=provider)

OpenAI-Compatible Providers

Use OpenAIChatModel with custom provider:

• Azure AI, DeepSeek, Fireworks AI
• GitHub Models, Heroku
• LiteLLM, Ollama, Perplexity
• Together AI, Vercel AI Gateway

OpenRouter multimodal input (v1.60.0)

OpenRouterModel supports video_url inputs for multimodal requests.

Provider reliability updates (v1.62.0)

• Groq: retries now handle tool_use_failed responses even when tool name/args are missing.
• Google/OpenAI: refusal/content-filter handling is hardened for prompt-feedback/refusal flows.

New model IDs (v1.65.0)

• Google: adds support for gemini-3.1-flash-lite-preview.

OpenAI Data Retention (v1.52.0)

OpenAI models support an openai_store setting to control data retention.

OpenAI Reasoning Content (v1.52.0)

OpenAIChatModel preserves reasoning content in the provider field it was received in; keep it if you need reasoning traces.

Anthropic Settings (v1.56.0)

New Anthropic settings extend AnthropicModelSettings:

• anthropic_effort: 'low' | 'medium' | 'high' | 'max'
• anthropic_thinking: supports type='adaptive' (model-dependent)
• anthropic_betas: list of beta feature flags to enable

from pydantic_ai import ModelSettings
from pydantic_ai.models.anthropic import AnthropicModel

model = AnthropicModel(
    'claude-sonnet-4-5',
    settings=ModelSettings(
        anthropic_effort='high',
        anthropic_thinking={'type': 'adaptive'},
        anthropic_betas=['interleaved-thinking-2025-05-14'],
    ),
)

Model Identifiers

Format: <provider>:<model>

from pydantic_ai import Agent

# Simple string identifier
agent = Agent('openai:gpt-4o')
agent = Agent('anthropic:claude-sonnet-4-5')
agent = Agent('google:gemini-2.5-flash')
agent = Agent('xai:grok-4-1-fast-non-reasoning')

# Gateway prefix (if using AI gateway)
agent = Agent('gateway/openai:gpt-5')

Model.model_id (v1.59.0)

Model instances expose model_id for the normalized provider+model string:

from pydantic_ai.models.xai import XaiModel

model = XaiModel('grok-4-1-fast-non-reasoning')
print(model.model_id)  # xai:grok-4-1-fast-non-reasoning

Explicit Model Configuration

from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.models.anthropic import AnthropicModel

# With custom configuration
model = OpenAIChatModel(
    'gpt-4o',
    api_key='sk-...',
    base_url='https://custom.endpoint.com',
)

agent = Agent(model)

Custom Providers

from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

# Azure example
provider = OpenAIProvider(
    api_key='azure-key',
    base_url='https://your-resource.openai.azure.com',
)
model = OpenAIChatModel('gpt-4o', provider=provider)

Ollama (Local Models)

from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

provider = OpenAIProvider(base_url='http://localhost:11434/v1')
model = OpenAIChatModel('llama3.2', provider=provider)

agent = Agent(model)

Current releases also provide a native OllamaModel path. Prefer the native model integration when available, especially if you depend on structured output or capability detection against Ollama Cloud; keep the OpenAI-compatible base URL approach as the fallback interoperability pattern.

Fallback Models

Automatic failover between providers:

from pydantic_ai.models.fallback import FallbackModel
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.models.anthropic import AnthropicModel

openai = OpenAIChatModel('gpt-4o')
anthropic = AnthropicModel('claude-sonnet-4-5')

fallback = FallbackModel(openai, anthropic)
agent = Agent(fallback)

Response-Based Fallback (v1.69.0)

FallbackModel can now trigger failover based on the returned ModelResponse, not only raised exceptions.

• Use this for semantic failures such as truncated responses or built-in tool failures.
• This currently works only for non-streaming runs: run() and run_sync().
• If you pass only a response handler to fallback_on, it replaces the default exception-based fallback; include both when you need both behaviors.

from pydantic_ai import Agent, ModelAPIError
from pydantic_ai.messages import ModelResponse
from pydantic_ai.models.fallback import FallbackModel

def bad_finish_reason(response: ModelResponse) -> bool:
    return response.finish_reason in ('length', 'content_filter', 'error')

fallback = FallbackModel(
    'openai:gpt-5.2',
    'anthropic:claude-sonnet-4-5',
    fallback_on=[ModelAPIError, bad_finish_reason],
)

agent = Agent(fallback)

Bedrock Inference Profiles (v1.70.0)

Use bedrock_inference_profile when you need AWS Bedrock inference-profile routing while still keeping the base model name for capability detection and token counting.

from pydantic_ai import ModelSettings
from pydantic_ai.models.bedrock import BedrockModel

model = BedrockModel(
    'anthropic.claude-sonnet-4-5-20250929-v1:0',
    settings=ModelSettings(
        bedrock_inference_profile='arn:aws:bedrock:us-east-1:123456789012:inference-profile/my-profile'
    ),
)

When set, the inference-profile ARN is sent as the Bedrock modelId for API calls.

Concurrency Limiting (v1.54.0)

Limit concurrent requests per model or across a shared limiter.

from pydantic_ai import Agent, ConcurrencyLimiter
from pydantic_ai.models.concurrency import ConcurrencyLimitedModel, limit_model_concurrency

# Simple limit (max 5 concurrent requests)
model = ConcurrencyLimitedModel('openai:gpt-4o', limiter=5)
agent = Agent(model)

# Shared limiter across multiple models
shared = ConcurrencyLimiter(max_running=10, name='openai-pool')
model_a = ConcurrencyLimitedModel('openai:gpt-4o', limiter=shared)
model_b = ConcurrencyLimitedModel('openai:gpt-4o-mini', limiter=shared)

# Convenience wrapper
model_c = limit_model_concurrency('openai:gpt-4o', limiter=3)

````

Per-Model Settings

from pydantic_ai import ModelSettings

openai = OpenAIChatModel(
    'gpt-4o',
    settings=ModelSettings(temperature=0.7)
)
anthropic = AnthropicModel(
    'claude-sonnet-4-5',
    settings=ModelSettings(temperature=0.2)
)

fallback = FallbackModel(openai, anthropic)

Exception Handling

from pydantic_ai import ModelAPIError

try:
    result = agent.run_sync('Query')
except* ModelAPIError as exc_group:
    for exc in exc_group.exceptions:
        print(f"Model failed: {exc}")

Key Concepts

Testing Models

from pydantic_ai.models.test import TestModel
from pydantic_ai.models.function import FunctionModel

# Deterministic testing
test_model = TestModel()
agent = Agent(model=test_model)

# Custom function model
def my_model(messages, info):
    return ModelResponse(parts=[TextPart('test')])

agent = Agent(model=FunctionModel(my_model))

---

HTTP Request Retries

Built on tenacity library with httpx transport integration.

Installation

pip install 'pydantic-ai-slim[retries]'

Basic Setup

from httpx import AsyncClient, HTTPStatusError
from tenacity import retry_if_exception_type, stop_after_attempt, wait_exponential
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
from pydantic_ai.retries import AsyncTenacityTransport, RetryConfig, wait_retry_after

def create_retrying_client():
    def should_retry_status(response):
        if response.status_code in (429, 502, 503, 504):
            response.raise_for_status()

    transport = AsyncTenacityTransport(
        config=RetryConfig(
            retry=retry_if_exception_type((HTTPStatusError, ConnectionError)),
            wait=wait_retry_after(
                fallback_strategy=wait_exponential(multiplier=1, max=60),
                max_wait=300
            ),
            stop=stop_after_attempt(5),
            reraise=True
        ),
        validate_response=should_retry_status
    )
    return AsyncClient(transport=transport)

client = create_retrying_client()
model = OpenAIChatModel('gpt-4o', provider=OpenAIProvider(http_client=client))
agent = Agent(model)

Key Components

wait_retry_after

Respects HTTP Retry-After headers automatically:

from tenacity import wait_exponential
from pydantic_ai.retries import wait_retry_after

wait_strategy = wait_retry_after(
    fallback_strategy=wait_exponential(multiplier=2, max=120),
    max_wait=600  # Max 10 minutes
)

Common Patterns

Rate Limit Handling:

transport = AsyncTenacityTransport(
    config=RetryConfig(
        retry=retry_if_exception_type(HTTPStatusError),
        wait=wait_retry_after(fallback_strategy=wait_exponential(multiplier=1, max=60)),
        stop=stop_after_attempt(10),
        reraise=True
    ),
    validate_response=lambda r: r.raise_for_status()
)

Network Errors:

import httpx

transport = AsyncTenacityTransport(
    config=RetryConfig(
        retry=retry_if_exception_type((
            httpx.TimeoutException,
            httpx.ConnectError,
            httpx.ReadError
        )),
        wait=wait_exponential(multiplier=1, max=10),
        stop=stop_after_attempt(3),
        reraise=True
    )
)

AWS Bedrock Retries

Use boto3's built-in retry:

from botocore.config import Config

config = Config(retries={'max_attempts': 5, 'mode': 'adaptive'})

Best Practices

• Start conservative: 3-5 retries, reasonable waits
• Use exponential backoff
• Set maximum wait times
• Respect Retry-After headers
• Monitor retry rates in production
• Disable HTTP retries when using Temporal (use Temporal's retry policy)

---

OpenAI Provider Details

Configuration

# Environment variable (preferred)
export OPENAI_API_KEY='your-api-key'

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

# Explicit provider
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

model = OpenAIChatModel('gpt-4o', provider=OpenAIProvider(api_key='your-api-key'))

Custom Client

from openai import AsyncOpenAI
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

client = AsyncOpenAI(max_retries=3)
model = OpenAIChatModel('gpt-4o', provider=OpenAIProvider(openai_client=client))

Azure OpenAI

from openai import AsyncAzureOpenAI

client = AsyncAzureOpenAI(
    azure_endpoint='...',
    api_version='2024-07-01-preview',
    api_key='your-api-key',
)
model = OpenAIChatModel('gpt-4o', provider=OpenAIProvider(openai_client=client))

OpenAI Responses API

# By name
agent = Agent('openai-responses:gpt-4o')

# Explicit
from pydantic_ai.models.openai import OpenAIResponsesModel
model = OpenAIResponsesModel('gpt-4o')

Built-in Tools (via Responses API):

• Web search
• Code interpreter
• Image generation
• File search
• Computer use

Web search domain allowlist: when using the WebSearchTool with OpenAI, you can restrict sources via allowed_domains in the tool configuration.

from openai.types.responses import ComputerToolParam
from pydantic_ai.models.openai import OpenAIResponsesModel, OpenAIResponsesModelSettings

model_settings = OpenAIResponsesModelSettings(
    openai_builtin_tools=[ComputerToolParam(type='computer_use')]
)

# Usage stats streaming
model_settings = OpenAIResponsesModelSettings(
    continuous_usage_stats=True
)

Previous Response ID (Context Continuity)

from pydantic_ai.models.openai import OpenAIResponsesModelSettings

# Manual
result = agent.run_sync('Secret is 1234')
model_settings = OpenAIResponsesModelSettings(
    openai_previous_response_id=result.all_messages()[-1].provider_response_id
)
result = agent.run_sync('What is the secret?', model_settings=model_settings)

# Auto (recommended)
model_settings = OpenAIResponsesModelSettings(openai_previous_response_id='auto')
result2 = agent.run_sync('Explain?', message_history=result1.new_messages(), model_settings=model_settings)

OpenAI-Compatible Providers

from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

model = OpenAIChatModel(
    'model_name',
    provider=OpenAIProvider(base_url='https://custom-api.com', api_key='...')
)

Model Profile (Custom Schema Handling)

from pydantic_ai.profiles.openai import OpenAIModelProfile

model = OpenAIChatModel(
    'model_name',
    provider=OpenAIProvider(base_url='https://custom-api.com', api_key='...'),
    profile=OpenAIModelProfile(openai_supports_strict_tool_definition=False)
)

---

Anthropic Provider Details

Configuration

# Environment variable
export ANTHROPIC_API_KEY='your-api-key'

# By name
agent = Agent('anthropic:claude-sonnet-4-5')

# Explicit provider
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.providers.anthropic import AnthropicProvider

model = AnthropicModel('claude-sonnet-4-5', provider=AnthropicProvider(api_key='...'))

Cloud Platform Integrations

AWS Bedrock:

from anthropic import AsyncAnthropicBedrock
from pydantic_ai.providers.anthropic import AnthropicProvider

bedrock_client = AsyncAnthropicBedrock()  # Uses AWS env credentials
provider = AnthropicProvider(anthropic_client=bedrock_client)
model = AnthropicModel('us.anthropic.claude-sonnet-4-5-20250929-v1:0', provider=provider)

Google Vertex AI:

from anthropic import AsyncAnthropicVertex

vertex_client = AsyncAnthropicVertex(region='us-east5', project_id='your-project-id')
provider = AnthropicProvider(anthropic_client=vertex_client)
model = AnthropicModel('claude-sonnet-4-5', provider=provider)

Microsoft Foundry:

from anthropic import AsyncAnthropicFoundry

foundry_client = AsyncAnthropicFoundry(
    api_key='your-foundry-api-key',
    resource='your-resource-name',
)
provider = AnthropicProvider(anthropic_client=foundry_client)

Prompt Caching

Anthropic supports prompt caching to reduce costs. Maximum 4 cache points per request.

Cache Methods:

1. anthropic_cache_instructions=True — cache system prompt (5m or 1h TTL) 2. anthropic_cache_tool_definitions=True — cache tool definitions 3. anthropic_cache_messages=True — cache all messages automatically 4. CachePoint() — manual cache point marker

from pydantic_ai import Agent, CachePoint
from pydantic_ai.models.anthropic import AnthropicModelSettings

agent = Agent(
    'anthropic:claude-sonnet-4-5',
    system_prompt='Detailed instructions...',
    model_settings=AnthropicModelSettings(
        anthropic_cache_instructions=True,      # 1 cache point
        anthropic_cache_tool_definitions='1h',  # 1 cache point with 1h TTL
        anthropic_cache_messages=True,          # 1 cache point
    ),
)

# Manual cache point
result = agent.run_sync([
    'Long context from documentation...',
    CachePoint(),  # Cache everything up to this point
    'Question'
])

# Check cache usage
usage = result.usage()
print(f'Cache write: {usage.cache_write_tokens}')
print(f'Cache read: {usage.cache_read_tokens}')

Note: When using AsyncAnthropicBedrock, TTL is automatically omitted (Bedrock doesn't support explicit TTL).

---

Google Provider Details

1.97.0+ migration:

• google-gla: becomes google:.
• google-vertex: becomes google-cloud:.
• GoogleProvider(vertexai=True, ...) is replaced by the separate GoogleCloudProvider(...) class.

Configuration

# Generative Language API
export GOOGLE_API_KEY='your-api-key'

# By name
agent = Agent('google:gemini-2.5-pro')
agent = Agent('google:gemini-3.1-pro-preview')  # v1.63.0

# Google Cloud / Vertex AI
agent = Agent('google-cloud:gemini-2.5-pro')

Vertex AI Authentication

from pydantic_ai.models.google import GoogleModel
from pydantic_ai.providers.google import GoogleProvider, GoogleCloudProvider

# Application Default Credentials (recommended in GCP)
provider = GoogleCloudProvider()

# Service Account
from google.oauth2 import service_account
credentials = service_account.Credentials.from_service_account_file(
    'path/to/service-account.json',
    scopes=['https://www.googleapis.com/auth/cloud-platform'],
)
provider = GoogleCloudProvider(credentials=credentials, project='your-project-id')

# Custom location/project
provider = GoogleCloudProvider(location='asia-east1', project='your-gcp-project-id')

Model Garden (Non-Gemini Models)

# Access Llama, etc. from Model Garden
provider = GoogleCloudProvider(project='your-project-id', location='us-central1')
model = GoogleModel('meta/llama-3.3-70b-instruct-maas', provider=provider)

YouTube & File Upload

from pydantic_ai import Agent, VideoUrl, DocumentUrl
from pydantic_ai.models.google import GoogleModel
from pydantic_ai.providers.google import GoogleProvider

# YouTube URLs directly
agent = Agent(GoogleModel('gemini-2.5-flash'))
result = agent.run_sync([
    'What is this video about?',
    VideoUrl(url='https://www.youtube.com/watch?v=dQw4w9WgXcQ'),
])

# File upload via Files API
provider = GoogleProvider()
file = provider.client.files.upload(file='document.pdf')
result = agent.run_sync([
    'Summarize this document',
    DocumentUrl(url=file.uri, media_type=file.mime_type),
])

Model Settings

from google.genai.types import HarmBlockThreshold, HarmCategory
from pydantic_ai.models.google import GoogleModel, GoogleModelSettings

settings = GoogleModelSettings(
    temperature=0.2,
    max_tokens=1024,
    google_thinking_config={'thinking_level': 'low'},  # or 'thinking_budget': 0 to disable
    google_safety_settings=[{
        'category': HarmCategory.HARM_CATEGORY_HATE_SPEECH,
        'threshold': HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
    }]
)

Vertex AI Logprobs (v1.63.0)

Enable logprobs via GoogleModelSettings.google_logprobs and google_top_logprobs.

• Supported only for Vertex AI and non-streaming requests.
• Logprobs are surfaced in ModelResponse.provider_details['logprobs'].

from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel, GoogleModelSettings
from pydantic_ai.providers.google import GoogleCloudProvider

provider = GoogleCloudProvider(location='europe-west1')
settings = GoogleModelSettings(google_logprobs=True, google_top_logprobs=2)
model = GoogleModel('gemini-2.5-flash', provider=provider, settings=settings)

agent = Agent(model)
result = agent.run_sync('Write one sentence about the sky.')
logprobs = result.response.provider_details.get('logprobs')

---

Groq Provider Details

Fast inference on open-source models.

# Environment variable
export GROQ_API_KEY='your-api-key'

# By name
agent = Agent('groq:llama-3.3-70b-versatile')

# Explicit provider
from pydantic_ai.models.groq import GroqModel
from pydantic_ai.providers.groq import GroqProvider

model = GroqModel('llama-3.3-70b-versatile', provider=GroqProvider(api_key='...'))

---

Mistral Provider Details

# Environment variable
export MISTRAL_API_KEY='your-api-key'

# By name
agent = Agent('mistral:mistral-large-latest')

# Explicit provider
from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider

model = MistralModel('mistral-large-latest', provider=MistralProvider(api_key='...'))

# Custom endpoint
model = MistralModel(
    'mistral-large-latest',
    provider=MistralProvider(api_key='...', base_url='https://custom-endpoint')
)

---

OpenRouter Provider Details

Multi-model routing service with unified API.

# Environment variable
export OPENROUTER_API_KEY='your-api-key'

# By name (model format: provider/model)
agent = Agent('openrouter:anthropic/claude-3.5-sonnet')
agent = Agent('openrouter:openai/gpt-4o')

# Explicit provider
from pydantic_ai.models.openrouter import OpenRouterModel
from pydantic_ai.providers.openrouter import OpenRouterProvider

model = OpenRouterModel('anthropic/claude-3.5-sonnet', provider=OpenRouterProvider(api_key='...'))

App Attribution

provider = OpenRouterProvider(
    api_key='...',
    app_url='https://your-app.com',
    app_title='Your App',
)

Model Settings

from pydantic_ai.models.openrouter import OpenRouterModel, OpenRouterModelSettings

settings = OpenRouterModelSettings(
    openrouter_reasoning={'effort': 'high'},
    openrouter_usage={'include': True}
)
model = OpenRouterModel('openai/gpt-5', model_settings=settings)

Function Tools Reference

Tools let models perform actions and retrieve information during response generation.

Registration Methods

1. Decorator with Context

from pydantic_ai import Agent, RunContext

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

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

2. Decorator without Context

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

3. Via Agent Constructor

from pydantic_ai import Tool

def roll_dice() -> str:
    """Roll a die."""
    return str(random.randint(1, 6))

def get_name(ctx: RunContext[str]) -> str:
    """Get name."""
    return ctx.deps

# Auto-detect context
agent = Agent('openai:gpt-4o', tools=[roll_dice, get_name])

# Explicit context specification
agent = Agent('openai:gpt-4o', tools=[
    Tool(roll_dice, takes_ctx=False),
    Tool(get_name, takes_ctx=True),
])

Tool Schema

Parameters extracted from function signature. Docstrings provide descriptions.

Docstring Formats

Supports: google, numpy, sphinx (auto-detected)

@agent.tool_plain(docstring_format='google', require_parameter_descriptions=True)
def search(query: str, limit: int) -> str:
    """Search for items.

    Args:
        query: Search query string
        limit: Maximum results to return
    """
    return "results"

Single Parameter Simplification

If tool has single object parameter, schema is simplified:

from pydantic import BaseModel

class SearchParams(BaseModel):
    """Search parameters"""
    query: str
    limit: int = 10

@agent.tool_plain
def search(params: SearchParams) -> str:
    return f"Searching: {params.query}"

Tool Return Types

Any JSON-serializable type works:

@agent.tool_plain
def get_data() -> dict[str, list[int]]:
    return {"values": [1, 2, 3]}

@agent.tool_plain
def get_count() -> int:
    return 42

RunContext Properties

@agent.tool
async def my_tool(ctx: RunContext[MyDeps]) -> str:
    ctx.deps         # Dependencies
    ctx.retry        # Current retry count
    ctx.tool_name    # Name of this tool
    ctx.run_step     # Current run step
    ctx.usage        # Token usage so far

Tool hook notes (1.84.x)

• Internal output tools now skip tool hooks; do not rely on hook side effects for framework-managed output tools.
• For single-BaseModel tools, validated hook arguments are consistently passed as a dict shape instead of an ambiguous model/object form.

PrepareTools migration (v1.96.x)

Tool preparation is moving toward explicit capabilities rather than constructor sugar.

• Agent(..., prepare_tools=...) is now the deprecated path.
• Agent(..., prepare_output_tools=...) is also deprecated in favor of capability-based composition.
• Prefer PrepareTools(...) and PrepareOutputTools(...) capabilities so tool filtering/modification participates in the same ordering/wrapping model as the rest of your agent stack.

This matters most when tool preparation interacts with other capabilities, because the capability path makes ordering explicit instead of burying it in constructor kwargs.

As of 1.103.0, toolset prepare callbacks warn when they return None. Treat that warning as a likely bug in custom filtering/modification code; return the intended tool list or explicit empty result instead of falling through accidentally.

Tool Retries

````python from pydantic_ai import ModelRetry

@agent.tool(retries=3) def fetch_user(ctx: RunContext[Deps], user_id: int) -> dict: """Fetch user by ID.""" user = ctx.deps.db.get(user_id) if not user: raise ModelRetry(f"User {user_id} not found. Try different ID.") return user

Args Validator (v1.63.0)

Use args_validator to run custom, typed validation before a tool executes.

• The validator has the same (typed) signature as the tool.
• On validation failure, raise ModelRetry to ask the model for new arguments.
• Validation runs before the FunctionToolCallEvent is emitted; the event includes args_valid.

from pydantic_ai import Agent, ModelRetry, RunContext

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

def validate_user_id(ctx: RunContext[int], user_id: int) -> None:
    if user_id <= 0:
        raise ModelRetry('user_id must be a positive integer')

@agent.tool(args_validator=validate_user_id)
def get_user(ctx: RunContext[int], user_id: int) -> str:
    return f'User {user_id}'

If you inspect validated args inside hooks, align that code with the dict-shaped behavior above before upgrading shared tool infrastructure.

````

Custom Tool Configuration

from pydantic_ai import Tool

tool = Tool(
    my_function,
    takes_ctx=True,
    name='custom_name',           # Override function name
    description='Custom desc',    # Override docstring
    retries=2,
)

agent = Agent('openai:gpt-4o', tools=[tool])

Advanced Tool Returns

Control both return value and model content:

from pydantic_ai import Agent, ToolReturn, BinaryContent

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

@agent.tool_plain
def click_and_capture(x: int, y: int) -> ToolReturn:
    """Click at coordinates and show before/after screenshots."""
    before = capture_screen()
    perform_click(x, y)
    after = capture_screen()

    return ToolReturn(
        return_value=f'Clicked at ({x}, {y})',  # Tool result for model
        content=[  # Additional context (separate user message)
            'Before:',
            BinaryContent(data=before, media_type='image/png'),
            'After:',
            BinaryContent(data=after, media_type='image/png'),
        ],
        metadata={  # Not sent to LLM, available in your app
            'coordinates': {'x': x, 'y': y},
        }
    )

ToolReturn Fields

Multimodal Tool Results (v1.69.0)

When provider APIs support multimodal tool-result payloads, Pydantic AI now forwards those results directly instead of always splitting them into extra user-message parts.

• Keep using ToolReturn.content for images, docs, and other multimodal artifacts.
• Prefer provider-native multimodal flows when a downstream model can consume them directly.
• If you depend on provider-specific multimodal behavior, verify it with the target model rather than assuming every provider handles the same content types identically.

UploadedFile (v1.65.0)

Pydantic AI adds an UploadedFile object to support files uploaded to model providers. Use it when a provider requires a pre-upload step (instead of inlining raw bytes in every request).

Custom Tool Schema

For functions without proper documentation:

from pydantic_ai import Agent, Tool

def foobar(**kwargs) -> str:
    return kwargs['a'] + kwargs['b']

tool = Tool.from_schema(
    function=foobar,
    name='sum',
    description='Sum two numbers.',
    json_schema={
        'properties': {
            'a': {'description': 'first number', 'type': 'integer'},
            'b': {'description': 'second number', 'type': 'integer'},
        },
        'required': ['a', 'b'],
        'type': 'object',
    },
    takes_ctx=False,
)

agent = Agent('openai:gpt-4o', tools=[tool])

Dynamic Tools (Prepare)

Customize tool availability per-run:

from pydantic_ai import Agent, RunContext, ToolDefinition

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

async def only_if_42(
    ctx: RunContext[int], tool_def: ToolDefinition
) -> ToolDefinition | None:
    if ctx.deps == 42:
        return tool_def
    return None  # Hide tool

@agent.tool(prepare=only_if_42)
def hitchhiker(ctx: RunContext[int], answer: str) -> str:
    return f'{ctx.deps}{answer}'

Agent-wide prepare_tools

Filter or modify all tools at once:

from dataclasses import replace
from pydantic_ai import Agent, RunContext, ToolDefinition

async def turn_on_strict_if_openai(
    ctx: RunContext[None], tool_defs: list[ToolDefinition]
) -> list[ToolDefinition] | None:
    if ctx.model.system == 'openai':
        return [replace(td, strict=True) for td in tool_defs]
    return tool_defs

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

Tool Timeout

import asyncio
from pydantic_ai import Agent

# Default timeout for all tools
agent = Agent('openai:gpt-4o', tool_timeout=30)

@agent.tool_plain(timeout=5)  # Override per-tool
async def fast_tool() -> str:
    await asyncio.sleep(1)
    return 'Done'

On timeout: tool fails → retry prompt sent → counts toward retry limit.

Tool Execution & Retries

Validation Errors

Arguments validated by Pydantic → ValidationError → RetryPromptPart → LLM retries.

Explicit Retry

from pydantic_ai import ModelRetry

def my_tool(query: str) -> str:
    if query == 'bad':
        raise ModelRetry("Query 'bad' not allowed. Try again.")
    return 'Success!'

Parallel Tool Calls

Multiple tool calls run concurrently via asyncio.create_task.

# Force sequential execution for specific tool
@agent.tool_plain(sequential=True)
def must_run_alone() -> str:
    return 'Done'

# Or for entire run
with agent.sequential_tool_calls():
    result = await agent.run('...')

Limit Tool Calls

from pydantic_ai import UsageLimits

result = await agent.run('...', usage_limits=UsageLimits(tool_calls_limit=10))

---

Deferred Tools

Tools that cannot/should not be executed during the same agent run.

Use Cases

• Tool requires user approval first
• Tool depends on external service, frontend, or user
• Result takes longer than reasonable to keep agent running

Tool Approval (Human-in-the-Loop)

from pydantic_ai import (
    Agent,
    ApprovalRequired,
    DeferredToolRequests,
    DeferredToolResults,
    RunContext,
    ToolDenied,
)

agent = Agent('openai:gpt-4o', output_type=[str, DeferredToolRequests])

@agent.tool_plain(requires_approval=True)  # Always requires approval
def delete_file(path: str) -> str:
    return f'File {path!r} deleted'

@agent.tool
def update_file(ctx: RunContext, path: str, content: str) -> str:
    if path == '.env' and not ctx.tool_call_approved:
        raise ApprovalRequired(metadata={'reason': 'protected'})  # Conditional
    return f'File {path!r} updated'

# First run: ends with deferred requests
result = agent.run_sync('Delete file.txt and clear .env')
messages = result.all_messages()

assert isinstance(result.output, DeferredToolRequests)
requests = result.output

# Gather approvals and continue
results = DeferredToolResults()
for call in requests.approvals:
    if call.tool_name == 'delete_file':
        results.approvals[call.tool_call_id] = ToolDenied('Deletion not allowed')
    else:
        results.approvals[call.tool_call_id] = True  # Approve

# Second run: continue with approvals
result = agent.run_sync(
    'Continue',
    message_history=messages,
    deferred_tool_results=results,
)

External Tools

Tools executed by external service/frontend:

from pydantic_ai import Agent, CallDeferred, DeferredToolRequests, DeferredToolResults

agent = Agent('openai:gpt-4o', output_type=[str, DeferredToolRequests])

@agent.tool
async def long_task(ctx, query: str) -> str:
    task_id = schedule_background_task(query)  # Your task scheduler
    raise CallDeferred(metadata={'task_id': task_id})

# First run: ends with deferred calls
result = await agent.run('Calculate something complex')
requests = result.output
messages = result.all_messages()

# Wait for results (e.g., poll background worker)
task_results = await wait_for_tasks(requests)

# Build results
results = DeferredToolResults()
for call in requests.calls:
    task_id = requests.metadata[call.tool_call_id]['task_id']
    results.calls[call.tool_call_id] = task_results[task_id]

# Continue with results
result = await agent.run(message_history=messages, deferred_tool_results=results)

DeferredToolRequests

Returned when agent run ends with deferred tools:

DeferredToolResults

Provide results/approvals to continue:

---

Toolsets

Collections of tools that can be registered with an agent.

FunctionToolset

from pydantic_ai.toolsets import FunctionToolset

toolset = FunctionToolset()

@toolset.tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

agent = Agent('openai:gpt-4o', toolsets=[toolset])

Combining Toolsets

from pydantic_ai.toolsets import CombinedToolset

combined = CombinedToolset([toolset1, toolset2, mcp_server])
agent = Agent('openai:gpt-4o', toolsets=[combined])

Filtering Tools

from pydantic_ai.toolsets import FilteredToolset

def filter_fn(ctx, tool_def):
    return tool_def.name.startswith('safe_')

filtered = FilteredToolset(toolset, filter_fn)
# Or chain: toolset.filtered(filter_fn)

Prefixing Tool Names

from pydantic_ai.toolsets import PrefixedToolset

prefixed = PrefixedToolset(toolset, 'math_')
# Or chain: toolset.prefixed('math_')

Renaming Tools

from pydantic_ai.toolsets import RenamedToolset

renamed = RenamedToolset(toolset, {'new_name': 'original_name'})
# Or chain: toolset.renamed({'new_name': 'original_name'})

Approval Required

from pydantic_ai.toolsets import ApprovalRequiredToolset

def needs_approval(ctx, tool_def, args):
    return tool_def.name == 'delete_file'

approved = ApprovalRequiredToolset(toolset, needs_approval)
# Or chain: toolset.approval_required(needs_approval)

Dynamic Toolsets

@agent.toolset
def dynamic_tools(ctx: RunContext[Deps]):
    toolset = FunctionToolset()
    if ctx.deps.is_admin:
        @toolset.tool
        def admin_action(): ...
    return toolset

External Toolsets

For tools executed by upstream service/frontend:

from pydantic_ai.toolsets import ExternalToolset
from pydantic_ai import ToolDefinition, DeferredToolRequests

external = ExternalToolset([
    ToolDefinition(
        name='ui_confirm',
        description='Show confirmation dialog',
        parameters_json_schema={'type': 'object', ...},
    )
])

agent = Agent('openai:gpt-4o', toolsets=[external], output_type=[str, DeferredToolRequests])

Third-Party Toolsets

Tool Categories

---

Structured Output Reference

Agent's output_type defines the structured data the model must return.

Supported Types

• Pydantic BaseModel
• dataclass
• TypedDict
• Scalar types (str, int, list[str], etc.)
• Union types (Foo | Bar)
• Output functions

Basic Usage

from pydantic import BaseModel
from pydantic_ai import Agent

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

agent = Agent('openai:gpt-4o', output_type=CityLocation)
result = agent.run_sync('Where were the 2012 Olympics?')
print(result.output)  # CityLocation(city='London', country='UK')

Multiple Output Types

class Box(BaseModel):
    width: int
    height: int
    units: str

# List of types or union
agent = Agent(
    'openai:gpt-4o',
    output_type=[Box, str],  # can return Box or ask for clarification
)

Output Functions

End run with function result instead of passing to model:

from pydantic_ai import Agent, ModelRetry

def run_sql(query: str) -> list[dict]:
    """Execute SQL and return results."""
    if 'DROP' in query.upper():
        raise ModelRetry("DROP not allowed. Try SELECT.")
    return db.execute(query)

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

Output Modes

v1.64.0 adds template=False on PromptedOutput and NativeOutput to disable schema prompt injection when you need tighter control over prompt content.

Tool Output (Default)

from pydantic_ai import ToolOutput

agent = Agent(
    'openai:gpt-4o',
    output_type=ToolOutput(MyModel, name='return_data', strict=True),
)

Native Output

from pydantic_ai import NativeOutput

agent = Agent(
    'openai:gpt-4o',
    output_type=NativeOutput([Fruit, Vehicle]),
)

Text Output

Process plain text through function:

from pydantic_ai import TextOutput

def split_words(text: str) -> list[str]:
    return text.split()

agent = Agent('openai:gpt-4o', output_type=TextOutput(split_words))

Output Validators

@agent.output_validator
async def validate(ctx: RunContext[Deps], output: MyModel) -> MyModel:
    if not output.is_valid:
        raise ModelRetry("Invalid output, try again")
    return output

Streaming Partial Output

@agent.output_validator
def validate(ctx: RunContext, output: str) -> str:
    if ctx.partial_output:
        return output  # Skip validation for partial
    if len(output) < 50:
        raise ModelRetry('Too short')
    return output

Custom JSON Schema

from pydantic_ai import StructuredDict

PersonDict = StructuredDict(
    {
        'type': 'object',
        'properties': {
            'name': {'type': 'string'},
            'age': {'type': 'integer'},
        },
        'required': ['name', 'age'],
    },
    name='Person',
)

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

---

Built-in Tools

Native tools executed by model providers (not by Pydantic AI).

Available Built-in Tools

Web Search Example

from pydantic_ai import Agent, WebSearchTool, WebSearchUserLocation

agent = Agent(
    'anthropic:claude-sonnet-4-0',
    builtin_tools=[
        WebSearchTool(
            search_context_size='high',
            user_location=WebSearchUserLocation(
                city='San Francisco',
                country='US',
            ),
            blocked_domains=['spam-site.net'],
            max_uses=5,  # Anthropic only
        )
    ],
)

result = agent.run_sync('What is the biggest AI news this week?')

Code Execution Example

from pydantic_ai import Agent, CodeExecutionTool

agent = Agent('anthropic:claude-sonnet-4-0', builtin_tools=[CodeExecutionTool()])
result = agent.run_sync('Calculate the factorial of 15.')

Image Generation Example

from pydantic_ai import Agent, BinaryImage, ImageGenerationTool

agent = Agent(
    'openai-responses:gpt-5',
    builtin_tools=[
        ImageGenerationTool(
            quality='high',
            size='1024x1024',
            output_format='png',
        )
    ],
    output_type=BinaryImage,
)

result = agent.run_sync('Generate an image of a sunset.')

File Search (RAG) Example

from pydantic_ai import Agent, FileSearchTool
from pydantic_ai.models.openai import OpenAIResponsesModel

model = OpenAIResponsesModel('gpt-5')

# Upload files to vector store first
file = await model.client.files.create(file=open('doc.txt', 'rb'), purpose='assistants')
vector_store = await model.client.vector_stores.create(name='my-docs')
await model.client.vector_stores.files.create(
    vector_store_id=vector_store.id,
    file_id=file.id
)

agent = Agent(
    model,
    builtin_tools=[FileSearchTool(file_store_ids=[vector_store.id])]
)

result = await agent.run('What does the document say about X?')

MCP Server Tool Example

from pydantic_ai import Agent, MCPServerTool

agent = Agent(
    'anthropic:claude-sonnet-4-5',
    builtin_tools=[
        MCPServerTool(
            id='deepwiki',
            url='https://mcp.deepwiki.com/mcp',
            # authorization_token='...',  # If required
            # allowed_tools=['tool1', 'tool2'],
        )
    ]
)

result = agent.run_sync('Tell me about pydantic/pydantic-ai repo.')

Key Notes

• Built-in tools executed by provider, not Pydantic AI
• Not all providers support all tools
• Google: Cannot use built-in + function tools together
• OpenAI: Web search/MCP require Responses API (openai-responses:)
• Access results via result.response.builtin_tool_calls

---

Common Tools

Pre-built tools that Pydantic AI executes (not provider-side).

DuckDuckGo Search

pip install "pydantic-ai-slim[duckduckgo]"

from pydantic_ai import Agent
from pydantic_ai.common_tools.duckduckgo import duckduckgo_search_tool

agent = Agent(
    'openai:gpt-4o',
    tools=[duckduckgo_search_tool()],
    instructions='Search DuckDuckGo for the query and return results.',
)

result = agent.run_sync('Top AI news this week')

Tavily Search

Paid service with free credits. Requires API key.

pip install "pydantic-ai-slim[tavily]"

import os
from pydantic_ai import Agent
from pydantic_ai.common_tools.tavily import tavily_search_tool

api_key = os.getenv('TAVILY_API_KEY')

agent = Agent(
    'openai:gpt-4o',
    tools=[tavily_search_tool(api_key)],
    instructions='Search Tavily for the query and return results.',
)

result = agent.run_sync('Tell me top GenAI news with links.')

Exa Neural Search

Neural search engine for high-quality results. Paid with free credits.

pip install "pydantic-ai-slim[exa]"

Individual Tools:

import os
from pydantic_ai import Agent
from pydantic_ai.common_tools.exa import exa_search_tool

api_key = os.getenv('EXA_API_KEY')

agent = Agent(
    'openai:gpt-4o',
    tools=[exa_search_tool(api_key, num_results=5, max_characters=1000)],
    instructions='Search the web using Exa.',
)

result = agent.run_sync('Latest developments in quantum computing')

Using ExaToolset (recommended):

import os
from pydantic_ai import Agent
from pydantic_ai.common_tools.exa import ExaToolset

api_key = os.getenv('EXA_API_KEY')

toolset = ExaToolset(
    api_key,
    num_results=5,
    max_characters=1000,  # Limit text for token control
    include_search=True,  # Web search (default: True)
    include_find_similar=True,  # Find similar pages (default: True)
    include_get_contents=False,  # Full content retrieval
    include_answer=True,  # AI answers with citations (default: True)
)

agent = Agent('openai:gpt-4o', toolsets=[toolset])
result = agent.run_sync('Find recent AI papers and summarize findings.')

Exa Tools Available:

Common vs Built-in Tools