Agent Zero 工具基类深度解析：Tool 与 Response 契约、执行生命周期与自定义工具开发指南

Agent Zero 是一个 AI Agent 框架，其所有"能力"都通过工具（Tool）对外暴露：从搜索、调度到代码执行、浏览器操作，无一例外。本篇文章以框架核心模块 helpers/tool.py.dox.md 为骨架，深入剖析 Tool 基类与 Response 响应契约的设计意图、运行时调用链（before_execution → execute → after_execution 三段式生命周期）、进度上报与日志埋点机制，并结合仓库内真实工具源码与测试用例，给出从零开发一个自定义工具的完整实战步骤。读完本文，你将掌握 Agent Zero 工具系统的运行原理，并能独立编写、注册和验证属于自己的插件工具。

一、模块定位：tool.py 在框架中的角色

在 Agent Zero 的 helpers/ 目录中，tool.py 是所有 Agent 工具的基础设施层。它的职责高度聚焦：

• 定义统一的工具基类 Tool：规定每个工具必须实现的执行入口与可选的生命周期钩子；
• 定义统一的响应契约 Response：规定工具向 Agent 主循环返回结果的固定数据结构；
• 提供进度上报、日志埋点、参数展示等跨工具复用的通用能力。

从依赖关系看，tool.py 只依赖 abc、dataclasses、typing 等标准库，以及 agent.Agent、helpers.extension、helpers.print_style、helpers.strings 等框架内部模块（见 helpers/tool.py 头部 import），自身不包含任何业务逻辑——这正是它作为"公共底座"被 tools/ 下全部默认工具和各插件工具共同继承的原因。

值得强调的是，helpers/ 目录被刻意设计为扁平结构（flat），因此 helpers/tool.py.dox.md 这份文件级 DOX 文档必须与 tool.py 源码保持同步：每当公共函数、类、持久化行为、路径/安全假设或跨模块契约发生变化，都必须同步更新该文档，确保实现与契约说明不脱节。

二、核心契约一：Response 数据类

Response 是工具与 Agent 主循环之间的"标准信封"，定义在 helpers/tool.py：

@dataclass
class Response:
    message: str
    break_loop: bool
    additional: dict[str, Any] | None = None

三个字段各有明确语义：

字段	类型	含义
message	str	工具返回给 Agent 的文本结果，会被写入对话历史供 LLM 继续推理
break_loop	bool	是否终止当前消息循环。True 表示把 message 作为最终回复直接返回给用户；False 表示继续让 Agent 推理下一轮
additional	dict | None	附加数据，会随工具结果一并写入历史；框架内部也会用它携带响应式（Responses API）的额外输出项

break_loop 是控制 Agent 行为节奏的关键开关，仓库中有大量使用范例：

• 内置的响应工具 tools/response.py 在拿到非空的 text/message 参数时返回 Response(message=message, break_loop=True)，把模型的最终回答直接交还给用户并结束循环；
• 示例工具 agents/_example/tools/example_tool.py 返回 Response(message=..., break_loop=False)，让 Agent 拿到结果后继续下一轮推理；
• 未知工具兜底类 tools/unknown.py 同样返回 break_loop=False，把"未找到工具"的提示喂回给 Agent，让它换一种方式继续尝试。

一个实用的判断标准是：工具的结果若只是中间产物（查到了什么、执行了什么），设 break_loop=False；若它本身就是对用户的最终答复，设 break_loop=True。

三、核心契约二：Tool 基类与完整生命周期

Tool 是抽象基类，其构造器与全部方法如下（见 helpers/tool.py）：

class Tool:
    def __init__(self, agent, name, method, args, message, loop_data=None, **kwargs):
        self.agent = agent
        self.name = name
        self.method = method
        self.args = args
        self.loop_data = loop_data
        self.message = message
        self.progress: str = ""

    @abstractmethod
    async def execute(self, **kwargs) -> Response:
        pass

    async def set_progress(self, content: str | None): ...
    def add_progress(self, content: str | None): ...
    async def before_execution(self, **kwargs): ...
    async def after_execution(self, response: Response, **kwargs): ...
    def get_log_object(self): ...
    def nice_key(self, key: str): ...

构造器参数的含义如下：

参数	含义
agent	当前 Agent 实例，工具通过它访问历史、日志、上下文等
name	工具名，如 search_engine
method	工具方法名（同一工具文件可定义多个方法），无则传 None
args	LLM 解析出的工具参数字典
message	LLM 生成该工具请求时的原始消息
loop_data	当前消息循环的状态数据（LoopData）

3.1 execute：唯一的抽象方法

execute 用 @abstractmethod 标注，是所有子类必须实现的入口。Agent 主循环拿到工具响应后，会依据 response.break_loop 决定是否结束消息链（见 agent.py）：

if response.break_loop:
    self._clear_responses_pending_state()
    return response.message

3.2 三段式生命周期：before_execution → execute → after_execution

工具并非只调用 execute 一个方法。框架在 agent.py 的 _execute_tool_request 中编排了完整的执行流程：

