自定义工具开发与注册让 Skills 能力无限扩展

凌晨两点，DevOps 工程师小林面对第三个出故障的 Skill 陷入沉思。API 密钥散落在每个 Skill 文件的硬编码里，更新凭证要改 12 个文件。上一章我们费劲心力构建了安全调用层，但小林的团队需要的不只是“能调用”——他们需要一种可发现的、标准化的工具复用机制。三个小时后，小林用一套自制工具注册系统将 12 个 Skill 重新整合，凭证改动从 12 处缩减为 1 处。

读完上一章，你已经拥有一个可靠的外部 API 安全调用层。但当你开始构建第二个、第三个 Skill 时，会迅速发现瓶颈：同样的 HTTP 封装在每个 Skill 里重写一遍，文档字符串写得东倒西歪，想找一个“获取用户订单明细”的函数得全局搜索。本章的目标很明确——把你的安全调用能力抽象成标准化的、可发现的工具函数，让每个 Skill 都能像拼乐高一样组合使用。

你需要什么

开始之前，请确认以下环境已就绪：

预计时间：45 分钟（含代码编写、注册、热更新验证）

最终成果

读完并完成练习后，你将得到：

• 一个类型安全的工具注册中心，支持按标签搜索、语义化版本管理
• 一套标准化的工具函数契约模板，让每个工具的输入输出和副作用一目了然
• 一个无重启热更新机制，在不中断 Skill 服务的状态下加载新工具版本
• 一个可运行的示例，展示工具函数如何被 Skill 发现并调用

这套基础设施将成为你整个 Skill 生态的骨架。后续开发的每个能力包都只需“挂载”到这个骨架，而不是推倒重来。

为什么做这个

从上一章的结尾出发：我们已经有了一个带缓存和降级的 API 调用层。这个层面的问题是耦合——每个 Skill 直接 import 这个模块，出了问题要逐文件修改。

工具注册中心要解决的就三件事：

1. 发现（Discovery）：不用全局搜索代码，用标签和名称就能找到工具
2. 契约（Contract）：每个工具附带类型签名、文档、使用条件，LLM 调用时不会猜错参数
3. 演化（Evolution）：工具升级版本不中断已有 Skill，渐进式迁移

这三个能力对应本章的三个子节，我们一步步构建。

步骤一：实现工具注册表的基础骨架

先从最核心的注册表开始——它应该是一个单例，存储所有已注册工具的元信息。

1.1 创建注册表类

在项目里新建文件 tool_registry.py：

# tool_registry.py
import inspect
from typing import Any, Callable, Optional
from dataclasses import dataclass, field
from datetime import datetime, timezone

@dataclass
class ToolMeta:
    """工具元信息——注册中心存储的核心单元"""
    name: str                              # 工具唯一标识，如 "get_user_orders"
    description: str                       # 给 LLM 阅读的功能描述
    func: Callable                         # 实际可执行函数
    tags: set[str] = field(default_factory=set)  # 标签，如 {"user", "order"}
    version: str = "1.0.0"                 # 语义化版本
    input_schema: Optional[dict] = None    # 输入参数的 JSON Schema
    created_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

class ToolRegistry:
    """工具注册中心——单例模式，全局唯一"""
    _instance: Optional["ToolRegistry"] = None

    def __new__(cls) -> "ToolRegistry":
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._tools: dict[str, ToolMeta] = {}
        return cls._instance

    def register(self, meta: ToolMeta) -> None:
        """注册一个工具。同名工具会覆盖旧版本（先简单处理，步骤三再优化）"""
        if meta.name in self._tools:
            existing_ver = self._tools[meta.name].version
            print(f"[Registry] 警告: 工具 '{meta.name}' 已存在 (v{existing_ver})，将被 v{meta.version} 覆盖")
        self._tools[meta.name] = meta
        print(f"[Registry] 工具 '{meta.name}' v{meta.version} 已注册 | 标签: {meta.tags}")

预期结果：运行 python tool_registry.py 不会报错，类定义正常加载。

步骤二：定义工具函数的最佳契约

有了注册骨架，接下来定义一个标准化的工具函数结构。每个工具都应携带：

• 类型注解（Pydantic 或 Python typing）
• OpenAPI 风格的文档字符串
• 明确的幂等性标识

2.1 使用装饰器绑定注册

在 tool_registry.py 文件中追加装饰器：

# 追加在 ToolRegistry 类定义之后

_registry = ToolRegistry()  # 全局单例实例

