Hooks

Hooks let you intercept and modify agent behavior at every stage of a run — model requests, tool calls, streaming events — using simple decorators or constructor arguments. No subclassing needed.

The Hooks capability is the recommended way to add lifecycle hooks for application-level concerns like logging, metrics, and lightweight validation. For reusable capabilities that combine hooks with tools, instructions, or model settings, subclass AbstractCapability instead — see Building custom capabilities.

Quick start

Create a Hooks instance, register hooks via @hooks.on.* decorators, and pass it to your agent:

hooks_decorator.py

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

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 to the model')
    #> Sending 1 messages to the model
    return request_context


agent = Agent('test', capabilities=[hooks])
result = agent.run_sync('Hello!')
print(result.output)
#> success (no tool calls)

Registering hooks

Decorator registration

The hooks.on namespace provides decorator methods for every lifecycle hook. Use them as bare decorators or with parameters:

# Bare decorator
@hooks.on.before_model_request
async def my_hook(ctx, request_context):
    return request_context

# With parameters (timeout, tool filter)
@hooks.on.before_model_request(timeout=5.0)
async def my_timed_hook(ctx, request_context):
    return request_context

Multiple hooks can be registered for the same event — they fire in registration order.

Constructor kwargs

You can also pass hook functions directly to the Hooks constructor:

hooks_constructor.py

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


async def log_request(ctx: RunContext, request_context: ModelRequestContext) -> ModelRequestContext:
    print(f'Sending {len(request_context.messages)} messages to the model')
    #> Sending 1 messages to the model
    return request_context


agent = Agent('test', capabilities=[Hooks(before_model_request=log_request)])
result = agent.run_sync('Hello!')
print(result.output)
#> success (no tool calls)

Both sync and async hook functions are accepted. Sync functions are automatically wrapped for async execution.

On-demand hooks

Hooks is a capability, so it can be loaded on demand just like any other capability. This is useful for optional, user-requested behavior such as verbose request logging:

deferred_hooks_capability.py

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

request_logging_hooks = Hooks(
    id='request-logging',
    description='Use when the user asks for verbose request diagnostics.',
    defer_loading=True,
)


@request_logging_hooks.on.before_model_request
async def log_request(
    ctx: RunContext[None],
    request_context: ModelRequestContext,
) -> ModelRequestContext:
    print(f'Model request at step {ctx.run_step}: {len(request_context.messages)} messages')
    return request_context


agent = Agent('openai-responses:gpt-5.4', capabilities=[request_logging_hooks])

Pydantic AI skips hooks owned by a deferred Hooks instance until its capability is loaded.

Use on-demand hooks for optional behavior that only applies after the capability is loaded. For human-in-the-loop tool approval, pass requires_approval=True when registering a tool, raise ApprovalRequired for conditional approval, or wrap a toolset with ApprovalRequiredToolset.

Hook types

Run hooks

hooks.on.	Constructor kwarg	AbstractCapability method
before_run	before_run=	before_run
after_run	after_run=	after_run
run	run=	wrap_run
run_error	run_error=	on_run_error

Run hooks fire once per agent run. wrap_run (registered via hooks.on.run) wraps the entire run and supports error recovery.

Node hooks

hooks.on.	Constructor kwarg	AbstractCapability method
before_node_run	before_node_run=	before_node_run
after_node_run	after_node_run=	after_node_run
node_run	node_run=	wrap_node_run
node_run_error	node_run_error=	on_node_run_error

Node hooks fire for each graph step (UserPromptNode, ModelRequestNode, CallToolsNode).

Note

wrap_node_run hooks are called automatically by agent.run(), agent.run_stream(), and agent_run.next(), but not when iterating with bare async for node in agent_run:.

Model request hooks

hooks.on.	Constructor kwarg	AbstractCapability method
before_model_request	before_model_request=	before_model_request
after_model_request	after_model_request=	after_model_request
model_request	model_request=	wrap_model_request
model_request_error	model_request_error=	on_model_request_error

