登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›MCP Server 实践之旅第 3 站：MCP 协议亲和性的技术内幕 ...

MCP Server 实践之旅第 3 站：MCP 协议亲和性的技术内幕

敕码 2025-6-3 00:18:12
作者：柳下，西流
背景

在分布式架构设计中，请求亲和性是实现有状态服务高可用的核心技术，通过将具备相同会话标识的请求智能路由至固定计算节点，保障会话连续性及缓存有效性。然而在 Serverless 范式下，函数计算服务因其瞬时实例生命周期、自动弹性扩缩容、无状态化运行等特性，在亲和性保障层面面临原生架构冲突。本文将以 MCP Server 在函数计算平台的深度集成为研究载体，解构基于 SSE 长连接通信模型，剖析会话亲和、优雅升级等关键技术，揭示 Serverless 架构在 MCP 场景中的亲和性创新实践。
概念介绍

在系列文章首篇 MCP Server 实践之旅第 1 站：MCP 协议解析与云上适配 我们深入解析了 MCP 以及 SSE 协议，为了方便本文的阅读， 这里再简单介绍下 MCP 以及 SSE 协议。
MCP：作为开放标准协议，为 AI 应用构建了通用化上下文交互框架。可以将 MCP 想象成 AI 应用程序的 USB-C 接口。就像 USB-C 为设备连接各种外设和配件提供了标准化方式一样，MCP 为 AI 模型连接不同的数据源和工具提供了标准化方式。
MCP Server&Client：MCP 通过 Server+Client 在 AI 应用程序和数据之间搭起了一座桥梁，MCP Server 负责打通 data、tools，AI 应用程序通过 MCP Client 连接 MCP Server。其中 Client 与 Server 的通信基于 SSE 协议实现，而 MCP 亲和奥秘也恰恰隐藏在 SSE 协议之中，下文将详细介绍。
1.png

SSE协议：作为 HTTP/1.1扩展协议，SSE（Server-Sent Events）定义了结构化流式传输规范，允许服务器实时向客户端推送事件或数据更新。在接收到订阅请求后，服务器会保持连接开启，并通过 HTTP 流向客户端推送事件，其中每个流数据格式如下：
  1. event: <event-type>       // 事件类型（endpoint/message/close）
  2. data: <payload>           // 通过数据字段传递实际内容
  3. id: <message-id>          // 用于唯一标识事件
  4. retry: <milliseconds>     // 重连退避策略
复制代码
2.png

函数计算：函数计算是事件驱动的全托管计算服务。使用函数计算，您无需采购与管理服务器等基础设施，只需编写并上传代码或镜像。函数计算为您准备好计算资源，弹性地、可靠地运行任务，并提供日志查询、性能监控和报警等功能。
MCP交互协议解析

在解析 MCP 亲和性实现原理前，需重点解析其基于 SSE（Server-Sent Events）构建的通信框架。该协议通过定义标准化事件类型，实现了客户端-服务端的交互控制及会话保持机制，具体流程如下：

  • 会话建立阶段：

    • 客户端发起初始 SSE 连接请求；
    • 服务端通过 event:'endpoint' 事件响应，在 data 字段中嵌入唯一会话标识（Session ID）；

  • 请求保持机制

    • 后续所有客户端请求必须携带该 Session ID；
    • 服务端通过该标识验证请求来源的合法性；
    • 实现客户端与服务端实例的绑定关联（Session Affinity）；

  • 实例绑定校验：

    • 当 messages 请求的路由目标实例与SSE连接绑定实例不一致时；
    • 服务端将触发安全校验失败机制，返回 4xx Conflict 错误代码；

该设计通过事件驱动架构确保了会话状态的连续性，同时通过实例绑定校验机制保障了分布式环境下的请求一致性。需要注意的是，Session ID 的有效期与 SSE 连接生命周期严格绑定，连接中断后需重新进行会话协商，一个简单的交互流程示例如下：
3.png


  • Client 端发起一个 GET 请求，建立 SSE 长连接。（Connection1）
  • Server 端回复event：endpoint类型的事件，将 sessionId 信息放入 data 中返回。（Connection1）
  • Client 端使用第2步返回的 sessionId 信息发起首个 HTTP POST 请求。（Connection2）
  • Server 端迅速响应202，但无内容。（Connection2）
  • Server 端返回第3步请求的实际消息。（Connection1）
  • Client 端使用第2步返回的 sessionId 发起 HTTP POST 请求initialized作为确认。（Connection3）
  • Server 端迅速响应202，无内容。（Connection3）
  • Client 端使用第2步返回的 sessionId 发起 HTTP POST 请求list tools。（Connection4）
  • Server 端迅速响应202，无内容。（Connection4）
  • Server 端返回第8步请求的实际消息，即工具列表。（Connection1）
  • Client 端使用第2步返回的 sessionId 发起 HTTP POST 请求call tool。（Connection5）
  • Server 端迅速响应202，无内容。（Connection5）
  • Server 端返回第11步请求的实际消息，即工具调用结果。（Connection1）