1. 将工具实例写入 loop_data.current_tool，随后调用 handle_intervention()（支持人工介入打断）；
2. await tool.before_execution(**tool_args)：打印"正在使用工具"的提示、逐条展示参数、创建日志对象；
3. 再次 handle_intervention() 后，触发 tool_execute_before 扩展钩子（插件可在执行前改写/拦截）；
4. await tool.execute(**tool_args)：执行真正业务逻辑；
5. 触发 tool_execute_after 扩展钩子（插件可在执行后处理响应，例如脱敏、转发到 Telegram/WhatsApp/邮件等通道，参见 plugins/_telegram_integration/extensions/python/tool_execute_after/_50_telegram_response.py 等）；
6. await tool.after_execution(response)：把结果清洗后写入对话历史、打印"工具响应"、更新日志内容；
7. 依据 break_loop 决定是否结束循环；finally 中清空 current_tool。

因此，before_execution 与 after_execution 是可被重写的生命周期钩子——默认实现负责通用收尾，子类可以覆盖以定制行为。

3.3 默认 before_execution：参数展示与日志初始化

基类默认的 before_execution（helpers/tool.py）做两件事：

• 用 PrintStyle（深蓝底白字、加粗）打印 ${agent_name}: Using tool '${name}'；
• 遍历 self.args，对每个参数调用 tool_output_update 扩展钩子（允许插件改写展示值），再用 nice_key 美化键名、以流式 PrintStyle.stream 输出。

3.4 默认 after_execution：结果落历史与日志

默认的 after_execution（helpers/tool.py）做三件事：

1. sanitize_string(response.message.strip())：清洗工具输出文本（来自 helpers/strings.py 的 sanitize_string，负责剔除敏感/非法字符，保证写入历史的内容安全）；
2. self.agent.hist_add_tool_result(self.name, text, id=self.log.id, **(response.additional or {}))：把清洗后的结果连同 additional 附加数据写入对话历史；
3. 打印"Response from tool"并更新日志对象内容。

这里可以看出"工具日志"与"对话历史"是两条线：历史供 LLM 推理，日志供 WebUI 前端渲染展示。

3.5 get_log_object：构造前端日志条目

get_log_object（helpers/tool.py）在 before_execution 中被调用，用于在 agent.context.log 上登记一条 type="tool" 的日志：

• 日志 ID 用 uuid.uuid4() 生成，供 after_execution 回填内容；
• 标题根据是否有 method 区分：Using tool 'name:method' 或 Using tool 'name'，并带 icon://construction 图标前缀；
• 参数快照 kvps=self.args 一并写入，前端可展开查看。

3.6 进度上报：set_progress 与 add_progress

• set_progress(content)：异步方法，通过 tool_output_update 扩展钩子推送进度内容到前端（扩展可改写），并同步更新 self.progress；
• add_progress(content)：同步方法，仅在 content 非空时把字符串追加到 self.progress。

二者组合的典型用法是：长任务中先 add_progress 累积文本，再 await set_progress 一次性推送渲染，避免频繁 IO。实际使用可参考 plugins/_code_execution/tools/input.py、plugins/_a0_connector/tools/code_execution_remote.py 等对进度接口的调用。

3.7 nice_key：参数键名美化

nice_key（helpers/tool.py）把 snake_case 键名转成人类可读标题：首词首字母大写、其余词小写并用空格连接，例如 test_input → Test input、search_query → Search query，用于终端参数展示。

四、工具是如何被找到与实例化的

工具的分发逻辑在 agent.py 的 get_tool 方法（标注 @extension.extensible，可被插件扩展）：

1. 通过 subagents.get_paths(self, "tools", name + ".py") 在当前 Agent 的目录层级中查找 tools/<name>.py；
2. 找到文件后，用 extract_tools.load_classes_from_file(path, Tool) 动态加载其中继承 Tool 的类；
3. 取第一个类实例化；若未找到任何实现，则回退到 tools/unknown.py 中的 Unknown 兜底类；
4. 实例化时把 agent、name、method、args、message、loop_data 全部传入构造器。

在执行入口 _execute_tool_request 中还有一层 MCP 优先逻辑（agent.py）：先通过 helpers.mcp_handler.MCPConfig.get_instance().get_tool() 查询 MCP 服务器是否暴露同名工具，命中则直接使用 MCP 工具，未命中才回退到本地 get_tool。

五、从源码看真实工具的编写范式

5.1 最小示例：ExampleTool

agents/_example/tools/example_tool.py 展示了最简形态——只实现 execute：

class ExampleTool(Tool):
    async def execute(self, **kwargs):
        test_input = kwargs.get("test_input", "")
        print("Example tool executed with test_input: " + test_input)
        return Response(
            message="This is an example tool response, test_input: " + test_input,
            break_loop=False,
        )

注意：工具参数通过 **kwargs 获取，参数名需与系统提示词（agent.system.tool.example_tool.md）中声明的参数一致。该示例还提示开发者：工具文件放在 Agent 的 tools/ 目录下，并配套编写 prompts/ 中的工具说明文档，让 LLM 知道何时调用、如何填参。

