Capabilities and Hooks

Read this file when the user wants reusable agent behavior, provider-adaptive tools, or lifecycle interception.

Add Capabilities to an Agent

Capabilities bundle reusable behavior and compose automatically.

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

agent = Agent(
    'anthropic:claude-opus-4-6',
    name='capabilities_agent',
    capabilities=[
        Thinking(effort='high'),
        WebSearch(),
    ],
)

Provider-adaptive capabilities to reach for first:

• Thinking
• WebSearch
• WebFetch
• ImageGeneration
• MCP

For stricter safety handling, use RaiseContentFilterError to raise ContentFilterError whenever a model response has finish_reason='content_filter', including responses with partial text.

Use capabilities when the user wants behavior that should survive model/provider changes.

Enable Thinking Across Providers

Use the unified Thinking capability or the thinking model setting.

from pydantic_ai import Agent
from pydantic_ai.capabilities import Thinking

capability_agent = Agent('anthropic:claude-opus-4-6', name='capability_agent', capabilities=[Thinking(effort='high')])
settings_agent = Agent('anthropic:claude-opus-4-6', name='settings_agent', model_settings={'thinking': 'high'})

Supported effort values:

• True
• False
• 'minimal'
• 'low'
• 'medium'
• 'high'
• 'xhigh'

Intercept Agent Lifecycle with Hooks

Use Hooks for decorator-based lifecycle interception.

from pydantic_ai import Agent, RunContext, ToolDefinition
from pydantic_ai.capabilities import ValidatedToolArgs
from pydantic_ai.capabilities.hooks import Hooks
from pydantic_ai.messages import ToolCallPart
from pydantic_ai.models import ModelRequestContext

hooks = Hooks()


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


@hooks.on.before_tool_execute(tools=['send_email'])
async def audit_tool(
    ctx: RunContext[None],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
) -> ValidatedToolArgs:
    print(f'Executing {call.tool_name}')
    return args


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

Important hook families:

• run-level hooks
• node-level hooks
• model-request hooks
• tool-validation hooks
• tool-execution hooks
• event-stream hooks

From tool-validation and tool-execution hooks you can raise ModelRetry (the model should retry the call) or ToolFailed (the call is done and failed — the model sees the result and adapts, without consuming the retry budget) to redirect a tool call in one place instead of per tool.

At wrap boundaries, ModelRetry is control flow and bypasses on_model_request_error, on_tool_execute_error, and on_output_process_error. ToolFailed bypasses only on_tool_execute_error; from model-request or output-process hooks it is an ordinary exception passed to the corresponding error hook.

For deferrals (ApprovalRequired, CallDeferred), the rule is that a tool call can only be deferred once its arguments have been validated, since whoever resolves it is shown those arguments. So they are allowed from after_tool_validate, from wrap_tool_validate after handler() returns, and from every tool-execution hook — prefer before_tool_execute, since deferring after the tool body ran means its side effects happened and its result is discarded. Raising one from before_tool_validate, from wrap_tool_validate before handler(), or from on_tool_validate_error is a UserError. For a per-tool decision, use the tool's args_validator instead of a hook.

Use hooks when the user wants observability, auditing, or light interception without adding a new abstraction.

Publish and React to Capability Events

A capability announces things to other capabilities and to the host application with CapabilityEvent, not with the application-owned CustomEvent covered in AGENTS-CORE.md. The split is enforced: emitting a CustomEvent from a capability raises UserError, and so does emitting a CapabilityEvent from application code.

Define each event as a dataclass subclass carrying a namespace, and await ctx.emit(event) from an async capability hook or a tool the capability contributes. The namespace plus the event name form the serialized kind (workspace.file_read), which is the wire identifier, so a published capability should pin name= on each event. The payload can't reuse the envelope's field names: data, capability_id, tool_call_id, tool_name, and event_kind are rejected at class definition.

React with @on_event(SomeEvent, OtherEvent) on an async capability method; a bare @on_event sees every AgentStreamEvent. Listeners run in capability order, then definition order, and the emitting capability receives its own events. This is the way to coordinate two capabilities without a shared object between them.

Always name the classes when you can. They are what lets dispatch skip a capability without descending into it — a bare @on_event, or an overridden on_event(), opts that capability into every event in the run. Override listens_to(event) alongside a custom on_event() if you can report something narrower.

Capability events reach the agent run event stream but UI adapters do not forward them, since they are internal signals. To surface one to a frontend, react to it from application code and emit your own CustomEvent with the public payload.

For application-level listening, prefer @agent.on_event(SomeEvent) — it takes the same event classes, filtering and timeout= as Hooks.on.event without a separate capability, runs after the capabilities' own listeners, and survives an overridden root capability. Reach for a Hooks capability when you also want hook families other than events, or when you need to choose where it sits among the other capabilities.

An event class can declare dispatch='immediate' when listeners must act before the emitter continues — a decision event, where a listener mutates a field the emitter then reads. Default dispatch delivers the event at its stream position instead, keeping listener work off the emitter's latency path.

from dataclasses import dataclass
from typing import Any

from pydantic_ai import CapabilityEvent, RunContext
from pydantic_ai.capabilities import AbstractCapability, on_event


@dataclass(kw_only=True)
class FileReadEvent(CapabilityEvent, namespace='workspace', name='file_read'):
    path: str


class Watcher(AbstractCapability[Any]):
    @on_event(FileReadEvent)
    async def _on_read(self, ctx: RunContext[Any], event: FileReadEvent) -> None:
        if event.path.endswith('AGENTS.md'):
            ctx.enqueue('Follow the instructions in the discovered AGENTS.md file.')

