MCP协议2026-07-28深度拆解：从有状态双向RPC到无状态客户端驱动的范式跃迁

前言：当协议内核说换就换

2026年7月28日，Model Context Protocol（MCP）发布了自诞生以来最大的一次版本更新。

说"最大"不是标题党。这次改版把整个协议的运行逻辑翻了个底朝天：从维持会话状态的双向RPC，变成了每个请求自包含上下文的客户端驱动模型。协议版本号从2025-11-25跳到了2026-07-28，中间隔了8个月，核心架构却像是换了一个内核。

对于已经在生产环境跑着MCP的团队来说，这个升级意味着什么？是"小版本迭代，打打补丁就过去"的温和升级，还是"架构级重构，半条命搭进去"的伤筋动骨？

答案：是后者。但这个"伤筋动骨"恰恰是这次升级最有价值的地方。

本文将用庖丁解牛的方式，从五个核心变化出发，彻底拆解这次MCP大改版的设计动机、架构细节、对开发者的实际影响，以及如何用最小代价完成迁移。无论你是MCP服务端开发者、客户端实现者、还是AI Agent框架的架构师，这篇文章都会给你一个完整的路线图。

一、为什么要有这次"范式跃迁"？先聊聊MCP的老毛病

在深入技术细节之前，我们需要理解一个前置问题：旧版MCP的设计到底哪里不对劲，让官方不得不搞出这么大动静的breaking change？

1.1 会话状态：MCP最好的特性，也是最大的瓶颈

MCP从诞生起就采用了经典的"三次握手 + 会话粘滞"模式：

客户端 ──initialize──► 服务端
客户端 ◄──result+sessionID── 服务端
客户端 ──initialized──► 服务端

这三步走完之后，客户端和服务端之间才算"接上了"，之后的每次请求都带着这个session ID，服务端靠session ID来识别"你是谁、聊到哪了、有哪些能力"。

这个设计其实很优雅：

• 一次握手，双方把能力（tools、resources、prompts）都报一遍，清晰明了
• 有session就有上下文，服务端可以主动发请求（比如工具执行到一半缺参数，直接反问客户端）
• 会话粘滞意味着负载均衡的时候得小心——同一个客户端的请求最好打到同一台服务端，否则服务端不认人

但问题就出在这个"优雅"上面。

当你只有一个AI Agent在调用MCP工具的时候，这套机制运转良好。但当你的系统里有100个Agent、1000个Agent，或者你要把MCP服务端做成一个高可用、弹性伸缩的公共服务的时候，会话状态就成了分布式系统中最讨厌的东西：

• 水平扩容困难：你不能随便增加服务端实例，因为新实例没有客户端的会话状态，负载均衡策略必须做sticky session
• 故障恢复复杂：服务端挂了，客户端得重新走一遍握手流程，之前的上下文全部丢失
• 网关透明化受阻：想在MCP请求前面加一层API网关做限流、鉴权、监控？对不起，网关得解析请求体里的JSON才知道你在调什么工具——因为工具名不在HTTP头里
• 多租户隔离困难：同一个服务端实例服务多个租户，靠session来区分，但如果要跨实例路由，光靠session不够用了

说白了，旧版MCP是一个适合单实例、对等通信的协议，但不适合同一时代AI Agent大规模部署、云原生弹性伸缩、多租户网关管控的生产环境。

1.2 服务端主动请求：听起来很美，用起来很坑

旧版MCP还有一个很"超前"的设计：服务端可以主动给客户端发请求。

什么意思呢？举个例子：你让AI Agent调用一个"发送邮件"的工具。工具执行的时候服务端发现，诶，你没提供邮件内容，服务端可以直接"反向"给客户端发一个请求，要你补全邮件内容。

这个设计在理论上看似美好——减少往返次数，提高交互效率。但实际落地的时候：

• 需要服务端维持一个到客户端的长连接（或轮询通道）
• 在云原生环境下，客户端通常在用户设备或临时容器里，连接不稳定
• 很多部署场景下服务端和客户端根本不在同一个网络域，主动推送根本走不通
• 给网关、防火墙、负载均衡器的实现增加了巨大的复杂度

所以，新版MCP把这个特性直接删了——服务端再也不能主动说话。这不是技术退步，而是去掉了一个看起来很美但实际上根本无法在生产环境落地的糖衣。

1.3 总结：为什么是"范式跃迁"

这次升级的本质，是把MCP从**"一个对等的、有状态的、面向会话的通信协议"改造成"一个严格的、客户端驱型的、无状态的工具调用接口"**。

这不是打补丁，是换架构。

理解了这个动机，后面的所有细节变化你就都能自己想通了——每个改动都是为了同一个目标：让MCP成为一个真正适合云原生、大规模、多租户环境的标准工具调用协议。

二、核心变化一：无状态化——初始化握手成为历史

2.1 旧版握手三步曲

在旧版MCP里，客户端连接上服务端之后，必须走完整的握手流程：

// 步骤1：客户端发送 initialize
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-11-25",
    "capabilities": {
      "roots": {"listChanged": true},
      "sampling": {}
    },
    "clientInfo": {
      "name": "my-agent",
      "version": "1.0.0"
    }
  }
}

// 步骤2：服务端返回 result + sessionId
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2025-11-25",
    "capabilities": {
      "tools": {"listChanged": true},
      "resources": {"subscribe": true}
    },
    "serverInfo": {"name": "filesystem-server", "version": "2.1.0"},
    "sessionId": "sess_abc123def456"
  }
}

