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机制自动过期

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缓存减少80%以上的令牌消耗。工具发现延迟从秒级降至毫秒级。

异步锁保护并发发现

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

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？

@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 的适配模式

为什么需要适配层？

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、发送请求到处理响应：

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集成的核心类，负责初始化各层组件和处理工具调用：

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解析错误和工具调用异常，确保系统稳定性：

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缓存减少延迟和令牌消耗

3. 灵活适配：将MCP工具转换为LLM可用的格式

4. 完整的生命周期：从发现、获取Schema到调用，一套完整的系统

关键指标：

• Schema缓存减少80%以上的令牌消耗

• 工具发现延迟从秒级降至毫秒级（缓存命中）

• 支持并发工具调用和错误恢复

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