Capabilities

A capability is a reusable, composable unit of agent behavior. Instead of threading multiple arguments through your Agent constructor — instructions here, model settings there, a toolset somewhere else, a history processor on yet another parameter — you can bundle related behavior into a single capability and pass it via the [capabilities][pydantic_ai.agent.Agent.init] parameter.

Capabilities can provide any combination of:

• Tools — via toolsets or native tools
• Lifecycle hooks — intercept and modify model requests, tool calls, and the overall run
• Instructions — static or dynamic instruction additions
• Model settings — static or per-step model settings
• Models — static or adaptive model selection and application-specific model ID resolution

This makes them the primary extension point for Pydantic AI. Whether you're building a memory system, a guardrail, a cost tracker, or an approval workflow, a capability is the right abstraction.

Capabilities can be always-on or loaded by the model on demand. The capability index below covers Pydantic AI and Pydantic AI Harness. Third-party packages provide many more capabilities, and you can define your own declaratively or by subclassing. To run agents durably across failures, restarts, and long waits, see Durable Execution.

Available capabilities

Capabilities come from two packages, and they compose with each other and with capabilities you define. Core (pydantic-ai) ships the capabilities that require model or framework support: provider-native tools, provider APIs, and deep loop integration. Pydantic AI Harness, the official capability library and harness for Pydantic AI, ships everything else, from single capabilities to complete agents. The Package column says which; every entry links to its documentation.

Harnesses

Complete agent stacks as regular combined capabilities: one import gives you a working agent, and you can take either apart into the blocks below.

Execution environments

The workspace the agent acts in: the files it edits and the commands it runs, local or isolated.