// 步骤3：客户端发送 initialized 确认
{
  "jsonrpc": "2.0",
  "method": "initialized",
  "params": {}
}

三步缺一不可。整个过程中，双方交换了能力（capabilities）、版本（protocolVersion）、身份（clientInfo/serverInfo），服务端生成了一个session ID，后续所有请求都要带着这个ID。

新版直接把这三步全删了。

2.2 新版：无状态请求，协议层零握手

新版MCP的请求格式变成了这样：

// 每个请求都是独立的、自包含的
{
  "jsonrpc": "2.0",
  "id": 42,
  "method": "tools/call",
  "params": {
    "name": "send_email",
    "arguments": {"to": "user@example.com", "subject": "Report"}
  },
  "_meta": {
    "protocolVersion": "2026-07-28",
    "clientId": "agent-001",
    "operationId": "op_xyz789"
  }
}

注意看 _meta 字段——这就是新版MCP的上下文载体。客户端每次请求都把以下信息塞进去：

• protocolVersion：协议版本，替代了握手时的版本协商
• clientId：客户端身份标识，替代了sessionId的"我是谁"
• operationId：操作追踪ID，替代了会话级别的上下文关联
• 如果需要，还可以在 _meta 里放 OpenTelemetry 的链路追踪信息（traceparent、tracestate、baggage）

服务端拿到请求，解析 _meta，直接处理，不需要查任何"之前有没有人跟我建立过连接"。

2.3 这意味着什么？——三个真实场景的改变

场景一：水平扩容

# 旧版：服务端实例之间没有共享session，需要sticky session
# nginx配置：
upstream mcp_backend {
    least_conn;
    server mcp-1:8080;
    server mcp-2:8080;
    server mcp-3:8080;
    # 必须加这个，否则同一个客户端的请求可能打到不同实例
    ip_hash;  # 或使用cookie做session sticky
}

# 新版：随便打，哪个实例闲就扔哪个
upstream mcp_backend {
    least_conn;
    server mcp-1:8080;
    server mcp-2:8080;
    server mcp-3:8080;
    # 不需要sticky session了！每个请求自己带上下文
}

场景二：故障恢复

# 旧版：服务端重启后，客户端的session失效，必须重新握手
# 如果AI Agent正在执行一个多步骤任务，中途服务端挂了
# → 任务状态全部丢失 → 需要从任务起点重跑

# 新版：服务端重启后，客户端重发同样的请求就行
# 因为请求自己包含了完整上下文，服务端不需要"记得"之前的事
# 客户端可以自己维护业务级别的任务状态，服务端只负责执行工具

场景三：API网关

# 旧版：网关想根据"调用的工具名"做限流
# 必须解析请求体JSON：
async def mcp_gateway_proxy(request):
    body = await request.json()
    tool_name = body.get("params", {}).get("name")
    await rate_limiter.check(tool_name)  # 必须先解析完JSON才能限流
    
# 新版：工具名在HTTP头里，网关可以在解析body之前就做限流
async def mcp_gateway_proxy(request):
    tool_name = request.headers.get("Mcp-Name")  # 直接从头里读
    protocol_ver = request.headers.get("MCP-Protocol-Version")
    await rate_limiter.check(tool_name)  # 先限流，再代理body

2.4 业务状态 vs. 协议状态：别搞混了

新版删掉的是协议层面的会话状态，不是说你的AI Agent不能有业务状态。

比如说，购物车、浏览器会话、多轮对话上下文——这些业务状态你当然可以自己维护。区别在于：

• 协议层不管了：不再有session来帮你存这些
• 你自己管：把这些业务ID作为参数传进去就行

# 新版MCP推荐的"业务状态携带方式"
{
  "method": "tools/call",
  "params": {
    "name": "add_to_cart",
    "arguments": {
      "product_id": "SKU-12345",
      "quantity": 2,
      # 业务ID自己带，服务端通过业务逻辑来处理，不是协议层
      "cart_id": "cart_user_001_session_42"
    }
  },
  "_meta": {
    "protocolVersion": "2026-07-28",
    "clientId": "ecommerce-agent-001"
  }
}

这其实是更清晰的关注点分离：协议层做协议的事（传输、路由、鉴权），业务层做业务的事（购物车、会话、任务状态）。

三、核心变化二：主动能力发现——从"一次性报家底"到"按需查询"

3.1 旧版：握手时报家底

旧版MCP的初始化握手有一个隐含的好处——双方在握手阶段就把能力全部交换了：

• 服务端告诉客户端：我支持哪些工具（tools.list）、哪些资源（resources.list）、哪些提示模板（prompts.list）
• 客户端告诉服务端：我支持哪些根目录（roots.list）、我是否需要采样能力（sampling.create）

握手完成后，客户端对服务端的能力了如指掌，不需要再额外查询。

3.2 新版：按需主动查询

新版没有了初始化握手，客户端怎么知道服务端有什么能力？

答案是：主动查询，用 server/discover 方法。

// 客户端主动查询服务端能力
{
  "jsonrpc": "2.0",
  "id": 10,
  "method": "server/discover",
  "params": {},
  "_meta": {
    "protocolVersion": "2026-07-28",
    "clientId": "my-agent"
  }
}

// 服务端返回完整能力清单
{
  "jsonrpc": "2.0",
  "id": 10,
  "result": {
    "protocolVersion": "2026-07-28",
    "capabilities": {
      "tools": {"listChanged": true, "supported": true},
      "resources": {"subscribe": true, "supported": true},
      "prompts": {"supported": true}
    },
    "instructions": "这是filesystem MCP服务，支持读写本地文件..."
  }
}

