pydantic_ai.capabilities

aexit

@async

def __aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Once the outermost async with on this event loop exits, fail its queued requests and cancel its running submissions and checks.

Returns

MCP

Bases: NativeOrLocalTool[AgentDepsT]

MCP server capability.

The primary entry point for using MCP servers with Pydantic AI. Runs the MCP server locally — keeps credentials, hooks, and tracing under your control — and accepts any MCPToolset input (URL, fastmcp.Client, transport, in-process FastMCP server, script path, etc.) directly via local=.

Pass url= for HTTP-based servers; the same URL can also be advertised to providers that support native MCP via native=True. For non-URL local clients, omit url= and pass the client/toolset as local=. Pass native=True, local=False for strict native-only (no local at all — works without the mcp extra).

Attributes

url

The URL of the MCP server.

Required when using native MCP. Optional when using a local-only client via local=.

Type: str | None Default: url

authorization_token

Authorization header value for MCP server requests. Passed to both native and local.

Pass a function of the run context instead to connect each agent run as its own user, for example with a token from the run’s deps: it’s called at the start of each run, and the token it returns is used for that run on both the native and the local side. Returning None sends no Authorization header.

Type: str | MCPAuthorizationTokenFunc[AgentDepsT] | None Default: authorization_token

headers

HTTP headers for MCP server requests. Passed to both native and local.

Like authorization_token, this can be a function of the run context, called at the start of each run.

Type: dict[str, str] | MCPHeadersFunc[AgentDepsT] | None Default: headers

allowed_tools

Filter to only these tools. Applied to both native and local.

Type: list[str] | None Default: allowed_tools

tool_prefix

Prepended with _ to the names of the server’s tools, as in <prefix>_<tool>, so they don’t collide with other tools.

None (default) or False: keep the names the server gives its tools.
True: use the explicit id as the prefix. Requires id=; an id derived from the URL is never used.
A string: use that string as the prefix.

Applies to the local tools. allowed_tools keeps matching the server’s own names, and the toolset’s id doesn’t change. Native MCP tools are named by the model provider and aren’t prefixed.

Type: str | bool | None Default: tool_prefix

description

Description of the MCP server. Native-only; ignored by local tools.

Type: str | None Default: description

include_instructions

Whether to add the server’s instructions to the agent’s instructions.

On by default. The server writes this text, and the model receives it with the same authority as instructions you write yourself, so set this to False for servers you don’t trust, such as ones your end users supply.

Applies to the local toolset the capability builds from url= or a non-toolset local= input. A pre-built MCPToolset passed as local= keeps its own include_instructions, and native MCP is handled by the model provider.

Type: bool Default: include_instructions

Methods

from_spec

@classmethod

def from_spec(
    cls,
    url: str,
    *,
    native: MCPServerTool | bool = False,
    local: str | bool | None = None,
    id: str | None = None,
    authorization_token: str | None = None,
    headers: dict[str, str] | None = None,
    allowed_tools: list[str] | None = None,
    tool_prefix: str | bool | None = None,
    description: str | None = None,
    include_instructions: bool = True,
    defer_loading: bool = False,
) -> MCP[AgentDepsT]

Construct an MCP capability from spec-serializable args.

Restricts the runtime-wide local= union to the JSON/YAML-serializable subset (str | bool | None) so AgentSpec schema generation works, and requires url= (which is optional at runtime when local= is a concrete non-URL client). Non-serializable runtime values like fastmcp.Client, ClientTransport, or pre-built MCPToolset instances can still be passed to MCP(...) directly — they just can’t roundtrip through a spec file.

Returns

MCP[AgentDepsT]

ToolSearch

Bases: AbstractCapability[AgentDepsT]

Capability that provides tool discovery for large toolsets.

Tools marked with defer_loading=True are hidden from the model until discovered. Auto-injected into every agent — zero overhead when no deferred tools exist.

When the model supports native tool search (Anthropic BM25/regex, OpenAI Responses), discovery is handled by the provider: the deferred tools are sent with defer_loading on the wire and the provider exposes them once they’ve been discovered. Otherwise, discovery happens locally via a search_tools function that the model can call.

On providers that support a native “client-executed” surface (Anthropic, OpenAI), the discovery message is delivered append-only — prompt cache is preserved across discovery turns, so growing the message history with discovered-tool results does not invalidate the cached prefix.

from collections.abc import Sequence

from pydantic_ai import Agent, RunContext, Tool
from pydantic_ai.capabilities import ToolSearch
from pydantic_ai.tools import ToolDefinition


# Tools become deferred via `defer_loading=True`. They stay hidden from the model
# until tool search discovers them.
def get_weather(city: str) -> str:
    ...


weather_tool = Tool(get_weather, defer_loading=True)

# Default: native search on supporting providers, local keyword matching elsewhere.
agent = Agent('anthropic:claude-sonnet-4-6', tools=[weather_tool], capabilities=[ToolSearch()])

# Force a specific Anthropic native strategy; errors on providers that can't honor it.
agent = Agent(
    'anthropic:claude-sonnet-4-6',
    tools=[weather_tool],
    capabilities=[ToolSearch(strategy='regex')],
)

# Always run the local keyword-overlap algorithm, regardless of provider.
agent = Agent(
    'anthropic:claude-sonnet-4-6',
    tools=[weather_tool],
    capabilities=[ToolSearch(strategy='keywords')],
)

# Custom search function — used locally, and by provider-native "client-executed"
# modes when supported.
def my_search(
    ctx: RunContext, queries: Sequence[str], tools: Sequence[ToolDefinition]
) -> list[str]:
    return [
        t.name
        for t in tools
        if any(q.lower() in (t.description or '').lower() for q in queries)
    ]

agent = Agent(
    'anthropic:claude-sonnet-4-6',
    tools=[weather_tool],
    capabilities=[ToolSearch(strategy=my_search)],
)

Attributes

strategy

The search strategy to use.

None (default): let Pydantic AI pick the best strategy for the current provider — native on supporting models (Anthropic BM25, OpenAI server-executed tool search), local keyword matching elsewhere. On the legacy AsyncAnthropicBedrock client it’s Anthropic regex, so BM25 there needs an explicit 'bm25'. The choice may change in future versions.
'keywords': always use the local keyword-overlap algorithm. Still prompt-cache compatible on providers that expose a “client-executed” native surface (Anthropic, OpenAI): the algorithm rides the same defer_loading wire as a custom callable, so the tool list stays stable across discovery rounds and the cached prefix is preserved.
'bm25' / 'regex': force a specific Anthropic native strategy. Raises on providers that can’t honor the choice (including OpenAI, which has no named native strategies).
Callable (ctx, queries, tools) -> names: custom search function (sync or async). Used locally, and by the native “client-executed” surface on providers that support it (Anthropic custom tool-reference blocks, OpenAI execution='client').

Type: ToolSearchStrategy[AgentDepsT] | None Default: None

max_results

Maximum number of matches returned by the local search algorithm.

Type: int Default: 10

tool_description

Custom description for the model-facing search tool when search runs on our side.

Used for the local search_tools fallback and for providers with client-executed native tool search.

Type: str | None Default: None

parameter_description

Custom description for the queries parameter when search runs on our side.

Type: str | None Default: None

id

One-off: an agent has a single tool-discovery configuration, so the id is fixed by default.

Two of them resolve to one via combine. Pass a distinct id to keep both, or id=None for derived ids.

Type: str | None Default: 'tool_search'

function_tool_name

Reserved name of the local function tool used when tool search runs client-side.

Type: str Default: 'search_tools'

Methods

before_model_request

@async

def before_model_request(
    ctx: RunContext[AgentDepsT],
    request_context: ModelRequestContext,
) -> ModelRequestContext

Record tools unlocked by a capability load.

Returns

FileUnderstanding

Bases: AbstractCapability[AgentDepsT]

Describe the files the agent’s model cannot read, with a model that can.

Before each attempt at a request, every image, document, audio file or video in a user prompt or a tool result that the ModelProfile of the model serving that attempt says it does not accept is replaced by a text description of it, written by fallback_subagent_model. The model sees the description in the file’s place; files it accepts are sent as they are. With ModelFallback, each candidate is judged by its own profile, so one that reads the file gets it and only one that can’t gets the description. Without this capability, a file a tool returned that the model cannot read is replaced by a note saying so instead.

from pydantic_ai import Agent, DocumentUrl
from pydantic_ai.capabilities import FileUnderstanding

agent = Agent(
    'typesafe:jev-latest',
    output_type=bool,
    instructions='Is this document about animals?',
    capabilities=[FileUnderstanding(fallback_subagent_model='openai:gpt-5.6-sol')],
)
result = agent.run_sync([DocumentUrl('https://example.com/whatever.pdf')])

A text-like document (plain text, CSV, JSON, XML, YAML) is not described but inlined as it is, the way models that read text files do. A provider file reference (UploadedFile) is left alone.

A file is described once per conversation. A file given by URL is identified by its URL, and bytes by their content and media type. The description is recorded in the message history, under the framework’s reserved __pydantic_ai__ namespace in the metadata of the request that holds the file (which is not sent to the model), and later requests use that text instead of describing the file again. The downloaded text of a text document given by URL is recorded the same way, so it is not downloaded again. A turn that continues from message_history, another worker serving the next turn, and a durable run replaying its history all see the same description, so earlier requests keep the same content and the describer is not paid for twice. The UI adapters leave that namespace out of the history they send to a client, and sanitize_messages drops it from history a client sends back, so a conversation whose history the client holds describes the file again each turn, and a new description can differ, which changes earlier requests and costs the cached prefix.

The distinct files a request needs described are described concurrently. Each describing call is a run of its own, but its requests, tokens and cost are added to the agent run’s usage, and the run’s usage limits are checked before any is sent, as if the describing requests were the run’s next ones.

Attributes

fallback_subagent_model

The model of the subagent that describes the files, with its provider: 'openai:gpt-5.6-sol', or a Model.

Named like ImageGeneration and XSearch’s fallback_subagent_model: a separate agent run, not a replacement for the agent’s model.

Type: Model | KnownModelName | str

instructions

