9.5 实战：为 MiniHarness 集成 MCP

本节把 MCP 工具接入 MiniHarness 的统一执行管线。实现使用官方 MCP 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：

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，不写入配置文件：

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 协商得到的协议版本、能力和实现信息，不在客户端硬编码“协商成功”的值：

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 串行保护一次发现过程，并用新的发现结果整体替换旧映射，避免已经下线的工具残留：

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 注册表：

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 与本地工具共用安全执行器。运行方式：

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 只负责自身的策略和生命周期。