你甚至可以跳过这个查询，直接发业务请求。如果版本不兼容，服务端会返回400错误，并附带它支持的版本列表：

// 服务端不支持的版本 → 返回400
{
  "jsonrpc": "2.0",
  "id": null,
  "error": {
    "code": -32001,
    "message": "Unsupported protocol version",
    "data": {
      "supportedVersions": ["2026-07-28", "2025-11-25"],
      "received": "2024-01-01"
    }
  }
}

3.3 为什么这样改更好？

第一，按需查询减少了不必要的网络往返。

如果你的AI Agent只用一个工具（比如只调用 filesystem.read），你根本不需要知道服务端支持哪些提示模板、旧版握手的时候这些信息就白白传了。新版你可以只查询你真正关心的能力。

第二，更适合动态注册的服务端。

在MCP生态里，有一个"MCP Directory"的概念——你可以动态发现和连接各种MCP服务端。新版的 server/discover 是一个标准方法，任何服务端实现都必须支持，客户端可以统一用这个方法探测能力，而不需要针对每个服务端写特定的握手适配代码。

第三，支持能力协商的演进。

旧版的能力协商是在握手时一次性完成的，如果服务端后来新增了一个工具，客户端在当前会话里是不知道的（除非重新握手）。新版你可以随时调用 server/discover，动态更新你对服务端能力的认知。

3.4 能力发现的实战代码

import asyncio
import aiohttp
import json

class MCPSimpleClient:
    """新版MCP客户端，演示无状态请求 + 主动能力发现"""
    
    def __init__(self, base_url: str, client_id: str):
        self.base_url = base_url
        self.client_id = client_id
        self._capabilities = None
    
    def _make_request(self, method: str, params: dict = None):
        """构造无状态请求，每个请求都带完整上下文"""
        return {
            "jsonrpc": "2.0",
            "id": id(params) if params else 1,
            "method": method,
            "params": params or {},
            "_meta": {
                "protocolVersion": "2026-07-28",
                "clientId": self.client_id,
                # 新版直接支持OpenTelemetry链路追踪
                "traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
            }
        }
    
    async def discover_capabilities(self, session: aiohttp.ClientSession):
        """主动发现服务端能力，类似旧版的握手"""
        request = self._make_request("server/discover")
        
        async with session.post(
            f"{self.base_url}/mcp",
            json=request,
            headers={
                "Content-Type": "application/json",
                "MCP-Protocol-Version": "2026-07-28",
                "Mcp-Client-Id": self.client_id
            }
        ) as resp:
            result = await resp.json()
            if "error" in result:
                raise Exception(f"能力发现失败: {result['error']}")
            
            self._capabilities = result["result"]["capabilities"]
            print(f"发现服务端能力: {list(self._capabilities.keys())}")
            return self._capabilities
    
    async def call_tool(self, session: aiohttp.ClientSession, 
                        tool_name: str, arguments: dict):
        """调用工具前先检查能力，然后构造无状态请求"""
        if not self._capabilities:
            await self.discover_capabilities(session)
        
        if "tools" not in self._capabilities:
            raise Exception("服务端不支持tools能力")
        
        request = self._make_request(
            "tools/call",
            {"name": tool_name, "arguments": arguments}
        )
        
        # 新增的HTTP头，让网关在解析body之前就能路由和限流
        headers = {
            "Content-Type": "application/json",
            "MCP-Protocol-Version": "2026-07-28",
            "Mcp-Method": "tools/call",
            "Mcp-Name": tool_name,
            "Mcp-Client-Id": self.client_id
        }
        
        async with session.post(
            f"{self.base_url}/mcp",
            json=request,
            headers=headers
        ) as resp:
            result = await resp.json()
            if "error" in result:
                raise Exception(f"工具调用失败: {result['error']}")
            return result["result"]


async def main():
    client = MCPSimpleClient(
        base_url="https://mcp.example.com",
        client_id="my-agent-001"
    )
    
    async with aiohttp.ClientSession() as session:
        # 第一步：主动发现能力（替代旧版握手）
        await client.discover_capabilities(session)
        
        # 第二步：直接调用工具（无状态）
        result = await client.call_tool(
            session, "filesystem.read",
            {"path": "/etc/hostname"}
        )
        print(f"读取结果: {result}")


asyncio.run(main())

四、核心变化三：MRTR模式——服务端主动请求的优雅替代

4.1 旧版：服务端可以"插嘴"

旧版MCP有一个很特别的设计——服务端可以在工具执行过程中主动给客户端发请求。官方叫法是"服务端发起的请求"（server-initiated requests）。

这个机制的技术原理是：服务端在返回结果之前，可以先发一个"需要输入"的请求，客户端必须先处理这个请求，把结果填上，再重新调用原来的工具。

// 旧版：工具执行到一半，服务端主动要求输入
// 服务端返回：
{
  "jsonrpc": "2.0",
  "method": "notifications/message",
  "params": {
    "level": "info",
    "data": {
      "type": "input_required",
      "message": "请提供邮件内容",
      "requestId": "req_123"
    }
  }
}

// 客户端收到后，发送输入：
{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "send_email",
    "arguments": {
      "content": "用户补全的邮件内容"
    },
    "requestId": "req_123"  // 关联到之前的请求
  }
}

这个设计听起来很美好——减少往返次数，提高交互效率。但问题在于：

