Add metadata support to deferred tool exceptions

Enables CallDeferred and ApprovalRequired exceptions to carry arbitrary
metadata via an optional `metadata` parameter. The metadata is accessible
in DeferredToolRequests.metadata keyed by tool_call_id.

This allows tools to:
- Provide cost/time estimates for approval decisions
- Include task IDs for external execution tracking
- Store context about why approval is required
- Attach priority or urgency information

Backward compatible - metadata defaults to empty dict if not provided.

Author: cjohnhanson

docs/deferred-tools.md

@@ -320,6 +320,111 @@ async def main():
 
 _(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_
 
+## Attaching Metadata to Deferred Tools
+
+Both [`CallDeferred`][pydantic_ai.exceptions.CallDeferred] and [`ApprovalRequired`][pydantic_ai.exceptions.ApprovalRequired] exceptions accept an optional `metadata` parameter that allows you to attach arbitrary context information to deferred tool calls. This metadata is then available in the [`DeferredToolRequests.metadata`][pydantic_ai.tools.DeferredToolRequests.metadata] dictionary, keyed by the tool call ID.
+
+Common use cases for metadata include:
+
+- Providing cost estimates or time estimates for approval decisions
+- Including task IDs or tracking information for external execution
+- Storing context about why approval is required
+- Attaching priority or urgency information
+
+Here's an example showing how to use metadata with both approval-required and external tools:
+
+```python {title="deferred_tools_with_metadata.py"}
+from pydantic_ai import (
+    Agent,
+    ApprovalRequired,
+    CallDeferred,
+    DeferredToolRequests,
+    DeferredToolResults,
+    RunContext,
+    ToolApproved,
+    ToolDenied,
+)
+
+agent = Agent('openai:gpt-5', output_type=[str, DeferredToolRequests])
+
+
+@agent.tool
+def expensive_compute(ctx: RunContext, task_id: str) -> str:
+    if not ctx.tool_call_approved:
+        raise ApprovalRequired(
+            metadata={
+                'task_id': task_id,
+                'estimated_cost_usd': 25.50,
+                'estimated_time_minutes': 15,
+                'reason': 'High compute cost',
+            }
+        )
+    return f'Task {task_id} completed'
+
+
+@agent.tool
+async def external_api_call(ctx: RunContext, endpoint: str) -> str:
+    # Schedule the external API call and defer execution
+    task_id = f'api_call_{ctx.tool_call_id}'
+
+    raise CallDeferred(
+        metadata={
+            'task_id': task_id,
+            'endpoint': endpoint,
+            'priority': 'high',
+        }
+    )
+
+
+result = agent.run_sync('Run expensive task-123 and call the /data endpoint')
+messages = result.all_messages()
+
+assert isinstance(result.output, DeferredToolRequests)
+requests = result.output
+
+# Handle approvals with metadata
+for call in requests.approvals:
+    metadata = requests.metadata.get(call.tool_call_id, {})
+    print(f"Approval needed for {call.tool_name}")
+    print(f"  Cost: ${metadata.get('estimated_cost_usd')}")
+    print(f"  Time: {metadata.get('estimated_time_minutes')} minutes")
+    print(f"  Reason: {metadata.get('reason')}")
+
+# Handle external calls with metadata
+for call in requests.calls:
+    metadata = requests.metadata.get(call.tool_call_id, {})
+    print(f"External call to {call.tool_name}")
+    print(f"  Task ID: {metadata.get('task_id')}")
+    print(f"  Priority: {metadata.get('priority')}")
+
+# Build results with approvals and external results
+results = DeferredToolResults()
+for call in requests.approvals:
+    metadata = requests.metadata.get(call.tool_call_id, {})
+    cost = metadata.get('estimated_cost_usd', 0)
+
+    if cost < 50:  # Approve if cost is under $50
+        results.approvals[call.tool_call_id] = ToolApproved()
+    else:
+        results.approvals[call.tool_call_id] = ToolDenied('Cost too high')
+
+for call in requests.calls:
+    metadata = requests.metadata.get(call.tool_call_id, {})
+    # Simulate getting result from external task
+    task_id = metadata.get('task_id')
+    results.calls[call.tool_call_id] = f'Result from {task_id}: success'
+
+result = agent.run_sync(message_history=messages, deferred_tool_results=results)
+print(result.output)
+"""
+I completed task-123 and retrieved data from the /data endpoint.
+"""
+```
+
+_(This example is complete, it can be run "as is")_
+
+The metadata dictionary can contain any JSON-serializable values and is entirely application-defined. If no metadata is provided when raising the exception, the tool call ID will still be present in the `metadata` dictionary with an empty dict as the value for backward compatibility.
+
 ## See Also
 
 - [Function Tools](tools.md) - Basic tool concepts and registration

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -879,6 +879,7 @@ async def process_tool_calls(  # noqa: C901
         calls_to_run = [call for call in calls_to_run if call.tool_call_id in calls_to_run_results]
 
     deferred_calls: dict[Literal['external', 'unapproved'], list[_messages.ToolCallPart]] = defaultdict(list)
+    deferred_metadata: dict[str, dict[str, Any]] = {}
 
     if calls_to_run:
         async for event in _call_tools(
@@ -890,6 +891,7 @@ async def process_tool_calls(  # noqa: C901
             usage_limits=ctx.deps.usage_limits,
             output_parts=output_parts,
             output_deferred_calls=deferred_calls,
+            output_deferred_metadata=deferred_metadata,
         ):
             yield event
 
@@ -923,6 +925,7 @@ async def process_tool_calls(  # noqa: C901
         deferred_tool_requests = _output.DeferredToolRequests(
             calls=deferred_calls['external'],
             approvals=deferred_calls['unapproved'],
+            metadata=deferred_metadata,
         )
 
         final_result = result.FinalResult(cast(NodeRunEndT, deferred_tool_requests), None, None)
@@ -940,10 +943,12 @@ async def _call_tools(
     usage_limits: _usage.UsageLimits,
     output_parts: list[_messages.ModelRequestPart],
     output_deferred_calls: dict[Literal['external', 'unapproved'], list[_messages.ToolCallPart]],
+    output_deferred_metadata: dict[str, dict[str, Any]],
 ) -> AsyncIterator[_messages.HandleResponseEvent]:
     tool_parts_by_index: dict[int, _messages.ModelRequestPart] = {}
     user_parts_by_index: dict[int, _messages.UserPromptPart] = {}
     deferred_calls_by_index: dict[int, Literal['external', 'unapproved']] = {}
+    deferred_metadata_by_index: dict[int, dict[str, Any]] = {}
 
     if usage_limits.tool_calls_limit is not None:
         projected_usage = deepcopy(usage)
@@ -978,10 +983,12 @@ async def handle_call_or_result(
                 tool_part, tool_user_content = (
                     (await coro_or_task) if inspect.isawaitable(coro_or_task) else coro_or_task.result()
                 )
-            except exceptions.CallDeferred:
+            except exceptions.CallDeferred as e:
                 deferred_calls_by_index[index] = 'external'
-            except exceptions.ApprovalRequired:
+                deferred_metadata_by_index[index] = e.metadata
+            except exceptions.ApprovalRequired as e:
                 deferred_calls_by_index[index] = 'unapproved'
+                deferred_metadata_by_index[index] = e.metadata
             else:
                 tool_parts_by_index[index] = tool_part
                 if tool_user_content:
@@ -1020,7 +1027,10 @@ async def handle_call_or_result(
     output_parts.extend([user_parts_by_index[k] for k in sorted(user_parts_by_index)])
 
     for k in sorted(deferred_calls_by_index):
-        output_deferred_calls[deferred_calls_by_index[k]].append(tool_calls[k])
+        call = tool_calls[k]
+        output_deferred_calls[deferred_calls_by_index[k]].append(call)
+        if k in deferred_metadata_by_index:
+            output_deferred_metadata[call.tool_call_id] = deferred_metadata_by_index[k]
 
 
 async def _call_tool(

pydantic_ai_slim/pydantic_ai/exceptions.py

@@ -67,18 +67,30 @@ class CallDeferred(Exception):
     """Exception to raise when a tool call should be deferred.
 
     See [tools docs](../deferred-tools.md#deferred-tools) for more information.
+
+    Args:
+        metadata: Optional dictionary of metadata to attach to the deferred tool call.
+            This metadata will be available in `DeferredToolRequests.metadata` keyed by `tool_call_id`.
     """
 
-    pass
+    def __init__(self, metadata: dict[str, Any] | None = None):
+        self.metadata = metadata or {}
+        super().__init__()
 
 
 class ApprovalRequired(Exception):
     """Exception to raise when a tool call requires human-in-the-loop approval.
 
     See [tools docs](../deferred-tools.md#human-in-the-loop-tool-approval) for more information.
+
+    Args:
+        metadata: Optional dictionary of metadata to attach to the deferred tool call.
+            This metadata will be available in `DeferredToolRequests.metadata` keyed by `tool_call_id`.
     """
 
-    pass
+    def __init__(self, metadata: dict[str, Any] | None = None):
+        self.metadata = metadata or {}
+        super().__init__()
 
 
 class UserError(RuntimeError):

pydantic_ai_slim/pydantic_ai/tools.py

@@ -147,6 +147,12 @@ class DeferredToolRequests:
     """Tool calls that require external execution."""
     approvals: list[ToolCallPart] = field(default_factory=list)
     """Tool calls that require human-in-the-loop approval."""
+    metadata: dict[str, dict[str, Any]] = field(default_factory=dict)
+    """Metadata for deferred tool calls, keyed by tool_call_id.
+
+    This contains any metadata that was provided when raising [`CallDeferred`][pydantic_ai.exceptions.CallDeferred]
+    or [`ApprovalRequired`][pydantic_ai.exceptions.ApprovalRequired] exceptions.
+    """
 
 
 @dataclass(kw_only=True)