MCP RC 迁移实战:Roots/Sampling/Logging 废弃后的代码改造指南

MCP 2026-07-28 RC 是协议诞生以来最大重构。5 个 Breaking Changes、3 个功能废弃、从 stateful 到 stateless 的架构转型。本文用真实代码对比,逐个拆解每个改动怎么改、改完怎么验。

MCP 的 7 月 RC 不是小版本迭代,是协议诞生以来最大的一次重构。3000+ 现有 MCP Server 都需要迁移,Roots、Sampling、Logging 三个核心功能直接废弃。如果你在维护 MCP Server,现在就该动手了。

一、为什么这次升级必须重视

RC 锁定日期是 2026-05-21,正式版 2026-07-28。核心变化就一句话:从有状态变成无状态。

项目旧协议RC 协议
会话模型服务端维护 Session-Id,需要 sticky session完全无状态,任何请求可落到任意实例
初始化流程initialize 握手 → 拿 Session-Id → 后续携带每个请求自包含,协议版本在 _meta
影响范围3000+ MCP Server(MCP.Directory 统计)
废弃窗口12 个月(不是立即删除,但现在就要准备)

时间线很清晰:5 月锁定 → 7 月 28 日发布 → 12 个月后移除废弃功能。 Tier 1 SDK 需要在 10 周窗口内发布支持。

二、5 个 Breaking Changes 详解

每个 Breaking Changes 都有旧代码 vs 新代码对比,不是纯文字描述。

2.1 移除 initialize 握手(SEP-2575)

旧方式需要两步握手才能拿到 Session-Id,新方式每个请求直接带协议信息:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
# ❌ 旧方式:initialize 握手
async def handle_mcp_request(request):
    if request.method == "initialize":
        session_id = generate_session_id()
        sessions[session_id] = {"client": request.params}
        return {"sessionId": session_id, "serverInfo": {...}}
    
    # 后续请求必须携带 Session-Id
    session_id = request.headers.get("Mcp-Session-Id")
    if not session_id or session_id not in sessions:
        return error("Invalid session")

# ✅ 新方式:每个请求自包含
async def handle_mcp_request(request):
    # 不再需要 initialize,协议版本和客户端信息在 _meta 中
    protocol_version = request.meta.get("protocolVersion")
    client_info = request.meta.get("clientInfo")
    # 直接处理业务逻辑
    return process_request(request)

改造要点:删除 initialize/initialized 的处理逻辑,从 _meta 中读取协议版本和客户端信息。

2.2 移除 Mcp-Session-Id(SEP-2567)

这是最根本的变化——从有状态变无状态:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
# ❌ 旧方式:服务端维护会话状态
class McpServer:
    def __init__(self):
        self.sessions = {}  # 有状态,需要 sticky session
    
    def handle(self, request):
        sid = request.headers["Mcp-Session-Id"]
        session = self.sessions[sid]
        # session 上下文用于后续请求...

# ✅ 新方式:完全无状态
class McpServer:
    def handle(self, request):
        # 没有 session,每个请求独立处理
        # 需要状态?用应用层 handle
        basket_id = request.params.get("basket_id")
        # 状态通过参数显式传递,任何实例都能处理

改造要点:把状态管理从协议层移到应用层。如果你的 Server 依赖 Session-Id 存状态(比如购物篮、浏览器上下文),改用显式 handle(basket_idbrowser_id 等)。

2.3 新增路由头(SEP-2243)

每个请求必须携带 Mcp-MethodMcp-Name 头,让 LB/网关/限流器无需解析 body 即可路由:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
# ❌ 旧方式:网关需要解析 JSON body 才能路由
def route_request(request):
    body = json.loads(request.body)
    method = body.get("method")  # 必须解析 body
    if method.startswith("tools/"):
        return forward_to_tools_server(request)

