MCP 协议深度拆解：当 Anthropic 决定「把有状态连接扔进垃圾桶」——从双工会话到无状态核心，一个被 Linux 基金会托管的 AI 工具协议如何用「请求即自包含」重新定义 Agent 基础设施的终极形态

2026 年 7 月 28 日，Anthropic 联合 Linux 基金会发布了 MCP 2026-07-28 规范——这是该协议自 2024 年 11 月诞生以来规模最大、最颠覆性的一次修订。核心变化只有一个词：无状态。本文将从架构层面深度拆解这次更新的每一个技术细节，附完整代码实战，帮你理解为什么 MCP 正在从一个「本地小工具」蜕变为「企业级 Agent 基础设施」。

一、背景：MCP 的前世今生

1.1 什么是 MCP？

MCP（Model Context Protocol，模型上下文协议）是 Anthropic 于 2024 年 11 月推出的一个开放标准协议，用于规范化 AI 系统与外部工具及数据源的交互方式。简单来说，它是 AI Agent 的「USB 接口」——不管你的 Agent 是 Claude、GPT 还是本地 Llama，不管你要操作数据库、调 API 还是读文件，MCP 提供了一个统一的协议层来连接这一切。

MCP 的架构基于经典的 Client-Host-Server 三元组：

┌─────────────────────────────────────────────┐
│              Application Host               │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐     │
│  │ Client1 │  │ Client2 │  │ Client3 │     │
│  └────┬────┘  └────┬────┘  └────┬────┘     │
└───────┼────────────┼────────────┼───────────┘
        │            │            │
   ┌────▼────┐  ┌────▼────┐  ┌────▼────┐
   │ Server1 │  │ Server2 │  │ Server3 │
   │ (Files) │  │  (DB)   │  │ (APIs)  │
   └─────────┘  └─────────┘  └─────────┘

• Host：宿主应用（如 Claude Desktop、Cursor、自研 IDE），创建并管理多个 Client 实例
• Client：协议连接器，每个 Client 与一个 Server 保持 1:1 的通信关系
• Server：提供具体能力的服务（文件系统、数据库、API 网关等）

1.2 为什么需要这次大更新？

MCP 最初的设计定位是「本地工具」——让 Claude Desktop 能读本地文件、操作本地数据库。在这个场景下，有状态的双向连接完全够用：Client 启动一个 stdio 进程和 Server 通信，进程活着就一直有状态。

但问题来了。当 MCP 要走向企业级部署时，有状态架构暴露了致命缺陷：

• 无法横向扩展：每个请求绑定到特定的服务器实例，负载均衡器没法玩
• 连接脆弱性：一个 WebSocket 断了，整个会话上下文就丢了
• 资源消耗：每个连接都持有状态，内存和连接池开销随用户数线性增长
• 运维复杂度：有状态服务器的滚动更新、故障转移都极其困难

正如 MCP 首席维护者 David Soria Parra（Anthropic）所说：

「无状态协议核心是开发者呼声最高的功能之一。他们迫切希望借此提升 MCP 服务器的可靠性与可扩展性。」

二、核心变化一：从有状态到无状态

2.1 旧架构的问题

在 2025-11-25 版本（旧规范）中，MCP 是一个有状态的双向协议：

# 旧架构（有状态）
Client ──── initialize ────► Server
         │                  │
         │    capabilities  │
         │◄─────────────────│
         │                  │
Client ──── tools/call ────► Server    ← 依赖上面的 initialize 状态
         │                  │
         │    result        │
         │◄─────────────────│

问题在于：Server 需要「记住」initialize 阶段协商好的协议版本、客户端能力、会话上下文。这意味着：

1. 请求 A 和请求 B 必须路由到同一个 Server 实例
2. 如果实例重启，所有会话上下文丢失
3. 无法在请求之间自由切换 Server 实例

2.2 新架构：请求即自包含