What to write about each file. The default asks for a description someone who cannot see the file could answer questions from.

Type: str | None Default: None

id

One-off: an agent describes files one way, so the id is fixed by default.

Two of them resolve to one via combine, which keeps the last. Pass a distinct id to keep both, or id=None for derived ids.

Type: str | None Default: 'file_understanding'

Methods

prepare_model_request

@async

def prepare_model_request(
    ctx: RunContext[AgentDepsT],
    request_context: ModelRequestContext,
) -> ModelRequestContext

Describe the files the model serving this attempt can’t read, and send the descriptions in their place.

Each attempt is judged by its own model, so with ModelFallback a candidate that reads a file gets the file and only one that can’t gets its description.

Returns

PrepareOutputTools

Bases: AbstractCapability[AgentDepsT]

Capability that filters or modifies output tool definitions using a callable.

Mirrors PrepareTools for output tools. ctx.retry/ctx.max_retries reflect the output retry budget (max_output_retries), matching the output hook lifecycle.

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import PrepareOutputTools
from pydantic_ai.output import ToolOutput
from pydantic_ai.tools import ToolDefinition


async def only_after_first_step(
    ctx: RunContext, tool_defs: list[ToolDefinition]
) -> list[ToolDefinition]:
    return tool_defs if ctx.run_step > 0 else []


agent = Agent(
    'openai:gpt-5',
    output_type=ToolOutput(str),
    capabilities=[PrepareOutputTools(only_after_first_step)],
)

Capability

Bases: AbstractCapability[AgentDepsT]

Convenience capability for bundling instructions, tools, and toolsets without subclassing.

This groups related instructions, descriptions, function tools, and toolsets under a capability identity. Instructions passed via instructions= are available through get_instructions(); instructions is the decorator for registering instruction functions. The constructor accepts static or callable description= values. For model settings, lifecycle hooks, native tools, wrapper toolsets, or custom per-run logic, subclass AbstractCapability.

Attributes

description

Static description mirrored on the instance.

The constructor also accepts callable descriptions, stored internally and returned from get_description().

Type: str | None Default: description if isinstance(description, str) else None

toolsets

Toolsets to register with the agent. Combined via CombinedToolset when more than one is provided.

Type: Sequence[AgentToolset[AgentDepsT]] Default: resolved_toolsets

tools

Function tools to register with the agent.

Type: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, …]] Default: tools

Methods

init

def __init__(
    *,
    instructions: AgentInstructions[AgentDepsT] | None = None,
    toolsets: Sequence[AgentToolset[AgentDepsT]] | None = None,
    tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] = (),
    id: str | None = None,
    description: CapabilityDescription[AgentDepsT] | None = None,
    defer_loading: bool = False,
) -> None

Build a capability from instructions, tools, toolsets, and an optional description.

Returns

Parameters

instructions : AgentInstructions[AgentDepsT] | None Default: None

Static instructions and/or instruction function(s), available via get_instructions(). Pass an InstructionPart to declare a part’s name or mark it dynamic. Register more with the instructions decorator.

toolsets : Sequence[AgentToolset[AgentDepsT]] | None Default: None

Toolsets to register with the agent.

tools : Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, …]] Default: ()

Function tools to register with the agent.

id : str | None Default: None

Stable identifier for the capability. Required when defer_loading=True, so the model’s load_capability call can reference it.

description : CapabilityDescription[AgentDepsT] | None Default: None

Static string or callable description, returned from get_description(). For a deferred capability it is shown to the model so it can decide whether to load it.

defer_loading : bool Default: False

When True, the capability’s tools and instructions stay hidden until the model loads it on demand via the load_capability tool; requires id.

tool_plain

def tool_plain(func: ToolFuncPlain[ToolParams], /) -> ToolFuncPlain[ToolParams]
def tool_plain(
    *,
    name: str | None = None,
    description: str | None = None,
    retries: int | None = None,
    prepare: ToolPrepareFunc[AgentDepsT] | None = None,
    args_validator: ArgsValidatorFunc[AgentDepsT, ...] | None = None,
    docstring_format: DocstringFormat = 'auto',
    require_parameter_descriptions: bool = False,
    schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
    strict: bool | None = None,
    sequential: bool = False,
    requires_approval: bool = False,
    metadata: dict[str, Any] | None = None,
    tool_kind: ToolKind[KindArgsT, KindContentT],
    timeout: float | None = None,
    defer_loading: bool = False,
    include_return_schema: bool | None = None,
) -> KindedToolPlainDecorator[KindArgsT, KindContentT]
def tool_plain(
    *,
    name: str | None = None,
    description: str | None = None,
    retries: int | None = None,
    prepare: ToolPrepareFunc[AgentDepsT] | None = None,
    args_validator: ArgsValidatorFunc[AgentDepsT, ToolParams] | None = None,
    docstring_format: DocstringFormat = 'auto',
    require_parameter_descriptions: bool = False,
    schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
    strict: bool | None = None,
    sequential: bool = False,
    requires_approval: bool = False,
    metadata: dict[str, Any] | None = None,
    tool_kind: ToolPartKind | None = None,
    timeout: float | None = None,
    defer_loading: bool = False,
    include_return_schema: bool | None = None,
) -> Callable[[ToolFuncPlain[ToolParams]], ToolFuncPlain[ToolParams]]

Decorator to register a plain (no-RunContext) function tool on this capability.

Mirrors Agent.tool_plain: the tool is added to this capability’s function toolset and registered with the agent whenever the capability is active.

Returns

tool

def tool(
    func: ToolFuncContext[AgentDepsT, ToolParams],
    /,
) -> ToolFuncContext[AgentDepsT, ToolParams]
def tool(
    *,
    name: str | None = None,
    description: str | None = None,
    retries: int | None = None,
    prepare: ToolPrepareFunc[AgentDepsT] | None = None,
    args_validator: ArgsValidatorFunc[AgentDepsT, ...] | None = None,
    docstring_format: DocstringFormat = 'auto',
    require_parameter_descriptions: bool = False,
    schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
    strict: bool | None = None,
    sequential: bool = False,
    requires_approval: bool = False,
    metadata: dict[str, Any] | None = None,
    tool_kind: ToolKind[KindArgsT, KindContentT],
    timeout: float | None = None,
    defer_loading: bool = False,
    include_return_schema: bool | None = None,
) -> KindedToolDecorator[AgentDepsT, KindArgsT, KindContentT]
def tool(
    *,
    name: str | None = None,
    description: str | None = None,
    retries: int | None = None,
    prepare: ToolPrepareFunc[AgentDepsT] | None = None,
    args_validator: ArgsValidatorFunc[AgentDepsT, ToolParams] | None = None,
    docstring_format: DocstringFormat = 'auto',
    require_parameter_descriptions: bool = False,
    schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
    strict: bool | None = None,
    sequential: bool = False,
    requires_approval: bool = False,
    metadata: dict[str, Any] | None = None,
    tool_kind: ToolPartKind | None = None,
    timeout: float | None = None,
    defer_loading: bool = False,
    include_return_schema: bool | None = None,
) -> Callable[[ToolFuncContext[AgentDepsT, ToolParams]], ToolFuncContext[AgentDepsT, ToolParams]]

Decorator to register a function tool (taking RunContext) on this capability.

Mirrors Agent.tool: the tool is added to this capability’s function toolset and registered with the agent whenever the capability is active.

Returns

instructions

def instructions(
    func: Callable[[RunContext[AgentDepsT]], str | None],
    /,
) -> Callable[[RunContext[AgentDepsT]], str | None]
def instructions(
    func: Callable[[RunContext[AgentDepsT]], Awaitable[str | None]],
    /,
) -> Callable[[RunContext[AgentDepsT]], Awaitable[str | None]]
def instructions(func: Callable[[], str | None], /) -> Callable[[], str | None]
def instructions(
    func: Callable[[], Awaitable[str | None]],
    /,
) -> Callable[[], Awaitable[str | None]]
def instructions(
    func: Callable[[RunContext[AgentDepsT]], str | InstructionPart | None],
    /,
) -> Callable[[RunContext[AgentDepsT]], str | InstructionPart | None]
def instructions(
    func: Callable[[RunContext[AgentDepsT]], Awaitable[str | InstructionPart | None]],
    /,
) -> Callable[[RunContext[AgentDepsT]], Awaitable[str | InstructionPart | None]]
def instructions(
    func: Callable[[], str | InstructionPart | None],
    /,
) -> Callable[[], str | InstructionPart | None]
def instructions(
    func: Callable[[], Awaitable[str | InstructionPart | None]],
    /,
) -> Callable[[], Awaitable[str | InstructionPart | None]]
def instructions(
    *,
    name: str | None = None,
    on_change: InstructionChangePolicy = 'rewrite',
) -> Callable[[InstructionFunc[AgentDepsT]], InstructionFunc[AgentDepsT]]

Decorator to register an instructions function on this capability.

Mirrors Agent.instructions: the function may take RunContext (or no arguments), may be sync or async, and is appended to any instructions provided via the instructions= field.

Example:

from pydantic_ai import RunContext
from pydantic_ai.capabilities import Capability

cap = Capability[str](instructions='base instructions')

@cap.instructions
async def dynamic(ctx: RunContext[str]) -> str:
    return f'extra: {ctx.deps}'

Returns

Callable[[InstructionFunc[AgentDepsT]], InstructionFunc[AgentDepsT]] | InstructionFunc[AgentDepsT]

Parameters

func : InstructionFunc[AgentDepsT] | None Default: None

The instructions function to register.

name : str | None Default: None

An optional name for the instruction part this function produces, keyed as 'capability:<capability id>:<name>' on InstructionPart.id so an application can address this part specifically, where the capability’s own key addresses everything it contributes. Requires the capability to have an id — without one there is no source key to qualify the name against, so the part stays unaddressable. See instruction parts.

on_change : InstructionChangePolicy Default: 'rewrite'

Whether to rewrite the instruction prefix (the default) or append full replacements when this block changes. Appending requires an addressable instruction identity; otherwise it warns and rewrites. The function still runs on every request. Use 'append' for instructions that change mid-conversation, to keep the cached prefix; see appending instruction changes. Build the text only from state you control; see keep untrusted text out of instructions.

