第 19 章

MCP 集成架构

第19章：MCP 集成架构

Model Context Protocol（MCP）是 Anthropic 于 2024 年底提出的开放标准，旨在为 AI 模型与外部工具/数据源之间建立统一的通信接口。Hermes Agent 作为 MCP 客户端，能够无缝接入任意 MCP Server，将其能力纳入自身工具体系。本章深入解析 MCP 的核心概念、Hermes 的集成架构，以及实战接入 Playwright MCP Server 的完整流程。

19.1 MCP 核心概念

MCP（Model Context Protocol）的设计哲学是"工具即服务"：将任何外部能力包装为标准化的 MCP Server，AI 客户端通过统一协议发现和调用这些能力。

19.1.1 MCP 架构三要素

┌─────────────────────────────────────────────────────┐
│                    MCP 生态系统                       │
│                                                     │
│  ┌──────────────┐     MCP Protocol      ┌─────────┐ │
│  │  MCP Client  │ ◄──────────────────► │MCP Server│ │
│  │ (Hermes等)   │    JSON-RPC 2.0       │(工具提供者)│ │
│  └──────────────┘                      └─────────┘ │
│         │                                    │      │
│    发现工具                              提供工具      │
│    调用工具                              执行逻辑      │
│    接收结果                              返回结果      │
└─────────────────────────────────────────────────────┘

三个核心组件：

组件	角色	示例
MCP Client	发起工具发现和调用请求的一方	Hermes Agent, Claude Desktop
MCP Server	提供工具能力的服务端	Playwright MCP, GitHub MCP
MCP Protocol	基于 JSON-RPC 2.0 的通信规范	tools/list, tools/call 等方法

19.1.2 MCP 协议核心方法

// 工具发现：客户端请求服务端暴露的工具列表
// Request: tools/list
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

// Response
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "playwright_navigate",
        "description": "Navigate to a URL in the browser",
        "inputSchema": {
          "type": "object",
          "properties": {
            "url": {"type": "string"},
            "wait_until": {
              "type": "string",
              "enum": ["load", "domcontentloaded", "networkidle"],
              "default": "load"
            }
          },
          "required": ["url"]
        }
      },
      {
        "name": "playwright_screenshot",
        "description": "Take a screenshot of the current page",
        "inputSchema": {
          "type": "object",
          "properties": {
            "full_page": {"type": "boolean", "default": false},
            "element": {"type": "string", "description": "CSS selector"}
          }
        }
      }
    ]
  }
}

// 工具调用：客户端调用指定工具
// Request: tools/call
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "playwright_navigate",
    "arguments": {
      "url": "https://nousresearch.com",
      "wait_until": "networkidle"
    }
  }
}

// Response
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "Successfully navigated to https://nousresearch.com. Page title: NousResearch"
      }
    ],
    "isError": false
  }
}

19.1.3 MCP 传输层

MCP 支持三种传输方式：

传输方式	适用场景	特点
stdio	本地进程通信	最低延迟，无需网络
SSE (Server-Sent Events)	远程 HTTP 服务	单向流式推送，适合云端
WebSocket	实时双向通信	适合高频交互场景

19.2 Hermes 作为 MCP 客户端的工作原理

Hermes 内置了完整的 MCP 客户端实现，在启动时自动连接配置的 MCP Server，并将其工具纳入统一工具注册表。

19.2.1 MCP 客户端初始化流程

# Hermes MCP 客户端核心实现（简化版）
from hermes.mcp import MCPClient, MCPTransport
from hermes.tools import ToolRegistry