2026-07-28 版 MCP 的核心设计哲学是：每个请求都携带处理它所需的所有信息。

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": { "sql": "SELECT * FROM users LIMIT 10" },
    "_meta": {
      "io.modelcontextprotocol/protocolVersion": "2026-07-28",
      "io.modelcontextprotocol/clientInfo": {
        "name": "MyAgent",
        "version": "1.2.0"
      },
      "io.modelcontextprotocol/clientCapabilities": {
        "extensions": {
          "io.modelcontextprotocol/tasks": {}
        }
      }
    }
  }
}

注意看 _meta 字段——协议版本、客户端信息、能力声明全部内联在每个请求中。Server 不需要「记住」任何东西，因为每次请求都是自包含的。

规范原文的定义非常明确：

MCP 是一个无状态协议：处理请求所需的所有信息都包含在请求本身中。Server 独立处理每个请求；不应从之前的请求中推断状态，即使这些请求来自同一个连接或流。

具体约束：

2.3 代码实战：构建无状态 MCP Server

以下是一个 Python 实现的无状态 MCP Server 示例：

# stateless_mcp_server.py
import json
import uuid
from http.server import HTTPServer, BaseHTTPRequestHandler
from typing import Any

# 模拟的工具注册表
TOOLS = {
    "query_database": {
        "description": "Execute a SQL query against the database",
        "inputSchema": {
            "type": "object",
            "properties": {
                "sql": {"type": "string", "description": "SQL query to execute"}
            },
            "required": ["sql"]
        }
    },
    "read_file": {
        "description": "Read contents of a file",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path to read"}
            },
            "required": ["path"]
        }
    }
}

def handle_request(request: dict) -> dict:
    """
    无状态请求处理器：每次调用都是独立的，不依赖任何外部状态。
    所有需要的信息都在 request 中。
    """
    method = request.get("method")
    params = request.get("params", {})
    meta = params.get("_meta", {})
    
    # 从每个请求中提取协议版本和客户端能力
    protocol_version = meta.get(
        "io.modelcontextprotocol/protocolVersion"
    )
    client_info = meta.get("io.modelcontextprotocol/clientInfo", {})
    client_capabilities = meta.get(
        "io.modelcontextprotocol/clientCapabilities", {}
    )
    
    # 验证协议版本
    if protocol_version != "2026-07-28":
        return {
            "jsonrpc": "2.0",
            "id": request.get("id"),
            "error": {
                "code": -32022,
                "message": f"Unsupported protocol version: {protocol_version}"
            }
        }
    
    # 根据方法分派处理
    if method == "tools/call":
        tool_name = params.get("name")
        arguments = params.get("arguments", {})
        
        if tool_name not in TOOLS:
            return {
                "jsonrpc": "2.0",
                "id": request.get("id"),
                "error": {
                    "code": -32602,
                    "message": f"Unknown tool: {tool_name}"
                }
            }
        
        # 执行工具逻辑（无状态，不依赖之前的请求）
        result = execute_tool(tool_name, arguments)
        
        return {
            "jsonrpc": "2.0",
            "id": request.get("id"),
            "result": {
                "resultType": "complete",
                "content": result
            }
        }
    
    elif method == "server/discover":
        # 能力发现：返回服务器支持的版本和能力
        return {
            "jsonrpc": "2.0",
            "id": request.get("id"),
            "result": {
                "resultType": "complete",
                "serverInfo": {
                    "name": "MyStatelessServer",
                    "version": "1.0.0"
                },
                "capabilities": {
                    "tools": list(TOOLS.keys()),
                    "extensions": {}
                },
                "supportedVersions": ["2026-07-28"]
            }
        }
    
    else:
        return {
            "jsonrpc": "2.0",
            "id": request.get("id"),
            "error": {
                "code": -32601,
                "message": f"Method not found: {method}"
            }
        }


def execute_tool(name: str, args: dict) -> Any:
    """工具执行器（示例逻辑）"""
    if name == "query_database":
        sql = args.get("sql", "")
        # 实际实现中这里会连接数据库执行查询
        return {"rows": [{"id": 1, "name": "example"}], "affected": 1}
    
    elif name == "read_file":
        path = args.get("path", "")
        # 实际实现中这里会读取文件
        return {"content": f"Contents of {path}", "size": 1024}
    
    return None


class MCPHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length)
        
        try:
            request = json.loads(body)
            response = handle_request(request)
            
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps(response).encode())
        except json.JSONDecodeError:
            self.send_response(400)
            self.end_headers()
    
    def log_message(self, format, *args):
        pass  # 静默日志


if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", 8080), MCPHandler)
    print("MCP Server running on port 8080 (stateless mode)")
    server.serve_forever()

关键设计要点：

1. 每个请求都携带 _meta：协议版本、客户端信息、能力声明
2. 不维护任何连接级状态：没有 session 对象、没有会话存储
3. 工具执行纯函数化：相同输入 → 相同输出，不依赖历史请求

2.4 Go 语言实现

// mcp_handler.go
package main

import (
    "encoding/json"
    "fmt"
    "net/http"
)

type MCPRequest struct {
    JSONRPC string      `json:"jsonrpc"`
    ID      interface{} `json:"id"`
    Method  string      `json:"method"`
    Params  struct {
        Name      string                 `json:"name"`
        Arguments map[string]interface{} `json:"arguments"`
        Meta      map[string]interface{} `json:"_meta"`
    } `json:"params"`
}

type MCPResponse struct {
    JSONRPC string      `json:"jsonrpc"`
    ID      interface{} `json:"id"`
    Result  interface{} `json:"result,omitempty"`
    Error   interface{} `json:"error,omitempty"`
}

func handleMCP(w http.ResponseWriter, r *http.Request) {
    var req MCPRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, "Invalid JSON", 400)
        return
    }

    // 从 _meta 中提取每个请求的上下文（无状态！）
    meta := req.Params.Meta
    protocolVersion := meta["io.modelcontextprotocol/protocolVersion"]
    
    if protocolVersion != "2026-07-28" {
        resp := MCPResponse{
            JSONRPC: "2.0",
            ID:      req.ID,
            Error: map[string]interface{}{
                "code":    -32022,
                "message": fmt.Sprintf("Unsupported version: %v", protocolVersion),
            },
        }
        json.NewEncoder(w).Encode(resp)
        return
    }

    // 处理请求（纯函数，无外部状态）
    switch req.Method {
    case "tools/call":
        result := executeTool(req.Params.Name, req.Params.Arguments)
        resp := MCPResponse{
            JSONRPC: "2.0",
            ID:      req.ID,
            Result: map[string]interface{}{
                "resultType": "complete",
                "content":    result,
            },
        }
        json.NewEncoder(w).Encode(resp)
    
    case "server/discover":
        resp := MCPResponse{
            JSONRPC: "2.0",
            ID:      req.ID,
            Result: map[string]interface{}{
                "resultType": "complete",
                "serverInfo": map[string]interface{}{
                    "name":    "GoStatelessServer",
                    "version": "1.0.0",
                },
                "capabilities": map[string]interface{}{
                    "tools": []string{"query_database", "read_file"},
                },
                "supportedVersions": []string{"2026-07-28"},
            },
        }
        json.NewEncoder(w).Encode(resp)
    }
}

func executeTool(name string, args map[string]interface{}) interface{} {
    switch name {
    case "query_database":
        return map[string]interface{}{
            "rows":     []map[string]interface{}{{"id": 1}},
            "affected": 1,
        }
    case "read_file":
        path, _ := args["path"].(string)
        return map[string]interface{}{
            "content": fmt.Sprintf("File: %s", path),
            "size":    1024,
        }
    }
    return nil
}

三、核心变化二：基于请求头的路由

3.1 旧方案的痛点

旧版本的 MCP Client-Server 通信依赖连接绑定：

# 旧方案：连接绑定
Client ──[WebSocket]──► Load Balancer ──► Server Instance 1
                                          Server Instance 2
                                          Server Instance 3

Client 一旦连接到某个 Server 实例，就一直绑定在那个实例上。Load Balancer 只能在初始连接时做选择，后续请求无法路由。

3.2 新方案：Header-Based Routing

新规范引入了基于 HTTP 请求头的路由机制，配合无状态架构，彻底解耦了请求和实例：

POST /mcp HTTP/1.1
Host: mcp.example.com
Content-Type: application/json
Authorization: Bearer eyJhbGciOiJSUzI1NiIs...
X-MCP-Protocol-Version: 2026-07-28
X-MCP-Client-ID: my-agent-v2

