编程 MCP 2026-07-28 候选规范深度实战:从有状态会话到无状态基础设施——协议层无状态化、统一能力治理、长任务协作与全链路可信溯源的工程全解(2026)

2026-07-22 06:44:51 +0800 CST views 11

MCP 2026-07-28 候选规范深度实战:从有状态会话到无状态基础设施——协议层无状态化、统一能力治理、长任务协作与全链路可信溯源的工程全解(2026)

2026 年 7 月,全球 AI 智能体核心互联协议 MCP 发布了“2026-07-28 版本候选规范”(Release Candidate),官方将其定义为协议问世以来规模最大的一次系统性修订。本文从工程视角拆解这次修订的底层逻辑:为什么“有状态会话”曾经是正确设计,又为什么在 2026 年的云原生、多租户、长链路业务面前成了规模化瓶颈;新版如何用“协议层无状态化 + 统一能力治理 + 长任务协作 + 全链路可信溯源”四根支柱,把 MCP 从一个“AI 调工具的连接协议”升级为“可治理、可追溯、可弹性扩容的生产级智能体基础设施”。全文配 Python SDK 实战、可跑代码示例与性能优化清单。


一、背景:为什么 MCP 是 2026 年最重要的基础设施协议

如果你在 2024 年底关注过 AI 工程,一定记得那个让人又爱又恨的现状:每个大模型应用都要自己写一套对接 GitHub、数据库、文件系统、Slack、支付系统的适配代码。彼时行业里有个自嘲的说法——“你的 AI 项目 80% 的代码是在和各种工具的私有 API 吵架,只有 20% 在真正做智能”。

Anthropic 在 2024 年 11 月开源了 Model Context Protocol(MCP),本质是想做一件和当年 LSP(Language Server Protocol)类似的事:把“LLM 应用 ↔ 外部能力”的连接方式标准化。LSP 让一种编程语言能接入所有编辑器;MCP 想让一个 AI 应用能接入所有工具和数据源,而不必为每个工具重写胶水代码。

一年之后,MCP 已经事实成为 Agent 时代的“USB-C 接口”。但快速普及也暴露了协议本身的底层假设问题:它的初版是为单机、单进程、单用户、短会话设计的。当企业想把 MCP 跑在 Kubernetes 上、要支撑上千租户、要处理一笔需要几十轮交互的信贷尽调时,原来的“有状态会话”模型开始处处别扭。

这就是为什么 2026-07-28 版本候选规范被官方称为“史上最大修订”。它不是加几个新接口,而是动了协议的地基:

  • 协议层无状态化:取消会话绑定,所有请求自包含,服务端不再需要锁死在某台机器上;
  • 统一能力治理:新增能力发现、缓存、限流、审计的标准;
  • 长任务协作:用多轮交互、标准化任务引擎、MCP 应用组件支撑长周期业务流程;
  • 全链路可信溯源:落地完整 JSON Schema 与 W3C 链路追踪规范,叠加强化 OAuth 授权。

先把一个关键认知说在前面:规范发布 ≠ 旧协议立刻下线 ≠ 所有 SDK 自动升级。官方明确现有 MCP Server 不会在 7 月 28 日突然停止工作,RC 阶段是给 SDK 维护者和企业做验证用的。所以这篇文章既讲“现在怎么写”,也讲“接下来怎么迁移”。


二、核心概念:先吃透 MCP 的协议模型

在谈修订之前,必须先把当前协议(以 2025-11-25 稳定版为基准)的骨架吃透。所有新特性都是在这副骨架上的演进。

2.1 JSON-RPC 2.0 是底层

MCP 的全部消息都是 JSON-RPC 2.0。它是有状态连接 + 能力协商的协议,但传输的每一帧都是标准 JSON-RPC:request / response / notification(无 id 的通知)/ error

一个最小化的 initialize 握手长这样(客户端 → 服务端):

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-11-25",
    "capabilities": {
      "tools": { "listChanged": true },
      "resources": { "subscribe": true, "listChanged": true },
      "sampling": {}
    },
    "clientInfo": { "name": "my-agent-host", "version": "1.4.0" }
  }
}

