传输层

模型上下文协议 (MCP) 支持多种传输层，以适应不同的部署场景和技术要求。

概述

MCP 设计为传输层无关的协议，可以在多种通信机制上运行：

• stdio - 标准输入/输出流
• Server-Sent Events (SSE) - HTTP 流式传输
• WebSocket - 双向实时通信

stdio 传输

概述

stdio 传输使用标准输入和输出流进行通信，适用于本地进程间通信。

特点

• 简单性: 最简单的传输机制
• 本地性: 仅适用于本地进程
• 性能: 低延迟，高吞吐量
• 调试: 易于调试和监控

消息格式

{
  "jsonrpc": "2.0",
  "method": "tools/list",
  "id": 1
}

实现示例

Python 客户端

import subprocess
import json

# 启动服务器进程
process = subprocess.Popen(
    ["python", "server.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 发送请求
request = {
    "jsonrpc": "2.0",
    "method": "initialize",
    "id": 1,
    "params": {
        "protocolVersion": "2025-06-18",
        "capabilities": {}
    }
}

process.stdin.write(json.dumps(request) + "\n")
process.stdin.flush()

# 读取响应
response = process.stdout.readline()
result = json.loads(response)

TypeScript 服务器

import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';

const server = new Server(
  {
    name: "example-server",
    version: "0.1.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

const transport = new StdioServerTransport();
await server.connect(transport);

Server-Sent Events (SSE) 传输

概述

SSE 传输使用 HTTP 流式传输，支持服务器到客户端的实时数据推送。

特点

• HTTP 兼容: 基于标准 HTTP 协议
• 防火墙友好: 使用标准 HTTP 端口
• 单向流: 服务器到客户端的推送
• 自动重连: 浏览器自动重连机制

端点设计

GET /sse - 建立 SSE 连接
POST /message - 发送消息到服务器

实现示例

Python 服务器 (FastAPI)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
import asyncio

app = FastAPI()

@app.get("/sse")
async def sse_endpoint():
    async def event_stream():
        while True:
            # 发送心跳
            yield f"data: {json.dumps({'type': 'ping'})}\n\n"
            await asyncio.sleep(30)
    
    return StreamingResponse(
        event_stream(),
        media_type="text/plain",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

@app.post("/message")
async def handle_message(message: dict):
    # 处理客户端消息
    return {"status": "received"}

JavaScript 客户端

const eventSource = new EventSource('/sse');

eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log('收到消息:', data);
};

// 发送消息到服务器
async function sendMessage(message) {
    const response = await fetch('/message', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
        },
        body: JSON.stringify(message)
    });
    return response.json();
}

WebSocket 传输

概述

WebSocket 提供全双工通信，支持客户端和服务器之间的实时双向消息交换。

特点

• 双向通信: 客户端和服务器都可以主动发送消息
• 低延迟: 无需 HTTP 请求/响应开销
• 实时性: 适合实时应用场景
• 连接状态: 持久连接，需要处理断线重连

实现示例

Node.js 服务器

const WebSocket = require('ws');
const { Server } = require('@modelcontextprotocol/sdk/server/index.js');

const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', async (ws) => {
    const server = new Server({
        name: "websocket-server",
        version: "0.1.0",
    }, {
        capabilities: { tools: {} },
    });

    // 设置 WebSocket 传输
    const transport = {
        async send(message) {
            ws.send(JSON.stringify(message));
        },
        onMessage: null,
        onClose: null,
    };

    ws.on('message', (data) => {
        const message = JSON.parse(data.toString());
        if (transport.onMessage) {
            transport.onMessage(message);
        }
    });

    ws.on('close', () => {
        if (transport.onClose) {
            transport.onClose();
        }
    });

    await server.connect(transport);
});

Python 客户端

import asyncio
import websockets
import json

async def websocket_client():
    uri = "ws://localhost:8080"
    
    async with websockets.connect(uri) as websocket:
        # 发送初始化请求
        init_request = {
            "jsonrpc": "2.0",
            "method": "initialize",
            "id": 1,
            "params": {
                "protocolVersion": "2025-06-18",
                "capabilities": {}
            }
        }
        
        await websocket.send(json.dumps(init_request))
        
        # 监听响应
        async for message in websocket:
            data = json.loads(message)
            print(f"收到消息: {data}")

asyncio.run(websocket_client())

传输层选择指南

stdio

适用场景:

• 本地开发和测试
• 命令行工具集成
• 简单的进程间通信

优点:

• 实现简单
• 性能优秀
• 调试容易

缺点:

• 仅限本地使用
• 不支持网络部署

SSE

适用场景:

• Web 应用集成
• 需要服务器推送的场景
• 防火墙限制环境

优点:

• HTTP 兼容
• 浏览器原生支持
• 自动重连

缺点:

• 单向通信
• 需要额外的 POST 端点

WebSocket

适用场景:

• 实时应用
• 需要双向通信
• 高频消息交换

优点:

• 双向实时通信
• 低延迟
• 高效率

缺点:

• 实现复杂
• 需要处理连接管理

安全考虑

认证和授权

// WebSocket 认证示例
const ws = new WebSocket('ws://localhost:8080', {
    headers: {
        'Authorization': 'Bearer ' + token
    }
});

传输加密

• stdio: 依赖操作系统安全
• SSE: 使用 HTTPS
• WebSocket: 使用 WSS (WebSocket Secure)

输入验证

所有传输层都应该验证输入消息格式和内容。

最佳实践

1. 错误处理: 实现健壮的错误处理机制
2. 重连策略: 对于网络传输，实现指数退避重连
3. 消息队列: 处理连接中断时的消息缓存
4. 监控: 实现连接状态和性能监控
5. 超时处理: 设置合适的超时时间

相关资源

• 基础协议
• 生命周期管理
• 安全最佳实践
• 架构概述