# ✅ 新方式:从路由头直接判断
def route_request(request):
    mcp_method = request.headers.get("Mcp-Method")
    mcp_name = request.headers.get("Mcp-Name")
    
    # 无需解析 body,直接路由
    if mcp_method == "tools" and mcp_name == "call":
        return forward_to_tools_server(request)

注意:头和 body 不一致时,服务端必须拒绝请求。这是个安全设计——防止 header spoofing。

2.4 缓存元数据(SEP-2549)

tools/list 和 resource read 结果现在带 ttlMs + cacheScope,类似 HTTP 的 Cache-Control:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
# ❌ 旧方式:需要长连接 SSE 来发现列表变更
async def watch_tools(client):
    async for event in client.subscribe("notifications/tools/list_changed"):
        tools = await client.call("tools/list")
        update_tools_cache(tools)

# ✅ 新方式:缓存元数据,按 TTL 刷新
async def get_tools(client):
    result = await client.call("tools/list")
    tools = result["tools"]
    ttl_ms = result.get("meta", {}).get("ttlMs", 60000)
    cache_scope = result.get("meta", {}).get("cacheScope", "instance")
    # 在 ttl_ms 内使用缓存,过期再请求
    return tools

改造要点:移除 tools/list_changed 的 SSE 监听逻辑,改用 TTL 主动拉取。

2.5 多轮请求(SEP-2322)

InputRequiredResult 替代 SSE 流,客户端收集回答后带 inputResponses + requestState 重新发起请求:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# ❌ 旧方式:通过 SSE 流收集多轮输入
async def handle_tool_call(client, tool_call):
    async for chunk in client.stream(tool_call):
        if chunk.type == "input_required":
            user_input = await collect_user_input(chunk.prompt)
            # 通过同一个 SSE 流回传...
            await client.send_input(user_input)

# ✅ 新方式:请求-响应模式,状态在 payload 中
async def handle_tool_call(client, tool_call):
    result = await client.call("tools/call", tool_call)
    
    if result.get("type") == "inputRequired":
        # 收集用户输入
        user_input = await collect_user_input(result["prompt"])
        # 带 inputResponses + requestState 重新发起
        result = await client.call("tools/call", {
            **tool_call,
            "inputResponses": [{"promptId": result["promptId"], "value": user_input}],
            "requestState": result["requestState"]
        })
    
    return result

关键设计:状态在 payload 中,任何实例都能处理(和无状态架构一致)。

三、3 个废弃功能 + 替代方案

这是每个 MCP Server 开发者都会遇到的问题。废弃 ≠ 立即删除,但 12 个月后就真没了。

3.1 Roots 废弃

Roots 让服务端声明文件系统根目录,客户端通过 roots/list_changed 通知变更。替代方案有三种:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# ❌ 旧方式:Roots 能力声明
server = McpServer(
    capabilities={
        "roots": {"listChanged": True}
    }
)

# 客户端通过 roots 通知路径变更
await client.notify("notifications/roots/list_changed")

# ✅ 替代方案 1:Tool 参数传路径
@tool("read_file", description="读取文件内容")
async def read_file(path: str):
    # 路径作为参数显式传递
    return open(path).read()

# ✅ 替代方案 2:Resource URI 声明
@resource("file://{path}")
async def get_file(path: str):
    return open(path).read()

# ✅ 替代方案 3:服务端配置文件
# 在 MCP Server 启动时读取配置
config = load_config("~/.mcp-server/config.yaml")
base_dir = config.get("base_dir", "/workspace")

改造建议:如果你的 Server 依赖 Roots 来确定工作目录,用 Tool 参数最简单。

3.2 Sampling 废弃

Sampling 让服务端通过 sampling/createMessage 请求客户端采样(即调用 LLM)。替代方案:服务端直接集成 LLM API。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# ❌ 旧方式:请求客户端采样
@tool("analyze_text")
async def analyze_text(text: str):
    # 服务端不能自己调模型,必须请求客户端
    result = await request_client_sampling(
        messages=[{"role": "user", "content": f"分析:{text}"}],
        model="claude-sonnet-4"
    )
    return result.content

