Building Pydantic AI Agents

Building AI Agents with Pydantic AI

Pydantic AI is a Python agent framework for building production-grade Generative AI applications. This skill provides patterns, architecture guidance, and tested code examples for building applications with Pydantic AI.

When to Use This Skill

Invoke this skill when:

• User asks to build an AI agent, create an LLM-powered app, or mentions Pydantic AI
• User wants to add tools, capabilities (thinking, web search), or structured output to an agent
• User asks to define agents from YAML/JSON specs or use template strings
• User wants to stream agent events, delegate between agents, or test agent behavior
• Code imports pydantic_ai or references Pydantic AI classes (Agent, RunContext, Tool)
• User asks about hooks, lifecycle interception, or agent observability with Logfire
• The agent design includes optional instructions, specialist workflows, long-tail tools, or any context the model does not need on most turns

Do not use this skill for:

• The Pydantic validation library alone (pydantic/BaseModel without agents)
• Other AI frameworks (LangChain, LlamaIndex, CrewAI, AutoGen)
• General Python development unrelated to AI agents

Quick-Start Patterns

Create a Basic Agent

from pydantic_ai import Agent

agent = Agent(
    'anthropic:claude-sonnet-4-6',
    name='hello_world_agent',
    instructions='Be concise, reply with one sentence.',
)

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

Add Tools to an Agent

import random

from pydantic_ai import Agent, RunContext

agent = Agent(
    'google:gemini-3-flash-preview',
    name='dice_game_agent',
    deps_type=str,
    instructions=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)


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


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


dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.output)
#> Congratulations Anne, you guessed correctly! You're a winner!

Structured Output with Pydantic Models

from pydantic import BaseModel

from pydantic_ai import Agent


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


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

Dependency Injection

from datetime import date

from pydantic_ai import Agent, RunContext

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


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


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


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

Testing with TestModel

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

my_agent = Agent('openai:gpt-5.2', name='my_agent', instructions='...')


async def test_my_agent():
    """Unit test for my_agent, to be run by pytest."""
    m = TestModel()
    with my_agent.override(model=m):
        result = await my_agent.run('Testing my agent...')
        assert result.output == 'success (no tool calls)'
    assert m.last_model_request_parameters.function_tools == []

Use Capabilities

Capabilities are reusable, composable units of agent behavior — bundling tools, hooks, instructions, and model settings.

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

agent = Agent(
    'anthropic:claude-opus-4-6',
    name='research_assistant_agent',
    instructions='You are a research assistant. Be thorough and cite sources.',
    capabilities=[
        Thinking(effort='high'),
        WebSearch(),
    ],
)

Add Lifecycle Hooks

Use Hooks to intercept model requests, tool calls, and runs with decorators — no subclassing needed.

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities.hooks import Hooks
from pydantic_ai.models import ModelRequestContext

hooks = Hooks()


@hooks.on.before_model_request
async def log_request(ctx: RunContext, request_context: ModelRequestContext) -> ModelRequestContext:
    print(f'Sending {len(request_context.messages)} messages')
    return request_context


agent = Agent('openai:gpt-5.2', name='hooks_agent', capabilities=[hooks])

For a custom capability hook that performs I/O under Temporal, DBOS, or Prefect, mark a fixed method with @durable_operation(name='...'). The required name becomes part of persisted durable-unit names, so keep it stable even if the Python method is renamed. For dynamically contributed handlers, return them from get_durable_operations() and invoke a typed handle with ctx.durable_operation(self, name, handler). Always set a stable capability id; without a durability capability both forms call the original async handler directly. Arguments and results must be serializable like durable tool inputs and outputs.

Define Agent from YAML Spec

Use Agent.from_file to load agents from YAML or JSON — no Python agent construction code needed.

from pydantic_ai import Agent

# agent.yaml:
# model: anthropic:claude-opus-4-6
# instructions: You are a helpful research assistant.
# capabilities:
#   - WebSearch
#   - Thinking:
#       effort: high

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

Realtime (speech-to-speech) sessions

For voice models that stream audio over a persistent connection (OpenAI Realtime, Azure OpenAI, Gemini Live, or xAI Grok Voice), use agent.realtime().session() instead of run(). It reuses the agent's tools and instructions and runs the tool loop for you. Stream input with send_audio/send, and iterate the session to consume the same part/event vocabulary as a streamed run — PartStartEvent / PartDeltaEvent / PartEndEvent carrying SpeechParts and ToolCallParts, plus FunctionToolCallEvent / FunctionToolResultEvent, plus realtime control events (RealtimeInputSpeechStartEvent, RealtimeInputSpeechEndEvent, RealtimeResponseInterruptedEvent, ...). Use RealtimeTurnCompleteEvent as the exchange boundary, when generation and tool work are complete. This is not always the end of audible speech: on WebRTC sidebands, track playback with RealtimeOutputSpeechStartEvent and RealtimeOutputSpeechEndEvent. Before passing raw microphone bytes to send_audio, convert them to mono PCM16 at session.audio_input_sample_rate; raw chunks carry no sample-rate metadata.

import anyio

