首页 / AI工具 / MCP 2026 协议升级迁移指南

你的 MCP Server 还在用 Session 吗?2026-07-28 协议大改,这几行代码必须换掉

MCP 2026-07-28 是自诞生以来最大规模的版本更新。initialize 握手没了,Mcp-Session-Id 没了,协议层彻底无状态化。你的 MCP Server 需要改什么?这份迁移指南帮你一步步搞定。

为什么这次更新是 Breaking Changes 级别?

2026年7月28日,Model Context Protocol(MCP)将正式发布 2026-07-28 版本规范。这不是一次小修小补——它是 MCP 自2024年11月诞生以来最大规模的协议修订。

核心变化可以用一句话概括:MCP 协议层从有状态变为无状态

在此之前,每次连接 MCP Server 都需要经历 initialize/initialized 握手,服务器返回 Mcp-Session-Id,后续所有请求必须携带这个 Session ID。这导致了一个致命的生产问题:负载均衡器需要做 sticky routing(会话亲和),多实例部署需要共享 Session Store,一次服务器重启就会丢失所有进行中的会话。

2026-07-28 版本一口气解决了这些问题。六个 SEP(Specification Enhancement Proposal)协同工作,移除了协议层的会话概念,让任何 MCP 请求可以落在任何服务器实例上,普通的 round-robin 负载均衡器即可工作。

核心变更详解

1. Streamable HTTP Transport:告别 GET /sse,统一 POST /message

在旧版协议中,Streamable HTTP Transport 使用两个端点:

新版本将所有通信统一到 单一 POST 端点(通常为 /mcp),服务器可以选择以 JSON Body 或 SSE 流的方式响应。如果服务器需要向客户端发起请求(如 elicitation),不再通过 SSE 推送,而是返回 InputRequiredResult,客户端再发起新一轮请求。

旧版请求流程:

POST /message HTTP/1.1
Content-Type: application/json

{"jsonrpc":"2.0","id":1,"method":"initialize",
 "params":{"protocolVersion":"2025-11-25","capabilities":{},
           "clientInfo":{"name":"my-app","version":"1.0"}}}

服务器返回 Mcp-Session-Id,后续请求必须携带:

POST /message HTTP/1.1
Mcp-Session-Id: 1868a90c-3a3f-4f5b
Content-Type: application/json

{"jsonrpc":"2.0","id":2,"method":"tools/call",
 "params":{"name":"search","arguments":{"q":"otters"}}}

新版请求流程(2026-07-28):

POST /mcp HTTP/1.1
MCP-Protocol-Version: 2026-07-28
Mcp-Method: tools/call
Mcp-Name: search
Content-Type: application/json

{"jsonrpc":"2.0","id":1,"method":"tools/call",
 "params":{"name":"search","arguments":{"q":"otters"},
           "_meta":{"io.modelcontextprotocol/clientInfo":{"name":"my-app","version":"1.0"}}}}

