MCP 2026-07-28 规范深度解剖：从「工具调用协议」到「生产级 Agent 基础设施」的最大升级

前言

2026年5月，Model Context Protocol（MCP）官方发布了 2026-07-28 规范候选版。官方罕见地将其定性为"协议问世以来规模最大的一次系统性修订"。最终规范计划于2026年7月28日正式发布。

这不是一次功能堆叠，而是一次范式转移。MCP 正从"让 AI 会调工具"的连接协议，走向可规模化部署、全链路可治理、调用全流程可追溯的生产级智能体基础设施。

对于正在构建 AI Agent、或者准备将 MCP 引入生产环境的开发者来说，这次升级直接影响了你未来一到两年的架构选型。本文将深入拆解这次升级的技术细节、生产含义，以及作为开发者你应该如何应对。

一、背景：MCP 解决了什么问题，又产生了什么问题

1.1 为什么需要 MCP

在 MCP 出现之前，AI 应用接入外部工具有多痛苦？用一个具体场景来描述：

假设你正在开发一个企业知识库问答系统。用户问："帮我查一下这批供应商的最新资质情况。"AI 需要：

1. 查询内部 ERP 的供应商主数据
2. 调用工商信息 API 核验资质
3. 读取合同管理系统中的履约记录
4. 访问风险数据库获取诉讼和处罚信息

在 MCP 出现之前，你大概需要这样写：

# 传统方案：每个工具写一套集成代码
if "供应商" in query and "资质" in query:
    # 方案A: 用 OpenAI Function Calling
    response = openai.ChatCompletion.create(
        functions=[{
            "name": "query_erp",
            "parameters": {...}
        }, {
            "name": "query_business_info",
            "parameters": {...}
        }, ...]
    )
    
    # 方案B: 用 Anthropic Tool Use
    response = anthropic.messages.create(
        tools=[{
            "name": "query_erp",
            "input_schema": {...}
        }, ...]
    )
    
    # 方案C: 用 Google Function Declaration
    response = gemini.generate_content(
        tools=[...]
    )

每换一个大模型提供商，你就得重写一遍 glue code。更痛苦的是，如果你的系统需要同时调用 5 个不同来源的工具，这 5 套接口定义可能散落在代码库的各个角落，维护成本极高。

MCP 的核心价值：统一工具描述格式，让 AI 应用和工具提供者解耦。 你只需要写一个 MCP Server 实现业务逻辑，然后任何支持 MCP 的客户端（Claude Desktop、Cursor、你的自定义 Agent）都能发现并调用它。

1.2 MCP v1.x 时代的三个核心局限

MCP 在 2024 年 11 月推出后迅速获得了广泛采纳，但也暴露出了三个显著的工程局限：

局限一：会话强绑定

v1.x 版本的远程 MCP Server 要求客户端先建立协议会话（initialize → initialized 握手），后续所有请求都依赖同一份 Session。这意味着：

# v1.x 时代的远程 MCP 调用流程
# Step 1: 先建立会话
session = await mcp_client.connect("https://api.example.com/mcp")
session_id = session.id  # 这个会话被绑定到某台服务器实例

# Step 2: 后续请求必须经过同一会话
result = await session.call_tool("query_erp", {...})  # 只能发到会话所在的服务器

对于企业级部署，这意味着：

• 负载均衡？不存在的——请求必须路由到持有 Session 的那个实例
• 滚动发布？不可能——杀掉旧实例会中断所有活跃会话
• 水平扩容？很难——你需要一个黏性会话的负载均衡器

局限二：能力裸列，没有治理

v1.x 只提供了 tools/list 接口，工具只是一个名称和参数 Schema 的列表。当你的 MCP Server 有 20 个工具时，AI 客户端拿到的是一张 20 项的清单。当你有 200 个工具时，这张清单就变成了噪声。更糟糕的是，没有任何机制让 AI 理解"哪个工具查当前数据，哪个查历史数据"、"哪些场景应该先做扫描再下钻"。

局限三：结果只有自然语言，没有证据链

当 AI 调用了多个工具、聚合了多个数据源之后，返回给用户的只是一段自然语言文本。用户没有办法验证"这个结论是怎么得出的"、"用了哪些数据"、"数据的时间点是什么"。在企业场景中，这带来了严重的合规和审计问题。

这三个局限，正是 2026-07-28 规范要解决的核心问题。

二、无状态核心：从"会话绑定"到"请求自包含"

2.1 什么是无状态核心

2026-07-28 规范的最大变化，是取消协议层的 Session 机制。

旧版（v1.x）的握手流程：

Client → Server: initialize (协议版本、能力声明)
Server → Client: initialized (协议版本、服务端能力)
[协议会话建立，后续所有请求携带 Mcp-Session-Id]

新版（v0.28+）的请求流程：

Client → Server: 一个 POST 请求，包含所有必要信息
Server → Client: 响应，同一个请求可以由任意服务实例处理

直观理解：每个请求自己携带完成处理所需的全部上下文，不需要服务器记住任何会话状态。

这在技术上对应两个具体变化：

1. 移除 initialize / initialized 握手：客户端不再需要先建立协议会话再发送工具调用
2. 移除 Mcp-Session-Id 头：请求中不再需要携带会话标识

2.2 为什么这对生产部署至关重要

让我们通过一个具体场景来理解无状态的价值。

场景：企查查 MCP 的生产压力