亲和性机制解析

宏观分类与核心价值
系统亲和性主要分为两大维度：

  • 节点亲和性：面向资源调度场景，确保工作负载优先部署至符合标签规则的节点。
  • 会话亲和性：面向请求路由场景，保障客户端流量持续定向到特定后端实例。
两类机制均通过属性一致性调度实现核心价值：

  • 提升局部资源复用率（如缓存命中）
  • 保障有状态业务连续性
  • 满足合规性数据路由要求
会话亲和性实现范式
而常见的会话亲和性主要有以下几类：

  • Cookie 植入模式：首请求时 LB、网关类服务注入含后端标识的 Set-Cookie 头，后续请求基于 Cookie 值进行会话绑定。适用 HTTP 无状态协议场景，且客户端缺乏显式标识信息。
  • 源 IP 哈希模式：基于 ClientIP 哈希值映射到后端特定节点，满足 TCP/UDP 四层流量及需要客户端级会话保持的场景。
  • Header 字段路由模式：预定义 Header 字段值提取（如 X-Session-ID），并哈希计算生成目标映射，支持多客户端标识共存场景，满足细粒度路由策略。
MCP SSE 亲和性架构特性
MCP SSE 采用双阶段协商机制：

  • 会话建立阶段：MCP Server 生成全局唯一 SessionID 并同步至客户端
  • 请求路由阶段：网关通过专有协议实时获取 Session-Node 映射关系
该模式需网关层与 MCP Server 间实现会话信息同步。相较于传统会话亲和方案，在获得精确路由控制能力的同时，需权衡协议交互带来的系统复杂度提升。
MCP ON FC 亲和调度设计

函数计算支持一键托管 MCP Server，并通过深度适配 MCP SSE 协议，提供了一种即开即用的 Serverless 亲和调度能力，帮助您实现 MCP 服务的 Serverless 托管能力，下面将详细介绍函数计算的亲和策略机制。
亲和策略

函数计算作为集调度、计算托管、免运维等特性于一身的 Serverless 服务，可将函数计算核心组件抽象为三部分：

  • Gateway：网关层，用户流量入口，负责接收用户请求、鉴权、流控等功能。
  • Scheduler：调度引擎层，负责将用户的请求调度到合适的节点和实例。
  • VMS：资源层，函数执行环境        。
当客户通过函数计算托管 MCP 服务并通过 MCP Client 发起请求时，可将用户请求分为两类：SSE 管控链路和 Message 数据链路。
SSE 管控链路（会话初始化）
4.png


  • Client 发起首个 SSE 请求路由到一台函数计算网关节点 Gateway1，网关节点权限校验通过后转发至调度模块 Scheduler。
  • 调度模块根据特定标识识别出请求类型为 SSE 时，将调度到一台可用实例。
  • 当请求和实例绑定时，实例将启动用户代码。
  • 用户代码启动完成后，会通过event：endpoint事件将 sessionId 放入 data 中，返回第一个数据包。
  • 在 response 返回经过 Gateway 网关层时，网关层将拦截 SSE 请求的首个回包，解析 SessionID 信息，并将 SessionID 和实例的映射关系持久化到DB。
Message 数据链路（请求处理）
5.png


  • Client 完成SSE请求后，将发起多个 Message 请求，由于函数计算网关节点无状态，Message 请求将打散到多个网关节点。
  • 当 Gateway 收到 Message 请求，将检查网关节点 cache 中是否存在 Message 请求携带的 SessionID 亲和信息，如果 cache 中无记录，将回源到 DB 获取相关数据。
  • Gateway 通过 cache 或 DB 拿到 SessionID 和实例的绑定关系时，将携带相关信息转发至调度模块。
  • 调度模块根据特定标识识别出请求类型为 Message 时，解析携带的实例信息，将请求定向调度到特定实例。
  • 当请求和实例绑定时，MCP Server 校验请求通过，将返回202通知 Client 请求接收成功，实际数据将通过 SSE 请求建立的连接返回。
