feat(poc) add hook system for tools that require approval (eg. hitl)

[mattijn]

This PR adds a hook system that lets you intercept tool calls for eg human approval before execution.

It's merely a proof of concept for myself with no intention to use it (yet), but maybe it is of interest here too?

I think I'm not keen on manually traversing agent graphs, so I tried to see if we can have a public agent.on.xx interface that works directly with observing the nodes.

The current implementation is a registry-based approach that handles hooks without requiring graph iteration.

I also explored using traitlets for reactive model/instruction changes (e.g., dynamically switching models or instructions based on user context). I think traitlets integration for dynamic agent configuration/observing still remains interesting for future exploration.

But to start somehwere, first this for before/after tool calling.

Example

from pydantic_ai import Agent, Tool, ModelRetry

# Create a tool that needs approval
async def plan_holiday(destination: str, duration: str):
    return f"Holiday plan completed: {destination} for {duration} days!"

my_tool = Tool(name="holiday_planner", function=plan_holiday)
agent = Agent(model=model, tools=[my_tool])

# Set up approval hook
async def request_approval(context):
    tool_call = context['tool_call']
    args = tool_call.args_as_dict()
    
    print(f"Approve holiday to {args['destination']} for {args['duration']} days?")
    user_input = input("y/n: ").lower().strip()
    
    if user_input == 'y':
        return None  # Allow execution
    else:
        return ModelRetry("Holiday denied by user")

# Attach the hook
agent.on.tool(my_tool).before = request_approval

# Run - hook will intercept and ask for approval
await agent.run("Plan a holiday to Switzerland for 7 days")

Current options for the agent.on API

• agent.on.tool(my_tool).before = func - Run before tool execution
• agent.on.tool(my_tool).after = func - Run after tool execution
• agent.on.tool(my_tool).error = func - Run on tool errors
• Return ModelRetry to deny execution and suggest alternatives
• Return None to allow execution
• Access tool args via context['tool_call'].args_as_dict()

How it works

When you set agent.on.tool(my_tool).before = func, the system:

1. Registers your hook in a central registry
2. Nodes query the registry during execution via ctx.deps.hook_registry
3. Hooks run automatically during tool execution (no graph iteration needed)
4. No graph modification - existing Pydantic-AI nodes work unchanged

Testing

See test_reactivenodes.ipynb for a working example with simple human-in-the-loop holiday planning approval notebook.

[DouweM]

@mattijn Thanks Mattijn! Focusing on the HITL tool approval feature for now, and not looking at the implementation yet (I agree we'll want hooks at some point for what it's worth): what do you think of #2581 and specifically the HITL approval example at

pydantic-ai/tests/test_agent.py

Line 4266 in 8989b5b

async def test_hitl_tool_approval():

?

Note that that assumes that you want to end an agent run when approval is needed, and then resume it once approval (or denial) is received, as HITL input can take a while to be provided and would typically come in a subsequent request as opposed to being done inline inside the tool function.

[mattijn]

I missed that PR, thanks for sharing @DouweM! Regarding public facing api in #2581, only concern I have is that:

@agent.tool_plain(requires_approval=True)

limit its semantics for merely human approval, but pre-execution interventions can be wider.
But as first step towards generic hook support👍

[DouweM]

@mattijn That feature was now merged into v1.0.0b1 and addresses the biggest use case people had been requesting hooks for! So for now I'm going to hold off on adding generic hook support until we get some more requests with specific use cases in mind. If you have any, feel free to file an issue :)