WrapperCapability

Bases: AbstractCapability[AgentDepsT]

A capability that wraps another capability and delegates all methods.

Analogous to WrapperToolset for toolsets. Subclass and override specific methods to modify behavior while delegating the rest.

When the wrapped capability returns a fresh instance from for_agent or for_run, the wrapper is rebound as a shallow copy holding the new wrapped: subclass state is carried over verbatim and __init__/__post_init__ are not re-run. Compute values derived from wrapped on access (e.g. via a property) rather than caching them at construction, so they can’t go stale across a rebind.

Methods

visit_and_replace

def visit_and_replace(
    visitor: Callable[[AbstractCapability[AgentDepsT]], AbstractCapability[AgentDepsT] | None],
) -> AbstractCapability[AgentDepsT] | None

Visit the wrapper first; a replaced or removed wrapper takes its subtree with it.

When the wrapper survives, the visit descends into wrapped and this wrapper is rebuilt around whatever remains; see AbstractCapability.visit_and_replace for the tree-walking contract.

Returns

AbstractCapability[AgentDepsT] | None

ModelFallback

Bases: AbstractCapability[AgentDepsT]

Attempt other models when the one serving a request fails or returns a rejected response.

The chain is the model selected for the step followed by this capability’s models, so it composes with an agent-level model and with SelectModel rather than replacing them:

from pydantic_ai import Agent
from pydantic_ai.capabilities import ModelFallback

# Try gpt-6.1-sol, then Fable 5.1.
agent = Agent('openai:gpt-6.1-sol', capabilities=[ModelFallback('anthropic:claude-fable-5-1')])

# No agent model: the first candidate is used for the first attempt.
agent = Agent(capabilities=[ModelFallback('openai:gpt-6.1-sol', 'anthropic:claude-fable-5-1')])

A model already attempted this step is skipped, so listing the agent’s own model as the first candidate is harmless rather than a wasted duplicate attempt.

Each attempt re-runs prepare_model_request for the candidate about to serve it, so the model’s own message preparation and token counting, and any preparation a capability does there, are redone per model instead of inherited. (The built-in compaction capabilities still run once per step, in before_model_request.) The request step itself is not repeated: before_model_request and wrap_model_request run once, and the model never sees a retry prompt.

With sticky=True, a conversation that fell back stays on the model that answered, in later steps and later runs continuing its message history, so the provider’s prompt cache keeps being hit and the failing model isn’t asked first every step.

Prefer this over FallbackModel inside an agent. FallbackModel remains the way to get the same behavior from a bare Model, outside an agent run.

Attributes

models

The models to attempt, in order, after the one selected for the step.

Type: list[Model | KnownModelName | str] Default: list(models)

sticky

Whether a conversation stays on the model that last answered it after falling back.

Type: bool Default: sticky

Methods

init

def __init__(
    *models: Model | KnownModelName | str,
    fallback_on: FallbackOn = (ModelAPIError, ContentFilterError, OutputTokenLimitExceeded),
    sticky: bool = False,
)

Initialize the fallback capability.

Parameters

models : Model | KnownModelName | str Default: ()

The models to attempt after the one selected for the request step. When the agent has no model of its own, the first of these is used for the first attempt.

fallback_on : FallbackOn Default: (ModelAPIError, ContentFilterError, OutputTokenLimitExceeded)

Conditions that trigger fallback to the next model. Defaults to (ModelAPIError, ContentFilterError, OutputTokenLimitExceeded): any API error, any refusal, meaning a response with finish_reason='content_filter', with or without partial text, and any response the output token limit cut off, meaning one with finish_reason='length'. ContentFilterError and OutputTokenLimitExceeded are only raised after a response is accepted, so in fallback_on they stand for that response. Leave ContentFilterError out to accept refusals instead: an empty one ends the run with ContentFilterError, and one with text is the output. Leave OutputTokenLimitExceeded out to accept cut-off responses, as when max_tokens is a deliberate cap: cut-off text is then the output. Accepts:

A tuple of exception types: (ModelRateLimitError, ModelUnavailableError)
An exception handler (sync or async): lambda exc: isinstance(exc, MyError)
A response handler (sync or async): def check(r: ModelResponse) -> bool
A sequence mixing all of the above: [ModelAPIError, exc_handler, response_handler]

Handler type is auto-detected by inspecting type hints on the first parameter. If the first parameter is hinted as ModelResponse, it’s a response handler. Otherwise (including untyped handlers and lambdas), it’s an exception handler.

Response handlers only apply to non-streamed requests: a streamed response has already reached the consumer by the time it could be judged, so it’s accepted as is, with a UserWarning. A stream that fails to open, or whose provider reports an error before its first event, still falls back. A streamed refusal or cut-off response is accepted too, without a warning: a stream’s finish reason is only known once it was streamed.

sticky : bool Default: False

Whether a conversation that fell back keeps using the candidate that answered, instead of starting each request step from the model selected for it. The agent records the model that answered as the response’s model_id, and the attempts that failed before it as failed_attempts, so this also holds for a later run continuing that history. Only a fallback makes the conversation sticky: an answer from a candidate chosen for its run, with no attempt failing before it, doesn’t. It only replaces the agent’s default model: a model passed to run(model=...) or override(model=...), or selected by another capability, such as SelectModel, is still attempted first. If the sticky candidate fails, the chain moves on to the models after it, then the step’s own model and the models before it.

for_agent

def for_agent(agent: AbstractAgent[AgentDepsT, Any]) -> ModelFallback[AgentDepsT]

Decide whether this capability also has to supply the agent’s model.

A capability’s model contribution outranks the agent’s own default, so contributing unconditionally would silently redirect Agent(model, capabilities=[ModelFallback(...)]) to the first fallback candidate instead of falling back to it. Only step in when the agent has no model of its own — a run-level model= or override(model=) outranks a contribution anyway, so those keep working either way.

Returns

ModelFallback[AgentDepsT]

for_run

@async

def for_run(ctx: RunContext[AgentDepsT]) -> ModelFallback[AgentDepsT]

Return a fresh copy so each run tracks its own chain position.

Returns

ModelFallback[AgentDepsT]

prepare_model_request

@async

def prepare_model_request(
    ctx: RunContext[AgentDepsT],
    request_context: ModelRequestContext,
) -> ModelRequestContext

Note the model about to be attempted, routing the step to the candidate the conversation is pinned to.

That’s the candidate that started a suspended response being resumed, or, with sticky, the one that last answered. Every attempt is noted here, whichever capability asked for it, so a model that was already attempted this step is never picked again.

Returns

from_spec

@classmethod

def from_spec(
    cls,
    models: KnownModelName | str | Sequence[KnownModelName | str],
    *,
    fallback_on: _SpecFallbackError | Sequence[_SpecFallbackError] = ('ModelAPIError', 'ContentFilterError', 'OutputTokenLimitExceeded'),
    sticky: bool = False,
) -> ModelFallback[Any]

Build from an agent spec, which names the models and the exceptions to fall back on.

Model names are kept as given and resolved when the agent attempts them, as a model name passed to ModelFallback(...) is, so a ResolveModelId capability’s aliases work in a spec too. fallback_on takes the names of exception classes in pydantic_ai.exceptions that subclass ModelAPIError, ContentFilterError or OutputTokenLimitExceeded; exception and response handlers can only be passed to ModelFallback(...) directly.

YAML form:

capabilities:

ModelFallback: [anthropic:claude-opus-5-5, openai:gpt-6.1-sol]
ModelFallback: models: [anthropic:claude-opus-5-5] fallback_on: [ModelRateLimitError, ModelUnavailableError] sticky: true

Returns

ModelFallback[Any]

CombinedCapability

Bases: AbstractCapability[AgentDepsT]

A capability that combines multiple capabilities.

A nested CombinedCapability is only a grouping: its children are merged into the enclosing one. Give it an id and defer_loading=True to load its children together instead: the model loads the group by its id, and loading it activates every child at once. Each child still keeps its own ordering position among all of the agent’s capabilities, as if the group were merged.

When any child returns a fresh instance from for_agent or for_run, the container is rebound via _rebound: a shallow copy holding the new children, with subclass state carried over verbatim and __init__/__post_init__ not re-run. Compute values derived from capabilities on access (e.g. via a property) rather than caching them at construction, so they can’t go stale across a rebind. _instruction_sources is the one thing that can’t be — flattening destroys what it records — so it’s carried across by _rebound instead, and swapping children any other way (replace()) silently rebuilds it from the flattened list.

Methods

visit_and_replace

def visit_and_replace(
    visitor: Callable[[AbstractCapability[AgentDepsT]], AbstractCapability[AgentDepsT] | None],
) -> AbstractCapability[AgentDepsT] | None

Visit each child and rebuild the container from the survivors.

A child the visitor removed is reported to _rebound so the composition view drops it too; see AbstractCapability.visit_and_replace for the tree-walking contract.

Returns

AbstractCapability[AgentDepsT] | None

HookTimeoutError

Bases: AgentRunError, TimeoutError

Raised when a hook function exceeds its configured timeout.

Instrumentation

Bases: AbstractCapability[Any]

Capability that instruments agent runs with OpenTelemetry/Logfire tracing.

When added to an agent via capabilities=[Instrumentation(...)], this capability creates OpenTelemetry spans for the agent run, model requests, and tool executions.

Other capabilities can add attributes to these spans using the OpenTelemetry API (opentelemetry.trace.get_current_span().set_attribute(key, value)).