Model request hooks fire around each LLM call. ModelRequestContext bundles model, messages, model_settings, and model_request_parameters. To swap the model for a given request, set request_context.model to a different Model instance.

To skip the model call entirely, raise SkipModelRequest(response) from before_model_request or model_request (wrap).

Note

These hooks fire once per model turn, even when a provider pauses mid-turn (Anthropic pause_turn) or returns a background response (OpenAI background mode) and the agent transparently continues it. before_model_request runs before the turn starts, wrap_model_request wraps the whole turn including any continuations, and after_model_request receives the single completed ModelResponse.

When a run resumes a suspended turn from message_history, before_model_request and wrap_model_request see that suspended ModelResponse as the last entry in request_context.messages: it’s the continuation seed that will be echoed back to the provider, mirroring what actually goes over the wire.

Tool validation hooks

hooks.on.	Constructor kwarg	AbstractCapability method
before_tool_validate	before_tool_validate=	before_tool_validate
after_tool_validate	after_tool_validate=	after_tool_validate
tool_validate	tool_validate=	wrap_tool_validate
tool_validate_error	tool_validate_error=	on_tool_validate_error

Validation hooks fire when the model’s JSON arguments are parsed and validated. All tool hooks receive call (ToolCallPart) and tool_def (ToolDefinition) parameters.

Note

Tool validation and execution hooks only fire for function tools. Internal output tools (used to deliver structured output) are not user-facing and are skipped.

To skip validation, raise SkipToolValidation(args) from before_tool_validate or tool_validate (wrap).

A tool call can only be deferred once its arguments have been validated, since whoever resolves the deferral is shown those arguments. So ApprovalRequired and CallDeferred can be raised from after_tool_validate (and from tool_validate after its handler() has returned), but raising them from before_tool_validate, from tool_validate before it calls handler(), or from tool_validate_error is a UserError. To decide per tool rather than per capability, use the tool’s args_validator.

Tool execution hooks

hooks.on.	Constructor kwarg	AbstractCapability method
before_tool_execute	before_tool_execute=	before_tool_execute
after_tool_execute	after_tool_execute=	after_tool_execute
tool_execute	tool_execute=	wrap_tool_execute
tool_execute_error	tool_execute_error=	on_tool_execute_error

Execution hooks fire when the tool function runs. args is always the validated dict[str, Any].

To skip execution, raise SkipToolExecution(result) from before_tool_execute or tool_execute (wrap).

Every execution hook can defer the call — the arguments are validated by this point — but raise ApprovalRequired/CallDeferred from before_tool_execute (or from tool_execute before it calls handler()). A deferral from after_tool_execute, or from tool_execute after handler() has returned, is accepted but happens too late to be useful: the tool function already ran, so its side effects happened and its result is discarded.

Output validation hooks

hooks.on.	Constructor kwarg	AbstractCapability method
before_output_validate	before_output_validate=	before_output_validate
after_output_validate	after_output_validate=	after_output_validate
output_validate	output_validate=	wrap_output_validate
output_validate_error	output_validate_error=	on_output_validate_error

Output validation hooks fire when structured output is parsed against the output schema. They do not fire for plain text or image output. All output hooks receive an output_context (OutputContext) parameter.

Note

During streaming, output validation hooks fire on every partial validation attempt as well as the final result. Output processing hooks fire only when partial validation succeeds, and on the final result. Check ctx.partial_output in your hooks to distinguish partial from final results and avoid expensive work on partials.

Output processing hooks

hooks.on.	Constructor kwarg	AbstractCapability method
before_output_process	before_output_process=	before_output_process
after_output_process	after_output_process=	after_output_process
output_process	output_process=	wrap_output_process
output_process_error	output_process_error=	on_output_process_error

Output processing hooks fire when the output is processed — extracting values, calling output functions, and running output validators.

