# 9.5 实战：为MiniHarness集成MCP

本节在MiniHarness框架的基础上，实现完整的MCP Client集成，包括动态工具发现、Schema缓存、权限管理。

## 9.5.1 架构设计

MiniHarness的MCP集成采用四层架构：

1. **MiniHarness Agent**：上下文管理和工具调用入口
2. **MCPToolAdapter**：将MCP工具转换为LLM可调用形式，处理错误和降级
3. **MCPToolRegistry + SchemaCache**：动态发现工具、缓存Schema和管理权限
4. **MCP Servers**：多种传输协议支持(stdio、HTTP)

完整代码见 `lab/mini_harness/mcp/integration.py`

## 9.5.2 核心设计决策

### 双层 Schema 缓存策略

为什么需要两层缓存？

* **内存缓存**：毫秒级访问，适合热工具
* **磁盘缓存**：跨进程持久化，TTL机制自动过期

```python
class ToolSchemaCache:
    """双层缓存：内存 + 磁盘"""
    async def get(self, server_id: str, tool_name: str):
        # 1. 检查内存缓存(毫秒级)
        if cache_key in self.memory_cache:
            if not expired:
                return cached

        # 2. 检查磁盘缓存(毫秒级,TTL)
        if disk_path.exists:
            cached = load_from_disk()
            self.memory_cache[cache_key] = cached  # 晋升
            return cached

        return None
```

**效果**：Schema缓存通常能减少重复工具描述带来的Token消耗，并降低工具发现延迟；具体节省比例取决于工具数量、Schema 大小和缓存命中率。

### 异步锁保护并发发现

多个Agent同时调用同一工具时的竞态条件：

```python
class MCPToolRegistry:
    async def discover_tools(self, force: bool = False):
        async with self.lock:  # 关键：保护并发发现
            # 检查是否需要重新发现(5分钟发现一次)
            if not force and (now - self.last_discovery_time) < 300:
                return self.tool_to_servers

            # 构建工具 -> Server映射
            # tool_name -> [server_ids]
            ...
```

**好处**：避免重复发现、减少Server负载、保证发现的一致性。

### MCPServerConfig 数据类

为什么用dataclass而不是dict？

```python
@dataclass
class MCPServerConfig:
    server_id: str
    server_name: str
    transport_type: str      # stdio | http
    endpoint: str
    enabled: bool = True
    priority: int = 0         # 多个Server提供同工具时的优先级
    timeout_seconds: int = 30
```

**设计理由**：

* 类型安全（IDE补全、类型检查）
* 默认值管理
* 易于扩展（如添加认证、速率限制）
* 可持久化为JSON

### MCPToolAdapter 的适配模式

为什么需要适配层？

```python
class MCPToolAdapter:
    """将MCP格式 -> LLM格式"""

    def _schema_to_llm_format(self, schema: CachedToolSchema):
        return {
            "name": schema.tool_name,
            "description": schema.description,
            "input_schema": schema.input_schema,
            # 可扩展：添加权限检查、速率限制、审计等
        }
```

**优点**：

* LLM SDK版本变化时，只需改适配层
* 易于添加权限检查、审计、降级逻辑
* 分离MCP协议和LLM SDK

## 9.5.3 关键代码片段

### 工具调用生命周期

以下代码展示工具调用的完整生命周期，从查找Server、发送请求到处理响应：

```python
async def call_tool(self, tool_name: str, arguments: Dict[str, Any]):
    # 第1步：查找提供工具的Server
    server_ids = self.tool_to_servers.get(tool_name, [])
    if not server_ids:
        return False, None, f"Tool {tool_name} not found"

    # 第2步：尝试第一个Server(可扩展为轮询/负载均衡)
    server_id = server_ids[0]
    config = self.servers[server_id]
    client = await self._get_client(server_id, config)

    # 第3步：发送请求
    response = await client.send_request(
        "tools/call",
        {"name": tool_name, "arguments": arguments}
    )

    # 第4步：处理响应
    if "error" in response:
        return False, None, response["error"]["message"]

    result = response.get("result", {})
    content = result.get("content", [])
    return True, content[0].get("text", ""), ""
```