函数计算通过无状态网关层与智能调度层的协同设计，在 Serverless 架构下创新实现了 MCP SSE 会话亲和性保障。SSE 管控链路借助首包拦截实现 SessionID 与实例的动态绑定，Message 数据链路则通过多级缓存与持久化存储确保请求精准路由。该架构既保留了函数计算的弹性优势，又攻克了无状态服务处理有状态请求的难题，为 MCP 场景提供了高可靠、低延迟的 Serverless 化解决方案，同时通过冷启动优化与智能扩缩容机制，实现资源效率与性能的最佳平衡。
MCP 场景会话配额控制体系

配额冲突建模分析
在 MCP 会话资源需求模型中，单个 Session 生命周期内存在两类并发需求：
6.png

而这种并发需求在传统配额分配中存在严重缺陷，需要引入一种动态预留的配额分配策略：
7.png

动态配额分配策略
为避免上述问题，函数计算引入 Session Quota 策略，即结合函数实例的并发度配置，限制每个实例最多绑定 Round (函数单实例多并发配置 / 10) 个 Session。如下流程所示：

  • 当函数配置了20并发时，可服务 20/10=2 个 Session 请求。
  • Client1 发起 SSE 请求时，分配了VM1实例，并占用1 Session Quota。
  • Client2 发起 SSE 请求时，Scheduler 计算 VM1 仍有1 Session Quota，成功和 VM1 再次完成绑定。
  • Client3 发起 SSE 请求时，Scheduler 计算 VM1 2个并发 Session Quota 已被2个 SSE 请求占用，无法再次绑定，则调度到新实例 VM2，完成实例绑定。
8.png

MCP 会话场景灰度优雅升级方案

函数计算支持 UpdateFunction 操作更新函数配置，在用户更新函数后，新的请求将路由到新配置拉起的实例，旧实例不再接收新请求，在处理完存量请求后后台自动销毁。如下图在 UpdateFunction 前，请求1-n路由到 VM1，UpdateFunction 后新请求路由到 VM2，VM1 在处理完存量请求后自动销毁。
9.png

在 MCP 场景下，数据请求从请求级无状态变为会话级绑定，在 UpdateFunction后，如果存量 Session 关联的请求路由到新实例，则新增无法识别到 SessionID 信息，返回错误。为解决这类问题，函数计算优雅更新能力从升级至有状态 Session 级别，在用户更新函数后，存量 Session 关联的请求仍路由到旧实例，新建 Session 请求路由至新实例，优雅实现 MCP 亲和场景下的升级需求。
10.png

压测

压测准备

我们以一个简单的 MCP Server 服务托管到函数计算作为压测对象， 函数代码如下：
  1. import random
  2. import asyncio
  3. from mcp.server.fastmcp import FastMCP
  4. from starlette.applications import Starlette
  5. from starlette.routing import Mount
  7. mcp = FastMCP("My App")
  9. @mcp.tool()
  10. async def add(a: int, b: int) -> int:
  11.     """计算两个整数的和（含150-1000ms随机延迟）"""
  12.     delay = random.uniform(0.15, 1.0)
  13.     await asyncio.sleep(delay)
  15.     print(f"add工具被调用，延迟 {delay:.3f}s")
  16.     return a + b
  18. app = Starlette(
  19.     routes=[
  20.         Mount("/", app=mcp.sse_app()),
  21.     ]
  22. )
