MCP深入解析：从理论到实践的数据流转

引言

在前面的文章中，我们深入探讨了MCP的架构和核心组件。理论总是抽象的，今天让我们通过一个真实的场景，来看看MCP中的数据究竟是如何流转的。

我们将以**“查询北京天气”**这个简单但完整的例子，带你走一遍MCP从用户提问到最终回复的全过程，包括stdio和sse两种传输方式的具体实现。

一、场景设定

1.1 我们要做什么？

• 用户提问：“北京今天天气怎么样？”
• LLM角色：Claude或GPT等大模型，负责理解意图、决策调用、整合结果
• MCP Server：一个提供get_weather工具的天气服务

1.2 两种部署方式

我们将演示两种典型的MCP部署方式：

• stdio类型：本地部署，Server作为子进程运行
• sse类型：远程服务，Server作为独立的HTTP服务

二、stdio类型：本地进程通信

2.1 配置与启动

// MCP配置文件 (如 claude_desktop_config.json)
{
  "mcpServers": {
    "weather": {
      "type": "stdio",
      "command": "python",
      "args": ["weather_server.py"]  // 启动一个Python写的天气服务
    }
  }
}

当MCP Host（如Claude Desktop）启动时，它会：

1. 读取这个配置
2. 创建一个MCP Client实例
3. Client执行 python weather_server.py 启动子进程
4. Server进程一直后台运行，等待通过stdin/stdout通信

2.2 Server端实现（简化版）

# weather_server.py - MCP Server端代码
import sys
import json

# Server一直循环读取stdin，处理请求
for line in sys.stdin:
    request = json.loads(line)  # 读取一行JSON请求
    
    # 判断是否是工具调用请求
    if request["method"] == "tools/call" and request["params"]["name"] == "get_weather":
        city = request["params"]["arguments"]["city"]
        
        # 模拟调用真实天气API
        weather_data = f"{city}今天晴天，15-25°C"
        
        # 构造JSON-RPC响应
        response = {
            "jsonrpc": "2.0",
            "id": request["id"],
            "result": {
                "content": [{"type": "text", "text": weather_data}]
            }
        }
        
        # 通过stdout写回响应（必须加换行符作为消息分隔）
        sys.stdout.write(json.dumps(response) + "\n")
        sys.stdout.flush()

2.3 完整数据流转时序（stdio）

[用户] 
  │
  ├─ 提问："北京今天天气怎么样？"
  │
  ▼
[LLM] 
  │
  ├─ 理解意图：用户想查北京天气
  ├─ 决策：需要调用 get_weather 工具
  ├─ 生成工具调用请求（函数名+参数）
  │
  ▼
[MCP Host]
  │
  ├─ 将工具调用请求转发给负责天气的Client
  │
  ▼
[MCP Client (weather)]
  │
  ├─ 构建JSON-RPC请求消息
  ├─ {
  │   "jsonrpc": "2.0",
  │   "id": "req-123",
  │   "method": "tools/call",
  │   "params": {
  │     "name": "get_weather",
  │     "arguments": {"city": "北京"}
  │   }
  │ }
  │
  ├─ 通过stdin写入（实际传输的字符串）：
  ├─ "{"jsonrpc":"2.0","id":"req-123","method":"tools/call","params":{"name":"get_weather","arguments":{"city":"北京"}}}\n"
  │
  ▼
[MCP Server进程 (weather_server.py)]
  │
  ├─ 从stdin接收到请求字符串
  ├─ 解析JSON，提取city="北京"
  ├─ 调用天气API获取数据
  ├─ 得到结果："北京今天晴天，15-25°C"
  ├─ 构建JSON-RPC响应
  ├─ {
  │   "jsonrpc": "2.0",
  │   "id": "req-123",
  │   "result": {
  │     "content": [
  │       {"type": "text", "text": "北京今天晴天，15-25°C"}
  │     ]
  │   }
  │ }
  │
  ├─ 通过stdout写回（添加换行符）：
  ├─ "{"jsonrpc":"2.0","id":"req-123","result":{"content":[{"type":"text","text":"北京今天晴天，15-25°C"}]}}\n"
  │
  ▼