{
  "jsonrpc": "2.0",
  "id": "req-042",
  "method": "tools/call",
  "params": {
    "name": "deploy_service",
    "arguments": {"service": "api-gateway", "env": "staging"},
    "_meta": {
      "io.modelcontextprotocol/protocolVersion": "2026-07-28",
      "io.modelcontextprotocol/clientCapabilities": {
        "extensions": {"io.modelcontextprotocol/tasks": {}}
      }
    }
  }
}

这意味着：

1. 任何实例都能处理任何请求：只要请求头正确，Server 实例可以随意替换
2. 负载均衡真正有效：每一轮请求都可以路由到不同实例
3. 故障转移零感知：某个实例挂了，下一个请求自动路由到其他实例
4. 滚动更新无中断：更新 Server 实例时，请求可以无缝切换

3.3 Nginx 配置示例

upstream mcp_servers {
    least_conn;
    server 10.0.0.1:8080;
    server 10.0.0.2:8080;
    server 10.0.0.3:8080;
}

server {
    listen 443 ssl;
    server_name mcp.example.com;

    location /mcp {
        # 无状态 MCP：每个请求独立路由
        proxy_pass http://mcp_servers;
        
        # 传递关键请求头
        proxy_set_header X-MCP-Protocol-Version $http_x_mcp_protocol_version;
        proxy_set_header X-MCP-Client-ID $http_x_mcp_client_id;
        proxy_set_header Authorization $http_authorization;
        proxy_set_header Host $host;
        
        # 超时配置（无状态请求通常很快）
        proxy_connect_timeout 5s;
        proxy_read_timeout 30s;
        
        # 禁用连接复用（无状态不需要）
        proxy_http_version 1.1;
        proxy_set_header Connection "";
    }
}

四、核心变化三：授权框架全面升级

4.1 旧方案的局限

旧版本的 MCP 授权实现比较原始——大部分 Server 用简单的 API Key 或环境变量，不适合企业级场景。

4.2 新规范：完整 OAuth 2.1 支持

2026-07-28 版 MCP 引入了完整的 OAuth 2.1 授权框架，基于以下 RFC 标准：

• OAuth 2.1 (draft-ietf-oauth-v2-1-13)
• RFC 9728：Protected Resource Metadata
• RFC 8414：Authorization Server Metadata
• RFC 9207：Authorization Server Issuer Identification
• OpenID Connect Discovery 1.0

完整授权流程如下：

┌─────────┐     ┌──────────┐     ┌──────────────┐     ┌──────────────┐
│ Browser │     │  Client   │     │  MCP Server  │     │ Auth Server  │
└────┬────┘     └─────┬─────┘     └──────┬───────┘     └──────┬───────┘
     │                │                   │                     │
     │  1. MCP请求(无token)               │                     │
     │ ──────────────►│ ─────────────────►│                     │
     │                │                   │                     │
     │                │  2. 401 + WWW-Authenticate               │
     │                │ ◄─────────────────│                     │
     │                │                   │                     │
     │                │  3. 获取 Protected Resource Metadata     │
     │                │ ─────────────────►│                     │
     │                │ ◄─────────────────│                     │
     │                │                   │                     │
     │                │  4. 获取 Auth Server Metadata            │
     │                │ ────────────────────────────────────────►│
     │                │ ◄────────────────────────────────────────│
     │                │                   │                     │
     │  5. 打开浏览器进行授权             │                     │
     │ ◄──────────────│                   │                     │
     │ ─────────────────────────────────────────────────────────►│
     │ ◄────────────────────────────────────────────────────────│
     │                │                   │                     │
     │  6. 回调返回授权码                 │                     │
     │ ──────────────►│                   │                     │
     │                │  7. 用授权码换取token                    │
     │                │ ────────────────────────────────────────►│
     │                │ ◄────────────────────────────────────────│
     │                │                   │                     │
     │                │  8. 带token的MCP请求                     │
     │                │ ─────────────────►│                     │
     │                │ ◄─────────────────│                     │