企查查智能体数据平台（agent.qcc.com）截至 2026 年 7 月已有 9 个 MCP Server、197 个工具。平台面对的调用方包括：

• 不同 AI 工具（WorkBuddy、QoderWork、QClaw 等）
• 多个合作平台
• 不同客户租户
• 大规模并发调用

在有会话绑定的旧版协议下：

• 某个 AI 客户端建立了 Session，这个 Session 被路由到了 Server 实例 A
• 如果 Server A 因为负载高、滚动发布或故障而不可用，这个 Session 就断了
• 客户端需要重新建立会话，但之前的状态（如果有的话）已经丢失

在无状态的新版协议下：

• 每个请求都是自包含的，可以被路由到任意一个健康的 Server 实例
• 负载均衡器可以根据实时的 CPU、内存、响应时间做最优路由
• 滚动发布时，任意实例可以随时下线，请求自动分发到其他实例

# 无状态 MCP 请求的典型结构
# 每个请求携带完整上下文，无需服务端维护会话状态

request = {
    "jsonrpc": "2.0",
    "id": 42,
    "method": "tools/call",
    "params": {
        "name": "business_search",
        "arguments": {
            "keyword": "企查查科技股份有限公司",
            "data_scope": "current"  # 明确数据范围
        },
        # v0.28+ 新增：请求级别的上下文信息
        "meta": {
            "trace_id": "uuid-for-correlation",
            "tenant_id": "enterprise-customer-123",
            "capabilities_requested": ["real-time-data"]
        }
    }
}

2.3 无状态与 HTTP 网关的整合

无状态设计带来的另一个重要变化：MCP 可以复用成熟的 HTTP 网关生态。

旧版限制：远程 MCP 使用 HTTP + SSE（Server-Sent Events），但 SSE 是长连接，Session 维护在服务端。这让大多数 API 网关（Kong、Nginx、Envoy）无法正确处理——它们习惯处理无状态的请求-响应交互。

新版优势：每个请求是独立的 HTTP POST + JSON 响应，可以被任何标准 HTTP 网关处理。

# 一个典型的 v0.28+ MCP 部署架构
services:
  mcp-gateway:
    # Envoy 或 Kong 网关处理路由、限流、鉴权
    image: envoy-mcp-gateway:latest
    ports:
      - "8080:8080"
    environment:
      UPSTREAM_CLUSTERS: "mcp-server-1,mcp-server-2,mcp-server-3"
      LOAD_BALANCER: "round_robin"
      RATE_LIMIT: "1000/minute"
  
  mcp-server-1:
    image: qcc-mcp-server:latest
    replicas: 3
    resources:
      limits:
        memory: "512Mi"
        cpu: "500m"
  
  mcp-server-2:
    image: legal-mcp-server:latest
    replicas: 2
  
  mcp-server-3:
    image: docparse-mcp-server:latest
    replicas: 2

在这个架构中：

• 网关负责路由（根据 tool name 或 server 名称分发到对应后端）
• 网关负责限流（每个租户的调用频率限制）
• 网关负责鉴权（JWT token 验证）
• 每个 Server 实例无状态运行，水平扩容完全自由

三、能力发现：从"列清单"到"可治理"

3.1 为什么工具数量是双刃剑

当 MCP Server 只有 10 个工具时，tools/list 返回的清单是有用的。但当你的 Server 有 50 个、100 个、200 个工具时，清单就变成了噪声。

企查查 MCP 提供了 197 个工具，覆盖工商、股权、风险、知识产权、经营、历史、司法、法规等多个领域。对于 AI Agent 来说，直接面对 197 个工具选项，会产生几个具体问题：

问题一：选错工具

AI 看到 "company_search"、"company_basic_query"、"company_info_get" 三个工具，它怎么知道应该用哪个？旧版 MCP 的描述通常是这样的：

{
  "name": "company_search",
  "description": "查询企业信息",
  "inputSchema": {
    "type": "object",
    "properties": {
      "keyword": {"type": "string"}
    }
  }
}

这个描述没有说明：

• 这个工具查的是当前数据还是历史数据？
• 搜索结果是模糊匹配还是精确匹配？
• 返回的数据字段有哪些，和 company_basic_query 有什么区别？
• 在什么场景下应该用这个工具而不是另一个？

问题二：重复调用

AI 可能对同一个数据维度调用多个相似工具，造成浪费。更严重的是，可能产生矛盾的结果——两个工具查出的同一字段值不一样，AI 不知道该信哪个。

问题三：上下文污染

一次任务调用了几十个工具，每个工具返回的数据都塞进上下文窗口，很快就把宝贵的上下文空间填满了。AI 实际上在做"暴力搜索"而不是"精准下钻"。

3.2 能力发现机制的工程实现

2026-07-28 规范引入的能力发现机制，核心是让工具描述从"清单"升级为"可被系统化理解的语义网络"。

新增的 server/discover 接口

// 新版工具描述（简化示例）
{
  "name": "risk_scan",
  "description": "对企业进行综合风险扫描，返回各风险维度的数量和分布",
  "inputSchema": {...},
  "outputSchema": {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "properties": {
      "dimension": {
        "type": "string",
        "enum": ["legal", "financial", "operational", "compliance"]
      },
      "count": {"type": "integer"},
      "severity": {"type": "string", "enum": ["low", "medium", "high", "critical"]}
    }
  },
  // v0.28+ 新增：能力语义描述
  "annotations": {
    "dataScope": "current",           // 当前数据 vs 历史数据
    "dataFreshness": "realtime",       // 实时 vs T+N 延迟
    "capabilityType": "scanning",      // 工具能力类型：查询/扫描/分析
    "requiresAnchor": true,            // 是否需要先完成主体锚定
    "prerequisites": [],               // 前置工具依赖
    "mutuallyExclusiveWith": ["risk_query_individual"], // 不应同时调用
    "cacheTtlMs": 300000               // 缓存建议：5分钟
  }
}