[MCP Client]
  │
  ├─ 从stdout读取一行响应
  ├─ 解析JSON，提取result
  ├─ 将结果返回给Host
  │
  ▼
[MCP Host]
  │
  ├─ 将工具结果注入LLM的上下文
  │
  ▼
[LLM]
  │
  ├─ 整合工具返回的数据
  ├─ 生成自然语言回复："北京今天晴天，15-25°C，适合外出活动。"
  │
  ▼
[用户] ← 看到最终回复

2.4 stdio方式的关键点

• 一次启动，多次调用：Server进程启动后一直运行，每次调用都通过已建立的管道通信
• 消息分隔：每个JSON消息必须以换行符(\n)分隔
• 双向通信：同一个管道既用于发送请求，也用于接收响应

三、sse类型：远程HTTP通信

3.1 配置与启动

{
  "mcpServers": {
    "weather_remote": {
      "type": "sse",
      "url": "https://api.weather.com/mcp/sse"  // 远程天气服务
    }
  }
}

这里假设远程Server已经作为一个Web服务运行在api.weather.com上，Client只需连接即可。

3.2 Server端实现（FastAPI版）

# remote_weather_server.py - 远程MCP Server
from fastapi import FastAPI, Request
from sse_starlette import EventSourceResponse
import asyncio
import json

app = FastAPI()

# 存储活跃的SSE连接，key为client_id
connections = {}

@app.get("/mcp/sse")
async def sse_endpoint(client_id: str):
    """Client首先连接这个SSE端点，建立长连接"""
    
    # 为这个client创建一个消息队列
    queue = asyncio.Queue()
    connections[client_id] = queue
    
    async def event_generator():
        # 第一步：发送endpoint事件，告诉Client往哪里发请求
        yield {
            "event": "endpoint",
            "data": f"/mcp/message?client_id={client_id}"
        }
        
        # 第二步：持续等待，当有响应需要推送时发送
        while True:
            response = await queue.get()
            yield {
                "event": "message",
                "data": json.dumps(response)
            }
    
    return EventSourceResponse(event_generator())

@app.post("/mcp/message")
async def message_endpoint(request: Request, client_id: str):
    """Client通过POST请求将工具调用发送到这里"""
    
    # 解析请求体
    body = await request.json()
    
    # 处理工具调用
    if body["method"] == "tools/call" and body["params"]["name"] == "get_weather":
        city = body["params"]["arguments"]["city"]
        
        # 调用天气API
        weather_data = f"{city}今天晴天，15-25°C"
        
        # 构造响应
        response = {
            "jsonrpc": "2.0",
            "id": body["id"],
            "result": {
                "content": [{"type": "text", "text": weather_data}]
            }
        }
        
        # 通过之前建立的SSE连接将响应推送回去
        if client_id in connections:
            await connections[client_id].put(response)
    
    # 立即返回，表示已收到请求（实际响应通过SSE推送）
    return {"status": "ok"}

3.3 完整数据流转时序（sse）

[用户] 
  │
  ├─ 提问："北京今天天气怎么样？"
  │
  ▼
[LLM] 
  │
  ├─ 理解意图：用户想查北京天气
  ├─ 决策：需要调用 get_weather 工具
  ├─ 生成工具调用请求
  │
  ▼
[MCP Host]
  │
  ├─ 转发给负责远程天气的Client
  │
  ▼
[MCP Client (weather_remote)]
  │
  ├─ [首次调用] 检查是否已有SSE连接
  ├─ 没有连接，先建立SSE连接
  ├─ GET https://api.weather.com/mcp/sse?client_id=abc123
  │
  ▼
[远程Server]
  │
  ├─ 收到GET请求，建立SSE长连接
  ├─ 立即推送endpoint事件：
  ├─ {"event":"endpoint","data":"/mcp/message?client_id=abc123"}
  │
  ▼