4.3 代码实战：OAuth 2.1 授权集成

# mcp_auth.py
import httpx
import json
from urllib.parse import urlencode, urlparse, parse_qs

class MCPAuthManager:
    """MCP OAuth 2.1 授权管理器"""
    
    def __init__(self, server_url: str):
        self.server_url = server_url
        self.access_token = None
        self.refresh_token = None
    
    async def discover_auth_server(self) -> dict:
        """Step 1: 通过 Protected Resource Metadata 发现授权服务器"""
        async with httpx.AsyncClient() as client:
            # RFC 9728: Protected Resource Metadata
            metadata_url = f"{self.server_url}/.well-known/oauth-protected-resource"
            resp = await client.get(metadata_url)
            resource_metadata = resp.json()
            
            # 获取授权服务器 URL
            auth_server_url = resource_metadata["authorization_servers"][0]
            
            # RFC 8414: Authorization Server Metadata
            as_metadata_url = f"{auth_server_url}/.well-known/oauth-authorization-server"
            resp = await client.get(as_metadata_url)
            as_metadata = resp.json()
            
            return as_metadata
    
    async def get_access_token(self, auth_code: str, 
                                code_verifier: str,
                                as_metadata: dict) -> dict:
        """Step 2: 用授权码换取访问令牌"""
        async with httpx.AsyncClient() as client:
            token_data = {
                "grant_type": "authorization_code",
                "code": auth_code,
                "code_verifier": code_verifier,
                "resource": self.server_url,  # RFC 8707 Resource Indicators
            }
            
            resp = await client.post(
                as_metadata["token_endpoint"],
                data=token_data,
                auth=(as_metadata.get("client_id"), None)
            )
            
            tokens = resp.json()
            self.access_token = tokens["access_token"]
            self.refresh_token = tokens.get("refresh_token")
            
            return tokens
    
    async def make_mcp_request(self, method: str, 
                                params: dict) -> dict:
        """Step 3: 带授权头的 MCP 请求"""
        request_body = {
            "jsonrpc": "2.0",
            "id": "req-001",
            "method": method,
            "params": {
                **params,
                "_meta": {
                    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                    "io.modelcontextprotocol/clientCapabilities": {
                        "extensions": {
                            "io.modelcontextprotocol/tasks": {}
                        }
                    }
                }
            }
        }
        
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.server_url}/mcp",
                json=request_body,
                headers={
                    "Authorization": f"Bearer {self.access_token}",
                    "Content-Type": "application/json",
                    "X-MCP-Protocol-Version": "2026-07-28"
                }
            )
            
            if resp.status_code == 401:
                # 处理 scope challenge（升级授权）
                www_auth = resp.headers.get("WWW-Authenticate", "")
                required_scope = self._extract_scope(www_auth)
                raise ScopeChallengeError(required_scope)
            
            return resp.json()
    
    def _extract_scope(self, www_auth: str) -> str:
        """从 WWW-Authenticate 头中提取所需 scope"""
        parts = www_auth.split(",")
        for part in parts:
            if "scope=" in part:
                return part.split("scope=")[1].strip('"')
        return ""

五、核心变化四：Tasks 扩展——异步长任务

5.1 为什么需要 Tasks？

在有状态架构中，长任务（CI 流水线、批量数据处理、人工审批）可以通过保持连接来跟踪进度。但在无状态架构下，每个请求都是独立的——你需要一个机制来「跨请求」跟踪任务状态。

Tasks 扩展就是这个机制。

5.2 Tasks 工作流程

Client                          Server
  │                               │
  │  tools/call (long operation)  │
  │ ─────────────────────────────►│
  │                               │  创建持久任务
  │  CreateTaskResult             │
  │  {taskId, status: working,    │
  │   pollIntervalMs: 2000}       │
  │ ◄─────────────────────────────│
  │                               │
  │  tasks/get (taskId)           │
  │ ─────────────────────────────►│
  │  Task {status: working}       │
  │ ◄─────────────────────────────│
  │                               │
  │  tasks/get (taskId)           │
  │ ─────────────────────────────►│
  │  Task {status: input_required,│
  │        inputRequests: {...}}   │
  │ ◄─────────────────────────────│
  │                               │
  │  tasks/update (inputResponses)│
  │ ─────────────────────────────►│
  │  ack                          │
  │ ◄─────────────────────────────│
  │                               │
  │  tasks/get (taskId)           │
  │ ─────────────────────────────►│
  │  Task {status: completed,     │
  │        result: {...}}         │
  │ ◄─────────────────────────────│

