企业官网建设流程全解析

Pydantic AI 延迟工具实战：审批工具、外部执行与两种解析路径

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

Pydantic AI 的延迟工具（Deferred Tools）机制面向如下场景：模型要调用的工具不能、也不应在当前 agent 运行中同步执行——它需要人工审批、依赖前端或外部服务提供结果，或由后台 worker 完成耗时任务。本文以 docs/deferred-tools.md 为核心，结合仓库源码实现，完整讲解如何声明需审批工具与外部工具、如何使用内联 handler（HandleDeferredToolCallscapability）与"结束运行→新运行带回结果"两条解析路径，以及如何在流式事件流中观察这些延迟调用，帮助读者直接落地 human-in-the-loop 审批与外部任务交接。

为什么需要延迟工具

在 agent 系统中，工具调用往往不是"可以立即安全执行"的。根据 docs/deferred-tools.md 的界定，至少有以下几类场景，工具不应或不能在同一 agent run、同一 Python 进程中同步执行：

• 需要用户先批准：如删除文件、转账等敏感操作，模型不能直接执行；
• 结果依赖外部方：结果需要由上游服务、前端或用户提供；
• 结果生成耗时过长：不值得让 agent 进程一直挂着等待。

为支持这些用例，Pydantic AI 提供延迟工具概念，分为两种：

• 需要审批的工具（require approval）
• 外部执行的工具（executed externally）

关键设计在于：当模型调用延迟工具时，run 不会崩溃或挂死，而是被"暂停"，由调用方选择两条解析路径之一。

两种解析路径：内联 handler 与两段式 run

路径一：内联解析（inline）

使用带 handler 的HandleDeferredToolCallscapability 解析部分或全部待处理调用。agent run 在单次调用内继续执行，无需结束再重启。适合解析器（审批闸门、外部服务客户端）与 agent 同进程的场景。

路径二：结束运行，由新运行带回结果（stop-the-world）

run 以DeferredToolRequests输出对象结束，其中携带延迟调用的信息；调用方收集审批/结果后，发起一次新的 agent run，同时传入原 run 的 message history 和一个DeferredToolResults对象。这次 follow-up 是独立的 agent run，拥有自己的run_id（不要复用被暂停 run 的 run_id），通过conversation_id保持暂停/恢复的关联。适合解析器位于 agent 进程之外的场景——例如 UI 适配器把待处理调用呈现给用户、拿到回复后再启动 follow-up run。

两条路径可以组合：handler 可以解析一部分调用，让其余的作为DeferredToolRequests输出冒泡，交由外层调用方处理。

stop-the-world 流程的前提：DeferredToolRequests必须在Agent的output_type中，以便正确推断 agent run 输出的可能类型。如果你的 agent 也可能在没有延迟工具的上下文中使用、又不想到处处理这个类型，可以改为在调用 [agent.run()]、[agent.run_sync()]、[agent.run_stream()] 或 [agent.iter()] 时传output_type参数。注意出于类型推断原因，运行时的output_type会覆盖构造时声明的类型，因此需显式把原始输出类型一并包含进去。

这一行为有源码佐证：在工具执行管线 _tool_execution.py 中，若出现延迟调用但DeferredToolRequests不在输出类型里，框架会直接报错，提示将其加入输出类型。

从源码结构看这两个数据类的设计意图也很清晰（pydantic_ai_slim/pydantic_ai/_deferred.py）：

• DeferredToolRequests持有三个字段：calls（待外部执行的工具调用，list[ToolCallPart]）、approvals（待审批的工具调用，list[ToolCallPart]）和metadata（按tool_call_id键控的dict[str, dict[str, Any]]，见 L37-L42）；
• DeferredToolResults的calls字段映射 tool call ID 到任意值 /ToolReturn/ 异常，approvals字段映射 tool call ID 到布尔 /ToolApproved/ToolDenied（见 L166-L173）；
• 其to_tool_call_results()方法在转换为管线内部格式时，会把True/False归一化为ToolApproved/ToolDenied，并把外部调用的普通值包装为ToolReturn（L181-L205）。

