> 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-san-bu-fen-xi-tong-ji-cheng-yu-gong-cheng-shi-jian/09_mcp/9.1_protocol_design.md).

# 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. 调用 `server/discover` 一次性拿到 Server 信息与能力（老版本 Server 不支持该方法时回退到直接探测），再按需调用 `tools/list`、`resources/list`、`prompts/list` 拉取原语清单
3. 缓存 Schema 以减少运行时开销
4. 监控 Server 的健康状态

### 无状态协商：每个请求自带元数据

当前版本的 MCP 是无状态协议：`initialize` 请求与 `notifications/initialized` 通知已经移除，不存在握手阶段。客户端把协议版本和能力声明放进 **每一个** 请求的 `params._meta`：

```python
CLIENT_META = {
    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
    "io.modelcontextprotocol/clientCapabilities": {
        "elicitation": {}
    },
    "io.modelcontextprotocol/clientInfo": {
        "name": "HarnessMCPClient",
        "version": "1.0.0"
    }
}

def build_request(method: str, params: dict, request_id: int) -> dict:
    """所有 MCP 请求统一在这里构造，确保 _meta 不会漏发"""
    return {
        "jsonrpc": "2.0",
        "id": request_id,
        "method": method,
        "params": {**params, "_meta": dict(CLIENT_META)}
    }
```

`_meta` 中各字段的约定：

| 字段                                           | 类型                               | 是否必需    |
| -------------------------------------------- | -------------------------------- | ------- |
| `io.modelcontextprotocol/protocolVersion`    | 字符串，如 `"2026-07-28"`             | 必需      |
| `io.modelcontextprotocol/clientCapabilities` | ClientCapabilities               | 必需      |
| `io.modelcontextprotocol/clientInfo`         | Implementation（`name`、`version`） | 可选，建议发送 |
| `io.modelcontextprotocol/logLevel`           | LoggingLevel                     | 可选      |

对 Harness 实现者而言，有四条硬性约束：

* **漏发即报错**：缺少必需字段的请求会得到 JSON-RPC `-32602`（HTTP 状态码 400）。因此能力声明必须收敛到统一的请求构造函数，而不是散落在各个调用点。
* **能力不足有专门错误**：Server 需要某项客户端能力而客户端没有声明时，返回 `MissingRequiredClientCapabilityError`（`-32021`，HTTP 400），`data.requiredCapabilities` 给出缺失的能力列表，Harness 可据此打印可诊断的配置提示。
* **Server 不得推断状态**：Server 不能依据同一连接上的历史请求推断状态；任何跨请求的状态都必须是显式标识符，由客户端每次带上。一个长期存活的 stdio 进程 **不是** 会话。
* **Server 信息随结果返回**：Server 应在每个结果的 `_meta` 中放入 `io.modelcontextprotocol/serverInfo`，Harness 可以顺带刷新注册表中的 Server 元数据。

此外，每个结果都必须携带 `resultType`，取值为 `"complete"` 或 `"input_required"`；客户端遇到 **缺失** `resultType` 的结果（来自旧版 Server）时，必须按 `"complete"` 处理。

### 用 `server/discover` 替代探测式发现

`server/discover` 是 Server 必须实现、客户端可选调用的发现入口。请求只带 `_meta`，一次调用即可拿到过去要靠 `tools/list` + `prompts/list` + `resources/list` 反复探测才能拼出的 Server 画像：

```python
async def discover(server: MCPServer) -> dict:
    request = build_request("server/discover", {}, generate_request_id())
    result = (await server.call(request))["result"]

    return {
        "versions": result["supportedVersions"],     # 支持的协议版本列表
        "capabilities": result["capabilities"],      # Server能力
        "instructions": result.get("instructions"),  # 可选的使用说明
        "ttl_ms": result["ttlMs"],                   # 本结果的新鲜度提示（毫秒）
        "cache_scope": result["cacheScope"],         # "public" 或 "private"
        "info": result["_meta"]["io.modelcontextprotocol/serverInfo"],
    }
```

它对 Harness 有两点直接价值：注册阶段用一次 RPC 取代三四次列表探测；在 stdio 场景下，它还是判断对端讲新版还是旧版协议的探针。

### 向后兼容：2025-11-25 的握手模型

> ⚠️ **本小节只适用于对接尚未升级的旧版 Server**。新写的客户端与 Server 之间不需要握手。