# ✅ 新方式:服务端直接集成 LLM
from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

@tool("analyze_text")
async def analyze_text(text: str):
    # 服务端自己调模型,不再依赖客户端
    response = client.messages.create(
        model="claude-sonnet-4",
        max_tokens=1024,
        messages=[{"role": "user", "content": f"分析:{text}"}]
    )
    return response.content[0].text

改造要点:移除 sampling/createMessage handler,接入 OpenAI/Anthropic/其他 LLM provider 的 SDK。API Key 通过环境变量注入。

3.3 Logging 废弃

协议层的 logging/setLevel 控制日志级别,替代方案:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# ❌ 旧方式:协议层日志控制
@method("logging/setLevel")
async def set_log_level(level: str):
    # 通过 MCP 协议控制日志级别
    logging.getLogger("mcp").setLevel(level)

@method("notifications/message")
async def log_message(level: str, message: str, logger: str = None):
    # 日志通过协议通知发送
    await send_notification("notifications/message", {
        "level": level,
        "message": message,
        "logger": logger
    })

# ✅ 替代方案 1:stdio 用 stderr
import sys
import logging

logger = logging.getLogger("mcp-server")
handler = logging.StreamHandler(sys.stderr)  # stdio 模式用 stderr
handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# ✅ 替代方案 2:远程用 OpenTelemetry
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otel-collector:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("mcp-server")

改造建议:stdio 模式直接用 stderr + 标准 logging;HTTP/SSE 模式用 OpenTelemetry 结构化日志,和 SEP-414 的分布式追踪配合。

四、迁移检查清单

按时间线拆解,每一步都有具体的检查项和验证方法。

时间检查什么怎么改改完怎么验
现在(6月前)读 draft spec,审计 auth 代码标记所有 initializeSession-Id、Roots/Sampling/Logging 调用点grep -rn "initialize|Session-Id|roots|sampling|logging" src/
6 月用 RC SDK 重建,验证路由头升级 SDK,添加 Mcp-Method/Mcp-Name抓包验证每个请求都带路由头,头和 body 一致
7 月部署无状态变体,测试自动扩缩移除 session 管理,状态改应用层 handle多实例部署,任意请求落到任意实例都能正常返回
7月28日+声明新协议版本,保持废弃功能可用protocolVersion 更新为 RC 版本12 个月内新旧功能并行,监控废弃功能调用量

具体验证命令:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
# 1. 审计代码中的废弃功能调用
grep -rn "initialize\|initialized\|Session-Id\|roots\|sampling\|logging/setLevel" src/

# 2. 检查是否还有 session 状态
grep -rn "sessions\[" src/  # 如果有,需要改造成应用层 handle

# 3. 验证路由头
curl -X POST http://localhost:3000/mcp \
  -H "Content-Type: application/json" \
  -H "Mcp-Method: tools" \
  -H "Mcp-Name: list" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'

# 4. 测试无状态:发同一个请求到不同实例
for port in 3001 3002 3003; do
  curl -s "http://localhost:$port/mcp" \
    -H "Mcp-Method: tools" -H "Mcp-Name: list" \
    -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | jq '.result.tools | length'
done

五、总结

这次 RC 升级的核心逻辑就一条:无状态 = 水平扩展友好。

三个关键认知:

  1. 废弃 ≠ 删除。 12 个月窗口内旧功能仍可用,但新开发应该直接用新方案
  2. 状态不是消失了,是下沉到应用层。 Session-Id 移除后,你的 basket_idbrowser_id 等显式 handle 更清晰
  3. 现在就开始。 7 月 28 日发布后,Tier 1 SDK 会在 10 周内跟进,你的 Server 应该在 SDK 生态更新前完成适配

建议:先跑一遍上面的 grep 审计,标记所有需要改的地方。然后按检查清单的时间线逐步推进。别等到正式版发布才动手——那时候你会同时面对 SDK 升级 + 协议适配 + 用户反馈,手忙脚乱。