Agents Core

Read this file when the user needs the core Agent workflow: creating agents, choosing output types, using dependencies, defining specs, selecting models, or choosing how to run/stream an agent.

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)

Pass an explicit name= to each agent: it labels the agent's run span in Logfire. When omitted, the name is inferred from the variable the agent is assigned to and falls back to 'agent' when it can't be (e.g. agents kept in a list or dict), which matters once more than one agent runs in the same app.

Structured Output with Pydantic Models

Use output_type=MyModel when the model should return validated structured data.

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)

If the user is choosing between output modes:

• output_type=str for plain text
• output_type=MyModel for structured output
• TextOutput for custom text parsing
• NativeOutput or ToolOutput when they need explicit output-mode control

Dependency Injection

Use deps_type=... plus RunContext[...] when tools or instructions need app state.

from pydantic_ai import Agent, RunContext

agent = Agent('openai:gpt-5.2', name='greeting_agent', deps_type=str)


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

Use @agent.tool when the tool needs RunContext. Use @agent.tool_plain when it does not.

Define Agents Declaratively with Specs

Use YAML or JSON specs when configuration should live outside Python code.

model: anthropic:claude-opus-4-6
instructions: "You are helping {{user_name}} with research."
capabilities:
  - WebSearch
  - Thinking:
      effort: high

from dataclasses import dataclass

from pydantic_ai import Agent


@dataclass
class UserContext:
    user_name: str


agent = Agent.from_file('agent.yaml', deps_type=UserContext)
result = agent.run_sync('Find recent papers on AI safety', deps=UserContext(user_name='Alice'))

Template strings are part of the spec flow, so route template-string questions here too.

Choose or Configure Models

Model strings use the "provider:model-name" format.

Examples:

• openai:gpt-5.2
• anthropic:claude-sonnet-4-6
• google:gemini-3-pro-preview

Use a model instance instead of a string when the user needs provider-specific constructor arguments.

Run Methods and Streaming

Pick a run method based on the interaction pattern:

• run() for async runs that complete normally
• run_sync() for synchronous scripts and notebooks
• run_stream() for streaming final output
• run_stream_sync() for sync streaming
• run_stream_events() when the caller needs the typed event stream directly
• iter() when the caller needs step-by-step control over the agent loop

Use event_stream_handler= with run() or run_stream() when the user wants progress updates without manually consuming the event stream. The stream includes model deltas, tool call/result events, and framework events such as EnqueuedMessagesEvent when queued messages enter run history.

Realtime sessions do not use event_stream_handler; iterate the session to consume realtime-only RealtimeEvent members.

from collections.abc import AsyncIterable

from pydantic_ai import Agent, AgentStreamEvent, FunctionToolCallEvent, RunContext

agent = Agent('openai:gpt-5.2', name='streaming_agent')


async def stream_handler(ctx: RunContext, events: AsyncIterable[AgentStreamEvent]):
    async for event in events:
        if isinstance(event, FunctionToolCallEvent):
            print(f'Calling {event.part.tool_name}...')


async def main():
    await agent.run('Do the task', event_stream_handler=stream_handler)

Deferred tool calls also surface as batch-level events: DeferredToolRequestsEvent (once per batch of deferred calls, before any HandleDeferredToolCalls handler runs) and DeferredToolResultsEvent (when a handler resolves requests inline). Use these to tell a frontend the run is paused waiting for approvals or external calls.

To surface progress or intermediate results from an async tool into the same event stream without polluting the model's context, define a dataclass subclass of CustomEvent (its fields are the payload; the event name derives from the class name) and await ctx.emit(event). Sync tools cannot emit events. It reaches the event_stream_handler, run_stream_events(), iter() streaming, and the AG-UI/Vercel AI adapters; when emitted from a tool, its tool_call_id and tool_name are auto-stamped, and consumers use isinstance() against the class. Code driving agent.iter() can inject events by awaiting AgentRun.emit(). The payload can't reuse the envelope's own field names: data, tool_call_id, tool_name, and event_kind are rejected at class definition.

CustomEvent is for application-owned code only. Code that lives inside a capability must define namespaced CapabilityEvent subclasses instead; emitting either family from the other's side raises UserError. See CAPABILITIES-AND-HOOKS.md.

Custom events reach the AG-UI and Vercel AI frontends by default. For an event that should stay server-side (metrics, audit logs), declare the class ui=False — class IndexProgressEvent(CustomEvent, ui=False) — and every UI adapter skips it while in-process consumers still receive it. Declaring a ui field or ClassVar on an event class is rejected, since it would shadow that flag. The flag is class-level, not on the wire, so adapters also skip an UnknownCustomEvent (a class this process never imported): when events reach the frontend from another process, import their defining modules there or none of them are forwarded.

from dataclasses import dataclass

from pydantic_ai import Agent, CustomEvent, RunContext

agent = Agent('openai:gpt-5.2', name='progress_agent')


@dataclass(kw_only=True)
class ProgressEvent(CustomEvent):
    done: int
    total: int


@agent.tool
async def process(ctx: RunContext, count: int) -> str:
    for i in range(count):
        await ctx.emit(ProgressEvent(done=i + 1, total=count))
    return 'done'

Handle Provider Failures

Use FallbackModel when the user wants automatic provider or model failover.

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

fallback = FallbackModel(
    OpenAIChatModel('gpt-5.2'),
    AnthropicModel('claude-sonnet-4-6'),
)

agent = Agent(fallback, name='fallback_agent')

Good defaults:

• primary expensive/strong model, cheaper fallback for resilience
• same prompt/output contract across both models
• per-model settings only when the user actually needs them