Prompt-cache health is recorded on model-request spans using the pydantic_ai.cache.hit_ratio, pydantic_ai.cache.established_tokens, pydantic_ai.cache.collapsed, pydantic_ai.cache.missed_tokens, and pydantic_ai.cache.collapse_reason attributes. Collapses are classified as unexpected, ttl_expired, compacted, unknown, or unreported; only unexpected collapses emit a pydantic_ai.cache.collapse span event, so the event means the cacheable prefix moved while it should still have been warm. A sustained collapse emits the event once, until a healthy read-back re-stabilizes the cache. The established prefix is tracked per conversation and per provider, endpoint, and model, so the first request of a run that continues a conversation is judged against what the previous run cached. A request long enough to cache on a model that needs prompt caching configured, but has none configured, sets pydantic_ai.cache.not_enabled and emits a pydantic_ai.cache.not_enabled span event, once per conversation. An Anthropic request with cache breakpoints records the ones it asked for and the ones that fit the provider’s limit as pydantic_ai.cache.requested_breakpoints and pydantic_ai.cache.applied_breakpoints, as paths into the request as sent to Anthropic (after consecutive request parts are merged into one message), not into the message history.

Attributes

settings

OTel/Logfire instrumentation settings. Defaults to InstrumentationSettings(), which uses the global TracerProvider (typically configured by logfire.configure()).

Type: InstrumentationSettings Default: field(default_factory=(lambda: _default_settings()))

id

One-off: an agent has a single instrumentation configuration, so the id is fixed by default.

Two of them resolve to one via combine, which keeps the last. Pass a distinct id to keep both, or id=None for derived ids.

Type: str | None Default: 'instrumentation'

Methods

from_spec

@classmethod

def from_spec(
    cls,
    *,
    include_binary_content: bool = True,
    include_content: bool = True,
    include_model_request_parameters: bool = True,
    version: Literal[2, 3, 4, 5, 6] = DEFAULT_INSTRUMENTATION_VERSION,
    use_aggregated_usage_attribute_names: bool = True,
) -> Instrumentation

Build an Instrumentation capability from a YAML/JSON spec.

Accepts every serializable InstrumentationSettings option. The OTel tracer_provider and meter_provider fields can’t be expressed in YAML and default to the global providers (typically configured via logfire.configure()).

id is deliberately not accepted. An agent has one instrumentation configuration — which is what the class-level default id says — so there is nothing for a spec to name, and two Instrumentation capabilities resolve to one rather than colliding.

YAML form:

capabilities:

Instrumentation: {} # default settings
Instrumentation: version: 2 include_content: false

Returns

Instrumentation

for_run

@async

def for_run(ctx: RunContext[Any]) -> Instrumentation

Return a fresh copy for per-run state isolation.

Returns

Instrumentation

on_tool_validate_error

@async

def on_tool_validate_error(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: RawToolArgs,
    error: ValidationError | ModelRetry,
) -> ValidatedToolArgs

Emit an error span for a tool call whose argument validation failed.

Runs only after every other capability has declined to recover the error, so a recovered validation failure produces no span. The span keeps the execute_tool operation name so tracing backends group it with other tool spans, and sets pydantic_ai.tool.failure_stage: 'validation' to distinguish it from execution failures.

The span also carries the retry the failure becomes, as the pydantic_ai.tool.retry.* attributes: its cause, and its attempt against the tool’s budget unless that budget is exhausted, in which case the run ends instead and only the cause is recorded.

With content capture enabled, the span records the retried tool result built from the error. That is the exact result the model receives when the agent loop handles the failure; raw-mode callers (e.g. sandboxed dispatch via handle_call(wrap_validation_errors=False)) surface the raw exception to the calling code instead, and the recorded result is just the rendered description of the failure.

Returns

ValidatedToolArgs

wrap_output_process

@async

def wrap_output_process(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: Any,
    handler: WrapOutputProcessHandler,
) -> Any

Emit a span for output-function execution.

Output processing for plain validation (no function) is not span-worthy — the validated value is the model’s response itself, no user code ran. We open a span only when an output function will execute, regardless of whether the output arrived via a tool call. The span name reflects the function (or tool name when the function name is unavailable, e.g. union processors).

Returns

ImageGeneration

Bases: NativeOrLocalTool[AgentDepsT]

Image generation capability.

Uses the model’s native image generation when available. When the model doesn’t support it, the direct image generation API can take over. Which field it goes on follows what you hand over: an ImageGenerator carries settings of its own, so it goes on local beside the other implementations you supply; a bare ImageGenerationModel or a 'provider:model' name goes on fallback_image_model.

The fallback_subagent_model path is the other way to cover such a model: it runs an additional agent on an image-capable conversational model, so the image comes from that model’s native ImageGenerationTool. Use it when you want those native tool semantics. local also takes a fallback tool you write yourself. The three fields are alternatives: stating more than one raises UserError.

Portable dimensions and aspect_ratio settings are applied to the direct fallback using ImageGenerationSettings. Other fields configure the native ImageGenerationTool; configure provider-specific direct settings on an explicit ImageGenerator or ImageGenerationModel.

When passing a custom native instance or factory, its settings are also used for the fallback_subagent_model subagent; capability-level fields override any native settings. A static instance’s aspect_ratio is also inherited by the direct fallback.

Attributes

id

One-off: an agent searches, fetches or generates one way, so the id is fixed.

Declared here rather than only as an __init__ default so the class states it where _declares_default_id — and a reader — can see it.

Type: str | None Default: id

fallback_subagent_model

Model for a subagent to run when the agent’s model doesn’t support image generation natively.

Must be a model that supports image generation via the ImageGenerationTool native tool. This requires a conversational model with image generation support, not a dedicated image-only API. Examples:

'openai-responses:gpt-5.4' — OpenAI model with image generation support
'google:gemini-3-pro-image' — Google image generation model

Can be a model name string, Model instance, or a callable taking RunContext that returns a Model instance or model name string.

The model is kept as declared; the generate_image tool is derived from it and the capability’s settings each time the toolset is requested.

Type: ImageGenerationFallbackModel Default: fallback_model if fallback_model is not None else fallback_subagent_model

fallback_image_model

Direct image model to generate with when the agent’s model doesn’t support it natively.

Takes an ImageGenerationModel or a 'provider:model' string; a string without a provider prefix raises UserError, and so does an ImageGenerator, which carries settings of its own and belongs on local. The tool calls the direct image generation API rather than running a second agent, which is what fallback_subagent_model does.

Note which of the two image-model fields applies to which path: image_model names the model within the provider’s native tool and is unprefixed ('gpt-image-2'), while this one selects the direct model and carries the provider ('openai:gpt-image-2'). The direct fallback ignores image_model with a warning.

The model is kept as declared; the generate_image tool is derived from it and the capability’s settings each time the toolset is requested.

Type: ImageGenerationModel | str | None Default: fallback_image_model

action

Whether to generate a new image or edit an existing image.

Supported by: OpenAI Responses. Default: 'auto'.

The direct generator receives no reference images, so 'edit' raises UserError: at construction with native=False, and when the tool runs otherwise.

Type: Literal[‘generate’, ‘edit’, ‘auto’] | None Default: action

background

Background type for the generated image.

Supported by: OpenAI Responses.

The direct generator ignores it; set the provider-prefixed equivalent on the generator instead.

Type: Literal[‘transparent’, ‘opaque’, ‘auto’] | None Default: background

input_fidelity

Input fidelity for matching style/features of input images.

Supported by: OpenAI Responses. Default: 'low'.

The direct generator ignores it; set the provider-prefixed equivalent on the generator instead.

Type: Literal[‘high’, ‘low’] | None Default: input_fidelity

moderation

Moderation level for the generated image.

Supported by: OpenAI Responses.

The direct generator ignores it; set the provider-prefixed equivalent on the generator instead.

Type: Literal[‘auto’, ‘low’] | None Default: moderation

image_model

The image generation model to use.

Supported by: OpenAI Responses.

The direct fallback ignores it with a warning, because the generator on local or the model on fallback_image_model already names the model it generates with. image_model is unprefixed and names the model inside the native tool.

Type: ImageGenerationModelName | None Default: image_model

output_compression

Compression level for the output image.

Supported by: OpenAI Responses (jpeg/webp, default: 100), Google Cloud (jpeg, default: 75).

The direct generator ignores it; set the provider-prefixed equivalent on the generator instead.

Type: int | None Default: output_compression

output_format

Output format of the generated image.

Supported by: OpenAI Responses (default: 'png'), Google Cloud.

The direct generator ignores it; set the provider-prefixed equivalent on the generator instead.

Type: Literal[‘png’, ‘webp’, ‘jpeg’] | None Default: output_format

quality

Quality of the generated image.

Supported by: OpenAI Responses. gpt-image-2 and earlier models reject xhigh and max.

The direct generator ignores it; set the provider-prefixed equivalent on the generator instead.

Type: ImageQuality | None Default: quality

size

Size of the generated image for the native tool.

Supported by: OpenAI Responses ('auto', '1024x1024', '1024x1536', '1536x1024'), Google Gemini 3 Pro Image and later ('512' on Gemini 3.1 Flash Image only, '1K', '2K', '4K').

Direct image APIs use provider-prefixed size or resolution settings.

Type: ImageSize | None Default: size

dimensions

Exact direct-model output dimensions as (width, height) in pixels.

This is mutually exclusive with aspect_ratio: passing both alongside a direct generator raises UserError at construction. Only the direct generator can apply it, so pass native=False to guarantee it takes effect: with the default native=True the direct generator is dropped whenever the conversational model generates images natively, and the native tool has no equivalent — that request warns. The fallback_subagent_model path ignores it with a warning. Supported shapes are model-specific; see the Image Generation guide.

Type: ImageDimensions | None Default: dimensions

aspect_ratio

Aspect ratio for generated images.

Supported by: Google image-generation models (Gemini), OpenAI Responses (maps '1:1', '2:3', '3:2' to sizes).

Direct adapters map this to a canonical geometry supported by the selected model. Ratios the native tool also accepts apply on either path; the rest need the direct generator, so pass native=False to guarantee them, as for dimensions, and a request that takes the native path instead warns. Ratios outside the native vocabulary are ignored by the fallback_subagent_model path with a warning. See the ratio-to-dimensions matrix.

Type: ImageGenerationAspectRatio | None Default: aspect_ratio

local

Configure the local fallback tool.

Takes an ImageGenerator, which generates through the direct image generation API, or the Tool, toolset and callable shapes NativeOrLocalTool accepts, for a fallback you implement yourself. A generator carries settings of its own, which is why it belongs here; a bare ImageGenerationModel does not, and goes to fallback_image_model — passing one here raises UserError. Every string and local=True raise UserError too: there is no named local strategy, and a direct image model name is fallback_image_model’s to take.