Tools & native abilities {#tools-native-abilities}

Connections to systems outside the agent's workspace, and abilities the provider executes natively.

Web & research {#web-research}

Finding and reading things on the open web.

Reasoning, planning & delegation {#reasoning-planning-delegation}

How the agent thinks and divides the work.

Context management

How the agent spends its context window: the difference between an agent that degrades over a long run and one that doesn't, and between paying for tokens N times or once.

Knowledge & memory {#knowledge-memory}

What the agent knows and remembers, loaded when relevant instead of carried in every prompt.

Control & safety {#control-safety}

Bounding what the agent may do, and keeping it on-instructions.

Self-extension

Execution runtime

Outside the loop: how runs persist, survive failures, and get observed and configured in production.

Loop customization

Core also ships capabilities for customizing the agent loop itself, mostly for production servers:

The authoring primitives, [Capability][pydantic_ai.capabilities.Capability] for bundling behavior without subclassing and [Toolset][pydantic_ai.capabilities.Toolset] for wrapping an [AbstractToolset][pydantic_ai.toolsets.AbstractToolset], are covered below. ACP (experimental, Harness) serves any agent to editors like Zed over the Agent Client Protocol. Capabilities that can be declared in YAML/JSON agent specs are listed there.

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

agent = Agent(
    'anthropic:claude-fable-5',
    instructions='You are a research assistant. Be thorough and cite sources.',
    capabilities=[
        Thinking(effort='high'),
        WebSearch(local='duckduckgo'),
    ],
)

Instructions and model settings are configured directly via the instructions and model_settings parameters on Agent (or [AgentSpec][pydantic_ai.agent.AgentSpec]). Capabilities are for behavior that goes beyond simple configuration — tools, lifecycle hooks, and custom extensions. They compose well, especially when you want to reuse the same configuration across multiple agents or load it from a spec file.

Bundling behavior with Capability

You don't need a subclass to define a capability of your own: [Capability][pydantic_ai.capabilities.Capability] bundles instructions, function tools, and toolsets declaratively — think of it as defining a skill:

from pydantic_ai import Agent
from pydantic_ai.capabilities import Capability

refunds = Capability(
    id='refunds',
    description='Use for refund eligibility and refund status.',
    instructions='Always confirm the order ID before issuing a refund.',
)


@refunds.tool_plain
def refund_status(order_id: str) -> str:
    """Look up the refund status for an order."""
    return f'Order {order_id}: refund issued on 2026-05-01.'


agent = Agent('openai:gpt-5.6-sol', capabilities=[refunds])

Add defer_loading=True and the bundle becomes an on-demand capability that stays collapsed to a one-line catalog entry until the model loads it — like Agent Skills, which you can wrap in a Capability directly. See The Capability convenience class for the full API. For behavior beyond instructions, tools, and toolsets — lifecycle hooks, model settings, native tools — subclass [AbstractCapability][pydantic_ai.capabilities.AbstractCapability] as covered in Building Custom Capabilities.

Capability events

Reusable capabilities can publish typed [CapabilityEvent][pydantic_ai.messages.CapabilityEvent]s for coordination and observability. Give an event family a stable namespace, define each payload as a dataclass, and emit it from an async capability hook or capability-contributed tool by awaiting [ctx.emit()][pydantic_ai.tools.RunContext.emit]:

from dataclasses import dataclass
from typing import Any

from pydantic_ai import CapabilityEvent, RunContext
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.toolsets import AgentToolset, FunctionToolset

WORKSPACE = 'workspace'


@dataclass(kw_only=True)
class FileWriteEvent(CapabilityEvent, namespace=WORKSPACE):
    path: str
    bytes_written: int


workspace = FunctionToolset()


@workspace.tool
async def write_file(ctx: RunContext[Any], path: str, content: str) -> str:
    await ctx.emit(FileWriteEvent(path=path, bytes_written=len(content)))
    return f'Wrote {path}'


@dataclass
class Workspace(AbstractCapability[Any]):
    def get_toolset(self) -> AgentToolset[Any] | None:
        return workspace

(This example is complete, it can be run "as is")

Keep the namespace in a module-level constant and give it the capability's name or [id][pydantic_ai.capabilities.AbstractCapability.id]: every event in the family repeats it, and it's the prefix subscribers match on.

A capability publishes capability events specifically, and emitting an application [CustomEvent][pydantic_ai.messages.CustomEvent] from one raises a [UserError][pydantic_ai.exceptions.UserError]; see Which event type do I use? for the split. The payload cannot use the field names the envelope needs for itself: data, capability_id, tool_call_id, tool_name, and event_kind are rejected when the class is defined.

Pydantic AI stamps the emitting capability's run id as capability_id. For a capability you gave an id, that is the id. For one you did not, it is a handle the framework mints for that run — '<file_reader:4f3a9c>' — which differs every run and must not be matched on; give the capability an explicit id if a subscriber needs to recognize it. Events emitted by its tools also receive tool_call_id and tool_name. They surface on the agent run event stream but are internal coordination signals, so UI adapters do not forward them by default. A protocol adapter can override [handle_capability_event()][pydantic_ai.ui.UIEventStream.handle_capability_event] to map one onto its own protocol, building the payload itself as CapabilityEvent has no to_payload(). An application that wants to expose one to a frontend can instead listen for it with [@agent.on_event][pydantic_ai.agent.Agent.on_event] and emit an application CustomEvent carrying the public payload:

from dataclasses import dataclass

from pydantic_ai import Agent, CapabilityEvent, CustomEvent, RunContext

SEARCH_INDEX = 'search_index'


@dataclass(kw_only=True)
class IndexRebuiltEvent(CapabilityEvent, namespace=SEARCH_INDEX):
    documents: int


@dataclass(kw_only=True)
class SearchReadyEvent(CustomEvent):
    documents: int


agent = Agent('test')


@agent.on_event(IndexRebuiltEvent)
async def republish(ctx: RunContext, event: IndexRebuiltEvent) -> None:
    await ctx.emit(SearchReadyEvent(documents=event.documents))

(This example is complete, it can be run "as is")

The listener belongs to the application rather than to a capability, so it is one of the places application CustomEvents can be emitted. [Hooks.on.event][pydantic_ai.capabilities.Hooks] does the same job when you want a Hooks capability anyway — for the other hook families, or to choose where it sits among the other capabilities.

The namespace and event name form the serialized kind (for example, workspace.file_read), and the event name is derived from the class name unless you pass an explicit name=. A namespace is required: defining a CapabilityEvent subclass without one raises TypeError there and then, rather than letting an unnamespaced event reach the stream. You only give it once per family, though — an event subclassing another capability event inherits its namespace and contributes just its own name, so a shared base is the tidiest way to define a family — and a subclass can pass its own namespace= to move out of the one it inherited. Mark a base that only carries the namespace and fields common to the family abstract=True, and it stays out of the registry and can't be emitted itself, while its subclasses register as usual. Decorate it with @dataclass like any other event: an undecorated base contributes no fields at all, which is rejected rather than left to surface as a payload quietly missing them. The kind is the event's wire identifier, so renaming the class renames the tag with it, breaking compatibility wherever events outlive the emitting process — durable execution histories and caches, persisted event logs, subscribers matching on the kind. A capability published as a library should pin name= on each of its events. Kinds are registered when the class is defined and must be unique within the process; re-executing the same class definition (as when re-running a notebook cell) replaces the registration. Import the module defining an event before creating the adapter that deserializes it, as each pydantic TypeAdapter captures the kinds registered when it is created. Otherwise the event becomes an [UnknownCapabilityEvent][pydantic_ai.messages.UnknownCapabilityEvent] and a UserWarning is emitted, without losing payload fields; serializing it again preserves the wire representation so a later consumer can recover the typed event.

Reacting to events

Use [@on_event][pydantic_ai.capabilities.on_event] on an async capability method to react to selected event classes. For example, a repository-context capability can enqueue instructions immediately after a file-system capability reports reading a repository guidance file:

from dataclasses import dataclass
from typing import Any

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

REPO_CONTEXT = 'repo_context'


@dataclass(kw_only=True)
class FileReadEvent(CapabilityEvent, namespace=REPO_CONTEXT):
    path: str


@dataclass(kw_only=True)
class DirectoryListedEvent(CapabilityEvent, namespace=REPO_CONTEXT):
    path: str


class RepoContext(AbstractCapability[Any]):
    @on_event(FileReadEvent, DirectoryListedEvent)
    async def _follow_discovered_instructions(
        self,
        ctx: RunContext[Any],
        event: FileReadEvent | DirectoryListedEvent,
    ) -> None:
        if event.path.endswith('AGENTS.md'):
            ctx.enqueue(f'Follow the instructions in {event.path}.')

Filtering is explicit: the decorator uses isinstance against the classes passed to it. A bare @on_event receives the full [AgentStreamEvent][pydantic_ai.messages.AgentStreamEvent] union, including model response deltas, tool call and result events, deferred and enqueued-message events, custom events, and capability events.

Name the classes when you can. Beyond narrowing the event argument for the type checker, the classes are what let dispatch skip a capability without descending into it: a capability is only woken for events one of its listeners accepts. A bare @on_event — or an overridden [on_event()][pydantic_ai.capabilities.AbstractCapability.on_event], whose dispatch isn't knowable in advance — opts that capability into every event, and a capability that combines others reports the union of its children's. If you override on_event() and can describe what it dispatches, override [listens_to()][pydantic_ai.capabilities.AbstractCapability.listens_to] alongside it to say so.

Listeners run sequentially in capability order, and marked methods within one capability run in definition order. The emitting capability also receives its own events. By default, listeners run when the event reaches its position in the stream, so listener ordering always matches stream ordering and listener work does not add to the emitter's latency. For events emitted during tool execution, listeners run before the next model request. An event emitted during before_model_request may only reach listeners after that request begins, with the same as-soon-as-possible timing as [ctx.enqueue()][pydantic_ai.tools.RunContext.enqueue].

Decision events that need listener mutations before the emitter continues declare dispatch='immediate' on the event class:

Building on the workspace capability above, a write can announce itself before it happens and let a listener veto it:

from dataclasses import dataclass
from typing import Any

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

WORKSPACE = 'workspace'


@dataclass(kw_only=True)
class FileWriteStartEvent(
    CapabilityEvent, namespace=WORKSPACE, dispatch='immediate'
):
    path: str
    cancelled: bool = False
    cancel_reason: str | None = None

    def cancel(self, reason: str | None = None) -> None:
        self.cancelled = True
        self.cancel_reason = reason


workspace = FunctionToolset()


@workspace.tool
async def write_file(ctx: RunContext[Any], path: str, content: str) -> str:
    event = await ctx.emit(FileWriteStartEvent(path=path))  # (1)!
    if event.cancelled:  # (2)!
        return f'Refused to write {path}: {event.cancel_reason}'
    return f'Wrote {path}'


class ProtectGitDirectory(AbstractCapability[Any]):
    @on_event(FileWriteStartEvent)
    async def _veto_writes_to_git(
        self, ctx: RunContext[Any], event: FileWriteStartEvent
    ) -> None:
        if event.path.startswith('.git/'):
            event.cancel('.git/ is managed by the repository, not the agent')

1. emit returns only once every listener has run, and returns the same instance it was given.
2. So the decision is readable immediately after, on either the returned event or the emitter's own reference.

(This example is complete, it can be run "as is")

For immediate dispatch, Pydantic AI buffers the event before invoking listeners, but stream consumers only receive it once all listeners have run, so they never observe a decision half-made. Attribution is stamped on the event in place, so after await ctx.emit(event) the emitter can read event.cancelled off its own reference (the same instance emit returns), and an event emitted by a listener appears after the decision event in the stream. Inline events are still delivered exactly once, including when the same event instance is re-emitted. Stream-dispatch listeners run inside user-defined [wrap_run_event_stream()][pydantic_ai.capabilities.AbstractCapability.wrap_run_event_stream] wrappers.

!!! note Any on_event listener automatically enables streaming for an otherwise non-streaming agent.run(): model requests are made with the provider's streaming API so events exist to listen to. Providers treat streaming and non-streaming requests the same in almost all respects, but if you need to guarantee non-streamed requests, don't attach listeners.

Provider-adaptive tools

[WebSearch][pydantic_ai.capabilities.WebSearch], [WebFetch][pydantic_ai.capabilities.WebFetch], [ImageGeneration][pydantic_ai.capabilities.ImageGeneration], [XSearch][pydantic_ai.capabilities.XSearch], and [MCP][pydantic_ai.capabilities.MCP] each cover a single capability (web search, URL fetch, image generation, X search, MCP) across two implementations:

• Native — invoked by the model provider when the model supports it. The work happens on the provider's side (e.g. Anthropic's web search runs server-side, returning results inline).
• Local — runs in your Python process. Used when the model doesn't support the native tool; your code does the work (e.g. calling DuckDuckGo directly).

Because these capabilities contribute model-facing tools, their id, description, and defer_loading fields are meaningful: set description and defer_loading when that tool should stay hidden until the model loads the matching workflow with the load_capability tool. Each of these covers a single fixed concern, so id already defaults to a stable value ('web_search', 'web_fetch', 'image_generation', 'x_search'; MCP derives one from the server URL) — which is what durable execution identifies the toolset they contribute by, so it works unconfigured. Set id only to rename it, and see building custom capabilities for what happens when two share one. This includes [ImageGeneration][pydantic_ai.capabilities.ImageGeneration] when image generation should only be available for an image-specific workflow, whether it resolves to a native image tool or a fallback subagent tool.

A [Capability][pydantic_ai.capabilities.Capability] contributing several toolsets names each one '{capability_id}_{index}' — those are the ids durable execution registers them under.

Configure each side via the native= and local= kwargs. native= accepts True (use the capability's default native tool instance), False (disable native), an explicit instance like WebSearchTool(...) for fine-grained config, or a callable taking [RunContext][pydantic_ai.tools.RunContext] that returns a native tool or None (see Dynamic Configuration). A factory that returns None omits the native tool for that request. local= accepts True (the bundled local fallback, on capabilities that have one — WebSearch and WebFetch), False (disable local), a named strategy string where supported, or any callable, [Tool][pydantic_ai.tools.Tool], or [AbstractToolset][pydantic_ai.toolsets.AbstractToolset]. Optional installs needed for the local fallback are opt-in — the capability raises a [UserError][pydantic_ai.exceptions.UserError] at construction (with an install hint) when you ask for a local strategy whose extra isn't installed.

!!! note "None on a capability with a fallback_model" [XSearch][pydantic_ai.capabilities.XSearch] and [ImageGeneration][pydantic_ai.capabilities.ImageGeneration] route unsupported models to a subagent instead of a local tool. Once a fallback_model is set, a factory returning None no longer omits anything: the subagent tool stays offered to the model, even one that supports the native tool, and calling it raises [UserError][pydantic_ai.exceptions.UserError] — see X Search and Image Generation.

from pydantic_ai import Agent
from pydantic_ai.capabilities import MCP, ImageGeneration, WebFetch, WebSearch, XSearch

agent = Agent(
    'anthropic:claude-fable-5',
    capabilities=[
        # Native when supported; DuckDuckGo fallback on unsupported models
        WebSearch(local='duckduckgo'),
        # Native when supported; markdownify-based fallback on unsupported models
        WebFetch(local=True),
        # Native when supported; subagent fallback via `fallback_model`
        ImageGeneration(fallback_model='openai:gpt-5.6-sol'),
        # Native on xAI; on other models, explicitly delegate to an xAI model
        XSearch(fallback_model='xai:grok-4.3'),
        # Runs the MCP server locally by default; pass `native=True` to also advertise native MCP
        MCP('https://mcp.example.com/api'),
    ],
)

MCP defaults the other way from the others: because MCP carries credentials, it runs locally by default and you opt into native MCP with native=True. The others default to native and you opt into local with local=.

[XSearch][pydantic_ai.capabilities.XSearch] is slightly different from [WebSearch][pydantic_ai.capabilities.WebSearch] and [WebFetch][pydantic_ai.capabilities.WebFetch]: there is no default non-xAI fallback. If your agent is not running on an xAI model, set fallback_model explicitly to an xAI model that supports [XSearchTool][pydantic_ai.native_tools.XSearchTool].

Some constraint fields require the native tool (the bundled local fallback can't enforce them) — passing them locks the capability to the native path. If the model doesn't support the native tool, the capability raises a [UserError][pydantic_ai.exceptions.UserError].

# Limit to 5 searches per run — requires native (the local fallback can't track call count)
WebSearch(max_uses=5)

# Only fetch example.com — enforced locally when native is unavailable
WebFetch(allowed_domains=['example.com'], local=True)

Building your own

All five capabilities are subclasses of [NativeOrLocalTool][pydantic_ai.capabilities.NativeOrLocalTool], which you can use directly or subclass to build your own provider-adaptive tools. For example, to pair [CodeExecutionTool][pydantic_ai.native_tools.CodeExecutionTool] with a local fallback:

from pydantic_ai.native_tools import CodeExecutionTool
from pydantic_ai.capabilities import NativeOrLocalTool

cap = NativeOrLocalTool(native=CodeExecutionTool(), local=my_local_executor)

Third-party capabilities

Third-party packages publish capabilities of their own — see Third-Party Capabilities for the ecosystem, and Publishing capabilities for making your own capability available to others.