# 9.1 Harness中的MCP集成设计

MCP(Model Context Protocol)已成为行业标准。关于MCP协议的基础介绍、核心架构、三种原语(Tools、Resources、Prompts)和设计哲学，请参阅《Claude 指南》第四章。

> 💡 **基础参考**：关于 MCP 协议的基础介绍和完整开发指南，请参阅《Claude 指南》第四章(4.1-4.5)。本节重点讨论 Harness 框架中的 MCP 集成特性。

## 9.1.1 Harness对MCP的消费模式

Harness 作为一个多工具整合框架，与标准 MCP 客户端的主要差异在于 **消费策略** 和 **性能优化**。

在标准客户端中，所有MCP Server直接连接到主应用。Harness引入了一个中间抽象层。下图是架构示意，不是 MCP 规范定义的部署拓扑：

```
用户请求 → Harness调度器 → 工具抽象层 → [MCP Client] → MCP Server
                           ↓
                      状态机管理
```

**优势**：

* 跨Server的事务一致性：同一工作流中的多个MCP Server调用可以共享上下文
* 失败恢复：工具层可以捕获单个Server故障，而不影响整个工作流
* 审计日志：统一记录所有工具调用，便于合规性检查

## 9.1.2 MCP Server发现与注册

### 动态发现机制

Harness 在启动时扫描配置的 MCP Server，而非硬编码：

```yaml
# harness.yaml
mcp_servers:
  - name: filesystem
    command: "mcp-server-filesystem"
    args: ["--root", "/workspace"]
    transport: stdio

  - name: postgres
    command: "mcp-server-postgres"
    args: ["--connection-string", "postgresql://..."]
    transport: streamable_http
    health_check:
      interval: 30s
      timeout: 5s
```

Harness会：

1. 启动每个Server进程（或连接到HTTP端点）
2. 调用 `tools/list`, `resources/list`, `prompts/list` 进行特性发现
3. 缓存 Schema 以减少运行时开销
4. 监控Server的健康状态

### 版本协商

不同的Server可能支持不同版本的MCP协议。Harness在连接时进行握手：

```python
# Harness初始化时
init_message = {
    "jsonrpc": "2.0",
    "id": 0,
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-11-25",
        "capabilities": {
            "sampling": {},
            "roots": {"listChanged": True},
            "elicitation": {}
        },
        "clientInfo": {
            "name": "HarnessMCPClient",
            "version": "1.0.0"
        }
    }
}
```

Server 返回其支持的版本后，客户端还必须发送 `notifications/initialized` 通知，之后才能进入正常 `tools/list`、`resources/read` 等操作阶段。

## 9.1.3 工具调用的性能优化

在生产环境中，频繁地获取和解析工具Schema会成为性能瓶颈。本小节介绍Harness采用的几种关键优化策略，包括智能缓存机制和流式传输优化，以减少网络往返和内存占用。

### Schema缓存策略

每次调用工具前重新获取Schema会浪费大量往返。Harness 采用缓存+增量更新策略：

**启动时（冷启动）**：

* 一次性获取所有Server的tools/list
* 将Schema存入本地缓存（SQLite或内存）
* 计算Schema的哈希值，用于增量检测

**运行时（热启动）**：

* 在工具调用前，先检查缓存中的Schema版本
* 调用 `tools/list` 时，根据工具列表内容计算哈希
* 如果哈希不匹配，再做全量同步

**示例实现**：

```python
class ToolSchemaCache:
    def __init__(self, cache_path="~/.harness/tool_cache.db"):
        self.db = sqlite3.connect(cache_path)
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS tool_schemas (
                server_name TEXT,
                tool_name TEXT,
                schema TEXT,
                schema_version TEXT,
                cached_at INTEGER,
                PRIMARY KEY (server_name, tool_name)
            )
        """)

    def get_or_fetch(self, server: MCPServer, tool_name: str) -> dict:
        # 1. 先查缓存
        cached = self.db.execute(
            "SELECT schema, schema_version FROM tool_schemas WHERE server_name = ? AND tool_name = ?",
            (server.name, tool_name)
        ).fetchone()

        if cached:
            schema_json, cached_version = cached

            # 2. 检查本地缓存版本是否过期
            current_version = server.get_cached_schema_version(tool_name)
            if current_version == cached_version:
                return json.loads(schema_json)  # 命中缓存

        # 3. 缓存未命中,重新获取
        schema = server.fetch_tool_schema(tool_name)
        self.db.execute(
            "INSERT OR REPLACE INTO tool_schemas VALUES (?, ?, ?, ?, ?)",
            (server.name, tool_name, json.dumps(schema),
             server.get_cached_schema_version(tool_name), int(time.time()))
        )
        self.db.commit()
        return schema
```

上例中的 `schema_version` 是 Harness 自己维护的缓存元数据，不是 MCP `tools/list` 的标准字段。真实实现应根据工具列表哈希、`notifications/tools/list_changed` 通知或服务端自定义元数据来失效缓存。

### 流式传输优化

对于大数据量的Resource读取（如导入大文件），Harness需要处理流式响应。