[MCP Client]
  │
  ├─ 收到endpoint事件，知道后续POST请求的地址
  │
  ├─ 构建JSON-RPC请求消息（与stdio完全相同）
  ├─ {
  │   "jsonrpc": "2.0",
  │   "id": "req-456",
  │   "method": "tools/call",
  │   "params": {
  │     "name": "get_weather",
  │     "arguments": {"city": "北京"}
  │   }
  │ }
  │
  ├─ 发送HTTP POST请求：
  ├─ POST https://api.weather.com/mcp/message?client_id=abc123
  ├─ Content-Type: application/json
  ├─ 请求体: {"jsonrpc":"2.0","id":"req-456","method":"tools/call","params":{"name":"get_weather","arguments":{"city":"北京"}}}
  │
  ▼
[远程Server]
  │
  ├─ 收到POST请求
  ├─ 立即返回 {"status": "ok"} （表示已接收）
  ├─ 并行处理：解析请求，提取city="北京"
  ├─ 调用天气API获取数据
  ├─ 得到结果："北京今天晴天，15-25°C"
  ├─ 构建JSON-RPC响应（与stdio完全相同）
  ├─ {
  │   "jsonrpc": "2.0",
  │   "id": "req-456",
  │   "result": {
  │     "content": [
  │       {"type": "text", "text": "北京今天晴天，15-25°C"}
  │     ]
  │   }
  │ }
  │
  ├─ 通过SSE连接推送message事件：
  ├─ {"event":"message","data":"{\"jsonrpc\":\"2.0\",\"id\":\"req-456\",\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"北京今天晴天，15-25°C\"}]}}"}
  │
  ▼
[MCP Client]
  │
  ├─ 通过SSE连接实时接收到message事件
  ├─ 解析data中的JSON，提取result
  ├─ 将结果返回给Host
  │
  ▼
[MCP Host]
  │
  ├─ 将工具结果注入LLM上下文
  │
  ▼
[LLM]
  │
  ├─ 整合结果，生成回复："北京今天晴天，15-25°C，适合外出活动。"
  │
  ▼
[用户] ← 看到最终回复

四、两种方式的直观对比

阶段	stdio方式	sse方式
连接建立	python weather_server.py启动子进程	GET /mcp/sse建立长连接
请求发送	写入stdin: {...}\n	POST /mcp/message发送JSON
响应接收	从stdout读取一行	从SSE连接接收message事件
消息格式	完全相同的JSON-RPC 2.0	完全相同的JSON-RPC 2.0
进程状态	Server作为子进程运行	Server是独立Web服务
适用场景	本地工具、高频调用	远程服务、多客户端共享

五、关键洞察

5.1 统一的JSON-RPC核心

无论使用哪种传输方式，Client和Server之间交换的核心数据完全相同：

// 请求永远是这个形状
{
  "jsonrpc": "2.0",
  "id": "...",
  "method": "tools/call",
  "params": {"name": "...", "arguments": {...}}
}

// 响应永远是这个形状
{
  "jsonrpc": "2.0",
  "id": "...",
  "result": {"content": [...]}
}

5.2 传输层差异被完美封装

• stdio：通过管道传输，换行符分隔消息
• sse：通过HTTP+SSE传输，事件机制封装
• Client内部：根据配置选择不同的传输实现，但对上层Host完全透明

5.3 LLM的角色纯粹

LLM在整个过程中：

• 不知道：底层用的是stdio还是sse
• 不关心：Server是本地进程还是远程服务
• 只负责：理解意图、决策调用、整合结果

六、总结

通过"查询北京天气"这个真实场景，我们可以看到：

1. MCP的核心是消息格式的标准化，而不是传输方式的统一
2. stdio适合本地集成，通过进程管道实现高效通信
3. sse适合远程服务，通过HTTP+SSE实现双向通信
4. LLM始终位于决策中心，对底层传输方式完全无感

无论使用哪种传输方式，MCP都确保了：

• Client的实现可以复用（核心逻辑相同）
• 工具的调用方式统一（JSON-RPC格式）
• Host的管理简单（只需配置不同Server）

这正是MCP作为"AI工具的USB-C接口"的精妙之处——统一的协议标准，灵活的传输方式，让AI应用可以无缝连接任何工具。