A/NAI NATIVE ENGINEERING
INDEX
← 返回文章

SOURCE NOTE · TOOL SYSTEM

Lesson 3: Tools 子系统

工具不只是给 LLM 的函数列表。它还定义输入、权限、生命周期事件，以及结果如何回到下一轮对话。

2026.07.1310 MIN · ·

前两篇分别梳理了 OpenHarness 的 Query Loop 和 Agent Loop 生命周期。这一篇继续跟着模型返回的 tool_use 往下走：如何从零写一个工具、挂到注册表、让 LLM 调用它？工具执行时权限和 Hooks 又怎样介入？

tool是什么？一句话概括：

每个工具 = 名字 + 描述 + Pydantic 输入模型 + async execute()。

注册进 ToolRegistry 后，工具的 Schema 会发给 LLM；LLM 返回 tool_use 时，引擎查表、校验、鉴权、执行，再把结果写回对话。

LLM 看到工具列表（to_api_schema）
        │
        ▼
tool_use { name, input }
        │
        ▼
registry.get(name) → model_validate(input)
        │
        ▼
PRE_TOOL_USE → PermissionChecker → execute() → POST_TOOL_USE
        │
        ▼
ToolResult → 写回对话 → 下一轮 LLM

模型不会直接执行函数。它只会提出“我想调用这个工具，并传入这些参数”；是否找到工具、参数是否合法、是否要询问用户、是否允许产生副作用，都由 Runtime 决定。

INTERACTIVE TRACE · LOCAL SIMULATION

9 STAGES

参数有效、Hook 放行、权限允许，工具返回结果。

  1. 01

    ToolRegistry.to_api_schema()

    01 · 向 LLM 暴露工具

    name、description 与 input_model 的 JSON Schema 被组装成工具列表。tools: [{ name: "write_file", input_schema: { ... } }]
    PASS
  2. 02

    LLM

    02 · 提出 tool_use

    模型选择工具并生成输入；这仍然只是候选动作。tool_use: { name: "write_file", input: { path: "notes.md" } }
    PASS
  3. 03

    ToolRegistry.get()

    03 · 查找工具实例

    引擎根据 name 取回 BaseTool；未知名称会直接返回错误。tool: WriteFileTool
    PASS
  4. 04

    input_model.model_validate()

    04 · 校验输入

    Pydantic 将原始 input 解析为工具需要的参数模型。WriteFileInput(path="notes.md", content="...")
    PASS
  5. 05

    PRE_TOOL_USE

    05 · 运行前置 Hook

    Hook 可以记录、观察或阻断这次工具调用。pre_tool_use: pass
    PASS
  6. 06

    PermissionChecker.evaluate()

    06 · 判断权限

    结合 is_read_only、模式、路径与命令规则评估是否放行。permission: allow
    PASS
  7. 07

    BaseTool.execute()

    07 · 执行业务逻辑

    工具在单次 ToolExecutionContext 中完成真实动作，并返回 ToolResult。ToolResult(output="File written")
    PASS
  8. 08

    POST_TOOL_USE

    08 · 通知执行完成

    后置 Hook 用于记录和通知，不能再逆转已经发生的动作。post_tool_use: complete
    PASS
  9. 09

    ToolResultBlock

    09 · 写回对话

    执行结果成为下一轮 LLM 可以看到的新证据。role: "user" · tool_result appended
    PASS

核心文件

BaseTool

tools/base.py 里有四个最核心的类型：

BaseTool 本身很薄：

class BaseTool(ABC):
    name: str
    description: str
    input_model: type[BaseModel]

    async def execute(self, arguments, context) -> ToolResult:
        ...

    def is_read_only(self, arguments) -> bool:
        return False

    def to_api_schema(self) -> dict:
        ...

其中，name 是 LLM 调用时填写的名称；description 用来告诉 LLM 何时应该使用这个工具；input_model 则定义工具接受哪些参数，以及每个参数的类型和边界。

description 会和 name 一起发给 LLM，写得好不好直接影响模型是否会在正确时机调用工具。工具多、边界相近，或者模型能力一般时，描述就会变成工具设计的一部分。

最小tool例子：

内置里最简单的工具之一，完整逻辑不到 40 行：

class SleepToolInput(BaseModel):
    seconds: float = Field(default=1.0, ge=0.0, le=30.0)


class SleepTool(BaseTool):
    name = "sleep"
    description = "Sleep for a short duration."
    input_model = SleepToolInput

    def is_read_only(self, arguments: SleepToolInput) -> bool:
        return True

    async def execute(self, arguments, context) -> ToolResult:
        await asyncio.sleep(arguments.seconds)
        return ToolResult(output=f"Slept for {arguments.seconds} seconds")

