# 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(success=False,
                                 content=f"Tool not found")

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

        # 阶段 3:执行工具(execution)
        # 阶段 4:结果处理(result handling)
        result = await tool.call(input_params)
        return ToolResultBlock(
            success=result.success,
            # ... 其他结果字段
        )
```

四阶段设计的优点：

* **可观测性**：每个阶段可独立记录日志和监控
* **可扩展性**：插入速率限制、超时控制、重试逻辑等中间件
* **安全性**：权限检查在执行前进行，防止越权操作
* **容错性**：捕获任意异常并转换为统一的失败结果

完整实现参见 `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, error_type="PolicyViolation")

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


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET 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?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.