工具系统：自注册、按需暴露、安全调度

本章核心源码：tools/registry.py（335 行）、model_tools.py（577 行）、toolsets.py

定位：本章把 Hermes 的工具系统作为一个小型插件架构案例来写，涵盖三层发现机制、ToolRegistry 单例设计、toolset 分组、并发执行与安全调度。 前置依赖：第 4 章（AIAgent 内核）。适用场景：想理解工具如何被发现、注册、暴露给模型和安全调度。

为什么工具系统值得单独一章

Hermes 有数十个工具相关文件（含子目录），支持三类来源（内置、MCP、插件系统），可以按 toolset 粒度启用/禁用，并在执行时自动判断是否可以并行。这不是一个简单的函数调用表，而是一个完整的插件架构。

理解这个架构，你就能回答这些问题：

• 怎么添加一个新的内置工具？（答：至少三步，工具文件 + _discover_tools() 导入 + toolsets.py 接线）
• 怎么控制哪些工具暴露给模型？（答：toolset 过滤）
• 模型返回的 tool call 怎么被路由到正确的 handler？（答：ToolRegistry.dispatch）
• 多个工具调用怎么安全地并行？（答：路径冲突检测）

三层工具发现

工具发现发生在 model_tools.py 的模块加载时，分三层：

graph TD
    A["model_tools.py 被 import"] --> B["_discover_tools()"]
    B --> C["内置工具<br/>importlib.import_module()"]
    C --> D["tools/*.py 在 import 时<br/>调用 registry.register()"]
    
    A --> E["MCP 发现<br/>discover_mcp_tools()"]
    E --> F["读取 config.yaml → mcp_servers<br/>启动 MCP 进程"]
    F --> G["MCP 工具<br/>registry.register()"]
    
    A --> H["Plugin 发现<br/>discover_plugins()"]
    H --> I["扫描 user / project / entry_points<br/>加载插件"]
    I --> J["插件工具<br/>registry.register()"]
    
    D --> K["ToolRegistry<br/>单例"]
    G --> K
    J --> K

第 1 层：内置工具

_discover_tools()（model_tools.py:132）硬编码了 20 个内置工具模块的导入列表：

# model_tools.py:132-170
def _discover_tools():
    _modules = [
        "tools.web_tools",
        "tools.terminal_tool",
        "tools.file_tools",
        "tools.browser_tool",
        "tools.delegate_tool",
        "tools.mcp_tool",
        "tools.code_execution_tool",
        # ... 共 20 个（honcho_tools 已迁移为 memory provider 插件）
    ]
    for mod_name in _modules:
        try:
            importlib.import_module(mod_name)
        except Exception as e:
            logger.warning("Could not import tool module %s: %s", mod_name, e)

_discover_tools()  # 模块加载时立即执行

每个工具模块在 import 时调用 registry.register()。但对仓库内置工具来说，这还不是全部工作：