注意三个关键变化:

  1. 不再有 Mcp-Session-Id
  2. 新增 Mcp-MethodMcp-Name 路由头(必须携带
  3. 协议版本和客户端信息移入 _meta 对象

2. initialize 握手移除

initialize/initialized 握手彻底消失。原来在握手阶段交换的协议版本、客户端信息、能力声明,现在全部通过每个请求的 _meta 对象携带。

服务器必须实现新的 server/discover RPC 方法,用于向客户端通告自身支持的能力和身份。

3. HTTP-to-SSE 动态升级与 Multi Round-Trip Requests

旧版中服务器通过 SSE 长连接主动向客户端发起请求(如 sampling/createMessage)。新版本用 Multi Round-Trip Requests(MRTR) 替代了这个模式:

  1. 服务器处理请求时,如果需要用户输入,返回 InputRequiredResult,包含 inputRequests 和一个不透明的 requestState
  2. 客户端收集用户输入后,携带 inputResponses 和原始的 requestState 重新发起请求
  3. 任何服务器实例都能处理这个重试,因为所有状态都在 payload 中
{
  "resultType": "inputRequired",
  "inputRequests": {
    "confirm": {
      "type": "elicitation",
      "message": "确认删除 3 个文件?",
      "schema": { "type": "boolean" }
    }
  },
  "requestState": "eyJzdGVwIjoxLCJmaWxlcyI6WyJhIiwiYiIsImMiXX0="
}

4. 无状态化:服务器自主决定是否存储会话信息

协议层不再管理会话状态,但这不意味着你的应用必须是纯无状态的。如果你的应用确实需要跨调用保持状态,推荐的做法是:让工具方法返回一个显式句柄(handle),由模型在后续调用中作为普通参数传回

比如一个文件管理工具:

# 工具返回 basket_id 作为显式句柄
@mcp.tool()
def create_workspace(name: str) -> dict:
    basket_id = str(uuid4())
    workspaces[basket_id] = {"name": name, "files": []}
    return {
        "content": [{"type": "text", "text": f"工作区已创建,ID: {basket_id}"}],
        "structuredContent": {"basket_id": basket_id, "status": "created"}
    }

# 后续调用中模型将 basket_id 作为参数传入
@mcp.tool()
def list_files(basket_id: str) -> dict:
    workspace = workspaces.get(basket_id)
    if not workspace:
        raise ValueError(f"无效的工作区 ID: {basket_id}")
    return {
        "content": [{"type": "text", "text": json.dumps(workspace["files"])}]
    }

5. SDK Tier 体系

MCP 官方引入了 SDK 分层体系。Tier 1 SDK(TypeScript、Python、Go、C#)必须在 RC 窗口内完成适配。目前所有四个 Tier 1 SDK 都已发布 Beta 版本:

# Python SDK Beta
pip install "mcp[cli]==2.0.0b1"

# TypeScript SDK Beta
npm install @modelcontextprotocol/server@beta

# Go SDK Beta
go get github.com/modelcontextprotocol/go-sdk@v1.7.0-pre.1

TypeScript SDK 还提供了自动化迁移 codemod:

npx @modelcontextprotocol/codemod@beta v1-to-v2 .

6. 其他重要变更

实操:用 Python MCP SDK 适配新版本

新版 Streamable HTTP Server 示例

"""
MCP 2026-07-28 新版 Streamable HTTP Server 示例
使用 Python MCP SDK 2.0 Beta
"""

import json
import uuid
from typing import Any

from mcp.server.fastmcp import FastMCP
from mcp.server.streamable_http import StreamableHTTPServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

# 创建 FastMCP 实例(无状态模式)
mcp = FastMCP(
    name="stateless-demo-server",
    version="2.0.0",
)

# 工作区存储(应用层状态,非协议层)
workspaces: dict[str, dict] = {}


@mcp.tool()
def create_workspace(name: str) -> dict:
    """创建一个新的工作区,返回显式句柄 basket_id。"""
    basket_id = str(uuid.uuid4())[:8]
    workspaces[basket_id] = {"name": name, "files": []}
    return {
        "content": [
            {
                "type": "text",
                "text": (
                    f"工作区 '{name}' 已创建。\n"
                    f"basket_id: {basket_id}\n"
                    f"请在后续操作中使用此 ID。"
                ),
            }
        ],
        "structuredContent": {"basket_id": basket_id, "status": "created"},
    }


@mcp.tool()
def add_file(basket_id: str, filename: str, content: str) -> dict:
    """向指定工作区添加文件。"""
    workspace = workspaces.get(basket_id)
    if not workspace:
        raise ValueError(f"无效的 basket_id: {basket_id}")
    workspace["files"].append({"name": filename, "content": content})
    return {
        "content": [
            {
                "type": "text",
                "text": f"文件 '{filename}' 已添加到工作区 '{workspace['name']}'。"
            }
        ]
    }


@mcp.tool()
def list_files(basket_id: str) -> dict:
    """列出指定工作区中的所有文件。"""
    workspace = workspaces.get(basket_id)
    if not workspace:
        raise ValueError(f"无效的 basket_id: {basket_id}")
    if not workspace["files"]:
        return {
            "content": [
                {"type": "text", "text": "工作区中暂无文件。"}
            ]
        }
    file_list = "\n".join(f"- {f['name']}" for f in workspace["files"])
    return {
        "content": [
            {"type": "text", "text": f"工作区 '{workspace['name']}' 的文件:\n{file_list}"}
        ]
    }


# 挂载到 Streamable HTTP Transport
async def handle_streamable_http(request):
    """处理 Streamable HTTP 请求的 ASGI handler。"""
    transport = StreamableHTTPServerTransport(url="/mcp")
    await mcp.run_sse_async_server(transport)
    return await transport.handle_request(request)


app = Starlette(
    routes=[
        Route("/mcp", handle_streamable_http, methods=["POST"]),
    ]
)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

客户端迁移代码示例

"""
MCP 2026-07-28 客户端迁移示例
从旧版有状态客户端迁移到新版无状态客户端
"""

import asyncio
import json
from typing import Any

import httpx


# ============ 旧版客户端(2025-11-25)============

async def old_client_call(server_url: str):
    """旧版:需要先握手获取 session,后续请求携带 session ID。"""
    async with httpx.AsyncClient() as client:
        # 第一步:initialize 握手
        init_resp = await client.post(
            f"{server_url}/message",
            json={
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2025-11-25",
                    "capabilities": {},
                    "clientInfo": {"name": "my-client", "version": "1.0"},
                },
            },
        )
        init_data = init_resp.json()
        session_id = init_resp.json()["result"].get("session_id")
        headers = {"Mcp-Session-Id": session_id}

        # 第二步:发送 initialized 通知
        await client.post(
            f"{server_url}/message",
            json={
                "jsonrpc": "2.0",
                "method": "notifications/initialized",
            },
            headers=headers,
        )

        # 第三步:调用工具(依赖 session)
        tool_resp = await client.post(
            f"{server_url}/message",
            json={
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/call",
                "params": {
                    "name": "search",
                    "arguments": {"q": "otters"},
                },
            },
            headers=headers,
        )
        return tool_resp.json()


# ============ 新版客户端(2026-07-28)============

async def new_client_call(server_url: str):
    """新版:无状态,每个请求自包含,无需握手。"""
    client_meta = {
        "io.modelcontextprotocol/clientInfo": {
            "name": "my-client",
            "version": "2.0",
        }
    }

    headers = {
        "MCP-Protocol-Version": "2026-07-28",
        "Mcp-Method": "tools/call",
        "Mcp-Name": "search",
    }

    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {
            "name": "search",
            "arguments": {"q": "otters"},
            "_meta": client_meta,
        },
    }

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{server_url}/mcp",
            json=payload,
            headers=headers,
        )
        return resp.json()