See Output hooks for the full lifecycle, signatures, and details on how output validators interact with processing hooks.

Tool preparation

hooks.on.	Constructor kwarg	AbstractCapability method
prepare_tools	prepare_tools=	prepare_tools
prepare_output_tools	prepare_output_tools=	prepare_output_tools

Filters or modifies tool definitions the model sees on each step.

prepare_tools handles function tools; prepare_output_tools handles output tools separately, with ctx.max_retries reflecting the output retry budget. Both run as PreparedToolset wrappers — the result flows into the model’s request and ToolManager.tools, so filtering also blocks tool execution.

Deferred tool call hook

hooks.on.	Constructor kwarg	AbstractCapability method
deferred_tool_calls	deferred_tool_calls=	handle_deferred_tool_calls

Resolves deferred tool calls (approval-required or externally-executed) inline during a run. The hook receives a DeferredToolRequests and returns a DeferredToolResults (or None to decline). Multiple registered hooks accumulate: each receives the still-unresolved requests and can resolve some or all of them.

hooks_deferred_tool_calls.py

from pydantic_ai import Agent, DeferredToolRequests, DeferredToolResults, RunContext
from pydantic_ai.capabilities import Hooks

hooks = Hooks()


@hooks.on.deferred_tool_calls
async def auto_approve(
    ctx: RunContext, *, requests: DeferredToolRequests
) -> DeferredToolResults:
    return requests.build_results(approve_all=True)


agent = Agent('test', capabilities=[hooks])


@agent.tool_plain(requires_approval=True)
def delete_file(path: str) -> str:
    return f'File {path!r} deleted'

For pure application-level handler registration without other hooks, the dedicated HandleDeferredToolCalls capability is more concise — see Resolving deferred calls with a handler.

Event stream hooks

hooks.on.	Constructor kwarg	AbstractCapability method
run_event_stream	run_event_stream=	wrap_run_event_stream
event	event=	(per-event convenience)

run_event_stream wraps the full event stream as an async generator. event is a convenience — it fires for each individual event during a streamed run. Tool and model events flow through this stream, along with framework events such as EnqueuedMessagesEvent when queued messages enter run history:

hooks_event.py

from pydantic_ai import Agent, AgentStreamEvent, RunContext
from pydantic_ai.capabilities import Hooks

hooks = Hooks()
event_count = 0


@hooks.on.event
async def count_events(ctx: RunContext, event: AgentStreamEvent) -> AgentStreamEvent:
    global event_count
    event_count += 1
    return event


agent = Agent('test', capabilities=[hooks])

Tool hook filtering

Tool hooks (validation and execution) support a tools parameter to target specific tools by name:

hooks_tool_filter.py

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

hooks = Hooks()
call_log: list[str] = []


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


agent = Agent('test', capabilities=[hooks])


@agent.tool_plain
def send_email(to: str) -> str:
    return f'sent to {to}'


result = agent.run_sync('Send an email to test@example.com')
print(call_log)
#> ['audit: send_email']

The tools parameter accepts a sequence of tool names. The hook only fires for matching tools — other tool calls pass through unaffected.

Timeouts

Each hook supports an optional timeout in seconds. If the hook exceeds the timeout, a HookTimeoutError is raised:

hooks_timeout.py

import asyncio

from pydantic_ai import Agent, ModelRequestContext, RunContext
from pydantic_ai.capabilities import Hooks, HookTimeoutError

hooks = Hooks()


@hooks.on.before_model_request(timeout=0.01)
async def slow_hook(
    ctx: RunContext, request_context: ModelRequestContext
) -> ModelRequestContext:
    await asyncio.sleep(10)  # Will be interrupted by timeout
    return request_context  # pragma: no cover


agent = Agent('test', capabilities=[hooks])
try:
    agent.run_sync('Hello')