1. 新建 tools/*.py
2. 把模块加入 model_tools.py 的 _discover_tools() 导入列表
3. 把工具接入 toolsets.py，否则它即使注册成功也不一定会暴露给模型

也就是说，registry.register() 是注册动作，不是“让新内置工具真正可达”的全部步骤。

工具模块的注册模式如下：

# 以 tools/web_tools.py 为例（模式）
from tools.registry import registry

registry.register(
    name="web_search",
    toolset="web",
    schema={"description": "Search the web", "parameters": {...}},
    handler=web_search_handler,
    check_fn=lambda: bool(os.getenv("EXA_API_KEY")),
    requires_env=["EXA_API_KEY"],
    is_async=False,
    emoji="🔍",
)

关键设计：try/except 包裹每个模块导入。如果某个可选工具的依赖缺失（比如 fal_client 没装），只会 log warning，不会阻止其他工具加载。

第 2 层：MCP 工具

# model_tools.py:172-177
from tools.mcp_tool import discover_mcp_tools
discover_mcp_tools()

MCP（Model Context Protocol）工具来自外部进程。discover_mcp_tools() 读取 config.yaml 中的 mcp_servers 配置，启动 MCP server 进程，获取它们暴露的工具 schema，然后调用同一个 registry.register() 注册。

MCP 工具支持热更新：当 MCP server 发送 notifications/tools/list_changed 信号时，Hermes 会 deregister 旧工具并重新注册。ToolRegistry.deregister()（tools/registry.py:95）在移除工具时会清理 toolset check，避免残留。

第 3 层：插件系统

# model_tools.py:179-184
from hermes_cli.plugins import discover_plugins
discover_plugins()

Hermes 当前的插件发现不是单一路径，而是三路扫描：

• 当前 HERMES_HOME/plugins/ 下的用户插件（默认 profile 下表现为 ~/.hermes/plugins/）
• ./.hermes/plugins/ 下的项目插件（需开启 HERMES_ENABLE_PROJECT_PLUGINS）
• Python entry_points 暴露的 pip 插件

对工具系统而言，三类插件最后都会落到同一个事实：导入插件代码后，由插件自行调用 registry.register() 把工具注入注册表。

三层统一的价值

三种来源最终都注册到同一个 ToolRegistry 单例中。这意味着：

• 模型看到的是统一的工具列表，不区分来源
• 调度逻辑只需要一套
• 新增工具来源只要最终能调用 registry.register()，就不需要改动调度代码

ToolRegistry：核心抽象

ToolRegistry（tools/registry.py:48）是一个模块级单例（tools/registry.py:290）：

# tools/registry.py:290
registry = ToolRegistry()

它提供四组能力：

注册

# tools/registry.py:59-93
def register(self, name, toolset, schema, handler, check_fn=None,
             requires_env=None, is_async=False, description="", emoji="",
             max_result_size_chars=None):
    self._tools[name] = ToolEntry(...)

Schema 检索（带可用性过滤）

# tools/registry.py:116-143
def get_definitions(self, tool_names, quiet=False):
    result = []
    for name in sorted(tool_names):
        entry = self._tools.get(name)
        if entry.check_fn and not entry.check_fn():
            continue  # 可用性检查失败 → 跳过
        result.append({"type": "function", "function": {**entry.schema, "name": name}})
    return result

check_fn 是延迟执行的——只在构建 schema 列表时调用，不在注册时调用。这意味着环境变量的设置可以在注册之后、schema 检索之前完成。

调度

# tools/registry.py:149-166
def dispatch(self, name, args, **kwargs):
    entry = self._tools.get(name)
    if not entry:
        return json.dumps({"error": f"Unknown tool: {name}"})
    try:
        if entry.is_async:
            from model_tools import _run_async
            return _run_async(entry.handler(args, **kwargs))
        return entry.handler(args, **kwargs)
    except Exception as e:
        return json.dumps({"error": f"Tool execution failed: {type(e).__name__}: {e}"})

两个关键设计：

1. Async 桥接：异步 handler 通过 _run_async() 自动桥接到同步调用链
2. 异常捕获：所有异常被转换为 JSON error 格式返回，永远不会向上抛出导致编排层崩溃

ToolEntry：元数据容器

# tools/registry.py:24-45
class ToolEntry:
    __slots__ = (
        "name", "toolset", "schema", "handler", "check_fn",
        "requires_env", "is_async", "description", "emoji",
        "max_result_size_chars",
    )

__slots__ 优化内存。对这样一个会注册数十个工具的系统来说，slots 比普通实例 dict 更紧凑。

Toolset：分组与过滤

工具通过 toolset 字段分组。get_tool_definitions()（model_tools.py:234）负责按 toolset 过滤：

# model_tools.py:234（简化）
def get_tool_definitions(enabled_toolsets=None, disabled_toolsets=None, quiet_mode=False):
    # 确定最终的工具集合
    all_names = registry.get_all_tool_names()
    if enabled_toolsets:
        # 只保留指定 toolset 中的工具
        names = {n for n in all_names if registry.get_toolset_for_tool(n) in enabled_toolsets}
    elif disabled_toolsets:
        # 移除指定 toolset 中的工具
        names = {n for n in all_names if registry.get_toolset_for_tool(n) not in disabled_toolsets}
    else:
        names = set(all_names)
    return registry.get_definitions(names, quiet=quiet_mode)

用户通过 hermes tools 命令或 config.yaml 控制 toolset 的启用/禁用。

工具调度流程

当模型返回 tool_calls 时，编排层通过以下路径将调用分发到正确的 handler：

graph TD
    A["模型返回 tool_calls"] --> B["AIAgent._invoke_tool()"]
    B --> C{工具名称?}
    C -->|"todo/session_search/memory/clarify"| D["Agent 内置处理<br/>（需要 agent 内部状态）"]
    C -->|"memory_manager 动态工具"| E["MemoryManager.handle_tool_call()<br/>（外部 provider 注册）"]
    C -->|"delegate_task"| F["delegate_task()<br/>（spawn 子代理）"]
    C -->|"其他"| G["model_tools.handle_function_call()"]
    G --> H["ToolRegistry.dispatch()"]
    H --> I{is_async?}
    I -->|是| J["_run_async(handler)"]
    I -->|否| K["handler(args)"]
    J --> L["返回 JSON 结果"]
    K --> L
    D --> L
    E --> L
    F --> L

并发执行与安全调度

第 3 章提到编排器会判断工具是否可以并行执行。这个判断的核心逻辑在 _should_parallelize_tool_batch()（run_agent.py:265）：

# run_agent.py:214-235
_NEVER_PARALLEL_TOOLS = frozenset({"clarify"})  # 交互式工具，永不并行

_PARALLEL_SAFE_TOOLS = frozenset({              # 只读工具，总是可以并行
    "read_file", "search_files", "session_search",
    "skill_view", "skills_list", "vision_analyze",
    "web_extract", "web_search",
    "ha_get_state", "ha_list_entities", "ha_list_services",
})

_PATH_SCOPED_TOOLS = frozenset({"read_file", "write_file", "patch"})
_MAX_TOOL_WORKERS = 8

# run_agent.py:265（简化）
def _should_parallelize_tool_batch(tool_calls) -> bool:
    tool_names = [tc.function.name for tc in tool_calls]

    # 1. 有交互式工具 → 不并行
    if any(name in _NEVER_PARALLEL_TOOLS for name in tool_names):
        return False

    # 2. 全部是只读工具 → 并行
    if all(name in _PARALLEL_SAFE_TOOLS for name in tool_names):
        return True

    # 3. 有文件操作 → 检查路径是否重叠
    if any(name in _PATH_SCOPED_TOOLS for name in tool_names):
        paths = [extract_path(tc) for tc in tool_calls]
        if has_overlapping_paths(paths):
            return False  # 路径重叠 → 不并行
        return True

    return False  # 默认不并行

并行执行使用 ThreadPoolExecutor，最多 8 个并发线程（_MAX_TOOL_WORKERS = 8）。

结果大小的三层防御

设计哲学：三层防御对抗上下文溢出 工具结果管理有三层独立的防线，每层捕获不同的失败模式：

• 第 1 层（per-tool）：每个工具在注册时声明 max_result_size_chars，工具作者控制自己的输出预算。
• 第 2 层（per-result）：tool_result_storage.py 将超大结果持久化到磁盘并返回文件路径。如果沙箱写入失败，回退到内联截断并附带明确消息告诉模型"完整输出无法保存"——诚实的失败优于静默截断。
• 第 3 层（per-turn）：聚合轮次预算捕获 N 个中等大小的输出——它们单独通过了 per-tool 限制，但集体溢出了上下文窗口。

一个特殊情况：read_file 的 max_result_size_chars=float('inf')——固定的、不可配置的。这是为了防止一个已发现的无限循环：工具输出被持久化，模型通过 read_file 读取它，读取结果又被持久化，如此循环。来源：tools/budget_config.py:11-14。

危险命令检测

_is_destructive_command()（run_agent.py:254）对 terminal 工具的命令进行启发式检测：

# run_agent.py:237-262
_DESTRUCTIVE_PATTERNS = re.compile(r"""(?:^|\s|&&|\|\||;|`)(?:
    rm\s|rmdir\s|          # 删除
    mv\s|                  # 移动/重命名
    sed\s+-i|              # 原地编辑
    truncate\s|dd\s|shred\s|  # 覆写
    git\s+(?:reset|clean|checkout)\s  # Git 破坏性操作
)""", re.VERBOSE)

匹配到的命令会触发审批机制——通过 clarify_callback 让用户确认执行。

设计启示

Hermes 的工具系统展示了一个实用的插件架构模式：

1. 自注册消除了统一 registry 文件：每个工具在 import 时自注册；但对仓库内置工具来说，仍要接入 _discover_tools() 和 toolsets.py
2. 延迟可用性检查：check_fn 在 schema 检索时才执行，不在注册时执行——解耦了工具代码和运行环境
3. 异常隔离：工具导入失败、check 失败、执行失败都被捕获并降级为 log/JSON error，永远不崩溃编排层
4. 并行安全由数据决定：不需要每个工具声明"我能不能并行"——通过分析工具名和参数路径自动判断

第 7 章将选取 Terminal、File、Browser、MCP 四个代表性工具做深度剖面。

设计赌注回扣：工具系统的三层发现机制直接服务于 Run Anywhere 赌注——MCP 工具让 Hermes 可以接入任何实现了 MCP 协议的外部能力，而 pip 插件让社区可以扩展工具集，不受 Hermes 核心代码库的限制。

版本演化说明

本章核心分析基于 Hermes Agent v0.8.0（2026 年 4 月）。 ToolRegistry 作为独立模块早在 v0.3.0 之前就已出现，并不是 v0.7.0 才抽出来的。之后的演化重点主要在于把 MCP、插件系统和 toolset 过滤逐步接到同一条注册与调度链上。