复制代码
压测脚本 load_test.py 如下：
  1. # 并发 100 个 mcp client
  2. # 每次 client 建立一条 SSE 连接， 执行 call tool
  3. # 并且校验 call tool 的结果是符合预期的
  5. import asyncio
  6. from mcp.client.sse import sse_client
  7. from mcp import ClientSession
  8. import logging
  9. import time
  10. import random
  11. import traceback
  13. logging.basicConfig(level=logging.INFO)
  14. logger = logging.getLogger("LoadTestClient")
  16. class ErrorCounter:
  17.     def __init__(self):
  18.         self.count = 0
  19.         self._lock = asyncio.Lock()
  21.     async def increment(self):
  22.         async with self._lock:
  23.             self.count += 1
  25. err_counter = ErrorCounter()
  27. async def robust_client_instance(instance_id: int):
  28.     try:
  29.         async with sse_client(
  30.             "https://xxx-yyy-zzz.cn-hangzhou.fcapp.run/sse",
  31.             headers={
  32.                 "Authorization": "Bearer YOUR_API_KEY",
  33.                 "X-Instance-ID": str(instance_id),
  34.             },
  35.             timeout=10
  36.         ) as streams:
  37.             if streams is None:
  38.                 raise ConnectionError("SSE连接失败")
  40.             async with ClientSession(
  41.                 read_stream=streams[0],
  42.                 write_stream=streams[1],
  43.             ) as session:
  44.                 # 初始化及调用逻辑
  45.                 start = time.time()
  46.                 await asyncio.wait_for(session.initialize(), timeout=10.0)
  47.                 logger.info(f"实例{instance_id}; initialize 耗时: {time.time() - start}")
  49.                 try:
  50.                     # 设置call_tool超时
  51.                     start = time.time()
  52.                     random_number = random.randrange(1, 51)
  53.                     result = await asyncio.wait_for(
  54.                         session.call_tool(
  55.                             "add", {"a": instance_id, "b": random_number}
  56.                         ),
  57.                         timeout=10.0,
  58.                     )
  59.                     logger.info(
  60.                         f"实例{instance_id};  call_tool 耗时: {time.time() - start}"
  61.                     )
  63.                     if result.isError:
  64.                         logger.error(
  65.                             f"实例{instance_id}, call_tool 调用失败: {result.content}"
  66.                         )
  67.                         raise Exception(
  68.                             f"实例{instance_id}, call_tool 调用失败: {result.content}"
  69.                         )
  70.                     else:
  71.                         logger.info(
  72.                             f"实例{instance_id}, call_tool 调用成功: {result.content}"
  73.                         )
  74.                         assert (
  75.                             int(result.content[0].text)
  76.                             == instance_id + random_number
  77.                         )
  78.                 except asyncio.TimeoutError:
  79.                     logger.error(f"实例{instance_id} 操作超时")
  82.     except asyncio.TimeoutError:
  83.         await err_counter.increment()
  84.         logger.error(f"实例{instance_id} 操作超时")
  85.     except Exception as e:
  86.         await err_counter.increment()
  87.         logger.error(f"实例{instance_id} 失败: {str(e)}")
  88.         logger.debug(traceback.format_exc())
  90. async def main():
  91.     BATCH_SIZE = 100
  92.     tasks = [robust_client_instance(j) for j in range(0, BATCH_SIZE)]
  93.     await asyncio.gather(*tasks, return_exceptions=True)
  95.     print(f"总错误数: {err_counter.count}")
  97. if __name__ == "__main__":
  98.     asyncio.run(main())
复制代码
压测结果：

执行压测命令,  并行启动 3 个压测进程， 每次压测进程并发启动 100 mcp client,  每次 client 建立一条 SSE 连接， 执行 call tool,  并且校验 call tool 的结果是符合预期的
  1. python load_test.py & python load_test.py & python load_test.py & wait
复制代码
3 个压测进程， 每个进程中的 100 个并发的 mcp client 全部成功执行

表示 SSE 长连接和后续配合该会话的 HTTP 请求（call tool）在同一个函数实例，实现亲和行为
11.png

MCP Server 函数实例从 0 毫秒级扩容到 15 个实例，实现有状态实例水平横向扩展

根据官方文档 MCP SSE亲和性调度 描述，该函数设置的单实例并发度为 200，那么单个实例支持的 session 数目（即 SSE 长连接个数为 20个）， 300/20 = 15 实例
12.png

测试更新函数， MCP 会话灰度优雅升级

我们尝试在持续压测过程中，中途更新函数,   所有的 mcp client 的行为都符合预期无报错。
我们使用如下脚本模拟持续压测：
  1. ...
  2. # 压测（分50批）
  3. BATCH_SIZE = 100
  4. for i in range(0, 5000, BATCH_SIZE):
  5.     tasks = [robust_client_instance(j) for j in range(0, BATCH_SIZE)]
  6.     await asyncio.gather(*tasks, return_exceptions=True)
  7.     await asyncio.sleep(1)  # 批次间间隔
复制代码
总结

通过系列技术方案的精妙设计，函数计算为 MCP 场景构筑了 Serverless 化的完整技术基座。在 Serverless 服务范式与有状态业务需求之间架起智能桥梁，通过会话亲和调度引擎、优雅会话升级、动态配额熔断等机制，让 Serverless 的极致弹性与有状态服务的强一致性实现了深度融合。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
敕码
2025-6-3 00:18:12

0

粉丝关注

20

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