5.2 重写生命周期钩子：ResponseTool

tools/response.py 示范了如何覆盖基类钩子：

• 重写 execute：校验 text/message 参数非空后返回 break_loop=True 的响应，否则抛出 RepairableException（可修复异常，框架会让 LLM 修正重试）；
• 重写 before_execution 为空实现（pass），注释说明日志改由 live_response 扩展负责；
• 重写 after_execution：不再写历史，而是通过 loop_data.params_temporary["log_item_response"] 把消息标记为 finished。

这证明生命周期钩子是可插拔的：框架只保证默认行为，具体工具可以按需裁剪。

5.3 兜底工具：Unknown

tools/unknown.py 是找不到工具实现时的最后防线：它重新构建完整工具清单提示，加载 fw.tool_not_found.md 提示模板，把"该工具不存在 + 可用工具列表"作为 break_loop=False 的结果喂回给 Agent，引导模型自我纠错。

六、测试与验证：如何确认工具契约不被破坏

由于 Tool/Response 是全局契约，仓库通过多层级测试守护其稳定性：

• tests/test_tool_action_contracts.py 直接针对工具动作契约做测试：测试文件用 _FakeResponse（message/break_loop/additional 三个字段）与 _FakeTool 模拟 Tool/Response 契约，随后对 SkillsTool、SchedulerTool 等真实工具执行 asyncio.run(tool.execute(**tool.args)) 并断言返回结构——任何破坏契约的改动都会在此失败；
• tests/test_responses_architecture.py、tests/test_mcp_handler_multimodal.py、tests/test_browser_agent_regressions.py 等覆盖了工具在响应式架构、MCP 多模态、浏览器 Agent 等场景下的回归行为；
• DOX 文档 helpers/tool.py.dox.md 中列出的相关测试还包括 tests/test_default_prompt_budget.py、tests/test_dirty_json.py、tests/test_document_query_plugin.py、tests/test_fastmcp_openapi_security.py、tests/test_host_browser_connector.py、tests/test_a0_connector_prompt_gating.py——它们分别从提示词预算、JSON 解析、文档查询插件、OpenAPI 安全、主机浏览器连接等侧面验证工具生态的稳定性。

因此，修改 helpers/tool.py 或新增公共工具 API 时，必须同步运行上述针对性测试；若改动涉及认证、文件系统、WebSocket、隧道、上传或密钥处理等安全敏感区，还应跑对应的安全回归测试。

七、实战：从零编写一个可运行的自定义工具

综合以上契约，自定义工具的完整开发步骤如下：

第 1 步：创建工具文件。 在目标 Agent 的 tools/ 目录下新建 <tool_name>.py（例如复用示例结构 agents/_example/tools/example_tool.py），定义继承 Tool 的类：

from helpers.tool import Tool, Response

class MyLookupTool(Tool):
    async def execute(self, **kwargs):
        keyword = kwargs.get("keyword", "")
        result = do_lookup(keyword)  # 你的业务逻辑
        return Response(
            message=f"Lookup result for '{keyword}': {result}",
            break_loop=False,          # 中间结果，继续推理
            additional={"source": "my-lookup"},
        )

第 2 步：编写工具说明。 在 Agent 的 prompts/ 目录创建 agent.system.tool.<tool_name>.md，向 LLM 描述工具用途、参数（名称、类型、必填性）与调用时机，确保模型能正确生成参数 JSON。

第 3 步：按需重写生命周期钩子。 长任务用 add_progress 累积 + await set_progress 推送进度；需要在执行前后拦截/装饰时重写 before_execution/after_execution（参考 tools/response.py 的裁剪写法）。

第 4 步：运行与验证。 启动框架后向 Agent 下达触发该工具的任务，观察终端中的 Using tool '...' / Response from tool '...' 输出、WebUI 工具日志卡片与对话历史中的结果条目；如需回归验证，参考 tests/test_tool_action_contracts.py 的 _FakeTool + asyncio.run(tool.execute(...)) 模式编写单元测试。

注意：helpers/ 是扁平共享层，任何新增的公共辅助函数若被多个模块复用，都应优先沉淀在 helpers/ 中并保持契约文档同步；路径、认证、密钥、持久化、网络与子进程相关行为必须显式、受控，避免引入安全风险（见 DOX 文档的 Work Guidance 一节）。

八、小结

Agent Zero 的工具系统建立在一组极简但严格的契约之上：Response 用 message + break_loop 控制对话流向，Tool 用三段式生命周期把"参数展示 → 业务执行 → 结果落库"解耦为可重写钩子，而 get_log_object、set_progress/add_progress、nice_key 则提供了日志、进度与展示的统一基础设施。无论你是想理解框架内部运行机制，还是准备为 Agent 编写第一个自定义工具，helpers/tool.py 与 helpers/tool.py.dox.md 都是必须精读的起点——前者定义了行为，后者锁定了契约。