# ============ 会话管理最佳实践 ============

async def stateless_workflow(server_url: str):
    """
    无状态工作流最佳实践:
    使用显式句柄跨调用传递状态,而非依赖协议层 session。
    """
    client_meta = {
        "io.modelcontextprotocol/clientInfo": {
            "name": "workflow-client",
            "version": "2.0",
        }
    }

    async def call_tool(name: str, arguments: dict, call_id: int = 1) -> Any:
        resp = await httpx.AsyncClient().post(
            f"{server_url}/mcp",
            headers={
                "MCP-Protocol-Version": "2026-07-28",
                "Mcp-Method": "tools/call",
                "Mcp-Name": name,
            },
            json={
                "jsonrpc": "2.0",
                "id": call_id,
                "method": "tools/call",
                "params": {
                    "name": name,
                    "arguments": arguments,
                    "_meta": client_meta,
                },
            },
        )
        return resp.json()

    # 第一步:创建工作区,获取显式句柄
    result = await call_tool("create_workspace", {"name": "project-alpha"}, 1)
    basket_id = result["result"]["structuredContent"]["basket_id"]
    print(f"创建工作区,获取句柄: {basket_id}")

    # 第二步:使用句柄添加文件(可由任何服务器实例处理)
    await call_tool(
        "add_file",
        {"basket_id": basket_id, "filename": "main.py", "content": "print('hello')"},
        2,
    )

    # 第三步:使用句柄列出文件
    result = await call_tool("list_files", {"basket_id": basket_id}, 3)
    print(f"文件列表: {result['result']['content'][0]['text']}")


if __name__ == "__main__":
    asyncio.run(new_client_call("http://localhost:8000"))

向后兼容性处理方案

如果你的服务器在迁移窗口内需要同时支持新旧两版客户端,可以采用渐进式策略:

"""
渐进式兼容方案:同时支持 2025-11-25 和 2026-07-28 客户端
"""

from mcp.server.fastmcp import FastMCP
from mcp.shared.version import PROTOCOL_VERSION

mcp = FastMCP(name="compat-server", version="2.0.0")


async def middleware_handler(request):
    """中间件:检测协议版本,路由到对应处理逻辑。"""
    protocol_version = request.headers.get(
        "MCP-Protocol-Version", "2025-11-25"
    )

    if protocol_version == "2026-07-28":
        # 新版无状态处理:从 _meta 读取客户端信息
        body = await request.json()
        meta = body.get("params", {}).get("_meta", {})
        client_info = meta.get(
            "io.modelcontextprotocol/clientInfo", {}
        )
        print(f"[新版] 客户端: {client_info}")
        return await handle_new_protocol(request)

    else:
        # 旧版兼容处理:检查 session header
        session_id = request.headers.get("Mcp-Session-Id")
        if session_id:
            print(f"[旧版] Session: {session_id}")
            return await handle_legacy_protocol(request)

    return await handle_new_protocol(request)


