第三篇：工具系统详解

2.5万字·1小时4分钟·

ToolRegistry单例、6种终端后端、自定义工具

1. 工具注册机制

1.1 ToolRegistry 单例

Hermes Agent 的工具系统采用中心化注册表模式，核心是 tools/registry.py 中的 ToolRegistry 类：

# tools/registry.py 第 48-53 行
class ToolRegistry:
    """Singleton registry that collects tool schemas + handlers from tool files."""

    def __init__(self):
        self._tools: Dict[str, ToolEntry] = {}
        self._toolset_checks: Dict[str, Callable] = {}

在模块级别创建全局单例：

# tools/registry.py 第 290 行
registry = ToolRegistry()

整个进程只有一个 registry 实例，所有工具文件共享。这种设计避免了多实例带来的状态不一致问题。

1.2 ToolEntry 数据结构

每个注册的工具由 ToolEntry 类描述（第 24 行）：

# tools/registry.py 第 24-45 行
class ToolEntry:
    """Metadata for a single registered tool."""

    __slots__ = (
        "name", "toolset", "schema", "handler", "check_fn",
        "requires_env", "is_async", "description", "emoji",
        "max_result_size_chars",
    )

各字段含义：

字段	类型	说明
`name`	str	工具名称（全局唯一标识符）
`toolset`	str	所属工具集（分组标识）
`schema`	dict	OpenAI function calling 格式的 JSON Schema
`handler`	Callable	工具处理函数
`check_fn`	Callable	可用性检查函数（None 表示始终可用）
`requires_env`	list	所需环境变量列表
`is_async`	bool	是否为异步处理函数
`description`	str	工具描述
`emoji`	str	显示用表情符号
`max_result_size_chars`	int	结果最大字符数限制

使用 __slots__ 而非 __dict__，节省内存并防止动态属性注入。

1.3 register() 调用

工具注册通过 registry.register() 完成，在每个工具文件的模块级别调用：

# 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):
    """Register a tool. Called at module-import time by each tool file."""
    existing = self._tools.get(name)
    if existing and existing.toolset != toolset:
        logger.warning("Tool name collision: '%s' ...", name)
    self._tools[name] = ToolEntry(
        name=name, toolset=toolset, schema=schema,
        handler=handler, check_fn=check_fn,
        requires_env=requires_env or [],
        is_async=is_async, description=description or schema.get("description", ""),
        emoji=emoji, max_result_size_chars=max_result_size_chars,
    )
    if check_fn and toolset not in self._toolset_checks:
        self._toolset_checks[toolset] = check_fn

以终端工具为例，注册发生在文件末尾：

# tools/terminal_tool.py 第 1769-1777 行
registry.register(
    name="terminal",
    toolset="terminal",
    schema=TERMINAL_SCHEMA,
    handler=_handle_terminal,
    check_fn=check_terminal_requirements,
    emoji="💻",
    max_result_size_chars=100_000,
)

1.4 导入链（循环安全）

工具注册的导入链经过精心设计，避免循环依赖：

tools/registry.py          ← 不导入 model_tools 或任何工具文件
       ↑
