> 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.5_miniharness_mcp.md).

# 9.5 实战：为 MiniHarness 集成 MCP

本节把 MCP 工具接入 MiniHarness 的统一执行管线。实现使用[官方 MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk)，支持 stdio 与 Streamable HTTP，并明确处理初始化、认证、超时取消和关闭。

## 9.5.1 模块边界

实现不再把传输、认证、客户端、注册表和测试替身放在一个文件中：

| 文件                                    | 职责                                               |
| ------------------------------------- | ------------------------------------------------ |
| `lab/mini_harness/mcp/auth.py`        | Bearer Token 请求头；对象表示不会输出 Token                  |
| `lab/mini_harness/mcp/transports.py`  | 对官方 `stdio_client`、`streamable_http_client` 的薄封装 |
| `lab/mini_harness/mcp/client.py`      | `ClientSession` 生命周期、协商结果、请求超时和结果归一化             |
| `lab/mini_harness/mcp/integration.py` | Server 配置、工具发现、Schema 缓存、注册表和 LLM 适配器            |
| `lab/mini_harness/application.py`     | 本地工具和 MCP 工具共用的权限、护栏、重试与检查点入口                    |
| `lab/tests/fakes/`                    | 单元测试替身和真实协议测试 Server；不进入生产包                      |

生产路径不会创建 Mock Client。注册表默认工厂只创建基于官方 SDK 的 `MCPClient`；单元测试通过 `client_factory` 显式注入 Fake。

## 9.5.2 传输与配置

stdio 配置把可执行程序与参数分开，避免把整条命令交给 Shell：

```python
import sys

from mini_harness.mcp.integration import MCPServerConfig

stdio_server = MCPServerConfig(
    server_id="local-tools",
    server_name="Local tools",
    transport_type="stdio",
    endpoint=sys.executable,
    args=("servers/local_tools.py",),
    timeout_seconds=10,
)
```

Streamable HTTP 可以带固定请求头和 Bearer Token。真实凭据应来自受控环境变量或 Secret Store，不写入配置文件：

```python
import os

from mini_harness.mcp.auth import BearerTokenAuth
from mini_harness.mcp.integration import MCPServerConfig

http_server = MCPServerConfig(
    server_id="remote-tools",
    server_name="Remote tools",
    transport_type="streamable_http",
    endpoint="https://mcp.example.com/mcp",
    auth=BearerTokenAuth(os.environ["MCP_BEARER_TOKEN"]),
    timeout_seconds=15,
)
```

`http` 仍是 `streamable_http` 的兼容别名。当前适配层提供 Bearer Token 与自定义请求头；需要 OAuth 动态注册或 Token 刷新时，应接入官方 SDK 的 OAuth Provider，而不是把刷新逻辑塞进注册表。

## 9.5.3 客户端生命周期

MCP 普通请求必须发生在初始化完成之后。`MCPClient` 保留 Server 协商得到的协议版本、能力和实现信息，不在客户端硬编码“协商成功”的值：

```python
from mini_harness.mcp.client import MCPClient, MCPNotInitializedError
from mini_harness.mcp.transports import StdioTransport

client = MCPClient(
    StdioTransport(command=sys.executable, args=("servers/local_tools.py",)),
    timeout_seconds=10,
)

try:
    await client.list_tools()  # 抛出 MCPNotInitializedError
except MCPNotInitializedError:
    pass

metadata = await client.initialize()
print(metadata.protocol_version)
print(metadata.capabilities)

tools = await client.list_tools()
result = await client.call_tool("echo", {"text": "hello"})
await client.close()
```

调用顺序由官方 `ClientSession.initialize()` 完成；SDK 负责发送 `initialize` 和初始化通知。MiniHarness 在每个初始化或工具请求外增加截止时间，超时时取消等待并抛出 `MCPRequestTimeout`。关闭是幂等的，关闭后再请求会抛出 `MCPClientClosedError`。

## 9.5.4 工具发现与 Schema 缓存

`MCPToolRegistry` 串行保护一次发现过程，并用新的发现结果整体替换旧映射，避免已经下线的工具残留：