def register_tool(
    name: str,
    description: str,
    tags: Optional[set[str]] = None,
    version: str = "1.0.0",
):
    """装饰器：将函数自动包装为 ToolMeta 并注册

    示例:
        @register_tool(
            name="get_user_orders",
            description="根据用户 ID 获取最近 30 天的订单列表",
            tags={"user", "order"},
            version="1.0.0"
        )
        def get_user_orders(user_id: str, limit: int = 10) -> dict:
            ...
    """
    def decorator(func: Callable):
        # 提取输入参数的类型信息，生成基础 schema
        sig = inspect.signature(func)
        params_schema = {}
        for pname, param in sig.parameters.items():
            param_type = str(param.annotation) if param.annotation != inspect.Parameter.empty else "Any"
            default = param.default if param.default != inspect.Parameter.empty else "REQUIRED"
            params_schema[pname] = {"type": param_type, "default": default}

        meta = ToolMeta(
            name=name,
            description=description,
            func=func,
            tags=tags or set(),
            version=version,
            input_schema=params_schema,
        )
        _registry.register(meta)
        return func  # 保持原函数不变，装饰器仅做注册副作用
    return decorator

2.2 编写一个符合契约的工具函数

在 tools/example_tools.py 中编写示例工具：

# tools/example_tools.py
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(__file__)))

from tool_registry import register_tool

@register_tool(
    name="get_user_orders",
    description="""
    根据用户ID获取最近N天的订单列表。该操作为只读查询，幂等。
    参数:
    - user_id: 用户的唯一标识字符串，格式为 "USR-xxxxxxxx"
    - days: 查询最近多少天的订单，默认 30，最大 365
    返回:
    - dict 包含 "orders" 列表和 "total_price" 总计金额
    - 若用户不存在返回 {"error": "USER_NOT_FOUND", "orders": []}
    副作用: 无（只读查询）
    """,
    tags={"user", "order", "query", "read-only"},
    version="1.0.0"
)
def get_user_orders(user_id: str, days: int = 30) -> dict:
    # 实际实现略——通常是调用上一章的 secure_api 模块
    pass

预期结果：导入 tools/example_tools.py 时，控制台自动打印 [Registry] 工具 'get_user_orders' v1.0.0 已注册 | 标签: {'query', 'read-only', 'order', 'user'}。不需要手动调用任何注册函数。

注意：踩坑经验
装饰器参数里的 description 会被 Claude 模型用来判断“是否应该调用这个工具”。如果写得太简略（例如只写“获取订单”），Claude 可能错过调用时机。务必描述触发场景和适用条件，OpenAPI 风格的文档字符串是个好习惯。

步骤三：实现按标签搜索与版本管理

有了注册能力，下一步让工具可以被发现。在 ToolRegistry 类中追加搜索方法：

3.1 追加搜索接口

# 在 ToolRegistry 类中追加以下方法

    def find_by_tags(self, tags: set[str], match_all: bool = False) -> list[ToolMeta]:
        """按标签搜索工具

        Args:
            tags: 要匹配的标签集合
            match_all: True 时要求工具拥有全部标签；False 时至少匹配一个

        Returns:
            匹配的 ToolMeta 列表
        """
        results = []
        for tool in self._tools.values():
            if match_all:
                if tags.issubset(tool.tags):  # 工具标签包含了所有指定标签
                    results.append(tool)
            else:
                if tags & tool.tags:  # 至少有一个交集
                    results.append(tool)
        # 按名称排序，保证结果确定性
        results.sort(key=lambda t: t.name)
        return results

    def get_latest_version(self, base_name: str) -> Optional["ToolMeta"]:
        """获取指定基础名的最新版本工具"""
        # 找出所有匹配基础名的工具，按语义化版本排序
        candidates = [
            t for t in self._tools.values()
            if t.name == base_name
        ]
        if not candidates:
            return None
        # 简化版版本排序——实际生产环境建议用 packaging.version
        candidates.sort(key=lambda t: [int(x) for x in t.version.split(".")], reverse=True)
        return candidates[0]

3.2 测试搜索功能

在同一个 tool_registry.py 文件末尾追加测试代码：

# 测试代码——仅当直接运行本文件时执行
if __name__ == "__main__":
    # 注册几个测试工具
    @register_tool(name="read_order", description="读订单", tags={"order", "read"})
    def read_order(): ...

    @register_tool(name="write_order", description="写订单", tags={"order", "write"})
    def write_order(): ...

    @register_tool(name="send_email", description="发邮件", tags={"email", "notification"})
    def send_email(): ...

    # 测试按标签搜索
    registry = ToolRegistry()
    print("\n--- 搜索含有 'order' 标签的工具 ---")
    for tool in registry.find_by_tags({"order"}):
        print(f"  {tool.name} ({', '.join(tool.tags)})")
    # 预期输出: read_order (order, read) 和 write_order (order, write)，按名称排序

    print("\n--- 搜索同时含有 'order' AND 'write' 标签的工具 ---")
    for tool in registry.find_by_tags({"order", "write"}, match_all=True):
        print(f"  {tool.name} ({', '.join(tool.tags)})")
    # 预期输出: 仅 write_order

预期结果：运行 python tool_registry.py，先看到三条注册日志，然后看到精确的标签搜索结果。

步骤四：在不重启服务的情况下热更新工具