class HermesMCPManager:
    def __init__(self, config: HermesConfig):
        self.config = config
        self.clients: dict[str, MCPClient] = {}
        self.tool_registry = ToolRegistry()
    
    async def initialize(self):
        """启动时连接所有配置的 MCP Server"""
        for server_config in self.config.mcp_servers:
            client = await self._connect_server(server_config)
            self.clients[server_config.name] = client
            
            # 发现并注册工具
            tools = await client.list_tools()
            for tool in tools:
                self.tool_registry.register_mcp_tool(
                    tool=tool,
                    server_name=server_config.name,
                    client=client,
                )
            
            print(f"[MCP] 已连接 {server_config.name}，"
                  f"注册 {len(tools)} 个工具")
    
    async def _connect_server(self, config: MCPServerConfig) -> MCPClient:
        transport = MCPTransport.create(
            transport_type=config.transport,  # stdio / sse / websocket
            command=config.command,           # stdio: 启动命令
            url=config.url,                   # sse/ws: 服务地址
            env=config.env,                   # 环境变量
        )
        
        client = MCPClient(transport=transport)
        await client.connect()
        
        # 初始化握手
        await client.initialize(
            client_info={
                "name": "hermes-agent",
                "version": "4.0.0",
            }
        )
        
        return client

19.2.2 MCP 工具的透明代理

当 Agent 调用一个 MCP 工具时，Hermes 内部自动完成协议转换：

class MCPToolProxy:
    """将 MCP 工具包装为 Hermes 原生工具格式"""
    
    def __init__(self, mcp_tool: MCPToolDefinition, client: MCPClient):
        self.mcp_tool = mcp_tool
        self.client = client
    
    @property
    def name(self) -> str:
        # MCP 工具名加上服务器前缀，避免命名冲突
        return f"mcp_{self.client.server_name}_{self.mcp_tool.name}"
    
    @property
    def description(self) -> str:
        return f"[MCP:{self.client.server_name}] {self.mcp_tool.description}"
    
    async def run(self, params: dict, context) -> ToolResult:
        try:
            # 调用 MCP Server
            mcp_result = await self.client.call_tool(
                name=self.mcp_tool.name,
                arguments=params,
            )
            
            # 转换 MCP 响应格式为 Hermes ToolResult
            return ToolResult(
                success=not mcp_result.isError,
                output=self._extract_content(mcp_result.content),
                metadata={
                    "mcp_server": self.client.server_name,
                    "mcp_tool": self.mcp_tool.name,
                }
            )
        except MCPConnectionError as e:
            # 尝试重连
            await self.client.reconnect()
            raise ToolExecutionError(f"MCP 连接中断: {e}")
    
    def _extract_content(self, content: list) -> str | dict:
        """从 MCP 内容块中提取可用数据"""
        texts = [c["text"] for c in content if c["type"] == "text"]
        images = [c for c in content if c["type"] == "image"]
        
        if images and not texts:
            return {"type": "image", "data": images[0]["data"]}
        return "\n".join(texts)

19.3 连接外部 MCP Server 的配置方法

19.3.1 hermes_config.yaml 中的 MCP 配置

# hermes_config.yaml
mcp:
  enabled: true
  connection_timeout_seconds: 30
  tool_prefix: "mcp"      # MCP 工具名称前缀
  
  servers:
    # 方式一：stdio 进程（最常用）
    - name: playwright
      transport: stdio
      command: ["npx", "@playwright/mcp@latest"]
      env:
        PLAYWRIGHT_BROWSERS_PATH: "/usr/local/share/playwright"
      auto_restart: true
      health_check_interval_seconds: 60
    
    # 方式二：本地 Python MCP Server
    - name: file_system
      transport: stdio
      command: ["python", "-m", "mcp_server_filesystem", "--root", "/workspace"]
      env: {}
    
    # 方式三：远程 SSE 服务
    - name: company_internal_api
      transport: sse
      url: "https://mcp.internal.company.com/v1"
      headers:
        Authorization: "Bearer ${INTERNAL_API_TOKEN}"
      tls:
        verify: true
        cert_path: "/etc/ssl/internal.crt"
    
    # 方式四：WebSocket 服务
    - name: realtime_data
      transport: websocket
      url: "wss://data.example.com/mcp"
      reconnect_on_failure: true
      max_reconnect_attempts: 5

19.3.2 通过 CLI 动态添加 MCP Server

# 添加 Playwright MCP Server
hermes mcp add playwright \
  --transport stdio \
  --command "npx @playwright/mcp@latest"