1. 需要双向通信：服务端要能主动推消息给客户端，这在很多网络拓扑下根本不可行（NAT、防火墙、临时容器）
2. 复杂的状态管理：服务端要维护"这个请求目前卡在哪一步、需要什么输入"的状态
3. 与无状态架构冲突：如果你想让MCP服务端无状态化，这个机制就必须删掉

4.2 新版：MRTR，input_required + 客户端重试

新版把这个机制改成了 MRTR（Multi-Round-Trip Request，多轮往返请求）。

核心思想是：服务端不能主动说话，但它可以告诉你"你需要重新来一次"。

// 新版：工具执行时缺参数，服务端返回 input_required 状态
{
  "jsonrpc": "2.0",
  "id": 42,
  "result": {
    "type": "input_required",
    "message": "send_email工具需要补充邮件内容",
    "requiredFields": [
      {"name": "content", "description": "邮件正文内容"}
    ],
    "operationId": "op_email_001"
  }
}

// 客户端拿到后，向用户请求输入，然后重新调用
{
  "jsonrpc": "2.0",
  "id": 43,
  "method": "tools/call",
  "params": {
    "name": "send_email",
    "arguments": {
      "content": "用户填写的邮件正文内容",
      "operationId": "op_email_001"  // 关联之前的操作
    }
  },
  "_meta": {
    "protocolVersion": "2026-07-28",
    "clientId": "my-agent"
  }
}

4.3 MRTR 的设计哲学：网络往返比状态维护便宜

MRTR 模式有一个很重要的设计哲学：宁可多几次网络往返，也不想维护复杂的状态。

这其实是云原生时代的一个主流选择。像AWS的Step Functions、Azure的 Durable Functions，以及各种工作流引擎，都遵循同一个原则：状态外置，往返内联。把状态存在外部存储（数据库、Redis），让流程处理逻辑变成无状态的函数，每次调用独立执行，靠传入的状态ID来恢复上下文。

MCP新版也走的是这条路：

# 旧版思维：服务端维护复杂状态
class OldMCPServer:
    def __init__(self):
        # 服务端要维护一堆状态
        self.sessions = {}  # session_id → 会话状态
        self.pending_requests = {}  # request_id → 等待的输入
        self.active_operations = {}  # operation_id → 执行到哪一步
    
    def call_tool(self, request):
        # 执行到一半缺参数，修改状态，等待客户端补充
        self.pending_requests[request["id"]] = {
            "tool": "send_email",
            "step": "awaiting_content",
            "partial_args": request["params"]["arguments"]
        }
        # 返回需要输入的指示
        return {"type": "input_required", "message": "需要邮件内容"}


# 新版思维：服务端无状态，客户端维护业务状态
class NewMCPServer:
    def __init__(self):
        # 服务端不需要维护任何状态！
        # 所有状态都在请求的 arguments 里
        pass
    
    def call_tool(self, request):
        args = request["params"]["arguments"]
        
        # 检查必需参数
        if "content" not in args or not args["content"]:
            # 返回 input_required，让客户端重试
            return {
                "type": "input_required",
                "message": "send_email需要content参数",
                "requiredFields": [
                    {"name": "content", "description": "邮件正文"}
                ]
            }
        
        # 参数齐全，正常执行
        return self._do_send_email(args)


# 客户端维护业务状态（operationId）
class MCPAgentClient:
    def __init__(self):
        self.pending_operations = {}  # operation_id → 状态
    
    async def call_tool_with_retry(self, tool_name, args):
        operation_id = f"op_{uuid.uuid4().hex[:8]}"
        args["operationId"] = operation_id
        
        while True:
            result = await self.mcp_server.call_tool(tool_name, args)
            
            if result.get("type") == "input_required":
                # 需要用户/AI补充信息
                missing_fields = result["requiredFields"]
                user_input = await self.prompt_user_for_input(missing_fields)
                args.update(user_input)
                # 重试，同一个operationId
                continue
            else:
                return result

4.4 实际例子：支付确认流程的MRTR实现

"""
真实场景：用MRTR实现支付确认的用户交互流程

旧版：服务端主动发确认请求
新版：服务端返回input_required，客户端重试
"""

async def payment_tool_mcp_flow(mcp_client, order_id: str):
    """
    支付工具的完整MRTR流程
    
    步骤1: Agent调用支付工具，只提供了订单ID
    步骤2: 服务端返回input_required，要求确认金额和支付方式
    步骤3: Agent从用户处获取确认信息
    步骤4: Agent重新调用支付工具，补充完整参数
    步骤5: 服务端执行支付，返回结果
    """
    
    # === 步骤1: 首次调用，只有订单ID ===
    result = await mcp_client.call_tool(
        "payment",
        {"order_id": order_id}
    )
    
    # === 步骤2: 服务端返回 input_required ===
    # result = {
    #     "type": "input_required",
    #     "requiredFields": [
    #         {"name": "amount", "description": "支付金额"},
    #         {"name": "method", "description": "支付方式: alipay|wechat|card"},
    #         {"name": "confirmed", "description": "是否确认支付"}
    #     ]
    # }
    
    if result.get("type") == "input_required":
        fields = result["requiredFields"]
        
        # === 步骤3: 向用户请求确认 ===
        # 这里可以是LLM调用、用户界面、或者自动化逻辑
        confirmed = await ask_user_confirmation(
            f"确认支付订单 {order_id} 吗？"
        )
        
        amount = await fetch_order_amount(order_id)
        payment_method = confirmed.get("preferred_method", "alipay")
        
        # === 步骤4: 补充参数，重试 ===
        result = await mcp_client.call_tool(
            "payment",
            {
                "order_id": order_id,
                "amount": amount,
                "method": payment_method,
                "confirmed": True,
                "operationId": result.get("operationId")  # 关联操作
            }
        )
    
    # === 步骤5: 拿到最终结果 ===
    return result