大量已部署的 Server 仍在讲 2025-11-25：连接后先发 `initialize` 请求协商版本与能力，收到响应后再发 `notifications/initialized` 通知，之后才进入正常 `tools/list`、`resources/read` 等操作阶段。

```python
# 仅用于回退到 2025-11-25 旧版Server的兼容路径
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"
        }
    }
}
```

Harness 的探测顺序是“先按新版发，再按错误回退”：

1. 直接发一个带 `_meta` 的新版请求（HTTP 上还要带 `MCP-Protocol-Version` 头）。
2. 收到 HTTP 400 时检查响应体：如果是可识别的新版 JSON-RPC 错误（例如 `-32020` 头部与 `_meta` 版本不一致、`-32022` 版本不受支持并回带支持列表），说明对端是新版 Server，改正请求后重试；只有响应体为空或无法识别时，才回退到 `initialize` 握手。
3. stdio 场景下用 `server/discover` 做同样的探测。

反过来，一个只实现 2026-07-28 的 Server 在面对旧流量时，应对该端点上的 GET/DELETE 返回 `405 Method Not Allowed`，并忽略 `Mcp-Session-Id` 与 `Last-Event-ID`。

还有几项特性被标记为 **弃用但未移除**，期间它们继续可用。2026-07-28 随之引入了特性生命周期与弃用策略，规定弃用窗口最短 12 个月：Roots、Sampling、Logging 与 OAuth 2.0 动态客户端注册（改用 Client ID Metadata Documents）都是在本修订版被弃用的，[弃用特性登记表](https://modelcontextprotocol.io/specification/2026-07-28/deprecated)给出的最早移除时间是「2027-07-28 或之后发布的第一个修订版」。**2024-11-05 的 HTTP+SSE 传输是例外**：它自 2025-03-26 起就被标注为弃用，本次只是按新策略重新归类，登记表给出的最早移除时间是「SEP-2596 定稿后三个月」——比其余几项短得多，仍在依赖它的实现应优先迁移。迁移方向是：用工具参数或资源 URI 传入目录与文件，替代 Roots；直接调用模型厂商 API，替代 Sampling；写 stderr 或 OpenTelemetry，替代 Logging。需要注意的是 `logging/setLevel` 与 `notifications/roots/list_changed` 已经移除——日志级别改为在每个请求的 `_meta` 中用 `io.modelcontextprotocol/logLevel` 指定。

## 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"):
        # sqlite3 不会展开 ~，需要 expanduser
        self.db = sqlite3.connect(os.path.expanduser(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` 的标准字段。在 2025-11-25 修订版下，真实实现应根据工具列表哈希、`notifications/tools/list_changed` 通知或服务端自定义元数据来失效缓存；2026-07-28 修订版起，`tools/list`、`prompts/list`、`resources/list` 等列表结果必须携带 `ttlMs`（新鲜度提示，毫秒）与 `cacheScope`（`"public"` 或 `"private"`），面向新版服务端时应优先以这两个字段作为缓存存活时间与共享范围的依据，旧版服务端不返回它们时再回退到哈希与变更通知。

### 大资源读取与流式响应

对于大数据量的 Resource 读取（如导入大文件），Harness 需要同时处理 JSON 响应和 Streamable HTTP 的 SSE 响应；但 `resources/read` 的结果仍是 MCP JSON-RPC 消息，二进制内容应放在 `BlobResourceContents.blob` 的 base64 字符串中，而不是把 HTTP body 当作任意字节流直接透传。

生产实现应把超大文件拆成资源模板、分页、任务进度或外部对象存储引用，并正确解析响应的 `Content-Type`：

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

async def read_resource_from_server(server: MCPServer, uri: str) -> dict:
    """读取资源内容，兼容 application/json 与 text/event-stream 响应"""
    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:
        response.raise_for_status()
        content_type = response.headers.get("content-type", "")
        if content_type.startswith("application/json"):
            return json.loads(await response.aread())

        if content_type.startswith("text/event-stream"):
            async for event in parse_sse_events(response.aiter_lines()):
                if event.get("data"):
                    message = json.loads(event["data"])
                    if message.get("id") == request["id"]:
                        return message

        raise ValueError(f"Unsupported MCP response type: {content_type}")
```

## 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 连接。