from pydantic_ai import Agent
from pydantic_ai.messages import (
    PartDeltaEvent,
    PartEndEvent,
    SpeechPart,
    SpeechPartDelta,
)
from pydantic_ai.realtime import RealtimeSessionErrorEvent, RealtimeTurnCompleteEvent
from pydantic_ai.realtime.openai import OpenAIRealtimeModelSettings

agent = Agent(instructions='You are a helpful voice assistant.')


async def main(microphone_chunk: bytes):
    settings = OpenAIRealtimeModelSettings(openai_voice='alloy', turn_detection=False)
    async with agent.realtime(
        'openai:gpt-realtime', model_settings=settings
    ).session() as session:
        # The chunk must already be mono PCM16 at `session.audio_input_sample_rate`.
        await session.send_audio(microphone_chunk)
        await session.commit_audio()
        await session.create_response()
        # Input transcription can finish after the model exchange. Give it a
        # bounded grace period so a missing transcript cannot hang the session.
        turn_complete = user_turn_complete = False
        with anyio.move_on_after(None) as transcript_wait:
            async for event in session:
                match event:
                    case PartDeltaEvent(delta=SpeechPartDelta(audio_chunk=chunk)) if chunk:
                        ...  # play audio out
                    case PartEndEvent(part=SpeechPart(speaker='user', transcript=t)):
                        if t is not None:
                            print('user said:', t)
                        user_turn_complete = True
                    case RealtimeTurnCompleteEvent():
                        turn_complete = True
                        transcript_wait.deadline = anyio.current_time() + 1
                    case RealtimeSessionErrorEvent(message=message, recoverable=True):
                        # The connection remains usable, but this turn may not complete.
                        raise RuntimeError(message)
                if turn_complete and user_turn_complete:
                    break

    # A session builds ordinary ModelMessage history: hand it off to a text agent.
    notes = Agent('openai:gpt-5.2', instructions='Summarize.')
    await notes.run(message_history=session.all_messages())

Key facts for building realtime agents:

• A string sent with session.send() solicits a response: use respond=False to add passive text context. Images are context-only by default; use respond=True to ask for a response to an image. Never pair session.send('...') with session.create_response(), because that asks twice.
• History handoff is the marquee integration: session.all_messages() / session.new_messages() return real ModelMessages; seed with realtime(model, message_history=...).session(). Transcripts are what carry over; OpenAI and Azure can also replay retained transcript-less user audio, Gemini and xAI cannot, and assistant audio is never replayed. Streamed images all reach the provider, but history keeps a sampled (retain_images_every_n) and bounded (retain_images_max, default 100, oldest evicted first) record.
• No output_type: realtime models don't do structured output. Delegate hard work to a text agent behind a tool, or hand off history afterwards.
• Check the model profile before calling profile-gated methods: model.profile (a RealtimeModelProfile, the realtime counterpart to ModelProfile) reports supports_manual_turn_control, supports_interruption, supports_image_input, supports_output_truncation, and supports_session_seeding. OpenAI and Azure OpenAI support all of these; Gemini Live lacks supports_manual_turn_control, supports_interruption, and supports_output_truncation (automatic VAD only). Calling an unsupported method raises UserError up front.
• Turn detection: use the shared TurnDetection setting for sensitivity, prefix padding, and silence duration across providers. Use openai_turn_detection, xai_turn_detection, or google_vad only for finer provider-specific control; when present, they fully override the shared setting. Automatic detection is on by default (True); set turn_detection=False for push-to-talk (OpenAI/Azure/xAI only — Gemini has no manual turn controls and raises).
• Barge-in (the user speaking over the model): pass handle_barge_in=True to .session() and the session owns the local half — flushing the audio the user will never hear, truncating the provider's transcript to what was played, and adding a client cancel only on providers whose own turn detection isn't already cancelling. Off by default, and it needs playback to drain a single device-paced stream_audio() iterator (the position it tracks); with none or several it stands down. To keep the trigger yourself, session.interrupt(played_bytes=session.played_audio_bytes) gets the same treatment on your own signal. A playback layer that buffers ahead of the device makes played_audio_bytes read too far: count real device consumption and pass played_ms.
• Tools: every tool runs in the background, so a slow tool never blocks the session. Whether the model keeps speaking meanwhile is provider-specific (OpenAI/Azure do; Gemini needs google_async_tool_calls=True on a native-audio model). An unhandled tool exception is raised from session iteration; when only stream_audio() or stream_transcripts() is consumed, it ends those views and is raised when the session context closes. An on_tool_execute_error capability can return a replacement result or raise ModelRetry to keep the session running. To end the call from a tool, await ctx.realtime_session.close() for a clean hang-up (the tool does not resume and its call is recorded as interrupted), or call ctx.cancel() to make the session context raise RunCancelled.
• Browser WebRTC (OpenAI and Azure OpenAI): for browser voice agents, relay the browser's SDP offer server-side with agent.realtime(model).answer_webrtc_offer(sdp_offer) — the agent's resolved instructions and tools are baked in and the API key stays on the server — then attach a control-plane sideband with .session(provider_session=answer.session). The browser owns the audio; the sideband session runs tools and builds history (its audio methods raise, and audio_retention must stay 'transcript_only').

See the Realtime guide for the full walkthrough.