五、核心变化四：网关友好——请求头标准化与参数镜像

5.1 旧版：网关的噩梦

在旧版MCP里，如果你想在MCP服务端前面加一个API网关（做鉴权、限流、监控、路由），你会遇到一个很尴尬的问题：所有路由和限流的信息都在请求体里。

// 旧版请求：工具名在body里，不在HTTP头里
POST /mcp HTTP/1.1
Content-Type: application/json

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",    // ← 网关想按方法限流？得解析body
  "params": {
    "name": "read_file",     // ← 网关想按工具名限流？还得解析body
    "arguments": {"path": "/data/secret.txt"}
  }
}

这意味着网关必须：

1. 先把body读进来
2. 解析JSON
3. 找到 method 和 params.name
4. 再决定是放行还是拦截

对于高性能网关来说，这个开销不可忽视。而且如果body很大（传了大量文件内容作为参数），解析JSON的开销就更可观了。

5.2 新版：标准HTTP头当"路牌"

新版MCP引入了三个标准HTTP请求头，让网关可以在解析body之前就知道这次请求的关键信息：

MCP-Protocol-Version: 2026-07-28   ← 协议版本
Mcp-Method: tools/call             ← 调用的方法
Mcp-Name: read_file                ← 具体工具名

这三个头信息相当于请求的"外包装标签"，网关不用拆包就知道往哪送：

# 新版MCP网关：先读头，再决定是否放行到后端
from fastapi import FastAPI, Request, HTTPException, Depends
from ratelimit import limits
import aiohttp

app = FastAPI()

# 工具级别的限流规则（per tool）
TOOL_RATE_LIMITS = {
    "read_file": (100, 60),      # 100次/分钟
    "write_file": (20, 60),      # 写操作更少
    "execute_code": (10, 60),    # 执行代码最严格
    "payment": (5, 60),          # 支付操作最严格
}

@app.post("/mcp")
async def mcp_gateway(request: Request):
    # 第一步：读HTTP头（不需要解析body）
    protocol_version = request.headers.get("MCP-Protocol-Version")
    method = request.headers.get("Mcp-Method")
    tool_name = request.headers.get("Mcp-Name")
    client_id = request.headers.get("Mcp-Client-Id")
    
    # 第二步：版本校验（不读body）
    if protocol_version != "2026-07-28":
        raise HTTPException(
            status_code=400,
            detail=f"不支持的协议版本: {protocol_version}"
        )
    
    # 第三步：按工具名限流（不读body）
    if tool_name and tool_name in TOOL_RATE_LIMITS:
        limit, period = TOOL_RATE_LIMITS[tool_name]
        if not await check_rate_limit(client_id, tool_name, limit, period):
            raise HTTPException(status_code=429, detail="限流")
    
    # 第四步：鉴权（不读body）
    if not await verify_client(client_id, request.headers):
        raise HTTPException(status_code=401, detail="未授权")
    
    # 第五步：读body，代理到后端
    body = await request.json()
    
    # 验证头和body的一致性（防伪造）
    if body.get("method") != method or \
       (body.get("params", {}).get("name") != tool_name and tool_name):
        raise HTTPException(status_code=400, detail="头信息与body不一致")
    
    return await proxy_to_backend(body)

5.3 参数镜像：给网关开"透视挂"

有时候光有方法名还不够。比如你想按租户ID做多租户路由和权限控制，但 tenant_id 是工具参数里的字段，不在HTTP头里。

新版引入了 x-mcp-header 参数标记，来解决这个问题：

// 客户端调用：把参数标记为需要镜像到HTTP头
{
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": {
      "tenant_id": {
        "value": "tenant_001",
        "x-mcp-header": "Mcp-Param-Tenant-Id"  // 标记：镜像到HTTP头
      },
      "sql": "SELECT * FROM users LIMIT 10"
    }
  }
}

客户端在发送请求时，会自动把这个字段的值复制到HTTP头：

Mcp-Param-Tenant-Id: tenant_001   ← 自动生成的，来自参数标记

这样网关就可以直接读HTTP头来判断租户身份，而不用解析body：

# 网关的路由逻辑
@app.post("/mcp")
async def mcp_gateway(request: Request):
    tenant_id = request.headers.get("Mcp-Param-Tenant-Id")
    
    # 直接从HTTP头获取租户ID，多租户路由
    if tenant_id:
        # 不同租户路由到不同后端实例
        backend_url = TENANT_BACKENDS.get(tenant_id)
        if not backend_url:
            raise HTTPException(status_code=403, detail="租户不存在")
        
        return await proxy_to_tenant_backend(request, backend_url)

5.4 JSON Schema彻底放开用

旧版MCP虽然基于JSON Schema 2020-12，但限制很多，$ref、oneOf、anyOf、条件Schema这些高级特性用不了。

新版彻底放开了：