async def handle_new_protocol(request):
    """处理 2026-07-28 协议请求。"""
    return await mcp.streamable_http_handler(request)


async def handle_legacy_protocol(request):
    """处理旧版 2025-11-25 协议请求(兼容期使用)。"""
    # 在兼容期内,可以维护一个 session 映射表
    # 将旧版 session 的状态转化为新版 _meta 格式处理
    body = await request.json()
    session_id = request.headers.get("Mcp-Session-Id")

    # 将 session 中的信息注入到请求的 _meta 中
    if session_id and "_meta" not in body.get("params", {}):
        body.setdefault("params", {})["_meta"] = {
            "io.modelcontextprotocol/clientInfo": {
                "name": "legacy-client",
                "version": "1.0",
            }
        }

    return await mcp.streamable_http_handler(request)

迁移检查清单

按照优先级排序,建议在正式版发布前完成以下迁移步骤:

  1. [高] 移除 Session 依赖:搜索代码中的 Mcp-Session-IdsessionIdGeneratorsessionStore,将跨调用状态改为显式句柄模式
  2. [高] 移除 initialize 握手:搜索 onInitializeInitializeRequestSchema,将能力协商改为从 _meta 逐请求读取
  3. [高] 更新错误码:将 -32002 改为 -32602(Invalid Params)
  4. [中] 添加路由头:确保所有 POST 请求携带 Mcp-MethodMcp-Name
  5. [中] 实现 server/discover:新版的必选 RPC 方法
  6. [中] 添加缓存支持:在 list 和 read 响应中添加 ttlMscacheScope
  7. [低] 评估 Deprecated 特性:检查 Roots、Sampling、Logging 的使用,规划迁移路径

快速排查命令:

# 搜索 Session 相关代码
grep -rn "Mcp-Session-Id\|sessionIdGenerator\|sessionStore" src/

# 搜索 initialize 握手代码
grep -rn "initialized\|onInitialize\|InitializeRequestSchema" src/

# 搜索旧版错误码
grep -rn "32002" src/ test/

# 搜索已废弃特性
grep -rn "createMessage\|ListRootsRequest\|sendLoggingMessage" src/

常见问题 FAQ

Q1:我的旧版 MCP Server 在 2026-07-28 之后还能正常工作吗?

取决于你的客户端和服务器实现。协议版本 2025-11-25 的服务器不会被"自动关闭",但如果你的客户端升级到了新版 SDK,它会发送不符合旧版预期的请求格式(缺少握手、缺少 Session ID)。建议尽快迁移到新版 SDK Beta 并测试兼容性。官方提供了 12 个月的废弃窗口,已废弃的特性(Roots、Sampling、Logging)在此期间仍可使用。

Q2:无状态化之后,我该怎么保持用户的上下文?

协议层不再管理会话,但应用层完全可以自行管理。推荐两种模式:

两种模式都不依赖协议层的 Session,因此天然兼容无状态部署。

Q3:TypeScript SDK 的 codemod 能自动处理所有迁移吗?

不能覆盖所有场景。codemod 主要处理以下机械性变更:

但以下情况需要手动处理:

建议先运行 codemod 完成机械性变更,再手动处理架构性变更。

Q4:SDK Beta 版本可以直接用于生产吗?

不推荐。Beta 版本用于在 RC 窗口内验证迁移方案,API 可能仍有微调。建议在开发/测试环境中使用 Beta 版验证迁移,待 7 月 28 日正式版发布后切换到稳定版。Go 和 C# SDK 保持 API 兼容,迁移成本最低;Python 和 TypeScript 的变更较大,需要更多测试。

Q5:负载均衡器和网关需要做什么调整?

好消息是:配置更简单了。你不再需要配置 sticky sessions(会话亲和)或 session store。但需要确保:

  1. 网关/负载均衡器放行 Mcp-MethodMcp-Name 自定义头(不要过滤掉)
  2. 如果使用 Nginx/HAProxy,可以用这些头做路由规则,例如将 tools/callresources/read 路由到不同后端
  3. 可以使用标准 round-robin 算法,无需深度包检测

写在最后

MCP 2026-07-28 的无状态化改造是一次正确的架构演进。它让 MCP Server 可以像普通 HTTP API 一样部署和扩展,消除了生产环境中最痛的几个问题。迁移的核心思路很简单:把状态从协议层上移到应用层,用显式句柄替代隐式会话

如果你现在还在用 2025-11-25 的 SDK,立即安装 Beta 版跑一下迁移测试,比等到最后一刻要好得多。RC 窗口就是为了这个准备的。

参考资料