# 添加远程 MCP Server
hermes mcp add company-api \
  --transport sse \
  --url "https://mcp.company.com/v1" \
  --header "Authorization: Bearer $TOKEN"

# 查看已连接的 MCP Server 和工具
hermes mcp list

# 测试 MCP Server 连接
hermes mcp test playwright --tool playwright_navigate \
  --args '{"url": "https://example.com"}'

# 断开 MCP Server
hermes mcp remove playwright

19.3.3 以编程方式管理 MCP 连接

import asyncio
from hermes import HermesAgent
from hermes.mcp import MCPServerConfig, MCPTransportType

async def main():
    agent = HermesAgent()
    
    # 动态添加 Playwright MCP Server
    playwright_config = MCPServerConfig(
        name="playwright",
        transport=MCPTransportType.STDIO,
        command=["npx", "@playwright/mcp@latest"],
        env={"DISPLAY": ":0"},
    )
    
    await agent.mcp_manager.add_server(playwright_config)
    
    # 验证工具已注册
    tools = agent.tool_registry.list_tools(prefix="mcp_playwright")
    print(f"Playwright MCP 提供了 {len(tools)} 个工具:")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}")
    
    # 正常使用 Agent（Playwright 工具对 Agent 透明）
    result = await agent.run(
        "请访问 https://nousresearch.com 并截图，告诉我页面的主要内容"
    )
    print(result)

asyncio.run(main())

19.4 MCP 工具与原生工具的优先级

当 Hermes 的原生工具与 MCP 工具存在功能重叠时，需要明确的优先级规则。

19.4.1 优先级规则

优先级（从高到低）:

1. 显式指定（Agent 在 Prompt 中明确指定工具名）
2. 用户配置偏好（hermes_config.yaml 中的 tool_preference）
3. 原生工具（Hermes 内置）
4. MCP 工具（按注册顺序，先注册的优先）

19.4.2 配置工具优先级

# hermes_config.yaml
tool_resolution:
  # 对于浏览器操作，优先使用 Playwright MCP
  overrides:
    browser_navigate: "mcp_playwright_playwright_navigate"
    browser_screenshot: "mcp_playwright_playwright_screenshot"
    browser_click: "mcp_playwright_playwright_click"
  
  # 禁用原生工具（强制使用 MCP 版本）
  disabled_native_tools:
    - browser_navigate
    - browser_screenshot

19.4.3 冲突解析示例

# 工具名解析逻辑
class ToolResolver:
    def resolve(self, tool_name: str, context: AgentContext) -> Tool:
        # 1. 检查是否显式指定（带 mcp_ 前缀）
        if tool_name.startswith("mcp_"):
            return self.registry.get_mcp_tool(tool_name)
        
        # 2. 检查用户配置的 overrides
        if tool_name in self.config.tool_overrides:
            return self.registry.get(self.config.tool_overrides[tool_name])
        
        # 3. 查找原生工具
        native = self.registry.get_native(tool_name)
        if native and tool_name not in self.config.disabled_native:
            return native
        
        # 4. 查找 MCP 工具（模糊匹配）
        mcp_tools = self.registry.search_mcp(query=tool_name)
        if mcp_tools:
            return mcp_tools[0]  # 返回最相关的
        
        raise ToolNotFoundError(f"找不到工具: {tool_name}")

19.5 实战：接入 Playwright MCP Server

以下是完整的 Playwright MCP 集成实战，实现"自动化网页内容提取"功能。

19.5.1 安装配置

# 安装 Playwright MCP 包
npm install -g @playwright/mcp

# 安装 Playwright 浏览器
npx playwright install chromium

# 验证安装
npx @playwright/mcp@latest --version

19.5.2 Hermes 配置

# hermes_config.yaml
mcp:
  servers:
    - name: playwright
      transport: stdio
      command: ["npx", "@playwright/mcp@latest", "--browser", "chromium"]
      env:
        PLAYWRIGHT_HEADLESS: "true"

19.5.3 完整使用示例

import asyncio
from hermes import HermesAgent