静态注册只在启动时生效。生产环境下，我们往往需要修复一个工具函数后立即生效，而不是重启整个 Skill 运行时。这里用一个简单的文件监控 + 重新加载机制实现。

4.1 实现热更新加载器

新建文件 hot_reload.py：

# hot_reload.py
import importlib
import os
from pathlib import Path
from typing import Dict
from datetime import datetime
from tool_registry import ToolRegistry, _registry

class HotReloader:
    """监控指定目录下的工具模块，支持热加载新版本"""

    def __init__(self, tools_dir: str = "./tools"):
        self.tools_dir = Path(tools_dir).resolve()
        self._load_timestamps: Dict[str, float] = {}  # 模块路径 -> 上次加载时间戳
        self.registry = _registry  # 全局注册中心的引用

    def load_all(self):
        """首次加载 tools_dir 下所有 Python 文件"""
        for py_file in self.tools_dir.glob("*.py"):
            if py_file.name.startswith("_"):
                continue  # 跳过 __init__.py 等
            module_name = py_file.stem
            try:
                # 动态导入模块——模块中的 @register_tool 装饰器会自动注册到全局注册中心
                importlib.import_module(f"tools.{module_name}")
                self._load_timestamps[str(py_file)] = py_file.stat().st_mtime
                print(f"[HotReload] 初次加载: {module_name}")
            except Exception as e:
                print(f"[HotReload] 加载失败 {module_name}: {e}")

    def check_and_reload(self):
        """检查文件是否有更新，有则重新加载"""
        for py_file in self.tools_dir.glob("*.py"):
            file_path = str(py_file)
            current_mtime = py_file.stat().st_mtime
            if file_path not in self._load_timestamps:
                # 新文件，首次加载
                module_name = py_file.stem
                importlib.import_module(f"tools.{module_name}")
                self._load_timestamps[file_path] = current_mtime
                print(f"[HotReload] 发现新工具文件: {module_name}")
            elif current_mtime > self._load_timestamps[file_path]:
                # 已有的文件，但被更新了——重新加载
                module_name = py_file.stem
                reloaded = importlib.reload(
                    importlib.import_module(f"tools.{module_name}")
                )
                self._load_timestamps[file_path] = current_mtime
                print(f"[HotReload] 重新加载: {module_name} @ {datetime.fromtimestamp(current_mtime).isoformat()}")

4.2 在 Skill 启动脚本中集成热更新

在项目的 entrypoint 或 Skill 的 SKILL.md 引用的脚本中添加：

# skill_starter.py
import time
from hot_reload import HotReloader

reloader = HotReloader(tools_dir="./tools")
reloader.load_all()  # 启动时首次加载所有工具

# 在实际项目中，这里会有 Kafka 消费者或 HTTP 服务器的主循环
# 我们在每次处理请求前做一次检查（生产环境建议降低检查频率或用文件事件通知）
while True:
    reloader.check_and_reload()
    # ... 处理 Skill 请求的逻辑
    time.sleep(5)  # 实际项目的检查间隔可根据 SLA 调整

预期结果：在 tools/ 目录中新增或修改 .py 文件，5 秒内控制台输出重载日志。新版本的 @register_tool 装饰的函数自动覆盖旧版本。

回顾

从零到一搭建一套 Skills 工具生态的基础设施，你完成了：

这一整套代码不到 200 行，但解决的是上一章遗留的核心问题：安全调用能力从“一次性胶水代码”变成“可复用、可发现、可演化的标准化资产”。

现在回到小林的故事：凌晨 5 点，他用这套工具注册系统把 12 个 Skill 全部迁移完毕。新增一个 API 集成时，只需要写 30 行工具函数并打上标签。检索工具变成了 registry.find_by_tags({"notification"}) 而不是全局搜索 grep -r "send_email"。

行动清单

• [ ] 复制 tool_registry.py 骨架代码到项目，用你自己的一个现有工具函数测试 @register_tool 装饰器
• [ ] 至少给 3 个工具打上标签，测试 find_by_tags() 的 match_all 和普通模式
• [ ] 启动 HotReloader，修改其中一个工具函数并观察 5 秒内的重新加载日志
• [ ] 在 SKILL.md 的指令区写明“可用工具通过 registry.find_by_tags() 发现”，让 Claude 知道去哪里找
• [ ] 保持每个工具函数的 description 字段包含：触发场景、输入约束、副作用说明

下一章预告

工具生态铺开了，但能力越大风险越大。任何一个注册进来的工具函数都可能包含 os.system("rm -rf /") 这样的恶意代码——你无法保证团队每个人或社区贡献者都值得完全信任。

下一章 《代码执行沙箱是防止恶意 Skill 的最后防线》 将直面这个问题：我们将分析代码注入的攻击向量，搭建隔离执行环境，并通过文件系统和网络白名单约束每个工具的破坏半径。毕竟，一个能无限扩展的生态，首先得是一个不会炸毁自己的生态。