5.3 Tasks 状态机

5.4 代码实战：Tasks Server 实现

# mcp_tasks_server.py
import uuid
import time
import threading
from typing import Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum

class TaskStatus(Enum):
    WORKING = "working"
    INPUT_REQUIRED = "input_required"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

@dataclass
class Task:
    task_id: str
    status: TaskStatus
    result: Optional[Dict] = None
    error: Optional[Dict] = None
    input_requests: Optional[Dict] = None
    created_at: float = field(default_factory=time.time)
    ttl_ms: int = 300000  # 5分钟 TTL
    poll_interval_ms: int = 2000

class TaskManager:
    """无状态 Task 管理器
    
    关键设计：Task 状态存储在外部存储（Redis/数据库）中，
    Server 实例本身不持有任何状态，完全通过 taskId 引用。
    """
    
    def __init__(self):
        self.tasks: Dict[str, Task] = {}
        self._lock = threading.Lock()
    
    def create_task(self, request_id: Any) -> Task:
        """创建新任务并返回 task handle"""
        task_id = f"task-{uuid.uuid4().hex[:12]}"
        task = Task(
            task_id=task_id,
            status=TaskStatus.WORKING,
        )
        
        with self._lock:
            self.tasks[task_id] = task
        
        return task
    
    def get_task(self, task_id: str) -> Optional[Task]:
        """查询任务状态（无状态 Server 的核心操作）"""
        with self._lock:
            return self.tasks.get(task_id)
    
    def update_task_input(self, task_id: str, 
                          input_responses: Dict) -> bool:
        """处理客户端输入响应"""
        with self._lock:
            task = self.tasks.get(task_id)
            if not task or task.status != TaskStatus.INPUT_REQUIRED:
                return False
            
            # 将输入传递给工作线程继续处理
            task.status = TaskStatus.WORKING
            task.input_requests = None
            return True
    
    def complete_task(self, task_id: str, result: Dict):
        """标记任务完成"""
        with self._lock:
            task = self.tasks.get(task_id)
            if task:
                task.status = TaskStatus.COMPLETED
                task.result = result
    
    def fail_task(self, task_id: str, error: Dict):
        """标记任务失败"""
        with self._lock:
            task = self.tasks.get(task_id)
            if task:
                task.status = TaskStatus.FAILED
                task.error = error


# 使用示例
task_manager = TaskManager()

def handle_tools_call(request: dict) -> dict:
    """处理 tools/call 请求，可能返回 task handle"""
    tool_name = request["params"]["name"]
    
    if is_long_running_tool(tool_name):
        # 创建异步任务
        task = task_manager.create_task(request["id"])
        
        # 在后台线程中执行（实际中可能发到消息队列）
        threading.Thread(
            target=execute_long_tool,
            args=(task.task_id, request["params"]["arguments"])
        ).start()
        
        return {
            "jsonrpc": "2.0",
            "id": request["id"],
            "result": {
                "resultType": "task",
                "taskId": task.task_id,
                "status": "working",
                "ttlMs": task.ttl_ms,
                "pollIntervalMs": task.poll_interval_ms
            }
        }
    else:
        # 普通同步工具
        return execute_sync_tool(request)


def handle_tasks_get(request: dict) -> dict:
    """处理 tasks/get 轮询"""
    task_id = request["params"]["taskId"]
    task = task_manager.get_task(task_id)
    
    if not task:
        return {
            "jsonrpc": "2.0",
            "id": request["id"],
            "error": {"code": -32602, "message": "Task not found"}
        }
    
    result = {
        "taskId": task.task_id,
        "status": task.status.value,
    }
    
    if task.status == TaskStatus.COMPLETED:
        result["result"] = task.result
    elif task.status == TaskStatus.FAILED:
        result["error"] = task.error
    elif task.status == TaskStatus.INPUT_REQUIRED:
        result["inputRequests"] = task.input_requests
    
    return {
        "jsonrpc": "2.0",
        "id": request["id"],
        "result": {"resultType": "complete", **result}
    }