MCP本身支持通过 Streamable HTTP 的流式传输，但需要正确配置连接池：

```python
# 连接池配置
HTTP_CLIENT = httpx.AsyncClient(
    limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
    timeout=30.0
)

async def stream_resource_from_server(server: MCPServer, uri: str) -> AsyncIterator[bytes]:
    """流式读取资源内容,避免一次性加载到内存"""
    request = {
        "jsonrpc": "2.0",
        "id": generate_request_id(),
        "method": "resources/read",
        "params": {"uri": uri}
    }

    headers = {
        "Accept": "application/json, text/event-stream",
        "MCP-Protocol-Version": "2025-11-25",
    }
    if server.session_id:
        headers["MCP-Session-Id"] = server.session_id

    async with HTTP_CLIENT.stream("POST", server.endpoint, json=request, headers=headers) as response:
        async for chunk in response.aiter_bytes(chunk_size=8192):
            yield chunk
```

## 9.1.4 工具调用的状态机集成

在工作流执行中，MCP工具调用需要与Harness的状态机紧密配合。

### 状态转移中的工具调用

以下示例展示如何在Harness的状态机中集成MCP工具调用，使得工具调用与状态转移紧密配合：

```python
class ToolCallState(State):
    """工具调用状态"""

    def __init__(self, server_name: str, tool_name: str, params: dict):
        self.server = harness.get_mcp_server(server_name)
        self.tool_name = tool_name
        self.params = params

    async def execute(self, context: ExecutionContext) -> StateTransition:
        try:
            # 1. 准备工具调用请求
            request = {
                "jsonrpc": "2.0",
                "id": context.request_id,
                "method": "tools/call",
                "params": {
                    "name": self.tool_name,
                    "arguments": self.params
                }
            }

            # 2. 发送请求(带超时和重试)
            result = await self.server.call_with_retry(
                request,
                max_retries=3,
                timeout=30
            )

            # 3. 检查结果
            if "error" in result:
                return StateTransition(
                    next_state="error_handling",
                    context={"error": result["error"]}
                )

            # 4. 结果写入上下文
            context.set("tool_result", result["result"])

            return StateTransition(next_state="next_step")

        except TimeoutError:
            return StateTransition(
                next_state="retry_or_fallback",
                context={"reason": "timeout"}
            )
```

### 工具调用的原子性

在多步工作流中，需要确保工具调用的原子性。Harness使用一个简单但有效的机制：

```python
class AtomicToolExecution:
    """原子性工具执行"""

    def __init__(self, execution_id: str):
        self.execution_id = execution_id
        self.state_file = f"/tmp/harness/{execution_id}.state"

    async def execute(self, server: MCPServer, request: dict) -> dict:
        # 1. 检查是否已经执行过(幂等性)
        if os.path.exists(self.state_file):
            with open(self.state_file) as f:
                cached_result = json.load(f)
            return cached_result  # 重复执行,直接返回缓存

        # 2. 执行工具调用
        result = await server.call(request)

        # 3. 原子地写入状态(保证不会丢失)
        os.makedirs(os.path.dirname(self.state_file), exist_ok=True)
        with open(self.state_file + ".tmp", "w") as f:
            json.dump(result, f)
        os.rename(self.state_file + ".tmp", self.state_file)  # 原子rename

        return result
```

## 9.1.5 权限与安全隔离

Harness中的MCP Server运行在受限的沙箱中，权限由Harness策略引擎管理。

```yaml
# harness.yaml
permissions:
  filesystem_server:
    tools:
      read_file:
        allowed_paths: ["/workspace/**", "/tmp/**"]
        denied_paths: ["/etc/**", "/home/**"]
      write_file:
        allowed_paths: ["/workspace/**"]
        denied_paths: ["/**"]

  postgres_server:
    tools:
      query:
        allowed_tables: ["users", "orders"]
        denied_tables: ["audit_log", "secrets"]
      write:
        allowed_operations: ["INSERT", "UPDATE"]
        denied_operations: ["DROP", "DELETE"]
```

Harness在工具调用前，先验证权限：

```python
async def call_tool(server_name: str, tool_name: str, params: dict) -> dict:
    # 1. 检查权限策略
    if not harness.policy_engine.is_allowed(server_name, tool_name, params):
        raise PermissionDenied(f"Tool call denied by policy: {server_name}/{tool_name}")

    # 2. 执行工具调用
    return await harness.mcp_servers[server_name].call(tool_name, params)
```

## 9.1.6 本小节小结

Harness通过以下机制有效地集成了MCP：

1. **工具层隔离**：在应用逻辑和MCP Server之间引入抽象层
2. **性能优化**：Schema缓存、流式传输、连接复用
3. **状态机集成**：工具调用作为工作流的一部分，支持重试和原子性
4. **安全隔离**：细粒度的权限控制和沙箱隔离

更多关于MCP协议本身的深度讨论，以及MCP Server的实现指南，请参阅《Claude 指南》。Harness的特色在于如何在 **多工具、多工作流** 的场景中，高效且安全地管理这些MCP连接。


---

# 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.1_protocol_design.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.