服务端回以自己支持的能力与协议版本,随后客户端发送一个 notifications/initialized 通知,握手结束,进入可操作态。注意 protocolVersion日期版本号——这是 MCP 的一个聪明设计:协议用发布日期(如 2024-11-052025-03-262025-06-182025-11-25)作为版本,天然兼容、可追溯,客户端和服务端各自声明自己最高支持的版本,取交集协商。

2.2 Host / Client / Server 三层

MCP 是 client-host-server 架构:

  • Host(宿主):真正运行 LLM 的应用程序(比如一个 AI IDE、一个聊天客户端、一个 Agent 运行时)。它创建并管理多个 Client 实例,负责安全策略、用户授权、上下文聚合。
  • Client(客户端):Host 内部为每个 Server 创建的隔离连接器,1:1 对应一个 Server,维护一条有状态会话,负责双向路由消息、订阅与通知。
  • Server(服务端):提供具体能力与上下文的进程或服务,可以是本地子进程,也可以是远程 HTTP 服务。

这套设计有四条明确原则,值得每个实现者刻在脑子里:

  1. Server 要极容易构建——复杂编排交给 Host;
  2. Server 要高度可组合——各自聚焦、共享协议、即插即用;
  3. Server 不该看到整段对话,也看不到其他 Server——对话历史永远留在 Host,跨 Server 的交互由 Host 控制,安全边界由 Host 强制执行;
  4. 能力可渐进添加——核心协议只给最小必需功能,其余通过 capability 协商逐步开放。

2.3 四类原语(Primitives)

Server 向 Client 暴露三类能力,Client 向 Server 暴露一类:

方向原语含义典型方法
Server → ClientResources上下文与数据(给模型或用户用)resources/listresources/read
Server → ClientPrompts模板化消息与工作流prompts/listprompts/get
Server → ClientTools给模型执行的函数(= 代码执行)tools/listtools/call
Client → ServerSampling服务端发起、由 Host 代跑的 LLM 调用sampling/createMessage
Client → ServerRoots服务端询问“我该在哪些 URI/文件系统边界内工作”roots/list
Client → ServerElicitation服务端向用户请求补充信息elicitation/create

一个容易混淆的点:Tools 是给模型调用的,Resources 是给模型/用户读取的,Prompts 是给用户的模板。Tool 的 descriptioninputSchema(标准 JSON Schema)是模型决定“要不要调、怎么填参”的唯一依据——这也是为什么 MCP 反复强调:tool 的 description、annotations 都是不可信输入,除非来自受信任的 Server。

下面用官方 Python SDK(FastMCP)写一个最小可跑的 Server,把三类原语一次讲清:

# server.py  —— 需要: pip install "mcp[cli]"
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-server")

# 1) Tool: 模型可调用的函数,inputSchema 由类型注解自动生成
@mcp.tool()
def query_orders(user_id: str, status: str = "paid") -> list[dict]:
    """按用户和状态查询订单。status 取值: paid/pending/refunded。"""
    # 真实场景这里查数据库;演示直接返回
    return [
        {"order_id": "O-1001", "user_id": user_id, "status": status, "amount": 99.0},
        {"order_id": "O-1002", "user_id": user_id, "status": status, "amount": 21.5},
    ]

# 2) Resource: 可被读取的上下文,URI 模板可带参数
@mcp.resource("file://docs/{doc_id}")
def read_doc(doc_id: str) -> str:
    """读取一篇知识库文档的正文。"""
    return f"# 文档 {doc_id}\n这是 doc_id={doc_id} 的演示正文。"

# 3) Prompt: 给用户用的模板
@mcp.prompt()
def summarize_doc(doc_id: str) -> str:
    """生成‘总结某篇文档’的提示词模板。"""
    return f"请用三句话总结 file://docs/{doc_id} 的要点,并列出两个开放问题。"

if __name__ == "__main__":
    mcp.run()  # 默认 stdio 传输

跑起来后,任何兼容 MCP 的 Host 都能 tools/list 发现 query_orders,拿到它自动生成的 JSON Schema,再把模型决策出的参数通过 tools/call 传回来。这就是“标准化连接”的威力:你的工具只需写一次,Claude Desktop、各类 IDE、企业 Agent 平台都能直接接

