MCP 传输层

传输层是 MCP 架构的重要组成部分，负责管理客户端和服务器之间的通信通道。MCP 支持两种传输机制，分别适用于不同的部署场景。

概述

传输层处理连接建立、消息帧和通信参与者之间的安全通信。选择合适的传输方式对于 MCP 应用的性能和可维护性至关重要。

STDIO 传输

STDIO（标准输入/输出）传输使用进程的标准输入流和标准输出流进行通信。这是本地 MCP 服务器的默认传输方式。

工作原理

STDIO 传输的工作流程非常简单：

1. 主机进程启动 MCP 服务器作为子进程
2. 主机通过子进程的 stdin 发送 JSON-RPC 消息
3. 服务器通过 stdout 返回响应
4. 服务器可以通过 stderr 输出日志

Python 实现

使用 Python SDK 创建 STDIO 服务器：

import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# 创建服务器实例
app = Server("my-server")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="echo",
            description="返回输入的文本",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "要返回的文本"}
                },
                "required": ["text"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "echo":
        return [TextContent(type="text", text=arguments["text"])]
    raise ValueError(f"Unknown tool: {name}")

async def main():
    # 使用 STDIO 传输运行服务器
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

TypeScript 实现

使用 TypeScript SDK 创建 STDIO 服务器：

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "my-server",
  version: "1.0.0"
});

server.tool(
  "echo",
  { text: z.string().describe("要返回的文本") },
  async ({ text }) => ({
    content: [{ type: "text", text }]
  })
);

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  // 注意：使用 console.error 输出日志，不要使用 console.log
  console.error("MCP Server running on stdio");
}

main().catch(console.error);

重要注意事项

使用 STDIO 传输时，必须特别注意日志输出：

import sys

# ❌ 错误：stdout 用于 MCP 消息，写入会破坏协议
print("Processing request")

# ✅ 正确：写入 stderr 进行日志输出
print("Processing request", file=sys.stderr)

# ✅ 正确：使用 logging 模块
import logging
logging.basicConfig(stream=sys.stderr)
logging.info("Processing request")

为什么这很重要？因为 STDIO 传输使用 stdout 传输 JSON-RPC 消息，任何写入 stdout 的内容都会破坏消息格式，导致协议错误。

适用场景

STDIO 传输最适合以下场景：

• 本地开发工具（如 Claude Desktop、VS Code 扩展）
• 单用户应用
• 需要最佳性能的本地集成
• 快速原型开发

Streamable HTTP 传输

Streamable HTTP 传输支持远程服务器通信，使用 HTTP POST 进行请求，可选 Server-Sent Events（SSE）实现服务器到客户端的流式消息。

工作原理

HTTP 传输的工作流程：

1. 客户端通过 HTTP POST 发送 JSON-RPC 请求
2. 服务器返回 HTTP 响应，可以是立即返回或流式返回
3. 对于需要服务器主动推送的场景，使用 SSE 连接

Python 实现

使用 Python SDK 创建 HTTP 服务器：

from mcp.server import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

app = Server("http-server")

# 定义工具和资源...

# 创建 Starlette 应用
sse = SseServerTransport("/messages")

async def handle_sse(request):
    async with sse.connect_sse(
        request.scope,
        request.receive,
        request._send
    ) as streams:
        await app.run(
            streams[0],
            streams[1],
            app.create_initialization_options()
        )

async def handle_messages(request):
    await sse.handle_post_message(request.scope, request.receive, request._send)

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Route("/messages", endpoint=handle_messages, methods=["POST"])
    ]
)

认证支持

HTTP 传输支持标准的 HTTP 认证方法：

Bearer Token 认证：

from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware

def validate_token(token: str) -> bool:
    # 验证 token 逻辑
    return token == "valid-token"

async def auth_middleware(request, call_next):
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return JSONResponse({"error": "Unauthorized"}, status_code=401)
    
    token = auth[7:]
    if not validate_token(token):
        return JSONResponse({"error": "Invalid token"}, status_code=401)
    
    return await call_next(request)

OAuth 集成：

MCP 建议使用 OAuth 获取认证令牌。客户端可以在请求中包含 OAuth 访问令牌：

Authorization: Bearer <oauth_access_token>

适用场景

HTTP 传输最适合以下场景：

• 远程 API 服务
• 多用户企业应用
• 需要负载均衡的部署
• 跨网络的服务集成
• 需要 OAuth 等标准认证的场景

传输方式对比

特性	STDIO	HTTP
部署位置	本地	本地或远程
性能	最优（无网络开销）	较好（有网络延迟）
认证	不需要	支持 Bearer、OAuth 等
并发客户端	单一	多个
调试难度	较低	中等
日志处理	必须 stderr	标准 HTTP 日志

客户端配置

Claude Desktop 配置

在 Claude Desktop 的配置文件中配置 MCP 服务器：

STDIO 服务器：

{
  "mcpServers": {
    "local-tool": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

HTTP 服务器：

{
  "mcpServers": {
    "remote-tool": {
      "url": "https://api.example.com/mcp",
      "headers": {
        "Authorization": "Bearer your-token"
      }
    }
  }
}

VS Code 配置

在 VS Code 的 MCP 配置中：

{
  "servers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/dir"]
    }
  }
}

最佳实践

STDIO 传输最佳实践

1. 始终使用 stderr 输出日志：避免污染 stdout
2. 使用绝对路径配置：确保服务器可执行文件路径正确
3. 妥善处理进程生命周期：确保子进程正确启动和关闭
4. 使用日志库：使用 logging 模块而非 print

HTTP 传输最佳实践

1. 使用 HTTPS：保护传输数据安全
2. 实施认证：使用 OAuth 或 Bearer Token
3. 添加速率限制：防止滥用
4. 监控和日志：记录请求和错误
5. 优雅关闭：正确处理连接关闭

故障排除

STDIO 常见问题

问题：服务器无响应

检查项：

• 服务器脚本是否有执行权限
• 路径是否正确（使用绝对路径）
• 是否错误地写入了 stdout

问题：协议错误

检查项：

• 是否有 print 语句写入 stdout
• JSON-RPC 消息格式是否正确

HTTP 常见问题

问题：连接超时

检查项：

• 服务器 URL 是否正确
• 网络是否可达
• 是否有防火墙阻止

问题：认证失败

检查项：

• Token 是否有效
• Authorization 头格式是否正确
• Token 是否有过期时间

下一步

• 快速入门：构建你的第一个 MCP 服务器
• 工具系统：学习如何定义工具
• 调试指南：深入了解调试技巧