六、核心变化五：版本化扩展框架

6.1 为什么需要扩展？

MCP 核心协议保持极简——JSON-RPC 2.0 消息格式、基本的工具/资源/提示模板。但企业级场景需要更多能力：

• 长任务异步执行 → Tasks 扩展
• 交互式 UI 元素 → MCP Apps 扩展
• 结构化 Agent 指令 → Skills over MCP 扩展

6.2 扩展协商机制

扩展采用显式协商模式——Client 和 Server 都必须声明支持：

// Client 在每个请求中声明支持的扩展
{
  "_meta": {
    "io.modelcontextprotocol/clientCapabilities": {
      "extensions": {
        "io.modelcontextprotocol/tasks": {},
        "com.example/custom-extension": {"version": "1.0"}
      }
    }
  }
}

// Server 在 server/discover 中声明支持的扩展
{
  "capabilities": {
    "extensions": {
      "io.modelcontextprotocol/tasks": {},
      "io.modelcontextprotocol/apps": {}
    }
  }
}

6.3 MCP Apps 扩展

MCP Apps 允许 Server 在对话中渲染交互式 UI 元素：

{
  "jsonrpc": "2.0",
  "id": "resp-001",
  "result": {
    "resultType": "complete",
    "content": {
      "text": "Here's the deployment status:"
    },
    "extensions": {
      "io.modelcontextprotocol/apps": {
        "type": "chart",
        "data": {
          "labels": ["Mon", "Tue", "Wed", "Thu", "Fri"],
          "datasets": [{
            "label": "Deployments",
            "data": [12, 19, 8, 15, 22]
          }]
        }
      }
    }
  }
}

七、性能优化与生产部署

7.1 无状态架构的性能优势

7.2 生产环境部署架构

                    ┌──────────────┐
                    │  CDN/WAF     │
                    └──────┬───────┘
                           │
                    ┌──────▼───────┐
                    │ Load Balancer│
                    │ (Nginx/HAProxy)│
                    └──────┬───────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
        ┌─────▼─────┐ ┌───▼─────┐ ┌───▼─────┐
        │ MCP Server│ │MCP Server│ │MCP Server│
        │   (Pod 1) │ │ (Pod 2) │ │ (Pod 3) │
        └─────┬─────┘ └────┬────┘ └────┬────┘
              │            │            │
              └────────────┼────────────┘
                           │
                    ┌──────▼───────┐
                    │  Redis/DB    │
                    │ (Task State) │
                    └──────────────┘

关键配置：

# Kubernetes Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-server
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0  # 无状态服务可以做到零停机更新
  template:
    spec:
      containers:
      - name: mcp-server
        image: mcp-server:latest
        env:
        - name: MCP_PROTOCOL_VERSION
          value: "2026-07-28"
        - name: TASK_STORAGE_URL
          value: "redis://redis-cluster:6379"
        resources:
          requests:
            memory: "128Mi"  # 无状态服务内存需求大幅降低
            cpu: "100m"

7.3 缓存策略

无状态架构下的缓存优化：

# 缓存 server/discover 结果（因为它是幂等的）
import functools
import time

DISCOVER_CACHE = {}

def cache_discover(ttl_seconds=300):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = "discover"
            now = time.time()
            
            if cache_key in DISCOVER_CACHE:
                cached_time, cached_result = DISCOVER_CACHE[cache_key]
                if now - cached_time < ttl_seconds:
                    return cached_result
            
            result = await func(*args, **kwargs)
            DISCOVER_CACHE[cache_key] = (now, result)
            return result
        return wrapper
    return decorator

@cache_discover(ttl_seconds=300)
async def handle_discover(request: dict) -> dict:
    """幂等的发现请求，可以安全缓存"""
    return {
        "jsonrpc": "2.0",
        "id": request["id"],
        "result": {
            "resultType": "complete",
            "serverInfo": {"name": "CachedServer", "version": "1.0.0"},
            "capabilities": {
                "tools": ["query", "deploy", "monitor"],
                "extensions": {"io.modelcontextprotocol/tasks": {}}
            }
        }
    }