2.4 传输层演进:stdio → HTTP+SSE → Streamable HTTP

MCP 的传输(transport)也经历过一次重要换代:

  • stdio:本地子进程,通过标准输入输出收发换行分隔的 JSON-RPC。零网络、零鉴权负担,是本地工具首选。
  • HTTP+SSE(旧远程):服务端用 SSE 推消息,客户端用 POST 发消息。但“服务端到客户端”只能走一条长连接 SSE,难以代理、难以负载均衡。
  • Streamable HTTP(2025-03-26 引入,取代旧 SSE 传输):客户端用 POST 发 JSON-RPC(带 Accept: application/json, text/event-stream),服务端可以选择用 SSE 流式回;客户端用 GET 建立 SSE 用于服务端主动通知;会话用 Mcp-Session-Id 头标识。

正是这个 Mcp-Session-Id,埋下了本次修订要解决的“有状态”问题的种子——我们下一节细说。


三、架构分析:从“有状态会话”到“无状态基础设施”

这是整篇文章的核心。我们要理解的不只是“改了什么”,更是“为什么必须改”。

3.1 有状态会话到底卡在哪

当前远程传输(Streamable HTTP)下,一次 MCP 会话由 Mcp-Session-Id 绑定。服务端为了维护这个会话的状态(已协商的 capabilities、订阅关系、某些服务端缓存),逻辑上期望同一个会话的所有请求都落到同一台机器

在单机、单用户、短会话场景下这完全没问题,甚至是优点:状态留在内存,延迟低、实现简单。但一旦进入企业的云原生世界,三个矛盾立刻显现:

  1. 水平扩容困难:前面挂了负载均衡(LB)或网关,请求会被随机分发到不同 Pod。要么 LB 做会话亲和(sticky session),把某用户的请求钉在同一台——这直接牺牲了弹性;要么服务端把会话状态外置到 Redis 之类——成本与复杂度骤增。
  2. 多租户隔离脆弱:政企、金融客户的高并发混合调用,要求不同租户的会话彼此强隔离。有状态会话把“隔离”的责任压在了“请求必须落在正确的那台机器”上,而不是协议本身。
  3. 长链路业务难支撑:一笔信贷尽调可能要几十轮交互、跨多个工具、耗时数分钟到数小时。把这种“长任务”硬塞进一个临时会话,会话一超时、Pod 一重启,进度就丢了。

一句话总结:有状态会话把“规模化的难题”从协议层推给了运维层。能跑,但很贵、很脆。

3.2 支柱一:协议层无状态化

2026-07-28 候选规范的最大变化,是彻底取消会话绑定,让每个请求都自包含(self-contained)

“自包含”意味着:请求本身携带完成本次调用所需的全部上下文(身份、租户、所需能力、必要的状态句柄),服务端收到后不需要依赖“上一次请求落在同一台机器”的前提就能处理。于是:

  • 前面可以放心挂 LB / 网关,请求自由分发到任意节点;
  • 天然适配国内主流云服务商的负载均衡、弹性扩缩容;
  • 多租户调用可以靠请求里的租户标识做隔离,而不是靠“哪台机器”。

工程视角提醒:无状态化≠服务端不能有任何状态。它只是要求会话状态不再隐式地绑在传输层连接上。状态该外置就外置(数据库、对象存储、KV),但协议不再假定“同一会话必落同机”。这正是经典的无状态服务设计哲学,MCP 把它下沉到了协议层。

一个示意性的“无状态请求”——注意它不再依赖连接级别的 Mcp-Session-Id,而是把租户与任务句柄显式带在协议字段里:

{
  "jsonrpc": "2.0",
  "id": 42,
  "method": "tools/call",
  "params": {
    "tool": "query_orders",
    "arguments": { "user_id": "U-7788", "status": "paid" },
    "context": {
      "tenant": "acme-bank",
      "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
      "task": "task_9f3c"
    }
  }
}

(注:上面 context 字段是为说明“自包含”思想做的示意,并非旧版字段;候选规范的具体字段名以 7 月 28 日正式版为准。核心思想不变:把身份、租户、追踪、任务句柄显式放进每一条请求。)