// 新版：可以定义复杂的条件Schema
{
  "name": "process_payment",
  "description": "处理支付",
  "inputSchema": {
    "type": "object",
    "oneOf": [
      {
        "properties": {
          "method": {"const": "credit_card"},
          "card_number": {"type": "string", "pattern": "^[0-9]{16}$"},
          "cvv": {"type": "string", "pattern": "^[0-9]{3,4}$"},
          "expiry": {"type": "string"}
        },
        "required": ["method", "card_number", "cvv", "expiry"]
      },
      {
        "properties": {
          "method": {"const": "alipay"},
          "alipay_account": {"type": "string", "format": "email"}
        },
        "required": ["method", "alipay_account"]
      }
    ],
    "if": {
      "properties": {"method": {"const": "credit_card"}}
    },
    "then": {
      "properties": {
        "three_d_secure": {"type": "boolean", "default": true}
      }
    }
  }
}

六、核心变化五：长任务正式"转正"——Tasks扩展的规范化

6.1 旧版：Tasks是实验性功能，各家实现各异

在旧版MCP里，Tasks（长任务） 一直是个实验性功能，没有统一的规范。

什么是长任务？比如：

• 生成一段10分钟的视频
• 训练一个机器学习模型（耗时可能几小时）
• 批量导出一个大数据库（可能需要几十分钟）
• 编译一个大型项目

这些任务的特点是：调用后不能立刻返回结果。旧版MCP没有标准处理方式，各家MCP服务端的实现都不一样——有的用轮询，有的用SSE推送，有的干脆超时了事。

6.2 新版：Tasks成为官方扩展，规范完整

新版把Tasks提升为官方扩展（official extension），有完整的生命周期规范：

// 调用一个长任务
{
  "method": "tasks/start",
  "params": {
    "name": "train_model",
    "input": {
      "dataset": "s3://bucket/training-data.csv",
      "model_type": "xgboost",
      "hyperparameters": {"n_estimators": 1000}
    }
  },
  "_meta": {
    "protocolVersion": "2026-07-28",
    "clientId": "ml-agent-001"
  }
}

// 服务端立即返回任务对象（不阻塞）
{
  "result": {
    "task": {
      "id": "task_abc123",
      "status": "in_progress",
      "name": "train_model",
      "createdAt": "2026-07-28T10:00:00Z",
      "progress": {
        "current": 0,
        "total": 100,
        "unit": "epoch"
      },
      "statusUpdateIntervalMs": 5000  // 建议5秒后查询一次
    }
  }
}

之后客户端按照 statusUpdateIntervalMs 的建议间隔，轮询查询状态：

// 客户端轮询任务状态
{
  "method": "tasks/get",
  "params": {"id": "task_abc123"}
}

// 服务端返回最新状态
{
  "result": {
    "task": {
      "id": "task_abc123",
      "status": "in_progress",
      "progress": {"current": 47, "total": 100, "unit": "epoch"},
      "estimatedTimeRemainingMs": 318000
    }
  }
}

// 任务完成后
{
  "result": {
    "task": {
      "id": "task_abc123",
      "status": "completed",
      "progress": {"current": 100, "total": 100, "unit": "epoch"},
      "output": {
        "model_path": "s3://bucket/models/xgboost-v1.2.0.pkl",
        "metrics": {"accuracy": 0.94, "f1": 0.91}
      }
    }
  }
}

6.3 长任务实战：视频生成MCP服务

"""
用新版MCP Tasks扩展实现一个视频生成服务
"""

from fastapi import FastAPI, Request
from pydantic import BaseModel
import asyncio
import uuid

app = FastAPI()

class VideoGenerationTask(BaseModel):
    prompt: str
    duration_seconds: int
    resolution: str = "1080p"

@app.post("/mcp")
async def mcp_handler(request: Request):
    body = await request.json()
    method = body.get("method")
    
    if method == "tasks/start":
        return await handle_task_start(body)
    elif method == "tasks/get":
        return await handle_task_get(body)
    elif method == "tasks/cancel":
        return await handle_task_cancel(body)
    else:
        return {"error": {"code": -32601, "message": "Method not found"}}


# 任务状态存储（实际生产用Redis）
tasks_db = {}


async def handle_task_start(body: dict):
    params = body.get("params", {})
    task_name = params.get("name")
    task_input = params.get("input", {})
    
    task_id = f"task_{uuid.uuid4().hex[:12]}"
    
    # 创建任务对象
    task_obj = {
        "id": task_id,
        "status": "in_progress",
        "name": task_name,
        "input": task_input,
        "createdAt": "2026-07-28T10:00:00Z",
        "progress": {"current": 0, "total": 100, "unit": "percent"},
        "statusUpdateIntervalMs": 2000
    }
    
    tasks_db[task_id] = task_obj
    
    # 在后台启动真正的生成任务
    asyncio.create_task(run_video_generation(task_id, task_input))
    
    return {
        "jsonrpc": "2.0",
        "id": body.get("id"),
        "result": {"task": task_obj}
    }


async def run_video_generation(task_id: str, task_input: dict):
    """后台运行视频生成（模拟）"""
    total_steps = 100
    
    for step in range(total_steps + 1):
        await asyncio.sleep(0.1)  # 模拟处理时间
        
        task = tasks_db.get(task_id)
        if not task or task["status"] == "cancelled":
            return
        
        task["progress"] = {
            "current": step,
            "total": total_steps,
            "unit": "percent",
            "message": f"渲染帧 {step}/{total_steps}"
        }
        
        # 每20步输出一次详细日志
        if step % 20 == 0:
            print(f"任务 {task_id} 进度: {step}%")
    
    # 任务完成
    task = tasks_db[task_id]
    task["status"] = "completed"
    task["progress"] = {
        "current": 100, "total": 100, "unit": "percent"
    }
    task["output"] = {
        "video_url": f"https://cdn.example.com/videos/{task_id}.mp4",
        "duration_seconds": task_input.get("duration_seconds", 10),
        "resolution": task_input.get("resolution", "1080p"),
        "file_size_mb": 47.3
    }


