Agent工具调用机制解析：从注册到动态发现的全流程实现

1次阅读

共计 2363 个字符，预计需要花费 6 分钟才能阅读完成。

想象一个 CI/CD 流水线场景：当你的构建系统突然需要调用新部署的代码扫描工具时，如果每次新增工具都需要人工修改配置并重启服务，这种强耦合的设计会让系统变得脆弱且难以维护。更理想的模式是——Agent 能自动发现新工具并安全调用，这正是动态工具发现机制的核心价值。

另一个典型案例是数据分析平台：当用户临时上传一个自定义 Python 函数作为分析工具时，系统需要即时验证其输入输出格式、权限范围，并将其无缝接入现有工具调用链。这种灵活性要求工具管理系统具备完善的元数据描述和运行时发现能力。

集中式注册表 （Centralized Registry）
所有工具信息存储在单一数据库（如 MySQL/Redis）
优点：强一致性，易于实现权限管控
缺点：单点故障风险，扩展性受限
分布式注册表 （Distributed Registry）
采用 etcd/ZooKeeper 等分布式协调服务
优点：高可用性，天然支持多区域部署
缺点：实现复杂度高，需要处理最终一致性
混合式注册表 （Hybrid Registry）
核心工具用集中式，边缘工具用分布式
折中方案：平衡了稳定性和灵活性

每个工具必须声明完整的接口契约，例如：

# tools/scan_api/openapi.yaml
openapi: 3.0.0
info:
  title: Code Scanner
  version: 1.0.0
paths:
  /scan:
    post:
      parameters:
        - name: repo_url
          in: query
          required: true
          schema:
            type: string
            format: uri
      responses:
        '200':
          description: Scan report
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/ScanResult'
components:
  schemas:
    ScanResult:
      type: object
      properties:
        vulnerabilities:
          type: array
          items:
            $ref: '#/components/schemas/Vulnerability'

身份（Identity）：调用方必须提供有效认证凭证
权限（Permission）：检查 ACL（访问控制列表）中是否授权
范围（Scope）：验证请求参数是否在允许范围内（如项目 ID 匹配）
时效（TTL）：临时令牌需检查过期时间

from typing import Callable, TypeVar
from functools import wraps

T = TypeVar('T')

class ToolRegistry:
    _tools = {}

    @classmethod
    def register(
        cls,
        name: str,
        input_schema: dict,
        output_schema: dict
    ) -> Callable[[T], T]:
        def decorator(func: T) -> T:
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 运行时类型验证逻辑
                return func(*args, **kwargs)

            cls._tools[name] = {
                'function': wrapper,
                'input_schema': input_schema,
                'output_schema': output_schema,
                'last_heartbeat': time.time()}
            return wrapper
        return decorator

# 使用示例
@ToolRegistry.register(
    name="code_scanner",
    input_schema={"repo_url": {"type": "string", "format": "uri"}},
    output_schema={"vulnerabilities": {"type": "array"}}
)
def scan_code(repo_url: str) -> dict:
    # 实际扫描逻辑
    return {"vulnerabilities": []}

import threading
import time

class HealthChecker:
    @staticmethod
    def start():
        def _check_loop():
            while True:
                for name, tool in ToolRegistry._tools.items():
                    # 超过 30 秒未心跳视为不可用
                    if time.time() - tool['last_heartbeat'] > 30:
                        print(f"警告：工具 {name} 失去响应")
                time.sleep(10)

        threading.Thread(target=_check_loop, daemon=True).start()