except HookTimeoutError as e:
    print(f'Hook timed out: {e.hook_name} after {e.timeout}s')
    #> Hook timed out: before_model_request after 0.01s

Timeouts are set via the decorator parameter (@hooks.on.before_model_request(timeout=5.0)) or via the constructor when using kwargs.

Wrap hooks

Wrap hooks let you surround an operation with setup/teardown logic. In the hooks.on namespace, wrap hooks drop the wrap_ prefix — hooks.on.model_request corresponds to wrap_model_request:

hooks_wrap.py

from pydantic_ai import Agent, ModelRequestContext, RunContext
from pydantic_ai.capabilities import Hooks, WrapModelRequestHandler
from pydantic_ai.messages import ModelResponse

hooks = Hooks()
wrap_log: list[str] = []


@hooks.on.model_request
async def log_request(
    ctx: RunContext, *, request_context: ModelRequestContext, handler: WrapModelRequestHandler
) -> ModelResponse:
    wrap_log.append('before')
    response = await handler(request_context)
    wrap_log.append('after')
    return response


agent = Agent('test', capabilities=[hooks])
result = agent.run_sync('Hello!')
print(wrap_log)
#> ['before', 'after']

Hook ordering

Within a single Hooks instance, before_*, after_*, and on_*_error fire in registration order (the order they were defined or passed to the constructor). wrap_* nests as middleware, with the first-registered wrapper as the outermost layer.

Across multiple capabilities, the composition rules apply: before_* fires in capability order, after_* fires in reverse capability order, and wrap_* nests as middleware with the first capability outermost.

Hook timing also affects what is populated on RunContext. Early run and node hooks can fire before the current step’s tool manager and model request parameters have been assembled. At that point ctx.available_tool_names can still include tool-search discoveries reconstructed from history, but ctx.tools and current request parameters may be empty or reflect the previous step. before_model_request and later model-request hooks see the request about to be sent, including the current function tools, native tools, and model settings. Tool and output hooks see the state for the call or output currently being processed.

For on-demand capabilities, ctx.loaded_capability_ids updates as soon as the load_capability tool runs. Function tools, native tools, and model settings from the loaded capability appear on the next model request, while hooks owned by that capability can only run for hook points reached after the capability has loaded.

Error hooks

Error hooks (*_error in the hooks.on namespace, on_*_error on AbstractCapability) use raise-to-propagate, return-to-recover semantics:

• Raise the original error — propagates unchanged (default)
• Raise a different exception — transforms the error
• Return a result — suppresses the error

See Error hooks for the full pattern and recovery types.

Triggering retries with ModelRetry and failures with ToolFailed

Hooks can raise ModelRetry to ask the model to try again with a custom message — the same exception used in tool functions and output validators.

Model request hooks (after_model_request, wrap_model_request, on_model_request_error):

• The retry message is sent back to the model as a RetryPromptPart
• after_model_request: the original response is preserved in message history so the model can see what it said
• wrap_model_request: the response is preserved only if the handler was called
• Retries count against the output side of the agent’s retry budget

Tool hooks (before/after_tool_validate, before/after_tool_execute, wrap_tool_execute, on_tool_execute_error):

• Converted to tool retry prompts, same as when a tool function raises ModelRetry
• Retries count against the tool’s max_retries limit

Output hooks (before/after_output_validate, before/after_output_process, wrap_output_process, on_output_process_error):

• Converted to retry prompts, same as when an output function raises ModelRetry
• For tool output, retries count against the tool’s max_retries limit
• For text output, retries count against the output side of the agent’s retry budget

ModelRetry from wrap_model_request, wrap_tool_execute, or wrap_output_process is control flow and bypasses the corresponding on_*_error hook. ToolFailed is control flow only at the tool boundary, so it bypasses on_tool_execute_error. From model-request and output-process hooks, ToolFailed is an ordinary exception and is passed to on_model_request_error or on_output_process_error.