async def handle_task_get(body: dict):
    task_id = body.get("params", {}).get("id")
    task = tasks_db.get(task_id)
    
    if not task:
        return {
            "jsonrpc": "2.0",
            "id": body.get("id"),
            "error": {"code": -32002, "message": "Task not found"}
        }
    
    return {
        "jsonrpc": "2.0",
        "id": body.get("id"),
        "result": {"task": task}
    }


async def handle_task_cancel(body: dict):
    task_id = body.get("params", {}).get("id")
    task = tasks_db.get(task_id)
    
    if task:
        task["status"] = "cancelled"
        task["cancelledAt"] = "2026-07-28T10:15:00Z"
    
    return {
        "jsonrpc": "2.0",
        "id": body.get("id"),
        "result": {"task": task}
    }

七、扩展框架：插件化能力系统

7.1 旧版：能力是"平铺"的

旧版MCP的能力（tools、resources、prompts、sampling等）是平铺的，没有扩展机制。如果你想要新功能（比如Tasks），要么等官方加入，要么fork代码自己改。

7.2 新版：命名空间扩展框架

新版引入了插件化的扩展框架。每个扩展有自己的命名空间，可以独立演进：

// 新版扩展声明格式
{
  "capabilities": {
    // 核心能力（每个MCP服务端都必须实现）
    "tools": {"supported": true},
    "resources": {"supported": true},
    
    // 扩展能力（可选，需要双方协商）
    "extensions": {
      "tasks": {"version": "1.0", "supported": true},
      "mcp_apps": {"version": "1.0", "supported": true},
      "observability": {"version": "1.0", "supported": true}
    }
  }
}

MCP Apps 扩展是一个很好的例子——它让服务端可以在响应中返回可交互的UI组件（图表、表单、按钮），直接渲染在AI对话界面里：

// 服务端返回MCP Apps组件
{
  "result": {
    "type": "mcp_apps",
    "components": [
      {
        "type": "form",
        "id": "payment-form",
        "fields": [
          {"name": "amount", "type": "number", "label": "金额"},
          {"name": "method", "type": "select", "options": ["alipay", "wechat"]}
        ],
        "onSubmit": {"action": "submit_payment", "confirm": true}
      }
    ]
  }
}

八、弃用清单：这些功能要准备迁移了

新版MCP正式引入了功能生命周期管理——活跃（Active）、弃用（Deprecated）、移除（Removed）。从弃用到移除至少留12个月缓冲期。

这次被标记为弃用的功能：

九、迁移指南：双版本过渡，最稳的路线

9.1 迁移的复杂性评估

这次升级的破坏性（breaking changes）主要体现在：

✅ 删除了 initialize/initialized 握手流程
✅ 删除了 sessionId 管理
✅ 删除了服务端主动请求机制
✅ 改变了通知/订阅机制（两套并一套）
✅ 改变了Tasks的实验性地位（正式化）
⚠️  _meta字段是新增的，旧版服务端不认识
⚠️  新HTTP头是新增的，旧版客户端不发送

如果你有现成的MCP客户端或服务端，迁移复杂度取决于你目前用到了多少旧版特性：

• 只有tools调用：简单，改 _meta 字段，删握手即可
• 用了session状态：需要重构，把状态外置到客户端
• 用了服务端主动请求：需要改成MRTR模式
• 用了采样/roots/logging：需要迁移到对应的新方案

9.2 服务端迁移：兼容旧版的最小改动方案

"""
MCP服务端双版本兼容实现
同时支持2025-11-25（旧版）和2026-07-28（新版）
"""

class DualVersionMCPServer:
    def __init__(self):
        self.clients = {}  # clientId → 客户端信息
    
    async def handle_request(self, request: dict, headers: dict = None):
        """自动检测版本，走对应处理路径"""
        
        # 从_meta或headers获取版本
        protocol_version = (
            request.get("_meta", {}).get("protocolVersion") or
            (headers or {}).get("MCP-Protocol-Version")
        )
        
        if protocol_version == "2026-07-28":
            return await self.handle_v2026(request)
        else:
            return await self.handle_legacy(request)
    
    async def handle_v2026(self, request: dict):
        """新版处理路径：无状态"""
        method = request.get("method")
        
        if method == "server/discover":
            return self._discover_capabilities()
        
        elif method == "tools/call":
            args = request["params"]
            
            # MRTR: 检查必需参数
            if args.get("name") == "send_email":
                if not args.get("arguments", {}).get("content"):
                    return {
                        "type": "input_required",
                        "requiredFields": [
                            {"name": "content", "description": "邮件正文"}
                        ],
                        "operationId": f"op_{request['id']}"
                    }
            
            return await self._execute_tool(args)
        
        elif method == "tasks/start":
            return await self._start_long_task(request)
        
        elif method == "tasks/get":
            return await self._get_task_status(request)
    
    async def handle_legacy(self, request: dict):
        """旧版处理路径：维持原有行为"""
        method = request.get("method")
        
        if method == "initialize":
            session_id = f"sess_{id(request)}"
            self.clients[session_id] = {
                "capabilities": request.get("params", {}).get("capabilities", {}),
                "connected_at": "now"
            }
            return {
                "result": {
                    "protocolVersion": "2025-11-25",
                    "capabilities": self._get_capabilities(),
                    "sessionId": session_id
                }
            }
        
        elif method == "tools/call":
            # 旧版直接执行（不需要MRTR）
            return await self._execute_tool_legacy(request)
        
        # ... 其他旧版方法
    
    def _discover_capabilities(self):
        """新版：返回完整能力清单"""
        return {
            "protocolVersion": "2026-07-28",
            "capabilities": {
                "tools": {"supported": True, "listChanged": True},
                "resources": {"supported": True, "subscribe": True},
                "prompts": {"supported": True},
                "extensions": {
                    "tasks": {"version": "1.0", "supported": True}
                }
            },
            "instructions": "支持tools调用、资源订阅、长任务执行"
        }