### MiniHarness 集成入口

以下是MCP集成的核心类，负责初始化各层组件和处理工具调用：

```python
class MiniHarnessWithMCP:
    """集成MCP的MiniHarness核心类"""

    def __init__(self):
        self.schema_cache = ToolSchemaCache()
        self.registry = MCPToolRegistry(self.schema_cache)
        self.adapter = MCPToolAdapter(self.registry)

    async def initialize(self) -> None:
        """初始化MCP集成"""
        # 注册Server
        await self.registry.add_server(MCPServerConfig(
            server_id="filesystem",
            server_name="File System Server",
            transport_type="stdio",
            endpoint="./mcp_filesystem_server.py",
        ))

        # 发现工具
        tools_by_server = await self.registry.discover_tools()

    async def process_tool_call(
        self,
        tool_name: str,
        tool_input: str,
        agent_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """处理Agent的工具调用"""
        return await self.adapter.call_tool_from_llm(tool_name, tool_input, agent_id)
```

### 错误处理与降级

以下代码演示了如何处理JSON解析错误和工具调用异常，确保系统稳定性：

```python
async def call_tool_from_llm(
    self,
    tool_name: str,
    tool_input: str,
    agent_id: Optional[str] = None,
) -> Dict[str, Any]:
    """从LLM调用工具(处理输入解析)"""
    try:
        # 尝试解析JSON
        arguments = json.loads(tool_input)
    except json.JSONDecodeError as e:
        return {
            "success": False,
            "result": None,
            "error": f"Invalid JSON input: {e}",
        }

    # 调用工具,捕获异常
    try:
        success, result, error = await self.registry.call_tool(
            tool_name, arguments, agent_id
        )
        return {
            "success": success,
            "result": result,
            "error": error,
        }
    except Exception as e:
        return {
            "success": False,
            "result": None,
            "error": str(e),
        }
```

## 9.5.4 集成清单

使用此MCP集成时的检查清单：

**初始化**

* [ ] 创建MCPToolRegistry实例
* [ ] 创建ToolSchemaCache（选择缓存目录）
* [ ] 添加MCP Server配置

**工具发现**

* [ ] 调用discover\_tools()发现可用工具
* [ ] 验证tool\_to\_servers映射正确
* [ ] 检查缓存命中率

**工具调用**

* [ ] 从LLM获取工具名称和输入
* [ ] 调用adapter.call\_tool\_from\_llm()
* [ ] 处理成功和错误情况
* [ ] 记录调用结果（审计）

**性能优化**

* [ ] 启用Schema缓存
* [ ] 定期清理过期缓存（TTL机制）
* [ ] 监控缓存命中率(get\_stats())
* [ ] 考虑异步预热热工具Schema

**错误处理**

* [ ] 处理Server不可用的情况
* [ ] 实现工具调用超时
* [ ] 为关键工具配置fallback方案
* [ ] 记录详细的错误日志

**安全性**

* [ ] 验证MCP Server的身份
* [ ] 限制Agent可以调用的工具
* [ ] 记录所有工具调用（审计日志）
* [ ] 对高风险工具添加人工审批

## 9.5.5 本小节小结

通过MCPToolRegistry、SchemaCache和MCPToolAdapter的组合，MiniHarness实现了完整的MCP集成：

1. **动态发现**：自动发现多个MCP Server提供的工具
2. **智能缓存**：双层Schema缓存减少延迟和Token消耗
3. **灵活适配**：将MCP工具转换为LLM可用的格式
4. **完整的生命周期**：从发现、获取Schema到调用，一套完整的系统

关键指标：

* Schema缓存减少重复工具描述带来的Token消耗
* 工具发现延迟从秒级降至毫秒级（缓存命中）
* 支持并发工具调用和错误恢复

完整实现代码见 `lab/mini_harness/mcp/integration.py`


---

# 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-san-bu-fen-xi-tong-ji-cheng-yu-gong-cheng-shi-jian/09_mcp/9.5_miniharness_mcp.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.