Build a Custom Capability

Subclass AbstractCapability when the user wants reusable behavior that combines tools, hooks, instructions, or model settings into one package.

Reach for a custom capability when:

• the same bundle should be reused across multiple agents
• Hooks alone is not enough
• the behavior should be installable or declarative

Keep custom capabilities focused. If the user only needs one tool or one hook, do not introduce a capability.

Give a capability an id when its instructions should be addressable. You declare InstructionPart.name; the framework issues InstructionPart.id as an InstructionId, whose source identifies the agent, toolset, or capability and whose optional name identifies one named part. It renders and serializes as agent, toolset:<id>, capability:<id>, or one of those followed by :<name>. Compare the structured source and name in Python; use str(part.id) only when a persisted string key is required. Naming a part whose source has no id of its own leaves id as None: the name still travels with the part, but nothing addresses it.

Isolate Mutable State Per Run

Assume a capability instance may outlive and be reused across runs. By default, for_run() returns self, so every run uses that same instance. Mutating its fields from hooks, tools, dynamic callables, or other code executed during a run will therefore share state across sequential and concurrent runs.

Before adding an instance field, classify it as immutable configuration, a deliberately shared concurrency-safe resource, or per-run state. If code executed during a run mutates per-run state, override for_run() to return a fresh instance. Do not mutate and return self, and do not rely on clearing shared state in after_run(); overlapping runs can still interfere, and after_run() is skipped when a run produces no result. Use wrap_run() with try/finally when external resources need cleanup on errors or cancellation.

For dataclass capabilities, keep configuration in normal fields and declare per-run state with init=False so dataclasses.replace() preserves configuration while reinitializing the run state:

from __future__ import annotations

import logging
from dataclasses import dataclass, field, replace
from typing import Any

from pydantic_ai import AgentRunResult, RunContext
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.models import ModelRequestContext

logger = logging.getLogger(__name__)


@dataclass
class RequestCounter(AbstractCapability[Any]):
    label: str
    _requests: int = field(default=0, init=False, repr=False)

    async def for_run(self, ctx: RunContext[Any]) -> RequestCounter:
        return replace(self)

    async def before_model_request(
        self,
        ctx: RunContext[Any],
        request_context: ModelRequestContext,
    ) -> ModelRequestContext:
        self._requests += 1
        return request_context

    async def after_run(
        self,
        ctx: RunContext[Any],
        *,
        result: AgentRunResult[Any],
    ) -> AgentRunResult[Any]:
        logger.info('%s made %d model requests', self.label, self._requests)
        return result

replace() is shallow: do not mutate nested mutable configuration from any code executed during a run. Make that data run-local too, or copy it explicitly in for_run(). Capabilities with no mutable per-run state can keep the default for_run() implementation.

For every capability, consider whether defer_loading=True would improve the system by keeping instructions and tool schemas out of the eager context. Keep it eager only when the model benefits from that capability on most turns, when its hooks/settings must always apply, or when deferral would make capability selection unreliable.

Use for_agent(agent) when a capability needs the agent's model, name, or toolsets. Return a bound copy instead of mutating the original so one capability instance can safely be attached to multiple agents. The returned copy supplies all subsequent get_* contributions and hooks. Binding sees the constructor model exactly as supplied, including an unresolved string; default model inference happens afterwards only if the bound tree has no resolve_model_id() hook and model checking was not deferred. CombinedCapability and WrapperCapability bind their children automatically. Static per-run capabilities bind once per run. A CapabilityFunc result also binds before its own for_run() runs, while a specialized run-bound value returned by another capability's for_run() is not rebound.

Select a Model Dynamically

Implement get_model() when reusable policy should choose the model, or use SelectModel(selector) for the common callable-only case. Return a model or model ID for a static choice, or return a sync/async callable accepting ModelSelectionContext to choose before every request step. The context exposes the agent, run dependencies, lower-precedence configured model on step one (then the previous step's model), step number, messages, and accumulated usage. Keep get_model() cheap; put I/O in an async selector. Static choices are resolved once per run, while a selector runs once per new logical request step and not again for same-step continuation. A model-less agent can be bootstrapped by a selector because the callable is first evaluated during run setup, after dependencies and history are available.

Explicit run(model=...), run-spec, and agent.override(model=...) choices win and skip capability selection. Later capabilities override earlier model contributions. Same-step continuation remains pinned to its selected model; pass an explicit model when resuming a suspended provider-side request in another run.

Keep selection separate from construction. Use the resolve_model_id() hook, or the ResolveModelId convenience capability, when tenant, region, credentials, or another dependency controls how a selected string becomes a Model instance. Resolution uses the first non-None result in capability order; model selection uses the last non-None contribution.

Bootstrap strings use the post-for_agent, pre-for_run resolver chain. If for_run() replaces the capability, strings selected for step one and later use the replacement's resolver chain. Return a FallbackModel as the selected model when request failures, rather than routing policy, should trigger fallback.

Both hooks are eager: deferred capabilities do not select or resolve models. Run-spec capabilities can bootstrap a model-less agent, but CapabilityFunc and for_run() need an existing model to construct their RunContext; they can replace it starting on step one. Do not use adaptive selection with durable execution yet, and pass an explicit model when resuming a selector-backed suspended request in another run.

Defer Capability Loading

For capabilities on demand, load Capabilities on Demand. Use it when the user mentions deferred capabilities, capability progressive disclosure, defer_loading=True on a capability, or load_capability; also use it proactively when an agent design includes optional instructions, specialist workflows, long-tail tools, or context the model does not need on most turns.