9.3 客户端迁移：版本探测 + 降级策略

"""
MCP客户端：版本探测 + 智能降级
"""

class MCPDualClient:
    def __init__(self, server_url: str):
        self.server_url = server_url
        self.version = None
        self.capabilities = None
    
    async def connect(self):
        """
        双版本连接：先试新版，不行降级旧版
        """
        # 策略：优先新版
        test_request = self._make_request("server/discover")
        
        async with aiohttp.ClientSession() as session:
            try:
                resp = await session.post(
                    f"{self.server_url}/mcp",
                    json=test_request,
                    headers={
                        "MCP-Protocol-Version": "2026-07-28",
                        "Mcp-Client-Id": self.client_id
                    },
                    timeout=aiohttp.ClientTimeout(total=5)
                )
                
                if resp.status == 200:
                    result = await resp.json()
                    self.version = "2026-07-28"
                    self.capabilities = result.get("result", {})
                    print(f"✓ 成功连接到MCP服务端（新版 {self.version}）")
                    return
                    
            except Exception as e:
                print(f"新版连接失败，尝试旧版: {e}")
            
            # 降级到旧版：走握手流程
            await self._connect_legacy()
    
    async def _connect_legacy(self):
        """旧版连接：走initialize握手"""
        handshake = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": {"name": "my-agent", "version": "1.0.0"}
            }
        }
        
        async with aiohttp.ClientSession() as session:
            resp = await session.post(
                f"{self.server_url}/mcp",
                json=handshake
            )
            result = await resp.json()
            
            self.version = "2025-11-25"
            self.session_id = result["result"]["sessionId"]
            self.capabilities = result["result"]["capabilities"]
            
            # 发送 initialized 确认
            await session.post(
                f"{self.server_url}/mcp",
                json={"jsonrpc": "2.0", "method": "initialized", "params": {}}
            )
            
            print(f"✓ 降级连接到MCP服务端（旧版 {self.version}）")

9.4 迁移检查清单

□ 服务端：
  □ 添加 _meta 字段解析
  □ 添加新版HTTP头支持（MCP-Protocol-Version, Mcp-Method, Mcp-Name）
  □ 实现 server/discover 方法
  □ 把工具调用的参数校验改成返回 input_required（MRTR）
  □ 如果用了服务端主动请求 → 改成MRTR模式
  □ 如果用了Sampling/Roots → 评估迁移到客户端实现
  □ 添加 OpenTelemetry traceparent 传递
  □ 实现双版本兼容（旧版客户端也要服务）

□ 客户端：
  □ 删除 initialize/initialized 调用
  □ 删掉 sessionId 管理逻辑
  □ 每次请求添加 _meta 字段
  □ 实现 server/discover 能力发现
  □ 实现 MRTR 重试逻辑
  □ 添加新版HTTP头
  □ 实现版本探测和降级策略
  □ 业务状态外置（不依赖协议层session）

十、总结：MCP的无状态化对AI Agent生态意味着什么

10.1 核心收获

回顾这次MCP大改版，五个核心变化都指向同一个方向：

10.2 对AI Agent生态的影响

MCP这次升级的影响远不止"改改代码"这么简单。它代表了一个更深层的趋势：

AI Agent的协议栈正在从"对等通信"向"客户端-服务端"范式收敛。

过去的AI Agent框架倾向于让Agent和工具之间建立对等的双向通信——Agent可以调用工具，工具也可以回调Agent。但随着AI Agent进入生产环境，这种对等模型的弊端越来越明显：

• 网络拓扑复杂（NAT、防火墙、容器环境）
• 水平扩展困难（状态分布在多个组件间）
• 可观测性差（谁调用谁不清楚）

MCP新版的"无状态客户端驱动"模型，本质上是把Agent-工具关系拉回到了经典的"客户端-服务端"架构：客户端（Agent）发请求，服务端（工具）回响应，所有状态由客户端管理。

这个选择务实、成熟、工业化。

10.3 迁移时机建议

现在（2026年7-8月）：
  → 新项目：直接用新版（2026-07-28）
  → 现有项目：开始评估影响范围，制定迁移计划

6个月内（2026年底）：
  → 现有项目：完成双版本兼容实现
  → 监控MCP生态：留意主流框架（LangChain、AutoGen等）的升级支持

12个月后（2027年下半年）：
  → 可能开始有旧版功能移除的公告
  → 开始逐步淘汰旧版实现

最后一句话：这次升级的破坏性确实大，但它解决的是生产环境真正卡脖子的架构问题。早迁早解脱，晚迁要还债。

附录：完整请求/响应格式对照表

A. 初始化握手对比

B. 工具调用对比

C. 扩展机制对比

参考来源：

• MCP Protocol 2026-07-28 Specification
• Cloudflare MCP Changelog (2026-07-28 support)
• AgentCore Gateway MCP 2026-07-28 Implementation
• Anthropic MCP官方文档
• 各主流MCP服务端的升级适配文档