5.5 实战：MiniHarness 工具层实现

本节实现一个完整的 MiniHarness 工具层，包括工具协议、注册表、执行流水线和几个内置工具。核心设计原则是“发现、权限、执行、结果”四阶段流水线。

5.5.1 工具抽象与基类

工具层的核心是统一的 Tool 基类，定义了所有工具必须实现的接口。工具的设计遵循“契约优先”原则：

from abc import ABC, abstractmethod

class Tool(ABC):
    """工具基类 - 定义工具的通用契约"""

    @abstractmethod
    async def call(self, params: Dict[str, Any]) -> ToolResult:
        """执行工具的核心逻辑"""
        pass

    @abstractmethod
    def name(self) -> str:
        """工具的唯一标识符"""
        pass

    @abstractmethod
    def description(self) -> str:
        """工具描述"""
        pass

    @abstractmethod
    def input_schema(self) -> Dict[str, Any]:
        """JSON Schema 定义输入约束"""
        pass

    def check_permissions(self, context: Any) -> bool:
        """权限检查钩子 - 可被子类覆盖"""
        return True

关键设计决策：

1. 异步执行：所有工具使用 async/await，支持 IO 密集型操作

2. Schema 驱动：通过 JSON Schema 定义输入，实现自文档化和验证

3. 权限钩子：check_permissions() 允许子类实现细粒度权限控制

4. 统一结果类型：ToolResult 封装成功/失败/执行时间等元数据

完整实现参见 lab/mini_harness/tools/builtin.py。

5.5.2 工具执行流水线

执行流水线是系统的核心，实现了 4 阶段的工具调用流程：

class ExecutionPipeline:
    """执行流水线 - 控制工具的调用生命周期"""

    async def execute(self, tool_name: str,
                     input_params: Dict[str, Any]) -> ToolResultBlock:
        # 阶段 1:工具发现(lookup)
        tool = self.tool_registry.get(tool_name)
        if not tool:
            return ToolResultBlock(tool_name=tool_name, success=False,
                                 content=f"Tool not found: {tool_name}")

        # 阶段 2:权限检查(authorization)
        if not tool.check_permissions({}):
            return ToolResultBlock(tool_name=tool_name, success=False,
                                 content=f"Permission denied for tool: {tool_name}")

        # 阶段 3:执行工具(execution)
        # 阶段 4:结果处理(result handling)
        result = await tool.call(input_params)
        return ToolResultBlock(
            tool_name=tool_name,
            success=result.success,
            content=result.content,
            execution_time=result.execution_time,
            error_type=result.error_type,
        )

四阶段设计的优点：

• 可观测性：每个阶段可独立记录日志和监控

• 可扩展性：插入速率限制、超时控制、重试逻辑等中间件

• 安全性：权限检查在执行前进行，防止越权操作

• 容错性：捕获任意异常并转换为统一的失败结果

完整实现参见 lab/mini_harness/tools/builtin.py 中的 ExecutionPipeline 类。

5.5.3 工具注册表

工具注册表充当服务发现中心和 Schema 缓存：

class ToolRegistry:
    """工具注册中心 - 管理工具的生命周期"""

    def register(self, tool: Tool):
        """注册工具并缓存其 Schema"""
        name = tool.name()
        self.tools[name] = tool

        # 缓存 Schema 避免重复调用
        self.schema_cache[name] = {
            "name": name,
            "description": tool.description(),
            "input_schema": tool.input_schema()
        }

    def list_tools(self) -> List[Dict[str, Any]]:
        """返回所有工具的 Schema(用于 LLM 提示词)"""
        return list(self.schema_cache.values())

设计特点：

• Schema 缓存：减少重复的 Schema 构建，加速工具列表生成

• 统一发现：LLM 通过 list_tools() 获取全部可用工具及其约束

• 松耦合：执行流水线仅依赖注册表接口，不需关心工具实现细节

完整实现参见 lab/mini_harness/tools/registry.py。

5.5.4 内置工具示例

三个内置工具展示了不同的设计模式：

BashTool：命令执行工具，演示了长时间运行任务的超时处理

class BashTool(Tool):
    async def call(self, params: Dict[str, Any]) -> ToolResult:
        import shlex
        import subprocess

        command = params.get("command", "")
        timeout = params.get("timeout", 30)
        allowed_commands = {"echo", "pwd", "true", "false", "sleep"}
        argv = shlex.split(command)
        if not argv or argv[0] not in allowed_commands:
            return ToolResult(success=False, content="Command not in allowlist",
                              execution_time=0.0, error_type="PermissionError")

        try:
            result = subprocess.run(argv, shell=False,
                                  timeout=timeout, text=True,
                                  capture_output=True)
            return ToolResult(
                success=result.returncode == 0,
                # ... 其他结果字段
            )
        except subprocess.TimeoutExpired:
            return ToolResult(
                success=False,
                error_type="TimeoutError",
                # ... 其他结果字段
            )

FileReadTool 和 FileWriteTool：文件操作工具，演示了错误分类和限制（如 max_bytes）

这些工具的完整实现包括详细的错误处理、参数验证和边界情况处理，参见 lab/mini_harness/tools/builtin.py。

5.5.5 扩展路径

1. 中间件架构：在执行流水线中插入速率限制、重试、缓存层

2. 权限模型：为不同用户和工具组合配置细粒度权限

3. 超时和资源限制：为每个工具实例设置 CPU/内存配额

4. 并发执行：使用 asyncio.gather() 实现多工具并发调用

5. 工具依赖图：支持工具间的数据流依赖（DAG 执行）

6. 结果缓存：基于输入哈希缓存工具结果，支持失效策略

5.5.6 本节小结

MiniHarness 工具层的核心贡献：

1. 清晰的抽象边界：Tool 基类定义统一契约，工具之间互不依赖

2. 四阶段执行流水线：发现→权限→执行→处理，可观测且可扩展

3. Schema 驱动设计：工具通过 JSON Schema 自描述，LLM 可直接理解

4. 错误优先处理：所有失败路径返回统一的 ToolResult 类型

这些原则适用于任何规模的 AI Agent 系统工具层设计。