```python
from mini_harness.mcp.integration import MCPToolRegistry, ToolSchemaCache

registry = MCPToolRegistry(
    schema_cache=ToolSchemaCache(cache_dir=".cache/mcp", ttl_seconds=3600)
)
await registry.add_server(stdio_server)
await registry.add_server(http_server)

tool_to_servers = await registry.discover_tools(force=True)
success, result, error = await registry.call_tool(
    "echo", {"text": "hello"}, agent_id="agent-1"
)
await registry.close()
```

Schema 使用内存与 JSON 文件两层缓存，并带 TTL。缓存只减少重复的 `tools/list` 与 Schema 转换；它不缓存工具执行结果，也不能替代 Server 可用性检查。多个 Server 提供同名工具时，当前实现按 Server 优先级选择第一个，没有实现自动故障转移。

## 9.5.5 统一安全执行入口

MCP 工具不能绕过本地工具的安全策略。`HarnessApplication.execute_tool()` 先调用 `SecureToolExecutor`，只有权限、命令护栏、路径校验和必要批准全部通过，才进入 MCP 注册表：

```python
from mini_harness.application import HarnessApplication
from mini_harness.security.permissions import PermissionDecisionEngine, PermissionLevel
from mini_harness.security.secure_executor import SecureToolExecutor

permissions = PermissionDecisionEngine()
permissions.register_policy("remote_write", PermissionLevel.ASK)

application = HarnessApplication(
    mcp_registry=registry,
    secure_executor=SecureToolExecutor(
        permission_engine=permissions,
        user_approval_callback=ask_user,
    ),
)

result = await application.execute_tool(
    "remote_write",
    {"value": "approved data"},
    user_id="user-1",
    session_id="session-1",
    call_id="model-tool-call-id",
)
```

管线顺序是：

1. 执行权限和护栏判断；
2. 原子认领相同 `session_id + call_id` 的检查点；
3. 已完成记录直接回放，只有获得随机 Lease 的 Owner 才能写入 `started` 后执行；
4. 仅对配置的可恢复异常重试，并写入同一 Trace 的事件；
5. 成功返回后保存可重放结果，并校验完成者持有对应 Lease。

已完成的调用在恢复时直接复用结果，不重复副作用。并发请求若遇到另一个 Owner 的 `started` 记录，会得到明确的“正在执行或结果未知”错误；Owner 抛出异常时也保留 `started`，因为远端操作可能已经成功。该策略优先避免重复写入；它不能让“远端副作用 + 本地 JSON 检查点”获得跨系统原子性。需要严格一次性语义时，Server 仍须支持幂等键、事务或可查询的操作状态。

## 9.5.6 真实协议测试

`lab/tests/integration/test_mcp_lifecycle.py` 启动两个真实 Server：

* 子进程 stdio Server：验证初始化前拒绝、版本/能力协商、工具发现与调用；
* 本机回环 Streamable HTTP Server：验证 Bearer Token、401、调用与关闭；
* 慢工具：验证请求超时会取消客户端等待，随后仍能正常关闭。

`lab/tests/integration/test_application.py` 进一步验证 DENY 发生在副作用前、重试事件共享 Trace、检查点恢复不重复执行，以及 MCP 与本地工具共用安全执行器。运行方式：

```bash
cd lab
python -m pytest tests/integration/test_mcp_lifecycle.py tests/integration/test_application.py -q
```

## 9.5.7 集成检查清单

* [ ] stdio 的 `command` 与 `args` 分开配置，未使用 Shell 拼接
* [ ] HTTP 凭据来自 Secret Store 或环境变量，日志中不输出 Token
* [ ] 普通请求只在 `initialize()` 成功后发送
* [ ] 保存并观察协商得到的协议版本与 Server 能力
* [ ] 所有本地和 MCP 工具都从 `HarnessApplication` 进入
* [ ] 写操作使用稳定 `call_id`，Server 侧同时提供幂等保证
* [ ] 超时、认证失败、关闭和恢复路径均有测试
* [ ] 应用退出时调用 `registry.close()`

## 9.5.8 小结

MiniHarness 的 MCP 集成将协议细节限制在 SDK adapter 内，将工具发现限制在注册表内，并把安全、重试和恢复收口到应用入口。清晰边界比“自己实现一套 JSON-RPC”更重要：协议演进由官方 SDK 承担，Harness 只负责自身的策略和生命周期。