> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/05_tool_layer/5.5_miniharness_tools.md).

# 5.5 实战：MiniHarness 工具层实现

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

## 5.5.1 工具抽象与基类

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

```python
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 阶段的工具调用流程：

```python
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 缓存：

```python
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**：命令执行工具，演示了长时间运行任务的超时处理

```python
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 系统工具层设计。