A generator is kept as declared; the generate_image tool is derived from it and the capability’s settings each time the toolset is requested.

fallback_model

Deprecated alias for fallback_subagent_model.

Type: ImageGenerationFallbackModel

Methods

combine

@classmethod

def combine(
    cls,
    capabilities: Sequence[AbstractCapability[AgentDepsT]],
) -> AbstractCapability[AgentDepsT]

Merge like NativeOrLocalTool, except that dimensions is one value, not a collection.

The default merge unions two sequences, and a (width, height) pair’s entries are not independent: (1024, 1024) beside (1536, 1024) unions to (1024, 1536), a flipped orientation neither instance asked for, and two disjoint pairs union to a three-element tuple that is no size at all. It takes the later stated value instead, the rule the scalar fields already get.

Applied after the base merge because __post_init__ reads only whether dimensions is set, never what it is, and a merge never turns a stated pair into None.

Returns

AbstractCapability[AgentDepsT]

from_spec

@classmethod

def from_spec(
    cls,
    *,
    native: ImageGenerationTool | bool = True,
    local: Literal[False] | None = None,
    fallback_subagent_model: KnownModelName | str | None = None,
    fallback_image_model: str | None = None,
    action: Literal['generate', 'edit', 'auto'] | None = None,
    background: Literal['transparent', 'opaque', 'auto'] | None = None,
    input_fidelity: Literal['high', 'low'] | None = None,
    moderation: Literal['auto', 'low'] | None = None,
    image_model: ImageGenerationModelName | None = None,
    output_compression: int | None = None,
    output_format: Literal['png', 'webp', 'jpeg'] | None = None,
    quality: ImageQuality | None = None,
    size: ImageSize | None = None,
    dimensions: ImageDimensions | None = None,
    aspect_ratio: ImageGenerationAspectRatio | None = None,
    id: str | None = 'image_generation',
    defer_loading: bool = False,
    description: str | None = None,
    fallback_model: KnownModelName | str | None = None,
) -> ImageGeneration[AgentDepsT]

Construct from the JSON/YAML-serializable subset of the runtime API.

Runtime objects, such as the ImageGenerationModel that fallback_image_model also takes and the ImageGenerator, Tool, toolset and callables local takes, can be passed to ImageGeneration(...) directly but cannot be represented in an agent spec. A direct image model name is serializable and can be passed as fallback_image_model='provider:model'.

Returns

ImageGeneration[AgentDepsT]

CapabilityOrdering

Ordering constraints for a capability within a combined capability chain.

Capabilities follow middleware semantics: the first capability in the list is the outermost layer, wrapping all others. Declare ordering constraints via get_ordering to control a capability’s position in the chain regardless of how the user lists them.

When a CombinedCapability is constructed, it topologically sorts its children to satisfy these constraints, preserving user-provided order as a tiebreaker.

Attributes

position

Fixed position in the chain, or None for user-provided order.

Type: CapabilityPosition | None Default: None

wraps

This capability wraps around (is outside of) these capabilities in the middleware chain.

Each entry can be a capability type (matches all instances of that type via issubclass) or a specific capability instance (matches by identity via is).

Note: instance refs use identity (is) matching, so if a capability’s for_run returns a new instance, refs to the original will no longer match. Use type refs when the target capability uses per-run state isolation.

Type: Sequence[CapabilityRef] Default: ()

wrapped_by

This capability is wrapped by (is inside of) these capabilities in the middleware chain.

Each entry can be a capability type (matches all instances of that type via issubclass) or a specific capability instance (matches by identity via is).

Type: Sequence[CapabilityRef] Default: ()

requires

These types must be present in the chain (no ordering implied).

Type: Sequence[type[AbstractCapability[Any]]] Default: ()

AbstractCapability

Bases: ABC, Generic[AgentDepsT]

Abstract base class for agent capabilities.

A capability is a reusable, composable unit of agent behavior that can provide instructions, model settings, tools, and request/response hooks.

Lifecycle: capabilities are passed to an Agent at construction time, where most get_* methods are called to collect static configuration (instructions, model settings, toolsets, native tools). When for_run returns a replacement instance, that configuration is re-extracted from the replacement at run setup. The exception is get_wrapper_toolset, which is always called per-run during toolset assembly, and get_wrapper_model, called for every attempt at a model request. On each model request, wrap_model_request encloses the complete dynamic lifecycle: before_model_request once, then each attempt at the request, which runs prepare_model_request, the model call with on_model_request_error recovery, and after_model_request. A request usually makes one attempt, and another each time a hook raises RetryModelRequest.

See the capabilities documentation for built-in capabilities.

get_serialization_name and from_spec support YAML/JSON specs (via Agent.from_spec); they have sensible defaults and typically don’t need to be overridden.

Attributes

id

Optional identifier used to reference this capability within a run.

Must be unique within a run, not per instance: it identifies the capability across the run — including the fresh instance a for_run override may return — rather than a specific object.

Required when defer_loading=True. If omitted for an always-on capability, the run derives a local id from the class name.

Type: str | None Default: None

description

Description of the capability.

Type: str | None Default: None

defer_loading

If True, model-facing tools and instructions are hidden until the model explicitly loads the capability via the load_capability tool.

Model settings and lifecycle hooks are registered during run setup, but only apply or fire once the capability is loaded.

Requires a stable id so message history can identify the capability. A description or get_description override is optional and only adds routing context to the load catalog.

Type: bool Default: False

has_wrap_node_run

Whether this capability (or any sub-capability) overrides wrap_node_run.

Deprecated: wrap_node_run runs under every way of driving a run, so there is nothing left to test for.

Type: bool

has_wrap_run_event_stream

Whether this capability (or any sub-capability) overrides wrap_run_event_stream.

Type: bool

has_on_event

Whether this capability handles run events dynamically or with marked methods.

Type: bool

has_resolve_model_id

Whether this capability or a wrapped capability overrides resolve_model_id.

Type: bool

Methods

combine

@classmethod

def combine(
    cls,
    capabilities: Sequence[AbstractCapability[AgentDepsT]],
) -> AbstractCapability[AgentDepsT]

Combine capabilities that resolved to the same id into the one the run will use.

Two capabilities under one id name the same thing, so exactly one of them can be what that id refers to. The default merges them field by field: a value only one of them states is kept, and a value both state takes the later one.

That default needs no thought from most capabilities, because it follows from the id. Declaring a default id is the statement that an agent has one of these, so a repeat is one configuration stated twice and merging is what it meant. A capability that can legitimately appear several times declares no default id instead — and then this is never reached, because the run tells anonymous capabilities apart itself, and an id the user passed to such a capability is a name they chose, so passing it twice is reported as a collision rather than merged.

Override it when composing takes more than merging fields: NativeOrLocalTool rebuilds its native tool from the merged configuration, because that tool, not the capability, is what reaches the provider.

Only reached within one layer: a capability supplied for a run overrides its agent-level namesake outright rather than composing with it.

Returns

AbstractCapability[AgentDepsT] — The single capability the id refers to for this run.

Parameters

capabilities : Sequence[AbstractCapability[AgentDepsT]]

The two or more capabilities sharing an id, in application order. All are instances of cls; a shared id across different classes is always rejected, since no one class can say how it composes.

apply

def apply(visitor: Callable[[AbstractCapability[AgentDepsT]], None]) -> None

Run a visitor function on every capability this tree registers.

For a single capability, calls the visitor on itself. Overridden by CombinedCapability to recursively visit all child capabilities, and itself first when it is deferred.

Returns

visit_and_replace

def visit_and_replace(
    visitor: Callable[[AbstractCapability[AgentDepsT]], AbstractCapability[AgentDepsT] | None],
) -> AbstractCapability[AgentDepsT] | None

Run a visitor function on the same capabilities as apply, and replace them in this tree with its result.

Analogous to AbstractToolset.visit_and_replace, except that returning None removes the visited capability instead of replacing it.

Rewrites in place: containers and wrappers rebuild only the branches that changed, so what survives keeps its position in the hierarchy and a wrapper goes on wrapping whatever is left of its subtree. Rebuilding a tree from the flat list apply produces does neither: it loses the nesting, and re-adds a container’s children next to the wrapper that already contributes them.

Returns self when nothing changed, and None when the visitor removed everything.

For a single capability, returns the visitor’s result for itself. Overridden by CombinedCapability and WrapperCapability to rebuild their children; a custom capability that overrides apply because it holds children of its own should override this alongside it, or those children are invisible to callers rewriting the tree.

Returns

AbstractCapability[AgentDepsT] | None

listens_to

def listens_to(event: AgentStreamEvent) -> bool

Whether on_event would reach a listener for event.

Dispatch asks this before descending, so a capability that listens to a few event classes isn’t woken for every event in the run. The default reports True for any event a @on_event-marked method accepts, and for every event when on_event is overridden directly, since what an override dispatches to isn’t knowable here. Override this alongside on_event when you can report something narrower.

Returns

bool

get_serialization_name

@classmethod

def get_serialization_name(cls) -> str | None

Return the name used for spec serialization (CamelCase class name by default).

Return None to opt out of spec-based construction.

Returns

str | None

from_spec

@classmethod

def from_spec(cls, *args: Any, **kwargs: Any) -> AbstractCapability[Any]

Create from spec arguments. Default: cls(*args, **kwargs).

Override when __init__ takes non-serializable types.

Returns

AbstractCapability[Any]

get_ordering

def get_ordering() -> CapabilityOrdering | None

Return ordering constraints for this capability, or None for default behavior.

Override to declare a fixed position ('outermost' / 'innermost'), relative ordering (wraps / wrapped_by other capability types or instances), or dependency requirements (requires).

CombinedCapability uses these to topologically sort its children at construction time.

Returns

CapabilityOrdering | None

for_agent

def for_agent(agent: AbstractAgent[AgentDepsT, Any]) -> AbstractCapability[AgentDepsT]

Return the capability instance to use with an agent.