用 handler 解析延迟调用

推荐的延迟工具调用处理方式是注册一个HandleDeferredToolCallscapability：其 handler 接收DeferredToolRequests，返回解析了部分或全部请求的DeferredToolResults。工具执行管线会内联应用这些结果，agent run 在一次调用中继续执行，如同延迟工具正常返回了一样。

配置了 handler 后，DeferredToolRequests就不再需要声明为输出类型——除非你还希望未解析的调用冒泡给调用方（见下文）。

从实现看，HandleDeferredToolCalls是一个包装 handler 函数（sync 或 async 均可，内部用inspect.isawaitable判断是否需要 await）的轻量 dataclass，见 L51-L75：

• handler 返回包含部分/全部结果的DeferredToolResults时，对应调用被内联解析；
• handler返回None或在结果中省略某些调用时，下一个HandleDeferredToolCalls（或任何覆盖handle_deferred_tool_callshook 的 capability）获得处理机会，最终仍未解析的调用作为DeferredToolRequests输出冒泡。源码侧DeferredToolRequests.remaining()方法（L88-L99）负责计算结果与待处理请求的差集，全部解析则返回None；
• 允许冒泡的前提是把DeferredToolRequests加入 agent 的output_type——从而可以把内联处理与 stop-the-world 流程组合使用。

build_results()是便捷构造器：它验证每个 tool call ID 都对应正确类别的待处理请求（否则抛ValueError，见 L70-L80），并支持approve_all=True自动批准未显式列出的审批请求（填入默认ToolApproved()）。

from pydantic_ai import ( Agent, ApprovalRequired, CallDeferred, DeferredToolRequests, DeferredToolResults, RunContext, ToolDenied, ) from pydantic_ai.capabilities import HandleDeferredToolCalls async def handle_deferred( ctx: RunContext, requests: DeferredToolRequests ) -> DeferredToolResults: approvals: dict[str, bool | ToolDenied] = {} for call in requests.approvals: if call.tool_name == 'delete_file': approvals[call.tool_call_id] = ToolDenied('Deleting files is not allowed') else: approvals[call.tool_call_id] = True calls = {call.tool_call_id: f'(external result for {call.tool_name})' for call in requests.calls} return requests.build_results(approvals=approvals, calls=calls) agent = Agent( 'openai:gpt-5.2', capabilities=[HandleDeferredToolCalls(handler=handle_deferred)], ) @agent.tool_plain(requires_approval=True) def delete_file(path: str) -> str: return f'File {path!r} deleted' # (1)! @agent.tool def update_file(ctx: RunContext, path: str, content: str) -> str: if path == '.env' and not ctx.tool_call_approved: raise ApprovalRequired return f'File {path!r} updated: {content!r}' @agent.tool_plain async def send_to_worker(task: str) -> str: raise CallDeferred # (2)!

1. 这里永远不会执行到——handler 拒绝了这个调用，模型看到的是拒绝消息。
2. handler 为该外部调用提供结果，因此工具函数体只负责发出延迟信号。

如果你正在构建自定义 capability且需要自行解析审批或外部调用（例如暴露延迟工具的沙箱），请在你的 capability 上直接覆盖handle_deferred_tool_callshook，而不是再注册一个HandleDeferredToolCalls。同一个 hook 也可以通过Hookscapability 使用——见 Hooks。

下文将分别介绍 handler 可解析的两类延迟工具，以及每一类的 stop-the-world 替代流程。多个 capability 如何组合（含WrapperCapability与capabilities=[...]列表）见 Capabilities。

Human-in-the-Loop 工具审批

声明需审批的工具

如果工具函数总是需要审批，可以向@agent.tool装饰器、@agent.tool_plain装饰器、Tool类、FunctionToolset.tool装饰器或FunctionToolset.add_function()方法传入requires_approval=True。进入函数后，你可以假定该工具调用已经获得批准。

如果审批与否取决于工具调用的参数或 agent run context（依赖项、消息历史等），请在工具函数内抛出ApprovalRequired。RunContext.tool_call_approved属性在该调用已获批准时为True。