async def scrape_product_info():
    agent = HermesAgent(config_path="hermes_config.yaml")
    await agent.initialize()
    
    # Agent 自动使用 Playwright MCP 工具
    result = await agent.run("""
    请完成以下任务：
    1. 访问 https://www.amazon.com/dp/B0C9RD2FN9
    2. 截取商品主图
    3. 提取：商品名称、价格、评分、评论数量
    4. 以 JSON 格式返回提取的信息
    """)
    
    print(result)

# 输出示例：
# {
#   "product_name": "Echo Dot (5th Gen) Smart Speaker",
#   "price": "$49.99",
#   "rating": 4.7,
#   "review_count": 187432,
#   "screenshot": "base64_encoded_image..."
# }

asyncio.run(scrape_product_info())

19.5.4 Playwright MCP 提供的核心工具列表

mcp_playwright_playwright_navigate    — 导航到 URL
mcp_playwright_playwright_screenshot  — 截取页面截图
mcp_playwright_playwright_click       — 点击元素
mcp_playwright_playwright_fill        — 填写输入框
mcp_playwright_playwright_select      — 选择下拉选项
mcp_playwright_playwright_check       — 勾选复选框
mcp_playwright_playwright_evaluate    — 执行 JavaScript
mcp_playwright_playwright_wait        — 等待元素或条件
mcp_playwright_playwright_get_text    — 获取元素文本
mcp_playwright_playwright_get_html    — 获取页面 HTML
mcp_playwright_playwright_pdf         — 导出页面为 PDF
mcp_playwright_playwright_new_tab     — 打开新标签页

19.5.5 高级用法：多页面并行抓取

async def parallel_scrape(urls: list[str]) -> list[dict]:
    agent = HermesAgent(config_path="hermes_config.yaml")
    await agent.initialize()
    
    # 并行抓取多个页面
    tasks = []
    for url in urls:
        task = agent.run(f"""
        访问 {url}，提取页面标题、主要内容摘要（200字以内）、
        所有外部链接数量。以 JSON 返回。
        """)
        tasks.append(task)
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return [
        r if not isinstance(r, Exception) else {"error": str(r)}
        for r in results
    ]

19.6 MCP 安全最佳实践

# 生产环境 MCP 安全配置
mcp:
  security:
    # 工具白名单：只允许特定 MCP 工具被调用
    tool_whitelist:
      playwright:
        - playwright_navigate
        - playwright_screenshot
        - playwright_get_text
      # 禁止 playwright_evaluate（防止任意 JS 执行）
    
    # URL 过滤：防止 SSRF 攻击
    url_filters:
      allowed_domains:
        - "*.amazon.com"
        - "*.google.com"
      blocked_domains:
        - "169.254.169.254"  # AWS metadata endpoint
        - "localhost"
        - "*.internal"
    
    # 资源限制
    resource_limits:
      max_page_load_time_seconds: 30
      max_screenshot_size_mb: 5
      max_concurrent_pages: 3

19.7 小结

本章系统讲解了 Hermes Agent 的 MCP 集成架构：

MCP 核心概念：三要素（Client/Server/Protocol）、JSON-RPC 2.0 协议、三种传输层
Hermes MCP 客户端：初始化流程、MCPToolProxy 透明代理机制、协议格式转换
配置方法：YAML 声明式配置、CLI 动态管理、Python API 编程接入
优先级规则：四层解析链，支持工具 Override 和禁用原生工具
实战 Playwright：完整安装→配置→使用→安全配置流程

MCP 将 Hermes 从封闭的工具体系升级为开放的能力平台——任何第三方都可以通过发布 MCP Server 来扩展 Hermes 的能力边界。

思考题

MCP 使用 JSON-RPC 2.0 而非 REST API 的设计原因是什么？JSON-RPC 的双向通知机制（notifications）在 Agent 场景下有什么特别价值？
当一个 MCP Server 崩溃重启后，已经注册的工具在 Hermes 工具注册表中的状态如何处理？如何设计"工具健康检查"机制？
如果你要将公司内部的私有数据库系统包装为 MCP Server，应该如何设计认证机制和数据访问审计日志？

本章评分

4.6 / 5 (15 评分)