Called after the agent’s own configuration is available and before capability contributions are extracted. Constructor capabilities are bound once during agent construction; static run capabilities are bound once per run. Override this to inspect the agent and return an agent-bound copy. The default returns self.

A CapabilityFunc result is also bound before its own for_run hook. A specialized run-bound value returned by an ordinary capability’s for_run() is not bound again.

Capabilities in the innermost ordering tier (see get_ordering), i.e. durability capabilities, bind in a second phase, after the other capabilities’ contributed toolsets have been extracted, so agent.toolsets is complete when their for_agent wraps it. The flip side is that innermost capabilities can’t contribute toolsets of their own.

Returns

AbstractCapability[AgentDepsT]

for_run

@async

def for_run(ctx: RunContext[AgentDepsT]) -> AbstractCapability[AgentDepsT]

Return the capability instance to use for this agent run.

Called once per run, before get_*() re-extraction and before any hooks fire. Override to return a fresh instance for per-run state isolation. Under durable execution, worker processes re-derive this instance from the deserialized run context, so all per-run state must be derivable from ctx. Default: return self (shared across runs).

Return per-run instances here, but don’t acquire resources (connections, processes, temporary files): this runs before any run hook, so nothing releases them if the run fails before it starts. Acquire them in before_run or wrap_run, and release them in wrap_run.

Returns

AbstractCapability[AgentDepsT]

get_instructions

def get_instructions() -> AgentInstructions[AgentDepsT] | None

Return instructions to include in the system prompt, or None.

Return static instruction text, a dynamic instruction callable, or a sequence containing either. For dynamic per-request behavior, return a callable that receives RunContext or a TemplateStr — not a dynamic string.

This is read when the agent is built and the result is reused for every run, unless for_run returns a replacement, whose instructions are read for that run. To opt into on_change='append' for instructions that change mid-conversation, return a callable that returns an InstructionPart with a name and on_change='append' (for text that changes during a run), or return such a part directly from a replacement for_run builds (for text that changes between runs), from a capability with an id. See appending instruction changes. Build the text only from state you control; see keep untrusted text out of instructions.

When defer_loading is True, these instructions are resolved only after the model calls the load_capability tool for this capability.

Returns

AgentInstructions[AgentDepsT] | None

get_description

def get_description() -> CapabilityDescription[AgentDepsT] | None

Return a human-readable description of this capability, or None.

Surfaced to the model in the catalog shown with the load_capability tool when defer_loading is True.

Return a static description string or a callable that receives RunContext (or no arguments) when the deferred capability catalog is rendered. Default: return the static description field.

Returns

CapabilityDescription[AgentDepsT] | None

get_model_settings

def get_model_settings() -> AgentModelSettings[AgentDepsT] | None

Return model settings to merge into the agent’s defaults, or None.

Return a static ModelSettings dict when the settings don’t change between requests. Return a callable that receives RunContext when settings need to vary per step (e.g. based on ctx.run_step or ctx.deps).

When the callable is invoked, ctx.model_settings contains the merged result of all layers resolved before this capability (model defaults and agent-level settings). The returned dict is merged on top of that.

When defer_loading is True, these settings are registered up front but merge as an empty dict until the model calls the load_capability tool for this capability.

Returns

AgentModelSettings[AgentDepsT] | None

get_default_run_id

def get_default_run_id() -> str | None

Return a replay-stable run_id for a run started without one, or None to generate a UUID7.

Durable execution capabilities override this so a replayed or recovered run gets the same run_id it had the first time. When several capabilities return one, the one listed last wins.

Returns

str | None

get_default_conversation_id

def get_default_conversation_id(run_id: str) -> str | None

Return a replay-stable conversation_id for a run that has none to inherit, or None to generate a UUID7.

Durable execution capabilities whose run_id is replay-stable can derive it with conversation_id_from_run_id(), as the Temporal, DBOS and Prefect capabilities do.

Returns

str | None

get_run_clock

def get_run_clock() -> Callable[[], datetime] | None

Return a replay-safe clock to measure the run’s deadline against, or None for the system clock.

Durable execution capabilities return the engine’s deterministic clock, such as workflow.now() on Temporal.

Returns

Callable[[], datetime] | None

get_run_start_time

@async

def get_run_start_time(run_id: str) -> datetime | None

Return a replay-stable start time to count the timeout= of run run_id from, or None to read the run’s clock.

Returns

datetime | None

get_model

def get_model() -> AgentModel[AgentDepsT] | None

Return a static model, a per-step model selector, or None to make no selection.

A selector receives ModelSelectionContext and may be synchronous or asynchronous. Static selections are resolved once per run; selectors are evaluated before each new logical model request step. When several capabilities contribute a model, the last non-None selection wins. This differs from resolve_model_id(), where the first resolver to return a model wins.

See Selecting the model for precedence, bootstrap, and deferred-capability semantics.

Returns

AgentModel[AgentDepsT] | None

resolve_model_id

@async

def resolve_model_id(
    ctx: ModelResolutionContext[AgentDepsT],
    *,
    model_id: KnownModelName | str,
) -> Model | None

Resolve a model ID, or return None to defer.

Capabilities are tried in user-supplied order. When every capability returns None, the ID is passed to infer_model. The context provides the agent and actual run dependencies, so resolution can configure tenant-specific providers or look up models in a registry.

Returns

Model | None

get_toolset

def get_toolset() -> AgentToolset[AgentDepsT] | None

Return a toolset to register with the agent, or None.

Returns

AgentToolset[AgentDepsT] | None

get_native_tools

def get_native_tools() -> Sequence[AgentNativeTool[AgentDepsT]]

Return native tools to register with the agent.

Returns

Sequence[AgentNativeTool[AgentDepsT]]

get_workspace

def get_workspace(
    ctx: RunContext[AgentDepsT],
    *,
    ref: WorkspaceRef | None,
) -> WorkspaceBackend | None

Return the run’s workspace backend for ref, or None to leave it to another capability.

ref names an environment to continue in (from workspace= or the message history); None asks for a fresh one. Build the backend only, without I/O or side effects: it creates or attaches on first use. Capabilities passed to the run are asked before the agent’s, each list in order, before for_run; the first answer wins. Return None for a ref you don’t own. A workspace is chosen when the run starts, so a capability that supplies one can’t be deferred. Return a Workspace around the backend, such as ReadOnlyWorkspace(Workspace(backend)), to apply a policy.

Returns

WorkspaceBackend | None

get_wrapper_toolset

def get_wrapper_toolset(
    toolset: AbstractToolset[AgentDepsT],
) -> AbstractToolset[AgentDepsT] | None

Wrap the agent’s assembled toolset, or return None to leave it unchanged.

Called per-run with the combined non-output toolset (after the prepare_tools hook has already wrapped it). Output tools are added separately and are not included.

Unlike value-contribution methods such as get_instructions, this receives the already assembled toolset and is called each run (after for_run). When multiple capabilities provide wrappers, they follow middleware semantics: the first capability in the list wraps outermost (matching wrap_* hooks).

Use this to apply cross-cutting toolset wrappers like PreparedToolset, FilteredToolset, or custom WrapperToolset subclasses.

Returns

AbstractToolset[AgentDepsT] | None

get_wrapper_model

def get_wrapper_model(model: Model) -> Model | None

Wrap the model that serves a model request, or return None to leave it unchanged.

Called for every attempt at a model request, with the model that will send it: the one get_model, the run’s model= or a before_model_request hook selected, or the one a hook moved the attempt to by raising RetryModelRequest, as ModelFallback does. It’s called after every prepare_model_request hook has run, so the wrapper applies to whichever model serves each attempt. A FallbackModel is wrapped as a whole. Return a WrapperModel around model to change how requests are sent, as BatchModelRequests does. The returned model is entered (async with) around each attempt it serves and exited after it, also when the attempt fails, so set up per-request resources in its __aenter__; keep long-lived ones on the capability.

When multiple capabilities provide wrappers, they follow middleware semantics: the first capability in the list wraps outermost, as with get_wrapper_toolset. A wrapper that doesn’t call its wrapped model’s request(), such as BatchModelRequests, which sends requests through the batch API, skips the wrappers inside it: put behavior that should run once per model request in wrap_model_request instead.

Under durable execution, requests are sent from a durable unit (an activity, step or task), so this is called there too, with the model the unit rebuilds; in a Temporal activity, on the agent’s own capabilities rather than ones for_run returned. Return a wrapper that doesn’t do I/O when it’s created. The wrapper that’s applied outside the unit answers questions like continuation_delay without sending anything.

Not called for a capability with defer_loading set.

Returns

Model | None

prepare_tools

@async

def prepare_tools(
    ctx: RunContext[AgentDepsT],
    tool_defs: list[ToolDefinition],
) -> list[ToolDefinition]

Filter or modify function tool definitions for this step.

Receives function tools only. For output tools, override prepare_output_tools — it runs separately, with ctx.retry/ctx.max_retries reflecting the output retry budget instead of the function-tool budget.

Return a filtered or modified list. The result flows into both the model’s request parameters and ToolManager.tools, so filtering also blocks tool execution.

On a deferred capability this runs only once the capability is loaded, and then receives every function tool, as an always-on capability does. There is nothing to govern before that: an unloaded capability’s tools are neither advertised to the model nor callable, so no filtering here could change what the model can reach.

Returns

list[ToolDefinition]

prepare_output_tools

@async

def prepare_output_tools(
    ctx: RunContext[AgentDepsT],
    tool_defs: list[ToolDefinition],
) -> list[ToolDefinition]

Filter or modify output tool definitions for this step.

Receives only output tools. ctx.retry and ctx.max_retries reflect the output retry budget (agent-level max_output_retries), matching the output hook lifecycle.

Return a filtered or modified list. The result flows into both the model’s request parameters and ToolManager.tools, so filtering also blocks tool execution.

Returns

list[ToolDefinition]

before_run

@async

def before_run(ctx: RunContext[AgentDepsT]) -> None

Called before the agent run starts, inside wrap_run.

Observe-only; use wrap_run for modification.

A realtime session is a run. ContextVars set here are ambient in its instruction resolution, pump and tool tasks, and the caller’s async with block.

