传输
模型上下文协议(MCP)中的传输为客户端和服务器之间的通信提供基础。传输层负责处理消息如何发送和接收的底层机制。
消息格式
MCP 使用 JSON-RPC 2.0 作为其传输格式。传输层负责将 MCP 协议消息转换为 JSON-RPC 格式进行传输，并将接收到的 JSON-RPC 消息转换回 MCP 协议消息。
有三种类型的 JSON-RPC 消息：
请求
{
jsonrpc: "2.0",
id: number | string,
method: string,
params?: object
}响应
{
jsonrpc: "2.0",
id: number | string,
result?: object,
error?: {
code: number,
message: string,
data?: unknown
}
}通知
{
jsonrpc: "2.0",
method: string,
params?: object
}内置传输类型
MCP 包含两种标准传输实现：
标准输入/输出 (stdio)
stdio 传输通过标准输入和输出流实现通信。这对于本地集成和命令行工具特别有用。
使用 stdio 的场景：
- 构建命令行工具
- 实现本地集成
- 需要简单的进程间通信
- 使用 shell 脚本
| |
| |
async def stdio_server():
try:
# 创建用于双向通信的流
read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
async def message_handler():
try:
async with read_stream_writer:
# 消息处理逻辑
pass
except Exception as exc:
logger.error(f"消息处理失败：{exc}")
raise exc
async with anyio.create_task_group() as tg:
tg.start_soon(message_handler)
try:
# 返回用于通信的流
yield read_stream, write_stream
except Exception as exc:
logger.error(f"传输错误：{exc}")
raise exc
finally:
tg.cancel_scope.cancel()
await write_stream.aclose()
await read_stream.aclose()
except Exception as exc:
logger.error(f"初始化传输失败：{exc}")
raise exc服务器发送事件 (SSE)
SSE传输支持服务器到客户端的流式传输，同时使用HTTP POST请求实现客户端到服务器的通信。
适用场景：
- 仅需要服务器到客户端的流式传输
- 在受限网络环境下工作
- 实现简单的更新操作
const server = new Server({
name: "example-server",
version: "1.0.0"
}, {
capabilities: {}
});
const transport = new SSEServerTransport("/message", response);
await server.connect(transport);const client = new Client({
name: "example-client",
version: "1.0.0"
}, {
capabilities: {}
});
const transport = new SSEClientTransport(
new URL("http://localhost:3000/sse")
);
await client.connect(transport);from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route
app = Server("example-server")
sse = SseServerTransport("/messages")
async def handle_sse(scope, receive, send):
async with sse.connect_sse(scope, receive, send) as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
async def handle_messages(scope, receive, send):
await sse.handle_post_message(scope, receive, send)
starlette_app = Starlette(
routes=[
Route("/sse", endpoint=handle_sse),
Route("/messages", endpoint=handle_messages, methods=["POST"]),
]
)async with sse_client("http://localhost:8000/sse") as streams:
async with ClientSession(streams[0], streams[1]) as session:
await session.initialize()自定义传输
MCP让实现自定义传输变得简单。任何传输实现只需要符合Transport接口即可：
可以实现自定义传输用于：
- 自定义网络协议
- 专用通信通道
- 与现有系统集成
- 性能优化
interface Transport {
// Start processing messages
start(): Promise<void>;
// Send a JSON-RPC message
send(message: JSONRPCMessage): Promise<void>;
// Close the connection
close(): Promise<void>;
// Callbacks
onclose?: () => void;
onerror?: (error: Error) => void;
onmessage?: (message: JSONRPCMessage) => void;
}Note that while MCP Servers are often implemented with asyncio, we recommend
implementing low-level interfaces like transports with `anyio` for wider compatibility.
```python
@contextmanager
async def create_transport(
read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],
write_stream: MemoryObjectSendStream[JSONRPCMessage]
):
"""
Transport interface for MCP.
Args:
read_stream: Stream to read incoming messages from
write_stream: Stream to write outgoing messages to
"""
async with anyio.create_task_group() as tg:
try:
# Start processing messages
tg.start_soon(lambda: process_messages(read_stream))
# Send messages
async with write_stream:
yield write_stream
except Exception as exc:
# Handle errors
raise exc
finally:
# Clean up
tg.cancel_scope.cancel()
await write_stream.aclose()
await read_stream.aclose()
错误处理
传输实现应该处理各种错误场景：
- 连接错误
- 消息解析错误
- 协议错误
- 网络超时
- 资源清理
class ExampleTransport implements Transport {
async start() {
try {
// Connection logic
} catch (error) {
this.onerror?.(new Error(`Failed to connect: ${error}`));
throw error;
}
}
async send(message: JSONRPCMessage) {
try {
// Sending logic
} catch (error) {
this.onerror?.(new Error(`Failed to send message: ${error}`));
throw error;
}
}
}Note that while MCP Servers are often implemented with asyncio, we recommend
implementing low-level interfaces like transports with anyio for wider compatibility.
@contextmanager
async def example_transport(scope: Scope, receive: Receive, send: Send):
try:
# Create streams for bidirectional communication
read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
async def message_handler():
try:
async with read_stream_writer:
# Message handling logic
pass
except Exception as exc:
logger.error(f"Failed to handle message: {exc}")
raise exc
async with anyio.create_task_group() as tg:
tg.start_soon(message_handler)
try:
# Yield streams for communication
yield read_stream, write_stream
except Exception as exc:
logger.error(f"Transport error: {exc}")
raise exc
finally:
tg.cancel_scope.cancel()
await write_stream.aclose()
await read_stream.aclose()
except Exception as exc:
logger.error(f"Failed to initialize transport: {exc}")
raise exc最佳实践
在实现或使用MCP传输时：
- 正确处理连接生命周期
- 实现适当的错误处理
- 在连接关闭时清理资源
- 使用适当的超时设置
- 发送前验证消息
- 记录传输事件以便调试
- 在适当情况下实现重连逻辑
- 处理消息队列中的背压
- 监控连接健康状况
- 实现适当的安全措施
安全注意事项
在实现传输时：
身份验证和授权
- 实现适当的身份验证机制
- 验证客户端凭据
- 使用安全的令牌处理
- 实现授权检查
数据安全
- 使用TLS进行网络传输
- 加密敏感数据
- 验证消息完整性
- 实现消息大小限制
- 净化输入数据
网络安全
- 实现速率限制
- 使用适当的超时设置
- 处理拒绝服务场景
- 监控异常模式
- 实现适当的防火墙规则
传输调试
调试传输问题的建议：
- 启用调试日志
- 监控消息流
- 检查连接状态
- 验证消息格式
- 测试错误场景
- 使用网络分析工具
- 实现健康检查
- 监控资源使用
- 测试边缘情况
- 使用适当的错误跟踪