也可以在工具的args_validator中抛出它——args_validator在工具函数之前运行，让你在请人审批之前先拒绝无效参数。

如需对某个 toolset（如 MCP server）提供的工具调用要求审批，见ApprovalRequiredToolset文档。

实时会话注意：在 realtime session 中，审批必须内联解析，通常由HandleDeferredToolCallshandler 完成（capability hook 也可以）；没有任何解析器的调用会被每次拒绝。

!!! warning "审批不是针对不可信客户端的授权边界" 当你通过 UI 适配器对外提供服务时，审批决定由客户端随请求一起提交，适配器没有服务端记录来核对它发出过哪些工具调用。能触达该端点的客户端可以批准它自己发起的工具调用。Human-in-the-loop 审批防的是模型未经人工签核自行行动；它不能替代对适配器端点做认证、并在敏感操作的工具函数内部强制鉴权——工具函数无论调用如何进入历史都会执行。这一约束适用于任何接受客户端message_history的端点，不只是适配器——见 Trust boundary for client-supplied history 与 Trust model for client-submitted messages。

审批流程细节（stop-the-world 流程）

当模型调用了需要审批的工具，agent run 会以DeferredToolRequests输出对象结束，其approvals列表持有ToolCallPart，包含工具名、已校验的参数和唯一的 tool call ID。

收集到用户的批准/拒绝后，构造一个DeferredToolResults，其approvals字典把每个 tool call ID 映射为：

• 一个布尔值（True/False）；
• 一个ToolApproved对象（可选override_args，批准时可用它替换原始参数，见 L103-L109）；
• 或一个ToolDenied对象（可选自定义message供模型看到，见 L112-L121）。

还可以在DeferredToolResults上提供metadata字典：每个 tool call ID 映射到一个元数据字典，可在工具的RunContext.tool_call_metadata属性中取到。然后把这个DeferredToolResults对象与原 run 的 message history 一起，作为deferred_tool_results传给 agent 的任一运行方法。

下面是一个完整示例：所有文件删除、以及对受保护文件的更新都要求审批（该示例完整、可直接运行）：

from pydantic_ai import ( Agent, ApprovalRequired, DeferredToolRequests, DeferredToolResults, RunContext, ToolDenied, ) agent = Agent('openai:gpt-5.2', output_type=[str, DeferredToolRequests]) PROTECTED_FILES = {'.env'} @agent.tool def update_file(ctx: RunContext, path: str, content: str) -> str: if path in PROTECTED_FILES and not ctx.tool_call_approved: raise ApprovalRequired(metadata={'reason': 'protected'}) # (1)! return f'File {path!r} updated: {content!r}' @agent.tool_plain(requires_approval=True) def delete_file(path: str) -> str: return f'File {path!r} deleted' result = agent.run_sync('Delete `__init__.py`, write `Hello, world!` to `README.md`, and clear `.env`') messages = result.all_messages() assert isinstance(result.output, DeferredToolRequests) requests = result.output print(requests) """ DeferredToolRequests( calls=[], approvals=[ ToolCallPart( tool_name='update_file', args={'path': '.env', 'content': ''}, tool_call_id='update_file_dotenv', ), ToolCallPart( tool_name='delete_file', args={'path': '__init__.py'}, tool_call_id='delete_file', ), ], metadata={'update_file_dotenv': {'reason': 'protected'}}, ) """ results = DeferredToolResults() for call in requests.approvals: result = False if call.tool_name == 'update_file': # Approve all updates result = True elif call.tool_name == 'delete_file': # deny all deletes result = ToolDenied('Deleting files is not allowed') results.approvals[call.tool_call_id] = result result = agent.run_sync( 'Now create a backup of README.md', # (2)! message_history=messages, deferred_tool_results=results, ) print(result.output) """ Here's what I've done: - Attempted to delete __init__.py, but deletion is not allowed. - Updated README.md with: Hello, world! - Cleared .env (set to empty). - Created a backup at README.md.bak containing: Hello, world! If you want a different backup name or format (e.g., timestamped like README_2025-11-24.bak), let me know. """

