Hermes Agent 工具系统实战解析：40+ 工具为什么不用配置表？

tools/*.py          ← 各工具文件（模块导入时自注册）
       ↓ register()
tools/registry.py   ← 单例注册中心（ToolRegistry）
       ↓ get_definitions() / dispatch()
model_tools.py      ← 编排层（发现 + Schema 提供 + 调度）
       ↓ resolve_toolset()
toolsets.py         ← Toolset 定义（工具分组、组合）
       ↓
run_agent.py/cli.py ← 入口（消费工具定义，驱动 Agent Loop）

# 模块级单例
registry = ToolRegistry()

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

def register(self, name, toolset, schema, handler, ..., override=False):
    existing = self._tools.get(name)
    if existing and existing.toolset != toolset:
        # MCP-to-MCP 允许覆盖（服务器刷新场景）
        both_mcp = (existing.toolset.startswith("mcp-")
                    and toolset.startswith("mcp-"))
        if both_mcp:
            pass  # 允许
        elif override:
            pass  # 显式允许（插件场景）
        else:
            # 拒绝 - 防止意外覆盖内置工具
            logger.error("Tool registration REJECTED: ...")
            return

def tool_error(message, **extra) -> str:
    result = {"error": str(message)}
    if extra:
        result.update(extra)
    return json.dumps(result, ensure_ascii=False)

def tool_result(data=None, **kwargs) -> str:
    if data is not None:
        return json.dumps(data, ensure_ascii=False)
    return json.dumps(kwargs, ensure_ascii=False)

from tools.registry import registry, tool_error

registry.register(name="read_file", toolset="file", schema=READ_FILE_SCHEMA,
    handler=_handle_read_file, check_fn=_check_file_reqs,
    emoji="📖", max_result_size_chars=100_000)
registry.register(name="write_file", toolset="file", schema=WRITE_FILE_SCHEMA,
    handler=_handle_write_file, check_fn=_check_file_reqs,
    emoji="✍️", max_result_size_chars=100_000)
registry.register(name="patch", toolset="file", schema=PATCH_SCHEMA,
    handler=_handle_patch, check_fn=_check_file_reqs,
    emoji="🔧", max_result_size_chars=100_000)
registry.register(name="search_files", toolset="file", schema=SEARCH_FILES_SCHEMA,
    handler=_handle_search_files, check_fn=_check_file_reqs,
    emoji="🔎", max_result_size_chars=100_000)

def discover_builtin_tools(tools_dir=None):
    tools_path = Path(tools_dir) or Path(__file__).resolve().parent
    module_names = [
        f"tools.{path.stem}"
        for path in sorted(tools_path.glob("*.py"))
        if path.name not in {"__init__.py", "registry.py", "mcp_tool.py"}
        and _module_registers_tools(path)  # AST 分析
    ]
    for mod_name in module_names:
        importlib.import_module(mod_name)  # 触发自注册

Agent Loop → handle_function_call()
                ↓
            coerce_tool_args()        # 参数类型强转
                ↓
            pre_tool_call hook        # 插件拦截
                ↓
            registry.dispatch()       # 路由到处理器
                ↓
            async bridge (_run_async) # 异步桥接（如果需要）
                ↓
            post_tool_call hook       # 后置处理
                ↓
            transform_tool_result     # 结果转换
                ↓
            返回 JSON 字符串

try:
    if entry.is_async:
        return _run_async(entry.handler(args, **kwargs))
    return entry.handler(args, **kwargs)
except Exception as e:
    raw = f"Tool execution failed: {type(e).__name__}: {e}"
    sanitized = _sanitize_tool_error(raw)  # 脱敏处理
    return json.dumps({"error": sanitized})

def check_vision_requirements():
    """检查视觉分析所需依赖"""
    try:
        import some_vision_lib
        return True
    except ImportError:
        return False

_CHECK_FN_TTL_SECONDS = 30.0
_check_fn_cache: Dict[Callable, tuple[float, bool]] = {}

def _check_fn_cached(fn):
    now = time.monotonic()
    cached = _check_fn_cache.get(fn)
    if cached and (now - cached[0]) < _CHECK_FN_TTL_SECONDS:
        return cached[1]  # 命中缓存
    try:
        value = bool(fn())
    except Exception:
        value = False  # 异常 = 不可用
    _check_fn_cache[fn] = (now, value)
    return value

TOOLSETS = {
    "web": {
        "description": "Web research and content extraction tools",
        "tools": ["web_search", "web_extract"],
        "includes": []
    },
    "debugging": {
        "description": "Debugging and troubleshooting toolkit",
        "tools": ["terminal", "process"],
        "includes": ["web", "file"]  # 组合其他 toolset
    },
    "safe": {
        "description": "Safe toolkit without terminal access",
        "tools": [],
        "includes": ["web", "vision", "image_gen"]  # 纯组合，无直接工具
    },
}

from toolsets import create_custom_toolset

create_custom_toolset(
    name="my_workflow",
    description="Custom workflow toolset",
    tools=["my_tool", "web_search"],
    includes=["file"]  # 组合 file toolset
)

from tools.registry import registry, tool_error

READ_FILE_SCHEMA = {
    "name": "read_file",
    "description": "Read file contents...",
    "parameters": {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "File path to read"},
            "offset": {"type": "integer", "description": "Starting line number", "default": 1},
            "limit": {"type": "integer", "description": "Max lines to return", "default": 500},
        },
        "required": ["path"]
    }
}

def _handle_read_file(args, **kw):
    tid = kw.get("task_id") or "default"
    return read_file_tool(
        path=args.get("path", ""),
        offset=args.get("offset", 1),
        limit=args.get("limit", 500),
        task_id=tid
    )

registry.register(
    name="read_file",
    toolset="file",
    schema=READ_FILE_SCHEMA,
    handler=_handle_read_file,
    check_fn=_check_file_reqs,
    emoji="📖",
    max_result_size_chars=100_000,
)

registry.register(
    name="web_extract",
    toolset="web",
    schema=WEB_EXTRACT_SCHEMA,
    handler=lambda args, **kw: web_extract_tool(
        args.get("urls", [])[:5] if isinstance(args.get("urls"), list) else [],
        "markdown"),
    check_fn=check_web_api_key,
    requires_env=_web_requires_env(),
    is_async=True,         # 标记为异步
    emoji="📄",
    max_result_size_chars=100_000,
)

def _build_dynamic_schema_overrides() -> dict:
    """每次 get_definitions() 调用时执行，返回要覆盖的 Schema 字段"""
    overrides_params = {**DELEGATE_TASK_SCHEMA["parameters"]}
    overrides_params["properties"] = {
        k: dict(v) for k, v in
        DELEGATE_TASK_SCHEMA["parameters"]["properties"].items()
    }
    overrides_params["properties"]["tasks"]["description"] = (
        _build_tasks_param_description()  # 读取当前配置
    )
    return {
        "description": _build_top_level_description(),
        "parameters": overrides_params,
    }

registry.register(
    name="delegate_task",
    ...,
    dynamic_schema_overrides=_build_dynamic_schema_overrides,
)

{
    "name": "search_files",
    "description": "Search file contents or find files by name...",
    "parameters": {
        "type": "object",
        "properties": {
            "pattern": {"type": "string", "description": "Regex pattern..."},
            "target": {
                "type": "string",
                "enum": ["content", "files"],
                "default": "content"
            },
            "limit": {"type": "integer", "default": 50},
        },
        "required": ["pattern"]
    }
}

def _handle_write_file(args, **kw):
    if not args.get("path"):
        return tool_error("write_file: missing required field 'path'...")
    if "content" not in args:
        return tool_error("write_file: missing required field 'content'...")
    # ...

mcp_servers:
  filesystem:
    command: "npx"
    args: ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
    timeout: 120
  remote_api:
    url: "https://my-mcp-server.example.com/mcp"
    headers:
      Authorization: "Bearer sk-..."