Returns

after_run

@async

def after_run(
    ctx: RunContext[AgentDepsT],
    *,
    result: AgentRunResult[Any],
) -> AgentRunResult[Any]

Modify the run result after iteration or error recovery, inside wrap_run.

Not called when wrap_run returns without calling its handler, or recovers an error by returning its own result. It is called when on_run_error recovers a run-body failure.

It is also called when the handler produces a result while a cancellation is pending, but the run still ends cancelled after this hook returns. Put cancellation-safe cleanup in wrap_run (a try/finally around handler()), which does observe the CancelledError.

For a realtime session, the result is produced when the session closes; a transformed result becomes session.result before the caller leaves the async with boundary.

Returns

AgentRunResult[Any]

wrap_run

@async

def wrap_run(
    ctx: RunContext[AgentDepsT],
    *,
    handler: WrapRunHandler,
) -> AgentRunResult[Any]

Wrap the complete agent-run lifecycle.

handler() runs before_run, the run body with on_run_error recovery, and after_run. Returning without calling handler() skips that lifecycle.

If handler() raises and this method catches the exception and returns a result instead, the error is suppressed and the recovery result is used directly without after_run.

Note: if the caller cancels the run (e.g. by breaking out of an iter() loop, or by raising GeneratorExit or KeyboardInterrupt in its async with block, as closing a generator that owns the run does), this method receives an asyncio.CancelledError. Implementations that hold resources should handle cleanup accordingly. Cancellation is terminal: the hook may observe it and clean up, but cannot recover the run to success.

A realtime session is a run: handler() resolves when the session closes. ContextVars set before calling it are ambient in instruction resolution, pumps, tool tasks, and the caller’s block. Downward ContextVar propagation is one-way; keep bidirectional per-run state on the for_run copy’s instance attributes. Suppression and result transformation apply at the session’s async with boundary, after the caller may have observed events in real time.

Returns

AgentRunResult[Any]

on_run_error

@async

def on_run_error(
    ctx: RunContext[AgentDepsT],
    *,
    error: BaseException,
) -> AgentRunResult[Any]

Called inside wrap_run when the run body fails with an exception.

This is the error counterpart to after_run.

Raise the original error (or a different exception) to propagate it. Return an AgentRunResult to suppress the error, continue through after_run, and recover the run.

Recovery happens before control returns to wrap_run, so wrap hooks only observe failures that this hook does not recover.

Only failures of the run body reach this hook. An exception from before_run or after_run, including another capability’s, goes straight to wrap_run, so release resources acquired in before_run from wrap_run (a try/finally around handler()).

Cancellation is terminal: the hook may observe it and clean up, but cannot recover the run to success. A result returned for a cancellation is discarded, and wrap_run still sees the CancelledError.

Not called for GeneratorExit or KeyboardInterrupt.

For a realtime session, returning a recovery result sets session.result and suppresses the error at the caller’s async with boundary, after events may already have been observed.

Returns

AgentRunResult[Any]

before_node_run

@async

def before_node_run(
    ctx: RunContext[AgentDepsT],
    *,
    node: AgentNode[AgentDepsT],
) -> AgentNode[AgentDepsT]

Observe or replace a graph node before it executes, inside wrap_node_run.

Returns

AgentNode[AgentDepsT]

after_node_run

@async

def after_node_run(
    ctx: RunContext[AgentDepsT],
    *,
    node: AgentNode[AgentDepsT],
    result: NodeResult[AgentDepsT],
) -> NodeResult[AgentDepsT]

Modify a node result after execution or error recovery, inside wrap_node_run.

The result is the next node or End.

Not called for a node interrupted by cancellation — including a cancellation the node itself absorbed and completed through, which the framework re-asserts at the node boundary: cancellation skips downstream hooks. Put cancellation-safe cleanup in wrap_node_run (a try/finally around handler()), which does observe the CancelledError. (A hook that catches the CancelledError and calls Task.uncancel() takes over the cancellation bookkeeping for that boundary, so this hook does fire for that node — the run itself still ends cancelled at the next boundary.)

Returns

NodeResult[AgentDepsT]

wrap_node_run

@async

def wrap_node_run(
    ctx: RunContext[AgentDepsT],
    *,
    node: AgentNode[AgentDepsT],
    handler: WrapNodeRunHandler[AgentDepsT],
) -> NodeResult[AgentDepsT]

Wrap the complete node-run lifecycle.

handler(node) runs before_node_run, node execution with on_node_run_error recovery, and after_node_run, returning the next node or End. Returning without calling handler() skips that lifecycle; calling it multiple times runs the lifecycle each time.

Override to inspect or modify nodes before execution, inspect or modify the returned next node, call handler multiple times (retry), or return a different node to redirect graph progression.

Note: this hook fires however the run is driven — agent.run(), agent.run_stream(), an agent.iter() run advanced with agent_run.next(), and a bare async for node in agent_run: loop, which advances through next() too. The one exception is the final ModelRequestNode under run_stream(), which hands back the result mid-stream and so only fires before_node_run.

When using agent.run() with event_stream_handler, the handler wraps both streaming and graph advancement. agent.run_stream() is the documented exception: before_node_run fires before streaming, and the final streamed ModelRequestNode skips wrap_node_run and after_node_run; for non-final streamed nodes, the wrapper encloses only graph advancement, followed by the error and after hooks.

A cancelled run delivers asyncio.CancelledError through handler(). Cancellation is terminal: the hook may observe it and clean up, but cannot recover the run to success — even a returned End result is discarded once a cancellation is pending.

Returns

NodeResult[AgentDepsT]

on_node_run_error

@async

def on_node_run_error(
    ctx: RunContext[AgentDepsT],
    *,
    node: AgentNode[AgentDepsT],
    error: Exception,
) -> NodeResult[AgentDepsT]

Called inside wrap_node_run when core graph-node execution fails.

This is the error counterpart to after_node_run.

Raise the original error (or a different exception) to propagate it. Return a next node or End to recover and continue the graph.

Recovery happens before control returns to wrap_node_run, so wrap hooks only observe failures that this hook does not recover. In the agent.run_stream() exception described on wrap_node_run, recovery instead follows the post-stream wrapper.

Useful for recovering from UnexpectedModelBehavior by redirecting to a different node (e.g. retry with different model settings).

Returns

NodeResult[AgentDepsT]

on_event

@async

def on_event(ctx: RunContext[AgentDepsT], *, event: AgentStreamEvent) -> None

React to every event in the run’s event stream.

This includes model response stream events, tool events, deferred and enqueued-message events, CustomEvents, and CapabilityEvents. The default implementation dispatches to methods marked with on_event, in definition order.

Override this method for fully dynamic handling. Call super().on_event(...) to retain marked method dispatch. A capability receives events it emits itself. Events emitted by a listener enter the stream after the event being handled.

Returns

AsyncIterable[AgentStreamEvent]

wrap_run_event_stream

@async

def wrap_run_event_stream(
    ctx: RunContext[AgentDepsT],
    *,
    stream: AsyncIterable[AgentStreamEvent],
) -> AsyncIterable[AgentStreamEvent]

Wrap a run or realtime session’s consumer-facing event stream.

For classic runs, the wrapper is applied where each node’s stream is produced, so it fires however the run is driven — including under agent.iter() and when the caller streams a node itself with node.stream(). For realtime sessions, it wraps the async for event in session view. A wrapper must yield events appropriate for the stream it wraps.

Transformations affect only what the stream consumer sees. They never change realtime session history, tool execution, or the classic run’s accumulated response and output.

Note: when this method is overridden (or Hooks.on.event / Hooks.on.run_event_stream are registered), agent.run() and AgentRun.next() automatically enable streaming mode so this hook fires even without an explicit event_stream_handler.

Returns

before_model_request

@async

def before_model_request(
    ctx: RunContext[AgentDepsT],
    request_context: ModelRequestContext,
) -> ModelRequestContext

Called inside wrap_model_request once per request step, before any attempt at the model call.

Can modify the model, request-only messages, settings, and parameters. Persistent history changes must be made through RunContext.messages. model_request_parameters.instruction_parts is the source of truth for instructions: rewriting them changes what the model receives.

Exceptions propagate through the wrap chain and are not passed to on_model_request_error. A ModelRetry requests another model attempt and counts against the output retry budget. Work that depends on the model that will actually serve the request belongs in prepare_model_request, which runs before each attempt.

Returns

prepare_model_request

@async

def prepare_model_request(
    ctx: RunContext[AgentDepsT],
    request_context: ModelRequestContext,
) -> ModelRequestContext

Called before each attempt at a model request, once the serving model is known.

This is the hook for work that depends on which model is about to run: compaction, context-window fitting, per-profile message translation. Unlike before_model_request, which runs once per request step while the model may still change, request_context.model here is the model that will actually serve the request, and request_context.attempt says which attempt this is.

It runs again for every attempt a hook asks for by raising RetryModelRequest, so a capability never inherits preparation done for a different model. Anything that should happen once per request step regardless of how many models are tried — a history processor, an injected message — belongs in before_model_request instead.

Changes to request_context shape the request that goes on the wire for this attempt and are not persisted to the agent’s message history; the next attempt starts again from what before_model_request produced. Changes made through RunContext.messages, the run’s persistent history, do persist, and are not rolled back for a further attempt. Use them for state that doesn’t depend on the attempt, such as a record an attempt makes that later requests and turns should reuse.

Returns

after_model_request

@async

def after_model_request(
    ctx: RunContext[AgentDepsT],
    *,
    request_context: ModelRequestContext,
    response: ModelResponse,
) -> ModelResponse

Called inside wrap_model_request after each attempt’s model response or recovered error.

Can modify the response before wrap hooks post-process it and before further agent processing. A wrap hook that returns without calling its handler skips this hook.

Raise ModelRetry to reject the response and ask the model to try again. The original response is still appended to message history so the model can see what it said. Retries count against the output side of the agent’s retry budget.

Raise RetryModelRequest to reject the response and make another attempt, optionally on another model, as ModelFallback does. The rejected response is not added to message history, but its usage is still counted. A streamed response has already reached the consumer, so raising RetryModelRequest for one is a UserError.