八、从 2025-11-25 到 2026-07-28：完整对比

九、迁移指南：从旧版升级到新版

9.1 Server 迁移步骤

# 迁移前（旧版：有状态）
class OldMCPServer:
    def __init__(self):
        self.sessions = {}  # ❌ 有状态：需要维护会话
    
    def handle_request(self, request, connection_id):
        if connection_id not in self.sessions:
            self.sessions[connection_id] = self._init_session(request)
        session = self.sessions[connection_id]
        return self._process(request, session)

# 迁移后（新版：无状态）
class NewMCPServer:
    def __init__(self):
        self.tools = {}  # ✅ 无状态：只注册工具
        self._register_tools()
    
    def handle_request(self, request):
        # 从每个请求中提取上下文
        meta = request["params"].get("_meta", {})
        version = meta.get("io.modelcontextprotocol/protocolVersion")
        client_caps = meta.get("io.modelcontextprotocol/clientCapabilities", {})
        
        # 每个请求独立处理，不依赖外部状态
        return self._process(request, version, client_caps)

9.2 Client 迁移步骤

// 迁移前（旧版）
const client = new MCPClient();
await client.connect(serverUrl);  // ❌ 有状态连接
const result = await client.callTool("query", {sql: "..."});

// 迁移后（新版）
const client = new MCPClient({ version: "2026-07-28" });
// 无需长期连接，每个请求自包含
const result = await client.callTool("query", {
  sql: "...",
  _meta: {
    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
    "io.modelcontextprotocol/clientCapabilities": {
      extensions: { "io.modelcontextprotocol/tasks": {} }
    }
  }
});

十、总结与展望

MCP 2026-07-28 规范的核心转变——从有状态到无状态——不是一次简单的技术升级，而是一次架构范式的迁移。

这次更新解决了什么？

1. 企业级可扩展性：无状态架构让 MCP Server 可以像普通 Web 服务一样水平扩展
2. 可靠性提升：请求自包含意味着任何一个 Server 实例挂掉都不会影响整体服务
3. 生态统一：OAuth 2.1 授权框架让 MCP 可以对接 Entra、Okta 等企业身份系统
4. 能力扩展：Tasks 和 MCP Apps 扩展让 MCP 从「工具调用协议」进化为「Agent 基础设施」

对开发者意味着什么？

• Server 开发者：你的 MCP Server 现在可以部署在 Kubernetes 上，用 Nginx 做负载均衡，用 Redis 存任务状态
• Client 开发者：你的 AI 应用可以自由切换 MCP Server 实例，不再被连接绑定
• 平台方：MCP 可以作为 SaaS 产品的「AI 插件标准」，对接企业级身份系统
• 开源社区：扩展框架让社区可以在不修改核心协议的情况下增加新能力

下一步

MCP 的下一步演进方向可能包括：

1. Skills over MCP：让 Agent 的「技能文档」可以通过 MCP 发现和分发
2. 跨 Agent 互操作：不同 Agent 框架之间的 MCP Server 共享
3. 边缘部署：在 CDN 边缘节点部署轻量级 MCP Server
4. WebAssembly 支持：用 WASM 编写跨平台 MCP Server

正如 Linux 基金会智能体 AI 基金会（AAIF）所展示的：MCP 正在从 Anthropic 的「内部协议」变成整个 AI 行业的「公共基础设施」。当 OpenAI、Google、Microsoft、Amazon 都在同一个协议上构建 Agent 生态时，MCP 的无状态架构已经不是「可选项」——它是「必选项」。

参考资源：

• MCP 2026-07-28 官方规范：https://modelcontextprotocol.io/specification/2026-07-28
• Tasks 扩展规范：https://modelcontextprotocol.io/extensions/tasks/overview
• 授权框架：https://modelcontextprotocol.io/specification/2026-07-28/basic/authorization
• GitHub 仓库：https://github.com/modelcontextprotocol/specification