tools/*.py                 ← 导入 tools.registry（模块级 register 调用）
       ↑
model_tools.py             ← 导入 tools.registry + 所有工具模块
       ↑
run_agent.py, cli.py       ← 导入 model_tools

工具发现在 model_tools.py 的 _discover_tools() 函数中触发：

# model_tools.py 第 132-170 行
def _discover_tools():
    """Import all tool modules to trigger their registry.register() calls."""
    _modules = [
        "tools.web_tools",
        "tools.terminal_tool",
        "tools.file_tools",
        "tools.vision_tools",
        "tools.mixture_of_agents_tool",
        "tools.image_generation_tool",
        "tools.skills_tool",
        "tools.skill_manager_tool",
        "tools.browser_tool",
        "tools.cronjob_tools",
        "tools.rl_training_tool",
        "tools.tts_tool",
        "tools.todo_tool",
        "tools.memory_tool",
        "tools.session_search_tool",
        "tools.clarify_tool",
        "tools.code_execution_tool",
        "tools.delegate_tool",
        "tools.process_registry",
        "tools.send_message_tool",
        "tools.homeassistant_tool",
    ]
    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)

在标准工具发现之后，还有两层动态发现：

MCP 工具发现（第 173-177 行）：从配置的 MCP 服务器发现外部工具
插件发现（第 179-184 行）：加载用户/项目/Pip 插件注册的工具

2. 内置工具一览

2.1 按 Toolset 分类

Toolset	工具	源文件	说明
terminal	`terminal`	`terminal_tool.py`	命令执行（6 种后端）
file	`read_file`, `write_file`, `patch`, `search_files`	`file_tools.py`	文件操作
web	`web_search`, `web_extract`	`web_tools.py`	Web 搜索和内容提取
browser	`browser_navigate`, `browser_snapshot`, `browser_click`, `browser_type`, `browser_scroll`, `browser_back`, `browser_press`, `browser_get_images`, `browser_vision`, `browser_console`	`browser_tool.py`	无头浏览器自动化
vision	`vision_analyze`	`vision_tools.py`	图像分析
image	`image_generate`	`image_generation_tool.py`	图像生成
delegation	`delegate_task`	`delegate_tool.py`	子 Agent 委派
skills	`skills_list`, `skill_view`, `skill_manage`	`skills_tool.py`, `skill_manager_tool.py`	技能管理
memory	`memory`	`memory_tool.py`	持久记忆
todo	`todo`	`todo_tool.py`	任务列表管理
tts	`text_to_speech`	`tts_tool.py`	文本转语音
clarify	`clarify`	`clarify_tool.py`	向用户提问澄清
code_execution	`execute_code`	`code_execution_tool.py`	沙箱内程序化工具调用
cronjob	`cronjob`	`cronjob_tools.py`	定时任务
session_search	`session_search`	`session_search_tool.py`	历史会话搜索
send_message	`send_message`	`send_message_tool.py`	跨平台消息发送
homeassistant	`ha_get_state`, `ha_list_entities`, `ha_list_services`	`homeassistant_tool.py`	Home Assistant 集成
rl	10 个 RL 相关工具	`rl_training_tool.py`	强化学习训练环境
moa	`mixture_of_agents`	`mixture_of_agents_tool.py`	多 Agent 混合推理

2.2 可用性检查（check_fn）

每个 toolset 可以注册一个 check_fn，在工具加载时判断该工具是否可用：

# tools/registry.py 第 116-143 行
def get_definitions(self, tool_names: Set[str], quiet: bool = False) -> List[dict]:
    for name in sorted(tool_names):
        entry = self._tools.get(name)
        if entry.check_fn:
            try:
                check_results[entry.check_fn] = bool(entry.check_fn())
            except Exception:
                check_results[entry.check_fn] = False

例如，Web 工具需要 BRAVE_API_KEY 或 TAVILY_API_KEY，浏览器工具需要 Playwright 安装。check_fn 返回 False 时，工具不会出现在模型的可用工具列表中。

2.3 工具集过滤

用户可以通过配置启用或禁用特定工具集：

# model_tools.py 第 234-353 行
def get_tool_definitions(enabled_toolsets=None, disabled_toolsets=None, quiet_mode=False):
    # enabled_toolsets: 白名单模式，只包含指定工具集
    # disabled_toolsets: 黑名单模式，排除指定工具集
    # 两者都为 None: 加载所有可用工具

工具集名称支持新旧两种格式，通过 _LEGACY_TOOLSET_MAP（第 204-227 行）兼容旧配置。

3. 终端工具深度解析

3.1 六种执行后端

终端工具是 Hermes Agent 最核心的工具之一，支持 6 种执行后端，通过 TERMINAL_ENV 环境变量选择：

后端	环境类	源文件	适用场景
local	`LocalEnvironment`	`tools/environments/local.py`	本地直接执行（默认）
docker	`DockerEnvironment`	`tools/environments/docker.py`	容器隔离
modal	`ModalEnvironment` / `ManagedModalEnvironment`	`tools/environments/modal.py`	Modal 云沙箱
ssh	`SSHEnvironment`	`tools/environments/ssh.py`	远程服务器
singularity	`SingularityEnvironment`	`tools/environments/singularity.py`	HPC 环境
daytona	`DaytonaEnvironment`	`tools/environments/daytona.py`	Daytona 开发环境

所有后端继承自 tools/environments/base.py 中的基类，实现统一的 execute() 接口。

3.2 超时控制

终端工具实现了精细的超时控制：

# tools/terminal_tool.py 第 78-79 行
FOREGROUND_MAX_TIMEOUT = int(os.getenv("TERMINAL_MAX_FOREGROUND_TIMEOUT", "600"))

前台命令（同步等待结果）的最大超时为 600 秒（10 分钟），可通过环境变量调整。超时后，后台进程管理器会自动清理相关资源。

3.3 破坏性命令检测

Hermes Agent 在 Agent 层（run_agent.py）和工具层都实现了破坏性命令检测。Agent 层的检测使用正则表达式匹配：

# run_agent.py 第 240-264 行
_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
    )""", re.VERBOSE,
)
_REDIRECT_OVERWRITE = re.compile(r'[^>]>[^>]|^>[^>]')

def _is_destructive_command(cmd: str) -> bool:
    """Heuristic: does this terminal command look like it modifies/deletes files?"""
    if not cmd:
        return False
    if _DESTRUCTIVE_PATTERNS.search(cmd):
        return True
    if _REDIRECT_OVERWRITE.search(cmd):
        return True
    return False

检测到的破坏性模式：

rm / rmdir：删除文件/目录
mv：移动/重命名文件
sed -i：原地修改文件
truncate：截断文件
dd：直接磁盘操作
shred：安全删除
git reset/clean/checkout：Git 破坏性操作
> 输出重定向（但不包括 >> 追加）

在工具层，还有更深入的安全扫描：

# tools/terminal_tool.py 第 142-151 行
from tools.approval import (
    check_dangerous_command as _check_dangerous_command_impl,
    check_all_command_guards as _check_all_guards_impl,
)

安全审批流程整合了 Tirith 安全扫描器（tools/tirith_security.py），检测命令注入、路径遍历等攻击模式。

3.4 Sudo 密码处理

对于需要 sudo 权限的命令，终端工具实现了智能密码管理：

# tools/terminal_tool.py 第 448-499 行
def _transform_sudo_command(command):
    # 1. 将 "sudo" 重写为 "sudo -S -p ''"（从 stdin 读密码）
    # 2. 从环境变量或交互式提示获取密码
    # 3. 将密码作为 stdin 前缀传递给子进程

交互式模式下（HERMES_INTERACTIVE=1），如果未配置 SUDO_PASSWORD，会弹出密码输入框（45 秒超时），密码在会话内缓存。

4. 工具调度流程

sequenceDiagram
    participant LLM as LLM API
    participant Agent as AIAgent
    participant MT as model_tools
    participant Registry as ToolRegistry
    participant Tool as 工具 Handler

    LLM->>Agent: 返回 tool_calls [{name, arguments}]
    Agent->>Agent: IterationBudget.consume()
    Agent->>MT: handle_function_call(name, args)
    MT->>Registry: dispatch(name, args)
    Registry->>Tool: handler(**args)
    Tool-->>Registry: 执行结果
    Registry-->>MT: 返回结果
    MT-->>Agent: 格式化结果
    Agent->>Agent: 追加到 messages
    Agent->>LLM: 再次调用（含工具结果）

sequenceDiagram
    participant LLM as LLM API
    participant Agent as AIAgent
    participant T1 as 工具1 (read_file)
    participant T2 as 工具2 (web_search)

    LLM->>Agent: 返回 2 个 tool_calls
    Agent->>Agent: _should_parallelize_tool_batch() = true
    par 并行执行
        Agent->>T1: read_file(path)
        T1-->>Agent: 文件内容
    and
        Agent->>T2: web_search(query)
        T2-->>Agent: 搜索结果
    end
    Agent->>LLM: 再次调用（含两个工具结果）

4.1 handle_function_call() 路由

handle_function_call()（model_tools.py 第 459 行）是所有工具调用的统一入口：

# model_tools.py 第 459-548 行
def handle_function_call(function_name, function_args, task_id=None,
                         tool_call_id=None, session_id=None,
                         user_task=None, enabled_tools=None) -> str:
    # 1. 参数类型强制转换
    function_args = coerce_tool_args(function_name, function_args)

    # 2. Agent 循环拦截（todo/memory/session_search/delegate_task）
    if function_name in _AGENT_LOOP_TOOLS:
        return json.dumps({"error": f"{function_name} must be handled by the agent loop"})

    # 3. 插件 pre_tool_call 钩子
    invoke_hook("pre_tool_call", tool_name=function_name, args=function_args, ...)

    # 4. 路由到 ToolRegistry.dispatch()
    result = registry.dispatch(function_name, function_args, task_id=task_id, ...)

    # 5. 插件 post_tool_call 钩子
    invoke_hook("post_tool_call", tool_name=function_name, result=result, ...)

    return result

4.2 参数类型强制转换

LLM 经常返回字符串类型的数字和布尔值（"42" 而非 42）。coerce_tool_args() 自动处理这些类型不匹配：

# model_tools.py 第 372-456 行
def coerce_tool_args(tool_name, args):
    schema = registry.get_schema(tool_name)
    properties = (schema.get("parameters") or {}).get("properties")
    for key, value in args.items():
        if isinstance(value, str):
            prop_schema = properties.get(key)
            expected = prop_schema.get("type")
            coerced = _coerce_value(value, expected)
            if coerced is not value:
                args[key] = coerced

支持的转换：

"42" → 42（integer）
"3.14" → 3.14（number）
"true" → True（boolean）
联合类型（["integer", "string"]）按顺序尝试

4.3 异步桥接

许多工具的 handler 是异步函数（is_async=True），但 AIAgent 的主循环运行在同步上下文中。_run_async() 函数提供了统一的同步-异步桥接：

# model_tools.py 第 81-125 行
def _run_async(coro):
    """Run an async coroutine from a sync context."""
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None

    if loop and loop.is_running():
        # 在异步上下文中（Gateway/RL）→ 新线程运行
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
            future = pool.submit(asyncio.run, coro)
            return future.result(timeout=300)

    if threading.current_thread() is not threading.main_thread():
        # 工作线程（并行工具执行）→ 每线程持久事件循环
        worker_loop = _get_worker_loop()
        return worker_loop.run_until_complete(coro)

    # 主线程 → 共享持久事件循环
    tool_loop = _get_tool_loop()
    return tool_loop.run_until_complete(coro)

三种场景的处理策略：

场景	策略	原因
已有运行中的事件循环（Gateway）	新建线程	不能在运行中的循环内调用 `run_until_complete()`
工作线程（并行执行）	每线程持久循环	避免主线程循环竞争，防止缓存的异步客户端失效
主线程（CLI）	共享持久循环	避免重复创建/销毁循环导致 httpx 客户端报错

registry.dispatch() 中自动处理异步调用：

# tools/registry.py 第 149-166 行
def dispatch(self, name, args, **kwargs) -> str:
    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}"})

4.4 并行执行

当模型在单次响应中返回多个 tool_calls 时，_should_parallelize_tool_batch()（run_agent.py 第 267 行）判断是否可以并行执行。实际的并行调度逻辑也在此函数中完成，而 handle_function_call()（model_tools.py）仅负责单个工具的路由与执行：

并行安全工具白名单（第 219-231 行）：

_PARALLEL_SAFE_TOOLS = frozenset({
    "ha_get_state", "ha_list_entities", "ha_list_services",
    "read_file", "search_files", "session_search",
    "skill_view", "skills_list", "vision_analyze",
    "web_extract", "web_search",
})

永不并行工具（第 216 行）：

_NEVER_PARALLEL_TOOLS = frozenset({"clarify"})

路径作用域工具（第 233 行）：

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

并行判断逻辑：

批次中包含 clarify → 退化为顺序
路径作用域工具之间检查路径重叠（_paths_overlap()）：
- read_file("/a.txt") + read_file("/b.txt") → 可以并行
- read_file("/dir/a.txt") + write_file("/dir/b.txt") → 检查路径前缀
所有工具必须在白名单中或路径无冲突
最多 8 个并发线程

4.5 Agent 循环拦截

部分工具需要访问 Agent 级别的状态（如 TodoStore、MemoryStore），不能通过标准的 registry.dispatch() 执行：

# model_tools.py 第 364 行
_AGENT_LOOP_TOOLS = {"todo", "memory", "session_search", "delegate_task"}

当 handle_function_call() 遇到这些工具时，直接返回错误信息：

if function_name in _AGENT_LOOP_TOOLS:
    return json.dumps({"error": f"{function_name} must be handled by the agent loop"})

这些工具在 AIAgent.run_conversation() 的主循环中由 Agent 自行处理，绕过 model_tools.py 的路由。

5. 实战：开发自定义工具

5.1 最简工具示例

以下是一个完整的自定义工具实现，模拟一个"数据库查询"工具：

# tools/database_tool.py

"""
Database Query Tool for Hermes Agent.

Provides read-only SQL query execution against configured databases.
"""

import json
import logging
import os
from typing import Optional

from tools.registry import registry, tool_result, tool_error

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# 可用性检查
# ---------------------------------------------------------------------------

def check_database_requirements() -> bool:
    """Check if database connectivity is configured."""
    return bool(os.getenv("DATABASE_URL"))


# ---------------------------------------------------------------------------
# 工具 Schema（OpenAI function calling 格式）
# ---------------------------------------------------------------------------

DATABASE_QUERY_SCHEMA = {
    "name": "database_query",
    "description": (
        "Execute a read-only SQL query against the configured database. "
        "Returns results as JSON. Only SELECT statements are allowed. "
        "Use this tool to query business data, metrics, or reports."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "SQL query to execute (SELECT only).",
            },
            "database": {
                "type": "string",
                "description": "Database name (optional, uses default if omitted).",
            },
            "max_rows": {
                "type": "integer",
                "description": "Maximum number of rows to return (default: 100).",
            },
        },
        "required": ["query"],
    },
}


# ---------------------------------------------------------------------------
# 工具 Handler
# ---------------------------------------------------------------------------

def _handle_database_query(args: dict, **kwargs) -> str:
    """Execute a read-only SQL query."""
    query = args.get("query", "").strip()
    database = args.get("database")
    max_rows = args.get("max_rows", 100)

    # 安全校验：只允许 SELECT 语句
    normalized = query.upper().lstrip()
    if not normalized.startswith("SELECT") and not normalized.startswith("WITH"):
        return tool_error("Only SELECT queries are allowed for security reasons.")

    # 检测潜在危险操作
    dangerous_keywords = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "CREATE", "TRUNCATE"]
    for keyword in dangerous_keywords:
        if keyword in normalized.split():
            return tool_error(f"Query contains restricted keyword: {keyword}")

    try:
        # 这里替换为实际的数据库连接逻辑
        # 示例使用 sqlite3 作为演示
        import sqlite3

        db_url = os.getenv("DATABASE_URL", "example.db")
        conn = sqlite3.connect(db_url)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        cursor.execute(query)

        rows = cursor.fetchmany(max_rows)
        columns = [desc[0] for desc in cursor.description] if cursor.description else []

        results = [dict(zip(columns, row)) for row in rows]
        conn.close()

        return tool_result({
            "success": True,
            "rows": len(results),
            "columns": columns,
            "data": results,
            "truncated": len(results) >= max_rows,
        })

    except Exception as e:
        logger.error("Database query failed: %s", e)
        return tool_error(f"Query execution failed: {type(e).__name__}: {e}")


# ---------------------------------------------------------------------------
# 注册到 ToolRegistry
# ---------------------------------------------------------------------------

registry.register(
    name="database_query",
    toolset="database",
    schema=DATABASE_QUERY_SCHEMA,
    handler=_handle_database_query,
    check_fn=check_database_requirements,
    requires_env=["DATABASE_URL"],
    emoji="🗄️",
    max_result_size_chars=50_000,
)

5.2 注册流程详解

模块创建：将 database_tool.py 放入 tools/ 目录
发现注册：在 model_tools.py 的 _discover_tools() 中添加 "tools.database_tool"
导入时注册：当 _discover_tools() 导入该模块时，模块底部的 registry.register() 自动执行
可用性检查：check_database_requirements() 在 get_tool_definitions() 调用时执行
Schema 生成：工具的 JSON Schema 自动包含在 API 请求的 tools 参数中

5.3 异步工具示例

如果工具的 handler 是异步函数（例如使用 httpx 异步客户端），需要设置 is_async=True：

# tools/weather_tool.py

import json
import os
import httpx
from tools.registry import registry, tool_result, tool_error

WEATHER_SCHEMA = {
    "name": "weather_query",
    "description": "Query current weather for a location.",
    "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "City name or coordinates.",
            },
        },
        "required": ["location"],
    },
}

async def _handle_weather_query(args: dict, **kwargs) -> str:
    """Async weather query handler."""
    location = args.get("location", "").strip()
    api_key = os.getenv("WEATHER_API_KEY")

    if not api_key:
        return tool_error("WEATHER_API_KEY not configured")

    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.weatherapi.com/v1/current.json",
            params={"key": api_key, "q": location},
        )
        data = response.json()
        return tool_result({
            "location": data.get("location", {}).get("name"),
            "temperature": data.get("current", {}).get("temp_c"),
            "condition": data.get("current", {}).get("condition", {}).get("text"),
        })

def check_weather_requirements() -> bool:
    return bool(os.getenv("WEATHER_API_KEY"))

registry.register(
    name="weather_query",
    toolset="weather",
    schema=WEATHER_SCHEMA,
    handler=_handle_weather_query,
    check_fn=check_weather_requirements,
    requires_env=["WEATHER_API_KEY"],
    is_async=True,           # 标记为异步
    emoji="🌤️",
)

当 registry.dispatch() 检测到 is_async=True 时，会自动通过 _run_async() 进行桥接。

5.4 子 Agent 委派工具

delegate_tool.py（tools/delegate_tool.py）是最复杂的内置工具之一，它展示了如何生成独立的子 Agent：

核心设计原则：

隔离性：子 Agent 有独立的会话历史和终端会话
受限工具集：通过 DELEGATE_BLOCKED_TOOLS（第 30 行）阻止危险操作
并行执行：使用 ThreadPoolExecutor 并行运行多个子 Agent
深度限制：MAX_DEPTH = 2 防止无限递归委派

# tools/delegate_tool.py 第 30-39 行
DELEGATE_BLOCKED_TOOLS = frozenset([
    "delegate_task",   # 禁止递归委派
    "clarify",         # 子 Agent 不能与用户交互
    "memory",          # 子 Agent 不能写入共享记忆
    "send_message",    # 子 Agent 不能跨平台发消息
    "execute_code",    # 子 Agent 应逐步推理，而非写脚本
])

_DEFAULT_MAX_CONCURRENT_CHILDREN = 3
MAX_DEPTH = 2  # parent(0) → child(1) → grandchild rejected(2)

子 Agent 构建（第 224 行 _build_child_agent()）：

继承父 Agent 的 Provider 配置（或使用委派配置中的覆盖）
创建独立的 IterationBudget（默认 50 次）
构建专注的系统提示（_build_child_system_prompt()）
安装进度回调，将子 Agent 的工具调用显示在父 Agent 的界面上

5.5 MCP 工具扩展

除了 Python 工具文件，Hermes Agent 还支持通过 MCP（Model Context Protocol）接入外部工具服务器：

# model_tools.py 第 173-177 行
try:
    from tools.mcp_tool import discover_mcp_tools
    discover_mcp_tools()
except Exception as e:
    logger.debug("MCP tool discovery failed: %s", e)

MCP 工具在发现后通过 registry.register() 注册，与内置工具使用完全相同的调度路径。当 MCP 服务器发送 notifications/tools/list_changed 时，Hermes Agent 通过 registry.deregister() 移除旧工具并重新注册。

5.6 插件工具扩展

用户/项目/Pip 插件也可以注册工具：

# model_tools.py 第 179-184 行
try:
    from hermes_cli.plugins import discover_plugins
    discover_plugins()
except Exception as e:
    logger.debug("Plugin discovery failed: %s", e)

插件注册的工具同样通过标准的 toolset 过滤机制管理，不需要任何特殊处理。

调试指南

本节列出工具系统中最常见的故障及其排查方法。

工具注册失败

症状：工具在 hermes status 中不显示，或调用时返回 "Unknown tool" 错误。

排查步骤：

# 检查工具是否被 discover（查看启动日志中的 warning）
journalctl --user -u hermes-gateway | grep -i "could not import tool"

# 确认工具模块在 _discover_tools() 列表中
grep "tools\." model_tools.py | head -30

# 检查 registry.register() 是否被调用（在工具文件末尾搜索）
grep -n "registry.register" tools/<tool_name>.py

常见原因：工具文件未加入 model_tools.py 的 _discover_tools() 列表、模块导入时抛出异常（如缺少依赖包）、registry.register() 调用在条件分支中未被执行。检查 import chain 是否存在循环依赖。

工具执行超时

症状：工具调用长时间无返回，最终超时失败。

排查步骤：

# 检查终端工具超时配置
grep -E "TERMINAL_TIMEOUT|FOREGROUND_MAX_TIMEOUT" ~/.hermes/.env
grep -E "timeout" ~/.hermes/config.yaml

# 查看超时相关日志
grep -i "timeout\|timed?out" ~/.hermes/logs/agent.log | tail -20

调优方法：前台命令默认超时 600 秒（FOREGROUND_MAX_TIMEOUT），可通过环境变量 TERMINAL_MAX_FOREGROUND_TIMEOUT 调整。后台命令不受此限制。如果工具本身需要长时间运行，考虑使用后台执行模式或增大 terminal.timeout 配置值。

工具结果被截断

症状：工具返回的内容不完整，日志中出现 "truncated" 或 "max_result_size" 字样。

排查步骤：

# 检查 max_result_size_chars 配置
grep -n "max_result_size_chars" tools/<tool_name>.py

# 检查文件读取限制
grep -n "file_read_max_chars\|max_chars" tools/file_tools.py

# 查看截断日志
grep -i "truncat" ~/.hermes/logs/agent.log | tail -10

调优方法：每个工具注册时可指定 max_result_size_chars（如 terminal 工具默认 100,000 字符）。read_file 工具有独立的行数/字符数限制。如果截断影响了 Agent 判断，可在工具注册时增大限制值，或引导 Agent 使用分页读取策略。

异步桥接错误

症状：日志中出现 "Event loop is closed"、"RuntimeError: This event loop is already running" 等错误。

排查步骤：

# 搜索事件循环相关错误
grep -E "event loop|RuntimeError|_run_async" ~/.hermes/logs/agent.log | tail -20

# 检查是否在 Gateway 模式下运行（Gateway 有运行中的事件循环）
hermes status | grep -i gateway

常见原因：_run_async() 在三种场景下使用不同的策略（见第 4.3 节）。Gateway 模式下已有运行中的事件循环，此时桥接会新建线程执行 asyncio.run()。如果自定义工具中直接调用 asyncio.get_event_loop().run_until_complete() 而非通过 _run_async() 桥接，会触发此错误。确保异步工具通过 registry.register(is_async=True) 注册，让 dispatch() 自动处理桥接。

思考题

ToolRegistry 为什么设计为单例？如果改为依赖注入模式（在 AIAgent.__init__ 中传入 registry 实例），会有什么好处和挑战？考虑 MCP 动态工具和子 Agent 的场景。
coerce_tool_args() 只处理 string → int/float/bool 的转换。如果 LLM 返回嵌套对象中的类型错误（例如 {"items": ["1", "2"]} 应该是 {"items": [1, 2]}），当前设计能否处理？如何改进？
异步桥接中，_run_async() 在 Gateway 模式下为每个异步工具调用创建新的 ThreadPoolExecutor。在高并发场景下（每秒数十次工具调用），这会有什么性能影响？如何优化？
DELEGATE_BLOCKED_TOOLS 阻止了 execute_code，理由是"子 Agent 应逐步推理而非写脚本"。但在什么场景下这个限制会阻碍子 Agent 的效率？如何在不降低安全性的前提下放宽限制？
设计一个"持续集成"工具（ci_tool.py），支持触发构建、查看构建状态、获取构建日志。需要考虑哪些安全问题？如何处理长时间运行的构建任务（可能超过工具调用超时）？
终端工具的破坏性命令检测使用正则匹配。这种启发式方法有哪些假阳性（误报）和假阴性（漏报）的案例？你能构造出绕过检测的危险命令吗？如何改进检测策略？