能力分类体系的工程实践

企查查 MCP 将工具组织为五层能力体系：

┌─────────────────────────────────────────────────────────┐
│  全局约束层 (Global Constraints)                         │
│  回答：哪些事情不能猜、不能混、不能越过                  │
├─────────────────────────────────────────────────────────┤
│  SKILL 层 (Business Tasks)                             │
│  回答：怎样组合多个工具完成一个业务任务                   │
├─────────────────────────────────────────────────────────┤
│  Resources 层 (Stable Knowledge)                        │
│  回答：调用前需要理解什么                                │
├─────────────────────────────────────────────────────────┤
│  Server 层 (Capability Boundaries)                      │
│  回答：这些能力属于哪个专业范围                           │
├─────────────────────────────────────────────────────────┤
│  Tool 层 (Real-time Data)                               │
│  回答：数据从哪里来                                      │
└─────────────────────────────────────────────────────────┘

这个五层体系映射到 MCP 协议：

• Tool 层 → tools/call 提供原子数据能力
• Server 层 → 多个 Tool 的逻辑分组，AI 理解"这个 Server 解决什么问题域"
• Resources 层 → resources/list 提供稳定知识（术语表、数据字典、报告模板）
• SKILL 层 → MCP Apps（新版引入），封装可复用的业务流程
• 全局约束层 → annotations 和 constraints 字段，规定使用边界

3.3 工具治理的生产落地

工具数量多了之后，治理变得和工具本身一样重要。2026-07-28 规范在这一点上引入了几个关键能力：

缓存语义

// 工具级别的缓存声明
{
  "name": "company_basic_info",
  "annotations": {
    "cacheScope": "tenant",       // 租户级缓存（同一租户内复用）
    "ttlMs": 3600000,             // 缓存 1 小时
    "staleWhileRevalidate": true  // 返回旧数据的同时后台刷新
  }
}

对于企业数据查询场景，缓存策略直接影响成本。以企查查为例，每次 API 调用涉及积分消耗。如果 AI Agent 在一次会话中重复查询同一企业信息，有缓存机制可以节省大量调用成本。

限流和审计语义

// 新版网关层面的工具治理
{
  "tools": [...],
  "governance": {
    "rateLimits": {
      "default": "100/minute",
      "risk_scan": "20/minute",  // 高成本工具更严格的限流
      "company_search": "500/minute"  // 查询工具可以更宽松
    },
    "costTracking": {
      "enabled": true,
      "perToolCost": {
        "risk_scan": 10,      // 每个风险扫描消耗 10 积分
        "company_search": 2    // 每个查询消耗 2 积分
      }
    },
    "auditLog": {
      "required": true,
      "fields": ["tool", "arguments", "result_hash", "timestamp", "tenant"]
    }
  }
}

这些治理能力不是给 AI 用的，而是给 MCP 网关和平台运营者用的。网关可以基于这些元信息做限流、计费、审计，而不需要每个 Server 单独实现。

四、任务协作：从"一次调用"到"持续完成"

4.1 为什么单次调用不够用

在 MCP v1.x 的设计哲学中，每个工具调用是一个独立的事务：

用户: "帮我查一下这家公司的情况"
AI: → tools/call: company_search
AI: ← 返回企业基本信息
AI: → tools/call: risk_scan  
AI: ← 返回风险扫描结果
AI: → tools/call: shareholder_query
AI: ← 返回股东信息
AI: [整合所有结果，生成自然语言回复]

这个模式有三个明显的问题：

问题一：无法处理长任务

"帮我生成一份完整的尽调报告" 这样的任务，可能需要：

1. 30 分钟的数据采集
2. 多次用户确认（"这个股权结构你确认吗？"）
3. 分阶段的结果呈现

单次工具调用模式无法支持这种多轮交互、多阶段执行的长任务。

问题二：无法处理补充输入

在任务执行过程中，可能需要用户补充信息："您想查的是注册地在深圳的总部，还是所有分支机构？" 旧版协议没有标准的方式来处理这种暂停和补充输入。

问题三：任务状态无法持久化

如果 AI Agent 崩溃或会话中断，旧版协议下正在执行的任务就丢失了。用户不得不重新描述整个任务。

4.2 MCP Apps 与任务协作的新范式

2026-07-28 规范引入了 Tasks 和 MCP Apps 作为一等公民来解决这些问题。

Tasks：长任务的持久化状态管理

# 任务创建
task = {
    "id": "task-uuid-123",
    "name": "enterprise_due_diligence",
    "status": "in_progress",
    "stages": [
        {
            "id": "stage-1",
            "name": "主体锚定",
            "tools": ["company_search", "credit_code_validate"],
            "status": "completed",
            "output": {
                "confirmed_entity": "企查查科技股份有限公司",
                "credit_code": "91110108MA01XXXXX"
            }
        },
        {
            "id": "stage-2",
            "name": "综合风险扫描",
            "tools": ["risk_scan"],
            "status": "pending",
            "prerequisite": "stage-1"
        },
        {
            "id": "stage-3",
            "name": "报告生成",
            "tools": ["report_compile"],
            "status": "pending",
            "prerequisite": "stage-2"
        }
    ],
    "progress": {
        "completed": 1,
        "total": 3,
        "pct": 33
    }
}