3.3 支柱二:统一能力治理(发现 / 缓存 / 限流 / 审计)

旧版里,AI 只能“被动拿到一份工具清单”。工具一多(新闻里提到的企业场景有上百种数据工具),调用就乱:哪个工具该先调?哪些高频工具可以缓存结果?谁在疯狂刷接口?

候选规范新增了一套统一能力治理标准

  • 能力发现(Capability Discovery):网关/客户端能自动识别、分类、管控上百种工具,按业务域分组(基础查询 / 司法穿透 / 知识产权检索……)。
  • 缓存:对只读、幂等的 Resources/Tools 结果做标准化缓存,避免无差别重复调用。
  • 限流(Rate Limiting):按租户、按工具维度做分层限流,保护后端数据源。
  • 审计(Audit):每一次调用都留下结构化审计记录,满足金融、政务的强监管要求。

这意味着 MCP 网关从一个“透传代理”升级成了“能力治理平面”。一个最简的限流 + 缓存中间件思路(示意):

import time, functools
from collections import defaultdict

_cache: dict[str, tuple[float, object]] = {}
_hits: defaultdict[str, list[float]] = defaultdict(list)
RATE = 10.0  # 每租户每秒最多 10 次

def governed(tenant: str, cache_ttl: float = 30.0):
    def deco(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # 1) 限流
            now = time.time()
            window = [t for t in _hits[tenant] if now - t < 1.0]
            if len(window) >= RATE:
                raise RuntimeError(f"tenant {tenant} rate limited")
            window.append(now)
            _hits[tenant] = window
            # 2) 缓存(按 函数+参数 做 key)
            key = f"{fn.__name__}:{args}:{kwargs}"
            if key in _cache and now - _cache[key][0] < cache_ttl:
                return _cache[key][1]
            result = fn(*args, **kwargs)
            _cache[key] = (now, result)
            return result
        return wrapper
    return deco

3.4 支柱三:长任务协作(多轮交互 / 任务引擎 / 应用组件)

这是修订里最“面向业务”的部分。旧版 MCP 本质是“一问一答”的接口调用;但信贷风控、企业尽调、合规审查是长周期、多步骤、需要和用户反复确认的流程。

候选规范引入:

  • 多轮交互请求:一个任务可以拆成多步,中间向用户 elicitation(索取补充信息)、向 Host 做 sampling(调用 LLM 推理)。
  • 标准化任务引擎:长任务有统一的状态模型(pending / running / waiting_input / done / failed),可被查询、可断点续跑。
  • MCP 应用组件(MCP App Components):把一组工具 + 资源 + 提示 + 业务约束打包成一个“业务技能”,比如“企业核验”“受益所有人穿透”“风险扫描”,对外像一个组合能力。

一个长任务的“任务引擎”最小骨架(示意,体现状态机思想):

from enum import Enum
from dataclasses import dataclass, field

class TaskState(Enum):
    PENDING = "pending"
    RUNNING = "running"
    WAITING_INPUT = "waiting_input"
    DONE = "done"
    FAILED = "failed"

@dataclass
class Task:
    task_id: str
    tenant: str
    state: TaskState = TaskState.PENDING
    steps: list[str] = field(default_factory=list)
    cursor: int = 0
    result: dict = field(default_factory=dict)
    missing: list[str] = field(default_factory=list)

_TASKS: dict[str, Task] = {}

def create_task(tenant: str, steps: list[str]) -> Task:
    t = Task(task_id=f"task_{len(_TASKS)+1:04d}", tenant=tenant, steps=steps)
    _TASKS[t.task_id] = t
    return t

def advance(task_id: str, provided: dict | None = None) -> Task:
    """无状态服务也能续跑:任务状态外置在 _TASKS(生产用数据库)。"""
    t = _TASKS[task_id]
    if t.state == TaskState.WAITING_INPUT and provided:
        t.result.update(provided)
    while t.cursor < len(t.steps):
        step = t.steps[t.cursor]
        t.state = TaskState.RUNNING
        # 演示:某步需要用户补充材料
        if step == "collect_ubo" and "ubo_proof" not in t.result:
            t.state = TaskState.WAITING_INPUT
            t.missing = ["ubo_proof"]
            return t
        t.result[step] = f"done:{step}"
        t.cursor += 1
    t.state = TaskState.DONE
    return t

关键点:因为 3.2 的无状态化,这个任务引擎的状态可以安全地存在外部存储里,任意 Pod 收到 advance(task_id, ...) 都能继续推进——Pod 重启、扩容都不丢进度。这正是无状态化与长任务协作两块彼此咬合的地方。

3.5 支柱四:全链路可信溯源

企业最怕两件事:AI 给出结论却“无依据”,以及数据边界“糊里糊涂”。候选规范用三件套回应:

  • 完整 JSON Schema:所有 Tool 的输出固定为结构化格式,下游业务系统可自动校验,不再是一段自由文本让程序难以下手。
  • W3C 链路追踪(traceparent / tracestate):每一步调用都带上分布式追踪上下文,一次 AI 结论可以回溯到“哪一步、调了哪个工具、什么时间、返回了什么”。
  • 强化 OAuth 授权:每一次能力调用都要有清晰的授权主体与边界,叠加在 2025-11-25 已落地的 OAuth 2.0 / OIDC 发现之上。

把 traceparent 注入到 MCP 调用里,配合 OpenTelemetry,能让一条“用户问题 → 模型决策 → 工具调用 → 数据源返回 → 最终结论”的链路完全可观测。这块我们放到第五章性能与可观测性一起讲。


四、代码实战:从零搭一个可治理的 MCP Server

理论讲完,动手。下面用官方 Python SDK 搭一个贴近企业场景的 MCP Server:它暴露“订单查询”工具、“知识库文档”资源、以及一个“风控尽调”长任务。我们同时演示 stdio(本地)与无状态 HTTP(远程)两种跑法。

4.1 安装与最小 Server

pip install "mcp[cli]" httpx
# risk_mcp/server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("risk-desk")

@mcp.tool()
def query_orders(user_id: str, status: str = "paid") -> list[dict]:
    """按用户和状态查询订单。status: paid|pending|refunded。"""
    # 真实环境查库;这里演示返回结构化结果
    return [
        {"order_id": "O-1001", "user_id": user_id, "status": status, "amount": 99.0},
        {"order_id": "O-1002", "user_id": user_id, "status": status, "amount": 21.5},
    ]

@mcp.resource("kb://doc/{doc_id}")
def read_kb(doc_id: str) -> str:
    """读取知识库文档正文(用于给模型补充领域知识)。"""
    return f"# {doc_id}\nUBO 穿透要求:需追溯至最终自然人受益所有人,持股阈值 25%。"

@mcp.prompt()
def kyb_check(company: str) -> str:
    """生成 KYB 企业核验提示词模板。"""
    return (
        f"请对 {company} 执行 KYB 核验:\n"
        "1) 主体精准锚定(区分同名企业)\n"
        "2) 分维度扫描工商/司法/股权风险\n"
        "3) 标注数据时点(实时 vs 历史)\n"
        "4) 每一条结论附证据来源"
    )

if __name__ == "__main__":
    mcp.run(transport="stdio")

python risk_mcp/server.py,它就在 stdio 上监听。任何 Host 都能发现并调用它。这就是 MCP 的“Server 极容易构建”原则落地。

4.2 一个会“要材料”的工具(Elicitation 实战)

企业尽调经常需要向用户索取补充材料。MCP 的 elicitation 原语就是干这个的。下面用低层 SDK 演示服务端如何主动向用户请求补充信息:

# risk_mcp/elicitation_server.py
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool, TextContent, ElicitRequest, ElicitResult,
    JSONRPCMessage,
)
from mcp.shared.session import RequestResponder