Tool validation and execution hooks can also raise ToolFailed to report a failed tool result without consuming the tool’s retry budget. This has the same model-visible outcome and retry-budget behavior as raising ToolFailed from the tool function itself, and is useful when an error hook converts a third-party exception into a failure the model can see.

hooks_model_retry.py

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import Hooks
from pydantic_ai.exceptions import ModelRetry
from pydantic_ai.messages import ModelResponse
from pydantic_ai.models import ModelRequestContext

hooks = Hooks()


@hooks.on.after_model_request
async def check_response(
    ctx: RunContext,
    *,
    request_context: ModelRequestContext,
    response: ModelResponse,
) -> ModelResponse:
    if 'PLACEHOLDER' in str(response.parts):
        raise ModelRetry('Response contains placeholder text. Please provide real data.')
    return response


agent = Agent('test', capabilities=[hooks])
result = agent.run_sync('Hello')
print(result.output)
#> success (no tool calls)

By default, any exception other than ModelRetry or ToolFailed raised inside a tool escapes the tool boundary and aborts the entire run. A tool-execution hook lets you intercept these in one place — without editing every tool — and choose how each surfaces to the model. The distinction is the semantic one between requesting a retry and reporting a failure:

• raise ModelRetry for transient errors, where the same call might succeed if tried again;
• raise ToolFailed for definitive failures, where retrying won’t help and the model should see the result and adapt (choose another approach, tell the user, etc.).

The hook below makes that call based on an upstream status code — the per-error analogue of the MCP tool_error_behavior setting:

hooks_convert_tool_errors.py

from typing import Any

from pydantic_ai import Agent, RunContext, ToolCallPart, ToolDefinition, ToolReturnPart
from pydantic_ai.capabilities import Hooks
from pydantic_ai.exceptions import ModelRetry, ToolFailed
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart
from pydantic_ai.models.function import AgentInfo, FunctionModel


class UpstreamError(Exception):
    """Stand-in for an HTTP client error that carries the response status code."""

    def __init__(self, status_code: int, message: str):
        super().__init__(message)
        self.status_code = status_code


hooks = Hooks()


@hooks.on.tool_execute_error
async def convert_upstream_errors(
    ctx: RunContext[None],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: dict[str, Any],
    error: Exception,
) -> Any:
    if isinstance(error, UpstreamError):
        if error.status_code >= 500 or error.status_code == 429:
            # Transient: the same call might succeed, so ask the model to try again.
            raise ModelRetry(f'Upstream returned {error.status_code}, please try again.')
        # Definitive (e.g. 404, 403): retrying won't help — report it so the model can adapt.
        raise ToolFailed(f'Upstream returned {error.status_code}: {error}')
    raise error  # unrelated errors still abort the run


def model_fn(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    last_part = messages[-1].parts[-1]
    if isinstance(last_part, ToolReturnPart):
        return ModelResponse(parts=[TextPart(f'Could not fetch the document ({last_part.content}).')])
    return ModelResponse(parts=[ToolCallPart('get_document', {'doc_id': 42}, tool_call_id='call-1')])


agent = Agent(FunctionModel(model_fn), capabilities=[hooks])


@agent.tool_plain
def get_document(doc_id: int) -> str:
    raise UpstreamError(404, f'document {doc_id} not found')


result = agent.run_sync('Fetch document 42')
print(result.output)
#> Could not fetch the document (Upstream returned 404: document 42 not found).

Because the failure was raised as ToolFailed rather than ModelRetry, the model receives it as a ToolReturnPart with outcome='failed' and decides what to do next, instead of burning a retry on a call that can’t succeed.

When to use Hooks vs AbstractCapability

Use Hooks	Use AbstractCapability
Application-level hooks (logging, metrics)	Reusable, packaged capabilities
Quick one-off interceptors	Combined tools + hooks + instructions + settings
No configuration state needed	Complex per-run state management
Single-file scripts	Multi-agent shared behavior