Returns

ModelResponse

wrap_model_request

@async

def wrap_model_request(
    ctx: RunContext[AgentDepsT],
    *,
    request_context: ModelRequestContext,
    handler: WrapModelRequestHandler,
) -> ModelResponse

Wraps the complete model-request lifecycle.

Calling handler(request_context) runs before_model_request, then one or more attempts, each of which runs prepare_model_request, the model call with on_model_request_error recovery, and after_model_request. A hook makes another attempt by raising RetryModelRequest. Returning without calling handler skips that inner lifecycle. A wrapper may call it only once.

Raise ModelRetry to skip on_model_request_error and directly retry the model request with a retry prompt. If the handler was called, the model response is preserved in history for context (same as after_model_request).

Returns

ModelResponse

on_model_request_error

@async

def on_model_request_error(
    ctx: RunContext[AgentDepsT],
    *,
    request_context: ModelRequestContext,
    error: Exception,
) -> ModelResponse

Called when an attempt’s core model call fails with an exception.

This is the error counterpart to after_model_request. Recovery runs inside the wrapped handler, so wrap hooks observe the recovered response, not the handled exception. Exceptions from before_model_request, prepare_model_request, after_model_request, or wrap_model_request are outside this hook’s scope.

Raise the original error (or a different exception) to propagate it. Return a ModelResponse to suppress the error and use the response as if the model call succeeded. Raise ModelRetry to retry the model request with a retry prompt instead of recovering or propagating. Raise RetryModelRequest to make another attempt, optionally on another model, as ModelFallback does.

Not called for SkipModelRequest or ModelRetry.

Returns

ModelResponse

before_tool_validate

@async

def before_tool_validate(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: RawToolArgs,
) -> RawToolArgs

Modify raw args before validation.

The returned args feed validation and execution only: the tool call in the message history keeps the model’s arguments. call is that history part, so to record a rewrite there too, assign it to call.args as well. Also clear call.id, since OpenAI Responses replays its own stored copy of a call that keeps its item ID; keep the original in call.provider_details if it should stay visible.

Raise ModelRetry to skip validation and ask the model to redo the tool call.

A tool call can only be deferred once its arguments have been validated, so raising CallDeferred or ApprovalRequired here is a UserError. Defer from after_tool_validate, a tool’s args_validator, or before_tool_execute.

Returns

RawToolArgs

after_tool_validate

@async

def after_tool_validate(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
) -> ValidatedToolArgs

Modify validated args after validation or error recovery, inside wrap_tool_validate.

Raise ModelRetry to reject the validated args and ask the model to redo the tool call.

The arguments are valid by this point, so raising CallDeferred or ApprovalRequired here defers the call — the tool isn’t executed, and the deferral joins the run’s DeferredToolRequests with the validated arguments.

This hook also runs when the tool’s args_validator already deferred the call, so it stays a reliable gate on validated arguments: rejecting here wins over that deferral, deferring here replaces it, and the args returned here are the ones the deferred call carries.

Returns

ValidatedToolArgs

wrap_tool_validate

@async

def wrap_tool_validate(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: RawToolArgs,
    handler: WrapToolValidateHandler,
) -> ValidatedToolArgs

Wrap the complete tool-validation lifecycle.

handler() runs before_tool_validate, argument validation with on_tool_validate_error recovery, and after_tool_validate. Returning without calling handler() skips that lifecycle; calling it multiple times runs the lifecycle each time.

Deferring with CallDeferred or ApprovalRequired is allowed after handler() has returned, when the arguments are known to be valid; raising one before that is a UserError.

Returns

ValidatedToolArgs

on_tool_validate_error

@async

def on_tool_validate_error(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: RawToolArgs,
    error: ValidationError | ModelRetry,
) -> ValidatedToolArgs

Called inside wrap_tool_validate when core tool argument validation fails.

This is the error counterpart to after_tool_validate. Fires for ValidationError (schema mismatch) and ModelRetry (custom validator rejection).

Raise the original error (or a different exception) to propagate it. Return validated args to suppress the error and continue as if validation passed.

Recovery happens before control returns to wrap_tool_validate, so wrap hooks only observe failures that this hook does not recover.

Not called for SkipToolValidation, or when a tool’s args_validator raises CallDeferred or ApprovalRequired — those are control flow, not errors, and the call is deferred instead of executed.

Raising a deferral from this hook is a UserError: it only runs because validation failed, so there are no valid arguments to show whoever would resolve the deferral.

Returns

ValidatedToolArgs

before_tool_execute

@async

def before_tool_execute(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
) -> ValidatedToolArgs

Modify validated args before execution.

Raise ModelRetry to skip execution and ask the model to redo the tool call.

This is the hook to defer from: raising ApprovalRequired or CallDeferred here defers the call before the tool function runs, so nothing happens until it’s resolved.

Returns

ValidatedToolArgs

after_tool_execute

@async

def after_tool_execute(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
    result: Any,
) -> Any

Modify result after execution.

Raise ModelRetry to reject the tool result and ask the model to redo the tool call.

Deferring from here is accepted but rarely what you want: the tool function has already run, so its side effects happened and result is discarded. Defer from before_tool_execute instead.

Returns

wrap_tool_execute

@async

def wrap_tool_execute(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
    handler: WrapToolExecuteHandler,
) -> Any

Wrap the complete tool-execution lifecycle.

handler() runs before_tool_execute, tool execution with on_tool_execute_error recovery, and after_tool_execute. Returning without calling handler() skips that lifecycle; calling it multiple times runs the lifecycle each time.

Defer before calling handler(): a deferral raised after it has returned is accepted, but the tool function already ran and its result is discarded. Defer from before_tool_execute instead.

Returns

on_tool_execute_error

@async

def on_tool_execute_error(
    ctx: RunContext[AgentDepsT],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
    error: Exception,
) -> Any

Called inside wrap_tool_execute when core tool execution fails with an exception.

This is the error counterpart to after_tool_execute.

Raise the original error (or a different exception) to propagate it. Return any value to suppress the error and use it as the tool result. Raise ModelRetry to ask the model to redo the tool call instead of recovering or propagating.

Recovery happens before control returns to wrap_tool_execute, so wrap hooks only observe failures that this hook does not recover.

Not called for control flow exceptions (SkipToolExecution, CallDeferred, ApprovalRequired), retry signals (ToolRetryError from ModelRetry), or failure signals (ToolFailedError from ToolFailed). Use wrap_tool_execute to intercept retries or failures.

Returns

before_output_validate

@async

def before_output_validate(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: RawOutput,
) -> RawOutput

Modify raw model output before validation/parsing.

The primary hook for pre-parse repair and normalization of model output. Fires only for structured output that requires parsing: prompted, native, tool, and union output. Does not fire for plain text or image output.

For structured text output, output is the raw text string from the model. For tool output, output is the raw tool arguments (string or dict).

Raise ModelRetry to skip validation and ask the model to try again with a custom message.

During streaming, this hook fires on every partial validation attempt as well as the final result. Check ctx.partial_output to distinguish and avoid expensive work on partial results.

Returns

RawOutput

after_output_validate

@async

def after_output_validate(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: Any,
) -> Any

Modify validated output after successful parsing inside wrap_output_validate.

output is the semantic value the model was asked to produce — e.g., a MyModel instance for output_type=MyModel, or 42 for output_type=int, or the input to a single-arg output function. For multi-arg output functions, this is the dict of arguments (the genuine multi-value input).

Note: this differs from tool hooks (after_tool_validate), which always see dict[str, Any] — tool args follow the schema contract. Output hooks see the semantic output value, regardless of how it’s internally represented during validation.

Raise ModelRetry to reject the validated output and ask the model to try again.

Returns

wrap_output_validate

@async

def wrap_output_validate(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: RawOutput,
    handler: WrapOutputValidateHandler,
) -> Any

Wrap the complete output-validation lifecycle.

handler() runs before_output_validate, output validation with on_output_validate_error recovery, and after_output_validate. Returning without calling handler() skips that lifecycle; calling it multiple times runs the lifecycle each time.

Returns

on_output_validate_error

@async

def on_output_validate_error(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: RawOutput,
    error: ValidationError | ModelRetry,
) -> Any

Called inside wrap_output_validate when core output validation fails.

This is the error counterpart to after_output_validate.

Raise the original error (or a different exception) to propagate it. Return validated output to suppress the error and continue.

Recovery happens before control returns to wrap_output_validate, so wrap hooks only observe failures that this hook does not recover.

Returns

before_output_process

@async

def before_output_process(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: Any,
) -> Any

Modify validated output before processing (extraction, output function call).

output is the semantic value — e.g., a MyModel instance or 42, matching after_output_validate. For multi-arg output functions, it’s the dict of args. See after_output_validate for a full explanation of the semantic-value contract.

Raise ModelRetry to skip processing and ask the model to try again.

Returns

after_output_process

@async

def after_output_process(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: Any,
) -> Any

Modify result after output processing inside wrap_output_process.

Raise ModelRetry to reject the result and ask the model to try again.

Returns

wrap_output_process

@async

def wrap_output_process(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: Any,
    handler: WrapOutputProcessHandler,
) -> Any

Wrap the complete output-processing lifecycle.

handler() runs before_output_process, output processing with on_output_process_error recovery, and after_output_process. Returning without calling handler() skips that lifecycle; calling it multiple times runs the lifecycle each time.

During streaming, this fires only when partial validation succeeds, and on the final result. Check ctx.partial_output to skip expensive work on partial results.

Returns

on_output_process_error

@async

def on_output_process_error(
    ctx: RunContext[AgentDepsT],
    *,
    output_context: OutputContext,
    output: Any,
    error: Exception,
) -> Any

Called inside wrap_output_process when core output processing fails with an exception.

This is the error counterpart to after_output_process.

Raise the original error (or a different exception) to propagate it. Return any value to suppress the error and use it as the output.

Recovery happens before control returns to wrap_output_process, so wrap hooks only observe failures that this hook does not recover.

Not called for retry signals (ToolRetryError from ModelRetry).

Returns