app = Server("elicitation-demo")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [Tool(
        name="ubo_penetration",
        description="受益所有人穿透,缺材料时向用户索取。",
        inputSchema={"type": "object",
                     "properties": {"company": {"type": "string"}},
                     "required": ["company"]},
    )]

@app.call_tool()
async def call_tool(name: str, arguments: dict,
                    request_context) -> list[TextContent]:
    if name != "ubo_penetration":
        raise ValueError("unknown tool")
    company = arguments["company"]
    # 假设发现缺少 UBO 证明,向用户发起 elicitation
    req = ElicitRequest(
        method="elicitation/create",
        params={
            "message": f"对 {company} 做 UBO 穿透需要受益所有人证明,请提供。",
            "requestedSchema": {
                "type": "object",
                "properties": {"ubo_proof": {"type": "string",
                                             "title": "受益所有人证明链接"}},
                "required": ["ubo_proof"],
            },
        },
    )
    # 实际 SDK 中通过 session 发送并 await 用户响应;此处示意结构
    return [TextContent(type="text",
                        text=f"已向用户索取 {company} 的 UBO 证明。")]

说明:elicitation 在 2025-06-18 规范中引入,是“长任务协作”的前置能力之一。上面的低层写法展示了消息结构;用 FastMCP 时官方 SDK 会逐步封装更简洁的装饰器式 API,请以你所用的 SDK 版本为准。