模式很固定：

  1. 定义 XxxInput(BaseModel)。
  2. 定义 XxxTool(BaseTool)，填入 name、description、input_model。
  3. 实现异步 execute()，返回 ToolResult。
  4. 可选地覆盖 is_read_only()。

Pydantic 在这里同时解决两件事：从 input_model 生成 Schema，告诉 LLM 参数结构；并在真正执行前用同一份模型校验模型返回的输入。

ToolRegistry：注册与发现

ToolRegistry 提供的能力很直接：

class ToolRegistry:
    def register(self, tool: BaseTool) -> None: ...
    def get(self, name: str) -> BaseTool | None: ...
    def list_tools(self) -> list[BaseTool]: ...
    def to_api_schema(self) -> list[dict]: ...

默认工具在 create_default_tool_registry() 中显式 new + register，没有自动扫描目录：

def create_default_tool_registry(mcp_manager=None):
    registry = ToolRegistry()
    for tool in (BashTool(), FileReadTool(), ...):
        registry.register(tool)
    if mcp_manager is not None:
        # MCP 动态工具也 register 进来
        ...
    return registry

新增一个内置工具通常只有两步：① 新建 tools/my_tool.py；② 在 tools/__init__.py 的 create_default_tool_registry() 元组里加入 MyTool()。只写好工具类但没有注册，LLM 就永远不会看到它。

内置工具分类（按用途）

执行时序（引擎侧）

engine/query.py 中的 _execute_tool_call() 顺序固定：

  1. PRE_TOOL_USE Hook：可以 blocked，直接错误返回。
  2. registry.get(name)：未知工具返回错误。
  3. input_model.model_validate(tool_input)：参数非法返回错误。
  4. is_read_only() 加上路径 / 命令抽取，进入 PermissionChecker.evaluate()。
  5. 需要确认则弹出 permission_prompt；用户拒绝则返回错误。
  6. await tool.execute(parsed, ToolExecutionContext(...))。
  7. 输出过长可 offload；记录 carryover；组装 ToolResultBlock。
  8. POST_TOOL_USE Hook：通知用，不阻断。
PRE_TOOL_USE
  → registry.get(name)
  → model_validate(input)
  → PermissionChecker.evaluate()
  → execute()
  → POST_TOOL_USE

模型的输出永远只是候选动作。只有通过校验和授权，execute() 才会带来真实世界的副作用。

工具执行时的依赖数据流（一张图）

启动 Runtime
  tool_metadata = { mcp_manager, session_id, task_focus_state, ... }
  QueryEngine(tool_metadata=...)
        │
        ▼
  run_query(QueryContext(
      tool_registry=...,
      ask_user_prompt=...,
      tool_metadata=同一份 dict 引用,
  ))
        │
        ▼  LLM 要调工具
  _execute_tool_call(...)
        │
        ▼  现拼
  ToolExecutionContext(
      cwd=...,
      metadata={
        tool_registry,      ← 从 QueryContext 抄
        ask_user_prompt,    ← 从 QueryContext 抄
        **tool_metadata     ← 会话 dict 展开
      },
      hook_executor=...,
  )
        │
        ▼
  tool.execute(args, context)
        │
        ▼
  context.metadata.get("xxx")

ToolExecutionContext 是单次调用的最小环境：当前目录、Hook，以及 Runtime 明确交给工具使用的 metadata。需要 MCP manager、session id、任务焦点或用户提问回调时，工具从 context.metadata 取用，而不必依赖全局状态。

权限与 is_read_only

is_read_only() 只是工具对权限系统提供的一条信号，并不是最终授权。

另外还有显式 denied_tools / allowed_tools、路径规则、命令 deny 模式和内置敏感路径硬拦截。权限策略的主要入口在 permissions/checker.py。

除了内置工具还有哪类？注册一样吗？

相同点：最终都是 registry.register(BaseTool 实例)，执行路径一致。不同点：发现和实例化时机与方式不同。

Skills / MCP Resources 不是 ToolRegistry 里的普通 tool：Skills 用 skill 工具加载；Resources 用专用的 list/read 工具。

插件「二次注册」具体怎么做？

会话启动 build_runtime（ui/runtime.py）：

1) load_plugins → 每个插件 load_plugin
      enabled 时：_load_plugin_tools
        扫 tools/*.py → 动态 import
        找 BaseTool 子类 → instance = 类()
        放进 LoadedPlugin.tools（此时还没进 Registry）

2) create_default_tool_registry(mcp)
      第一次：内置 + MCP

3) for plugin in plugins
      if enabled and tools:
          for tool in plugin.tools:
              tool_registry.register(tool)

“二次”= 同一张表、再跑一轮 register，不是第二种 API。同名后注册覆盖先注册；禁用插件时 tools=[]，不会二次挂表。

源码精读：