这个任务结构意味着：

• 任务执行可以被暂停和恢复
• 每个阶段的输出被持久化，后续阶段可以直接引用
• 用户可以在任何阶段介入确认或补充信息
• 即使 Agent 重启，也可以通过 task ID 恢复执行

MCP Apps：可复用的业务流程封装

如果说 Tasks 是单个任务的状态管理，那么 MCP Apps 是更高层次的能力封装——将多个工具、多个任务组合为一个可复用的"应用"。

// MCP App 示例：企业尽调应用
{
  "name": "enterprise_kyb_app",
  "version": "1.0.0",
  "description": "企业 KYB（了解你的客户）核验应用",
  "stages": [
    {
      "name": "entity_anchor",
      "description": "确认要核验的企业主体",
      "requiresUserInput": true,
      "validation": {
        "requiredFields": ["company_name", "credit_code"],
        "uncertaintyThreshold": 0.7
      }
    },
    {
      "name": "risk_comprehensive_scan",
      "description": "综合风险扫描",
      "parallelTools": ["risk_scan", "litigation_query", "penalty_query"],
      "aggregation": "union"
    },
    {
      "name": "shareholder_analysis",
      "description": "股权结构分析",
      "requiresUserConfirmation": true,
      "confirmationPrompt": "股权穿透结果显示有VIE架构，请确认是否继续深入分析"
    },
    {
      "name": "report_generation",
      "description": "生成核验报告",
      "outputFormat": "structured_json_with_evidence",
      "evidenceFields": ["data_sources", "query_timestamps", "confidence_scores"]
    }
  ]
}

MCP Apps 的价值在于：企业可以把最佳实践封装为可复用的应用模板，而不需要每次都重新描述任务流程。

4.3 多轮交互的工程实现

新版 MCP 支持补充输入的标准机制：

# 用户补充输入
await client.send_input({
    "task_id": "task-uuid-123",
    "stage_id": "entity_anchor",
    "input": {
        "company_name": "企查查科技股份有限公司",
        "credit_code": "91110108MA01XXXXX",
        "confirmed_by": "user_id_456"
    }
})

# Agent 继续执行
await client.continue_task({
    "task_id": "task-uuid-123",
    "resume_from": "stage-2"
})

这个机制让 AI Agent 可以像人类一样"边做边问"，而不需要在上一步就把所有信息都问清楚。

五、证据链与可追溯：从"说了什么"到"依据能否还原"

5.1 企业场景的证据需求

在企业级 AI 应用中，AI 给出的结论往往需要被审计和追溯。考虑以下场景：

场景：银行信贷审批

信审员使用 AI Agent 辅助审查一家供应商的资质。AI 给出了"该供应商存在重大法律风险，建议拒绝"的结论。

三个月后，供应商投诉："凭什么拒绝我们？" 信审员需要回答：

• AI 的结论基于哪些数据？
• 数据的时间点是什么？
• 有没有可能数据已经过期？

旧版 MCP 的问题：工具调用结果只返回自然语言数据，没有任何元信息可以追溯。

新版 MCP 的改进：每个工具调用结果都携带结构化的证据元信息。

// v0.28+ 工具调用结果（带证据元信息）
{
  "result": {
    "data": {
      "company_name": "企查查科技股份有限公司",
      "credit_code": "91110108MA01XXXXX",
      "risk_level": "high",
      "litigation_count": 23,
      "penalty_count": 5
    },
    "_meta": {
      "evidence_id": "evidence-uuid-789",
      "data_source": "qcc_mcp_business_data",
      "query_timestamp": "2026-07-26T07:00:00+08:00",
      "data_timepoint": "current",
      "confidence": 0.95,
      "fields_used": ["litigation_count", "penalty_count", "risk_level"],
      "raw_response_hash": "sha256:abc123...",
      "limitations": [
        "仅覆盖中国大陆司法数据",
        "行政处罚数据可能存在 7 天更新延迟"
      ]
    }
  }
}

5.2 完整 JSON Schema 的生产价值

2026-07-28 规范将工具的输出 Schema 升级到完整的 JSON Schema 2020-12，这不只是格式升级，而是让结构化交付成为可能。

旧版 Schema（简化）：

{
  "type": "object",
  "properties": {
    "company_name": {"type": "string"},
    "risk_level": {"type": "string"}
  }
}

新版 Schema（完整）：

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "properties": {
    "company_name": {
      "type": "string",
      "description": "企业名称"
    },
    "risk_level": {
      "type": "string",
      "enum": ["low", "medium", "high", "critical"],
      "description": "综合风险等级"
    },
    "litigation_count": {
      "type": "integer",
      "minimum": 0,
      "description": "当前有效诉讼数量"
    },
    "last_updated": {
      "type": "string",
      "format": "date-time",
      "description": "数据更新时间"
    }
  },
  "required": ["company_name", "risk_level", "last_updated"],
  "additionalProperties": false
}

完整 Schema 的价值在于：

• 客户端可以自动验证返回数据的完整性
• AI Agent 可以知道哪些字段是必填的、哪些是可选的
• 可以自动生成类型安全的客户端代码
• 数据质量可以被系统化地监控

5.3 链路追踪：W3C Trace Context 集成

新版 MCP 引入了 W3C Trace Context 标准，让每次工具调用都可以被纳入分布式追踪体系：

# 工具调用时注入链路追踪上下文
result = await client.call_tool(
    "risk_scan",
    {"company_id": "qcc-123"},
    trace_context={
        "traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
        "tracestate": "congo=t61rcWkgMzE"
    }
)

# 链路追踪结构
trace = {
    "trace_id": "0af7651916cd43dd8448eb211c80319c",
    "span_id": "b7ad6b7169203331",
    "parent_span_id": "a1b2c3d4e5f6",  # 调用方 span
    "duration_ms": 234,
    "tools_in_chain": [
        "company_anchor",
        "risk_scan",
        "litigation_query",
        "penalty_query"
    ],
    "tokens_used": 18432,
    "errors": []
}

对于企业级部署，这意味着：

• 可以精确分析每次 AI 推理的耗时分布（哪个工具最慢？）
• 可以发现工具调用链中的错误和异常
• 可以审计每次 AI 决策背后的完整数据路径
• 可以对比不同 AI 模型在同一任务上的表现差异

六、企业级 MCP 架构：实战设计模式

6.1 一个完整的企业 MCP 部署架构

基于 2026-07-28 规范，一个生产级的 MCP 部署架构大概是这样的：

┌──────────────────────────────────────────────────────────────────────┐
│                           客户端层                                    │
│  Claude Desktop | Cursor | 自定义 Agent | 第三方平台                   │
└──────────────────────────────────────────────────────────────────────┘
                                  │
                                  ▼
┌──────────────────────────────────────────────────────────────────────┐
│                          MCP Gateway（Envoy/Kong）                      │
│  • JWT 鉴权 & 租户隔离                                                 │
│  • 限流（per-tenant, per-tool）                                        │
│  • 链路追踪（OpenTelemetry）                                           │
│  • 请求路由（根据 tool/server 分发）                                    │
│  • 响应缓存（基于 cacheTtlMs）                                         │
│  • 成本计量（per-tool-cost tracking）                                   │
└──────────────────────────────────────────────────────────────────────┘
           │                    │                    │
           ▼                    ▼                    ▼
┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐
│  Business MCP      │ │  Legal MCP        │ │  DocParse MCP    │
│  Server            │ │  Server          │ │  Server          │
│  ─────────────    │ │  ─────────────    │ │  ─────────────    │
│  185 tools        │ │  10 tools        │ │  2 tools        │
│  (company/risk/   │ │  (laws/cases/    │ │  (PDF→struct/   │
│   ip/opera...)    │ │   citations)      │ │   doc→struct)    │
└───────────────────┘ └───────────────────┘ └───────────────────┘
           │                    │                    │
           ▼                    ▼                    ▼
┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐
│  企业数据源        │ │  法律数据库        │ │  文档存储         │
│  (工商/司法/      │ │  (裁判文书/       │ │  (PDF/Word/      │
│   知识产权/...)   │ │   法规/...)       │ │   Excel)          │
└───────────────────┘ └───────────────────┘ └───────────────────┘

6.2 无状态 Server 的实现模式

在 2026-07-28 规范下，MCP Server 应该是完全无状态的。这意味着：

# 错误示范：Server 中存储会话状态
class MCPStatefulServer:
    def __init__(self):
        self.sessions = {}  # ❌ 不应该在 Server 中维护会话状态
        self.cache = {}    # ❌ 不推荐：限制水平扩容
    
    async def handle_request(self, request):
        session_id = request.headers.get("Mcp-Session-Id")  # ❌ v0.28 已移除
        if session_id in self.sessions:
            return self.sessions[session_id]

# 正确示范：无状态 Server
class MCPStatelessServer:
    def __init__(self, redis_url: str, upstream_db: str):
        self.redis = aioredis.from_url(redis_url)  # ✅ 状态外置到 Redis
        self.db = create_db_pool(upstream_db)
        # ✅ 无本地状态，可水平扩展
    
    async def handle_request(self, request: MCPRequest) -> MCPResponse:
        # 每个请求自包含所有必要信息
        trace_id = request.meta.trace_id  # 从请求中获取链路 ID
        tenant_id = request.meta.tenant_id  # 从请求中获取租户 ID
        
        # 状态从外部存储读取（Redis/DB）
        context = await self.redis.get(f"context:{trace_id}")
        
        # 执行工具逻辑
        result = await self.execute_tool(request.params)
        
        # 证据元信息
        result._meta = EvidenceMeta(
            trace_id=trace_id,
            timestamp=datetime.now().isoformat(),
            server_instance=self.instance_id  # 用于追踪哪台服务器处理的
        )
        
        return result

6.3 能力治理的实施清单

将能力治理落地到生产环境，建议按以下顺序推进：

Phase 1：工具分类（Week 1-2）

• 盘点所有现有工具，按"数据域"分类
• 标记每个工具的 dataScope（当前/历史）
• 标记每个工具的 capabilityType（查询/扫描/分析）
• 识别工具间的互斥关系

Phase 2：描述增强（Week 3-4）

• 为每个工具补充完整的 description
• 添加 prerequisites 声明
• 添加 cacheTtlMs 建议
• 完善 inputSchema 和 outputSchema

Phase 3：治理配置（Week 5-6）

• 配置限流规则（per-tool, per-tenant）
• 配置成本计量
• 启用审计日志
• 配置缓存策略

Phase 4：监控体系（Week 7-8）

• 接入 OpenTelemetry 链路追踪
• 设置工具调用成功率告警
• 设置响应时间 P99 告警
• 构建成本仪表盘

七、性能对比：v1.x vs v0.28

我们来做一个量化的对比分析：

八、迁移路径：从 v1.x 到 v0.28

8.1 客户端迁移

对于 MCP 客户端（如 Claude Desktop、Cursor、自定义 Agent），迁移主要是协议层面的：

# v1.x 客户端
class MCPv1Client:
    async def connect(self, url: str):
        self.session = await create_session(url)
        await self.session.initialize()  # v1.x 需要握手
        self.session_id = self.session.id
    
    async def call_tool(self, name: str, args: dict):
        return await self.session.call(
            "tools/call",
            {"name": name, "arguments": args},
            headers={"Mcp-Session-Id": self.session_id}
        )

# v0.28 客户端
class MCPv28Client:
    async def connect(self, url: str):
        # v0.28 不需要握手，直接可用
        self.base_url = url
    
    async def call_tool(self, name: str, args: dict):
        return await self.http.post(
            f"{self.base_url}/tools/call",
            {
                "jsonrpc": "2.0",
                "id": self._next_id(),
                "method": "tools/call",
                "params": {
                    "name": name,
                    "arguments": args,
                    "meta": {
                        "trace_id": uuid.uuid4().hex,
                        "tenant_id": self.tenant_id
                    }
                }
            }
        )

8.2 服务端迁移

对于 MCP Server，迁移需要注意兼容性问题：

class MCPv28CompatibleServer:
    async def handle_request(self, request: Request) -> Response:
        # 检测客户端协议版本
        client_version = request.headers.get("MCP-Version", "1.x")
        
        if client_version.startswith("1."):
            # 兼容 v1.x 客户端：维护 Session
            return await self._handle_v1(request)
        else:
            # v0.28+ 客户端：无状态处理
            return await self._handle_v28(request)
    
    async def _handle_v28(self, request: Request) -> Response:
        # 无状态处理：所有上下文从请求中获取
        result = await self._execute_tool(request.json)
        
        # 添加 v0.28 特有的元信息
        result["_meta"] = {
            "server_version": "0.28.0",
            "trace_id": request.params.meta.trace_id,
            "capabilities_discovered": True
        }
        
        return result

8.3 迁移检查清单

• 确认所有 MCP 客户端升级到支持 v0.28 的版本
• 服务端移除 Mcp-Session-Id 依赖
• 将会话级状态外置到 Redis/数据库
• 为每个工具补充 annotations 元信息
• 配置限流和审计规则
• 更新文档（工具描述、使用示例）
• 测试故障恢复场景
• 验证水平扩容能力

九、未来展望：MCP 的演进方向

9.1 协议成熟度的判断标准

2026-07-28 规范的发布，标志着 MCP 进入"生产成熟期"。判断一个协议是否达到生产成熟，有几个关键信号：

信号一：规范稳定 —— 协议的基本结构不再频繁变更，企业可以基于稳定规范做长期投入

信号二：治理能力完备 —— 不仅仅是"能用"，还要"管得住"。限流、审计、缓存、追踪等治理能力是生产部署的基础

信号三：生态丰富 —— 有足够多的客户端、Server、网关、中间件选择，企业不会被单一实现绑定

信号四：向后兼容 —— 新版本需要照顾存量部署，不能因为升级而破坏现有系统

从这几个标准看，MCP v0.28 是一个重要的里程碑，但生态的完全成熟还需要时间。

9.2 接下来值得关注的方向

方向一：MCP Apps 生态

随着 MCP Apps 的成熟，可能会出现"企业级 MCP App Store"——企业可以购买、组合标准化的 MCP Apps 来快速构建 Agent 能力。比如"反洗钱 KYC App"、"供应商评估 App"、"竞品分析 App"。

方向二：MCP 安全标准

企查查在实践中已经遇到了 MCP 安全的问题："大模型可能被恶意构造的工具描述或提示词操纵，执行超出预期的操作"。协议层面需要有更严格的能力边界描述和执行机制。

方向三：多模态 MCP

当前 MCP 主要处理文本工具调用。随着 AI Agent 能力的扩展，图像生成、视频处理、代码执行等能力的 MCP 标准化调用也是值得期待的方向。

结语

MCP 2026-07-28 规范的核心，不是增加了几种调用方式，而是推动了 MCP 从"工具连接协议"走向"生产级 Agent 基础设施"。无状态核心让远程 MCP 部署不再需要特殊的会话管理；能力发现机制让大规模工具集可以被系统化理解；任务协作和证据链让 AI Agent 能够完成复杂的企业级任务。

对于正在构建 AI Agent 系统的开发者，这次升级意味着：

• 如果你正在考虑将 MCP 引入生产环境，现在是好时机——协议已经准备好
• 如果你已经部署了 MCP v1.x，升级到 v0.28 可以获得显著的架构简化
• 如果你还在观望，2026-07-28 是一个值得关注的里程碑节点

唯一不变的是变化本身。MCP 的演进还在继续，但方向已经清晰：让 AI Agent 不仅能调用工具，而且能可靠地、可治理地、可持续地调用工具。

选题来源：MCP 2026-07-28 规范候选版技术解读
标签：MCP|AI Agent|协议标准化|生产级 AI|工具调用|无状态架构
关键字：MCP|Model Context Protocol|AI Agent|无状态核心|能力发现|工具治理|企业 AI|协议规范

附：MCP Server 实战开发——从零构建企业级 MCP Server

A.1 项目结构与依赖

让我们通过一个完整示例来展示如何构建一个符合 v0.28 规范的生产级 MCP Server。

# 项目结构
mcp-enterprise-server/
├── src/
│   ├── __init__.py
│   ├── server.py              # 主服务器入口
│   ├── tools/                 # 工具实现
│   │   ├── __init__.py
│   │   ├── business.py        # 企业数据工具
│   │   ├── risk.py            # 风险查询工具
│   │   └── legal.py           # 法律数据工具
│   ├── governance/             # 治理层
│   │   ├── __init__.py
│   │   ├── rate_limiter.py    # 限流器
│   │   ├── cost_tracker.py    # 成本计量
│   │   └── audit_logger.py    # 审计日志
│   ├── schemas/               # Schema 定义
│   │   ├── business.py
│   │   └── risk.py
│   └── utils/
│       ├── evidence.py         # 证据元信息生成
│       └── trace.py           # 链路追踪
├── pyproject.toml
├── Dockerfile
└── docker-compose.yaml

# pyproject.toml
[project]
name = "mcp-enterprise-server"
version = "0.28.0"
requires-python = ">=3.11"

dependencies = [
    "mcp>=1.0.0",
    "fastapi>=0.110.0",
    "uvicorn>=0.27.0",
    "redis>=5.0.0",
    "aioredis>=2.0.0",
    "httpx>=0.27.0",
    "pydantic>=2.6.0",
    "structlog>=24.0.0",
    "opentelemetry-api>=1.22.0",
    "opentelemetry-sdk>=1.22.0",
    "opentelemetry-instrumentation-fastapi>=0.43b0",
]

A.2 工具定义：完整的 Schema 与 Evidence

这是 v0.28 规范下工具定义的完整示例：

# src/schemas/business.py
from pydantic import BaseModel, Field
from typing import Optional, List, Literal
from datetime import datetime

class CompanySearchInput(BaseModel):
    keyword: str = Field(
        description="企业名称关键词（支持模糊匹配）",
        min_length=2,
        max_length=100
    )
    data_scope: Literal["current", "history", "all"] = Field(
        default="current",
        description="数据范围：current=当前在营，history=历史记录，all=全部"
    )
    province: Optional[str] = Field(
        default=None,
        description="省份筛选（行政区划代码，如 110000）"
    )
    limit: int = Field(
        default=10,
        ge=1,
        le=100,
        description="返回结果数量上限"
    )

class CompanySearchOutput(BaseModel):
    companies: List[dict] = Field(
        description="匹配的企业列表"
    )
    total: int = Field(description="符合条件的总企业数")
    search_id: str = Field(description="本次查询的唯一标识，用于关联后续查询")

class CompanySearchEvidence(BaseModel):
    """证据元信息：每个工具返回结果必须附带"""
    evidence_id: str
    trace_id: str
    data_source: str
    query_timestamp: datetime
    data_timepoint: str  # "current" | "history"
    confidence: float
    limitations: List[str]
    cache_hit: bool
    server_instance: str

A.3 限流器实现

# src/governance/rate_limiter.py
import time
import hashlib
from typing import Dict, Tuple
from collections import defaultdict
from dataclasses import dataclass
import structlog

logger = structlog.get_logger()

@dataclass
class RateLimitRule:
    """限流规则"""
    requests_per_minute: int
    requests_per_hour: int
    burst_size: int  # 允许的突发请求数

class TieredRateLimiter:
    """分层限流器：支持租户级、工具级的多层次限流"""
    
    def __init__(self, redis_url: str):
        self.redis_url = redis_url
        self._local_burst_cache: Dict[str, Tuple[int, float]] = {}
        self._rules: Dict[str, RateLimitRule] = {
            "default": RateLimitRule(100, 1000, 20),
            "company_search": RateLimitRule(500, 5000, 50),
            "risk_scan": RateLimitRule(20, 200, 5),
            "legal_case_query": RateLimitRule(100, 1000, 20),
        }
    
    async def check(
        self,
        tenant_id: str,
        tool_name: str,
        trace_id: str
    ) -> Tuple[bool, dict]:
        """
        检查请求是否允许通过
        返回: (是否允许, 限流元信息)
        """
        rule = self._rules.get(tool_name, self._rules["default"])
        now = time.time()
        
        # 突发限流（本地内存，毫秒级）
        burst_key = f"{tenant_id}:{tool_name}"
        if burst_key in self._local_burst_cache:
            count, window_start = self._local_burst_cache[burst_key]
            window_duration = now - window_start
            if window_duration < 1.0:  # 1秒窗口
                if count >= rule.burst_size:
                    return False, {
                        "reason": "burst_limit",
                        "retry_after_ms": int(1000 - window_duration * 1000)
                    }
                self._local_burst_cache[burst_key] = (count + 1, window_start)
            else:
                self._local_burst_cache[burst_key] = (1, now)
        else:
            self._local_burst_cache[burst_key] = (1, now)
        
        # 分钟级限流（Redis）
        minute_key = f"rl:minute:{tenant_id}:{tool_name}"
        minute_count = await self.redis.get(minute_key)
        if minute_count and int(minute_count) >= rule.requests_per_minute:
            ttl = await self.redis.ttl(minute_key)
            return False, {
                "reason": "minute_limit",
                "retry_after_ms": ttl * 1000
            }
        
        # 小时级限流（Redis）
        hour_key = f"rl:hour:{tenant_id}:{tool_name}"
        hour_count = await self.redis.get(hour_key)
        if hour_count and int(hour_count) >= rule.requests_per_hour:
            ttl = await self.redis.ttl(hour_key)
            return False, {
                "reason": "hour_limit",
                "retry_after_ms": ttl * 1000
            }
        
        # 记录请求
        pipe = self.redis.pipeline()
        pipe.incr(minute_key)
        pipe.expire(minute_key, 60)
        pipe.incr(hour_key)
        pipe.expire(hour_key, 3600)
        await pipe.execute()
        
        return True, {"allowed": True}

    def get_headers(self, metadata: dict) -> dict:
        """生成返回给客户端的限流头"""
        headers = {}
        if "retry_after_ms" in metadata:
            headers["Retry-After"] = str(metadata["retry_after_ms"] // 1000)
            headers["X-RateLimit-Retry-After-Ms"] = str(metadata["retry_after_ms"])
        return headers

A.4 证据元信息生成器

# src/utils/evidence.py
import hashlib
import uuid
from datetime import datetime, timezone
from typing import List, Optional
from dataclasses import dataclass, asdict

@dataclass
class Evidence:
    """MCP v0.28 证据元信息：每个工具调用结果的核心组成部分"""
    evidence_id: str
    trace_id: str
    data_source: str
    query_timestamp: str
    data_timepoint: str
    confidence: float
    limitations: List[str]
    fields_used: List[str]
    raw_response_hash: str
    server_instance: str
    protocol_version: str = "0.28.0"
    cache_hit: bool = False
    
    def to_meta(self) -> dict:
        """转换为返回给客户端的 _meta 字段"""
        return {
            "evidence_id": self.evidence_id,
            "trace_id": self.trace_id,
            "data_source": self.data_source,
            "query_timestamp": self.query_timestamp,
            "data_timepoint": self.data_timepoint,
            "confidence": self.confidence,
            "limitations": self.limitations,
            "fields_used": self.fields_used,
            "raw_response_hash": self.raw_response_hash,
            "server_instance": self.server_instance,
            "cache_hit": self.cache_hit,
            "protocol_version": self.protocol_version
        }

class EvidenceGenerator:
    def __init__(self, server_instance: str, redis_url: str):
        self.server_instance = server_instance
        self.redis = None  # 懒加载
    
    @staticmethod
    def hash_response(data: dict) -> str:
        """对响应数据生成哈希，用于审计"""
        import json
        normalized = json.dumps(data, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(normalized.encode()).hexdigest()[:16]
    
    async def generate(
        self,
        trace_id: str,
        tool_name: str,
        data: dict,
        data_source: str,
        data_timepoint: str,
        confidence: float,
        limitations: List[str],
        fields_used: List[str],
        cache_hit: bool = False
    ) -> Evidence:
        """生成证据元信息"""
        return Evidence(
            evidence_id=f"ev-{uuid.uuid4().hex[:12]}",
            trace_id=trace_id,
            data_source=data_source,
            query_timestamp=datetime.now(timezone.utc).isoformat(),
            data_timepoint=data_timepoint,
            confidence=confidence,
            limitations=limitations,
            fields_used=fields_used,
            raw_response_hash=self.hash_response(data),
            server_instance=self.server_instance,
            cache_hit=cache_hit
        )

A.5 主服务器入口

# src/server.py
import asyncio
import uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import structlog
from opentelemetry import trace

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

from src.governance.rate_limiter import TieredRateLimiter
from src.utils.evidence import EvidenceGenerator

logger = structlog.get_logger()
tracer = trace.get_tracer(__name__)

app = FastAPI(title="Enterprise MCP Server v0.28")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 全局组件（通过 lifespan 管理生命周期）
rate_limiter: TieredRateLimiter = None
evidence_gen: EvidenceGenerator = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global rate_limiter, evidence_gen
    import aioredis
    redis = await aioredis.from_url("redis://localhost:6379")
    rate_limiter = TieredRateLimiter("redis://localhost:6379")
    evidence_gen = EvidenceGenerator(
        server_instance=f"server-{uuid.uuid4().hex[:8]}",
        redis_url="redis://localhost:6379"
    )
    logger.info("server.started", instance=evidence_gen.server_instance)
    yield
    await redis.close()
    logger.info("server.stopped")

app.router.lifespan_context = lifespan

# ========== MCP v0.28 HTTP 传输 ==========

class ToolCallRequest(BaseModel):
    name: str
    arguments: dict
    trace_id: Optional[str] = None
    tenant_id: Optional[str] = None
    meta: Optional[dict] = None

@app.post("/tools/call")
async def call_tool(req: ToolCallRequest, request: Request) -> Response:
    trace_id = req.trace_id or uuid.uuid4().hex
    tenant_id = req.tenant_id or "anonymous"
    
    with tracer.start_as_current_span(f"tool.{req.name}") as span:
        span.set_attribute("trace_id", trace_id)
        span.set_attribute("tenant_id", tenant_id)
        span.set_attribute("tool_name", req.name)
        
        # Step 1: 限流检查
        allowed, limit_meta = await rate_limiter.check(
            tenant_id, req.name, trace_id
        )
        if not allowed:
            headers = rate_limiter.get_headers(limit_meta)
            return Response(
                status_code=429,