1. 可选的metadata参数可向延迟工具调用附加任意上下文，按tool_call_id键控，可从DeferredToolRequests.metadata访问。
2. 第二次 agent run 从第一次 run 中断处继续，提供审批结果，并可选携带新的user_prompt给模型补充指令。

从该示例的消息历史可以看到完整的暂停—恢复过程形态：第一个ModelRequest携带用户指令，ModelResponse发出三个ToolCallPart；follow-up run 先追加一条ModelRequest，其中包含两个审批相关调用的结果——被拒调用是带outcome='denied'的ToolReturnPart，批准的.env更新是普通ToolReturnPart——之后模型继续执行（创建README.md.bak备份并输出总结）。

!!! note "工具结果顺序" 工具结果遵循模型发出对应工具调用的顺序。在上面消息历史中，delete_file的被拒结果出现在update_file的.env结果之前，因为模型先发出的是delete_file。这是 v2 的有意行为变更：结果不再按工具类别分组，你看到的顺序即模型发出调用的顺序。

外部工具执行

如果一个工具调用的结果无法在发起调用的同一个 agent run 内生成，该工具即为外部工具。典型例子：由 web/app 前端实现的客户端工具，以及交给后台 worker 或外部服务执行、而不是让 agent 进程干等的耗时任务。

用 CallDeferred 做条件延迟

如果是否外部执行取决于调用参数、agent run context（如依赖项、消息历史）或任务预计耗时，可以定义工具函数并条件性地抛出CallDeferred异常。抛出前，工具函数通常先调度一个后台任务，并传递RunContext.tool_call_id，以便之后把结果匹配回该延迟调用。

从源码看，CallDeferred与ApprovalRequired都是携带可选metadata字典的轻量异常（exceptions.py），metadata 会以tool_call_id为键出现在DeferredToolRequests.metadata中——外部执行示例正是靠它带出task_id。工具执行管线在 _tool_execution.py 中显式捕获这两个异常并转入延迟分支。

与审批类似，工具的args_validator也可以抛出CallDeferred，这样只有参数合法的调用才会被交接出去。

用 ExternalToolset 处理基于 schema 的外部工具

如果一个工具总是外部执行，且其定义连同参数的 JSON schema 一起提供给你的代码，可以使用ExternalToolset。如果外部工具事先已知、但你手头没有参数的 JSON schema，也可以定义一个签名合适的工具函数，其唯一动作就是抛出CallDeferred。

外部执行流程细节

当模型调用外部工具时，agent run 以DeferredToolRequests输出对象结束，其calls列表持有ToolCallPart，包含工具名、已校验的参数和唯一 tool call ID。

当工具调用结果就绪后，构造DeferredToolResults，其calls字典把每个 tool call ID 映射为：

• 任意将返回给模型的值（内部自动包为ToolReturn）；
• 一个ToolReturn对象；
• 或调用失败时的异常：ModelRetry让模型重试，或ToolFailed把失败作为失败结果报告给模型（不消耗该工具的 retry 预算），由其决定如何继续。

随后把这个DeferredToolResults对象与原 run 的 message history 一起作为deferred_tool_results传给运行方法。

下面是一个完整示例：把耗时任务移到后台，任务完成后把结果返回给模型（运行前请确保导入asyncio并加上asyncio.run(main())，无需其他改动）：