4.3 无状态 HTTP 跑法(贴近候选规范方向)

要在远程、可水平扩容地运行,用 Streamable HTTP 传输,并在服务端把“会话状态”外置。FastMCP 支持 transport="streamable-http"

# risk_mcp/http_server.py
from fastmcp import FastMCP

mcp = FastMCP("risk-desk-http")

@mcp.tool()
def query_orders(user_id: str, status: str = "paid") -> list[dict]:
    """按用户和状态查询订单。"""
    return [{"order_id": "O-1001", "user_id": user_id, "status": status}]

if __name__ == "__main__":
    # streamable-http 传输:客户端 POST JSON-RPC,服务端可 SSE 回
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

而“无状态化”的工程落点在于:不要在进程内存里存会话级可变状态。把任务进度、用户授权、租户上下文都写到外部存储(如 Postgres / Redis),让每个请求可被任意 Pod 处理。配合前面的 Task 状态机,你就得到了候选规范描述的“请求自包含、可自由分发”的能力。

4.4 客户端怎么调(含 tracing 注入)

# client_demo.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    params = StdioServerParameters(command="python",
                                   args=["risk_mcp/server.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("发现工具:", [t.name for t in tools.tools])
            res = await session.call_tool(
                "query_orders",
                arguments={"user_id": "U-7788", "status": "paid"},
            )
            print("调用结果:", res.content)

if __name__ == "__main__":
    asyncio.run(main())

真实企业网关里,你会在 call_tool 之前把 traceparent 注入到请求上下文,让这次调用被 OpenTelemetry 捕获,串起整条链路——见下一章。


五、性能优化与工程避坑

MCP 跑在单机上 demo 很容易,跑成“生产级基础设施”则处处是坑。按候选规范的四根支柱,给一份实战清单。

5.1 连接与会话:能复用就别乱建

  • stdio 场景:Host 为每个 Server 起一个长生命周期子进程,复用同一条连接,不要“调一次起一个进程”。
  • HTTP 场景:用连接池(如 httpx.AsyncClientlimits),避免每次工具调用都 TLS 握手。
  • 无状态后更放心地扩容:因为请求自包含,前面挂 LB 不必做 sticky session,Pod 可以随意扩缩。

5.2 能力缓存:别反复 list

tools/listresources/listprompts/list 的结果在 Server 能力不变时应当缓存。FastMCP 有 listChanged 通知机制——Server 能力变化时主动推通知,客户端据此失效缓存。别在每次推理前都重新拉全量清单。

5.3 限流与背压:保护你的数据源

你的 MCP Server 背后往往是数据库、第三方 API、付费模型。务必在 Server 或网关层做分层限流(按租户、按工具、按数据源),并在超限时返回标准 JSON-RPC error(code: -32000 之类)而非雪崩。候选规范把“限流”列为标准能力,正是为此。

5.4 长任务:用进度通知,别让 Host 干等

长任务(尽调、批量扫描)不要“同步阻塞直到完成”。正确做法是:立即返回一个 task_id,过程中用 notifications/progress 推进度,Host 用任务引擎轮询或订阅状态。这样既不会超时,又能给用户实时反馈。

# 进度通知示意(FastMCP 提供 progress 回调)
from mcp.server.fastmcp import Context

@mcp.tool()
async def batch_scan(company: str, ctx: Context) -> dict:
    total = 100
    for i in range(total):
        await ctx.report_progress(progress=i, total=total)
        # ... 实际扫描逻辑
    return {"scanned": total, "company": company}

5.5 全链路可观测:把 traceparent 接上

给每一次 MCP 调用注入 W3C traceparent,并在 Server 内部用 OpenTelemetry 打 span。一条“用户问题 → 模型决策 → 工具调用 → 数据源”的链路就能在 Jaeger / Tempo 里完整回放,满足候选规范的“可信溯源”要求,也方便排障。

# 在 Server 入口从请求取 traceparent 并开启 span(示意)
from opentelemetry import trace
tracer = trace.get_tracer("risk-mcp")

def handle_call(traceparent: str, tool: str, args: dict):
    ctx = trace.get_current_span().get_span_context()
    with tracer.start_as_current_span(f"mcp.tools/call:{tool}") as span:
        span.set_attribute("mcp.tool", tool)
        span.set_attribute("mcp.tenant", args.get("tenant", "unknown"))
        return do_work(tool, args)

5.6 安全:四条红线

  1. 用户同意优先:调用任何 Tool、暴露任何用户数据前必须显式征得同意;tool 的 description 视为不可信。
  2. 最小权限:Server 只读它该读的;采样(sampling)时用户要能控制“发什么 prompt、Server 能看到什么结果”。
  3. 授权边界清晰:用强化后的 OAuth 2.0 / OIDC 做客户端与用户认证,租户隔离靠请求里的租户标识强制。
  4. 审计留痕:每次调用落结构化审计日志,关键时刻能回溯到“谁、在什么时间、调了什么、拿到什么”。

六、总结与展望:MCP 正在从“连接工具”变成“基础设施”

把这次 2026-07-28 候选规范的四根支柱连起来看,结论很清楚:MCP 的野心不再只是“让 AI 调通工具”,而是成为智能体时代的底层通行标准——可规模化部署、全链路可治理、调用全流程可追溯。

对工程师的三条落地建议:

  1. 现在就能做:用官方 SDK 把内部工具封装成 MCP Server,享受“写一次、处处接”的红利;stdio 用于本地、Streamable HTTP 用于远程。
  2. 为无状态化提前准备:新写 Server 时,不要在进程内存存会话级可变状态,把任务/授权/租户上下文外置。等到 7 月 28 日正式版落地,你几乎零成本切到无状态模式。
  3. 把治理当一等公民:能力发现、缓存、限流、审计、traceparent——这些不是“以后再说”,而是决定你的 MCP 平台能不能上生产的分水岭。

最后给一个判断框架,用来评估你自己的或第三方的 MCP 平台是否“达标”:

  • 能否依靠无状态架构实现弹性规模化、完成工具统一治理?
  • 能否提供清晰的行业语义与主体识别规则,让 AI 精准使用垂直数据?
  • 能否让每一条 AI 结论都溯源到原始数据、调用链路,满足审计与合规?

这三个“能否”,正是这一轮修订划出的新竞争主线:单纯比接口数量已经不够了,具备可规模化、语义适配、可信溯源三大能力的垂直数据基座,才是下一阶段的核心壁垒。

MCP 把“连接”做成了标准;而真正决定价值的,是你往这条标准上接进了多可信、多可治理、多贴近业务的能力。7 月 28 日之后,智能体工程的底座,会明显不一样。


参考:MCP 官方规范(modelcontextprotocol.io,基准版本 2025-11-25)、2026-07-28 版本候选规范公开解读,以及企业 MCP 平台工程实践。文中带“示意”字样的代码片段用于说明协议思想,具体字段以 7 月 28 日正式发布的规范与你所使用 SDK 的版本为准。

推荐文章

一文详解回调地狱
2024-11-19 05:05:31 +0800 CST
在 Nginx 中保存并记录 POST 数据
2024-11-19 06:54:06 +0800 CST
为什么大厂也无法避免写出Bug?
2024-11-19 10:03:23 +0800 CST
Vue中的样式绑定是如何实现的?
2024-11-18 10:52:14 +0800 CST
程序员茄子在线接单