import asyncio from dataclasses import dataclass from typing import Any from pydantic_ai import ( Agent, CallDeferred, DeferredToolRequests, DeferredToolResults, ModelRetry, RunContext, ) @dataclass class TaskResult: task_id: str result: Any async def calculate_answer_task(task_id: str, question: str) -> TaskResult: await asyncio.sleep(1) return TaskResult(task_id=task_id, result=42) agent = Agent('openai:gpt-5.2', output_type=[str, DeferredToolRequests]) tasks: list[asyncio.Task[TaskResult]] = [] @agent.tool async def calculate_answer(ctx: RunContext, question: str) -> str: task_id = f'task_{len(tasks)}' # (1)! task = asyncio.create_task(calculate_answer_task(task_id, question)) tasks.append(task) raise CallDeferred(metadata={'task_id': task_id}) # (2)! async def main(): result = await agent.run('Calculate the answer to the ultimate question of life, the universe, and everything') messages = result.all_messages() assert isinstance(result.output, DeferredToolRequests) requests = result.output print(requests) """ DeferredToolRequests( calls=[ ToolCallPart( tool_name='calculate_answer', args={ 'question': 'the ultimate question of life, the universe, and everything' }, tool_call_id='pyd_ai_tool_call_id', ) ], approvals=[], metadata={'pyd_ai_tool_call_id': {'task_id': 'task_0'}}, ) """ done, _ = await asyncio.wait(tasks) # (3)! task_results = [task.result() for task in done] task_results_by_task_id = {result.task_id: result.result for result in task_results} results = DeferredToolResults() for call in requests.calls: try: task_id = requests.metadata[call.tool_call_id]['task_id'] result = task_results_by_task_id[task_id] except KeyError: result = ModelRetry('No result for this tool call was found.') results.calls[call.tool_call_id] = result result = await agent.run(message_history=messages, deferred_tool_results=results) print(result.output) #> The answer to the ultimate question of life, the universe, and everything is 42.

1. 生成一个可独立于 tool call ID 追踪的任务 ID。
2. 可选的metadata参数传递task_id，便于之后与结果匹配，按tool_call_id键控，可从DeferredToolRequests.metadata访问。
3. 实际场景中，这一步通常发生在一个独立进程里——它轮询任务状态，或在所有挂起任务完成时被通知。

在事件流中观察延迟工具调用

与其他工具调用一样，延迟工具调用会向事件流发出FunctionToolCallEvent——但仅凭这个事件，流消费者无法得知该调用正挂起等待交互，也无法得知期望什么类型的交互。另有两种AgentStreamEvent携带这些上下文：

• DeferredToolRequestsEvent——每批延迟调用发出一次，携带DeferredToolRequests。它在任何HandleDeferredToolCallshandler 运行之前发出，因此消费方可以例如在 handler 等待期间通知前端"需要输入"。若没有任何 handler 解析全部请求，run 会以待处理请求作为其DeferredToolRequests输出结束。
• DeferredToolResultsEvent——handler 解析（部分）请求时发出，携带DeferredToolResults。被解析的调用随后走常规管线执行，各自发出FunctionToolResultEvent。当结果改为通过deferred_tool_results提供给新 run 时不会发出该事件——那种情况下调用方本来就已知结果。

这让"解析"与"呈现"解耦：handler 可以只包含纯解析逻辑（例如在 durable execution 工作流中等待信号），而流消费者负责与前端的所有通信，无需自行维护"哪些工具是交互式的"映射。在管线源码中，DeferredToolRequestsEvent的发出发生在延迟调用被批次化时（见 _tool_execution.py L1036）。

继续上面的 handler 示例：

from pydantic_ai import DeferredToolRequestsEvent, DeferredToolResultsEvent from deferred_tool_handler import agent async def main(): async with agent.run_stream_events( 'Delete `__init__.py`, write `Hello, world!` to `README.md`, and clear `.env`' ) as events: async for event in events: if isinstance(event, DeferredToolRequestsEvent): print(f'Approvals needed: {[call.tool_name for call in event.requests.approvals]}') #> Approvals needed: ['update_file', 'delete_file'] elif isinstance(event, DeferredToolResultsEvent): print(f'Resolved: {list(event.results.approvals)}') #> Resolved: ['update_file_dotenv', 'delete_file']

延伸阅读

• Function Tools — 工具基础概念与注册
• Advanced Tool Features — 自定义 schema、动态工具与执行细节（含args_validator、ToolReturn、ModelRetry、ToolFailed）
• Toolsets — 工具集合管理，包括外部工具用的ExternalToolset与审批用的ApprovalRequiredToolset
• Message History — 延迟工具的消息历史，含run_id/conversation_id的用法
• Realtime tools — 实时语音会话中的审批与延迟工具
• Capabilities — 多个 capability 的组合方式，含WrapperCapability与capabilities=[...]列表

关键类型与入口速查

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai