当前位置: 首页 > ds >正文

【LLM】深入解析MCP的三种传输方式实现

引言

在Model Context Protocol (MCP) 定义了多种传输方式以适应不同场景需求。本文将详细解析MCP的三种主要传输方式:标准输入输出(stdio)、服务器发送事件(SSE)和可流式HTTP(Streamable HTTP),帮助开发者根据实际需求选择最佳实现方案

1. 标准输入输出(stdio)模式

标准输入输出模式是最基础的传输方式,特别适合命令行工具和简单的交互场景。

实现逻辑


# 初始化服务器
app = Server("mcpServer")async def run_stdio():"""运行标准输入输出模式的服务器使用标准输入输出流(stdio)运行服务器,主要用于命令行交互模式Raises:Exception: 当服务器运行出错时抛出异常"""from mcp.server.stdio import stdio_serverasync with stdio_server() as (read_stream, write_stream):try:await app.run(read_stream,write_stream,app.create_initialization_options())except Exception as e:print(f"服务器错误: {str(e)}")raiseif __name__ == "__main__":asyncio.run(run_stdio())

核心特点:

  • 使用标准输入(stdin)和标准输出(stdout)进行通信

  • 实现简单,无需额外网络配置

  • 适合本地调试和CLI工具开发

  • 通过stdio_server()上下文管理器管理流资源

适用场景:

  • 本地开发调试

  • 命令行工具集成

  • 简单的进程间通信

2. 服务器发送事件(SSE)模式

SSE模式利用HTTP长连接实现服务器向客户端的单向实时通信。

该传输方式在 2024-11-05 协议版本开始替换为了 HTTP+SSE 传输,及第三中streamableHttp方式

实现逻辑

# 初始化服务器
app = Server("mcpServer")def run_sse():"""运行SSE(Server-Sent Events)模式的服务器启动一个支持SSE的Web服务器,允许客户端通过HTTP长连接接收服务器推送的消息服务器默认监听0.0.0.0:9000"""sse = SseServerTransport("/messages/")async def handle_sse(request):"""处理SSE连接请求Args:request: HTTP请求对象"""async with sse.connect_sse(request.scope, request.receive, request._send) as streams:await app.run(streams[0], streams[1], app.create_initialization_options())starlette_app = Starlette(debug=True,routes=[Route("/sse", endpoint=handle_sse),Mount("/messages/", app=sse.handle_post_message)],)uvicorn.run(starlette_app, host="0.0.0.0", port=9000)if __name__ == "__main__":run_sse()

核心特点:

  • 基于HTTP/1.1的长连接技术

  • 服务器可主动向客户端推送消息

  • 使用SseServerTransport处理SSE连接

  • 提供/sse端点建立连接,/messages/端点接收消息

适用场景:

  • 需要服务器主动推送但不需要双向通信的场景

3. 可流式HTTP(Streamable HTTP)模式

可流式HTTP模式提供了更灵活的HTTP双向通信能力。

实现逻辑

有状态的实现方式

# 初始化服务器
app = Server("mcpServer")def run_streamable_http(json_response: bool):event_store = InMemoryEventStore()# 使用我们的应用和事件存储创建会话管理器session_manager = StreamableHTTPSessionManager(app=app,event_store=event_store,json_response=json_response,)# 用于可流式HTTP连接的ASGI处理器async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:await session_manager.handle_request(scope, receive, send)@contextlib.asynccontextmanagerasync def lifespan(app: Starlette) -> AsyncIterator[None]:"""用于管理会话管理器生命周期的上下文管理器。"""async with session_manager.run():yield# 使用传输创建ASGI应用程序starlette = Starlette(debug=True,routes=[Mount("/mcp", app=handle_streamable_http)],lifespan=lifespan,)uvicorn.run(starlette, host="0.0.0.0", port=3000)if __name__ == "__main__":run_streamable_http(False)

无状态的实现方式

# 初始化服务器
app = Server("mcpServer")def run_streamable_http(json_response: bool):# 使用我们的应用和事件存储创建会话管理器session_manager = StreamableHTTPSessionManager(app=app,event_store=None,json_response=json_response,stateless=True,)# 用于可流式HTTP连接的ASGI处理器async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:await session_manager.handle_request(scope, receive, send)@contextlib.asynccontextmanagerasync def lifespan(app: Starlette) -> AsyncIterator[None]:"""用于管理会话管理器生命周期的上下文管理器。"""async with session_manager.run():yield# 使用传输创建ASGI应用程序starlette = Starlette(debug=True,routes=[Mount("/mcp", app=handle_streamable_http)],lifespan=lifespan,)uvicorn.run(starlette, host="0.0.0.0", port=3000)if __name__ == "__main__":run_streamable_http(False)

核心特点:

  • 有状态的方式使用InMemoryEventStore存储事件,另外实现的持久化方式

  • 通过StreamableHTTPSessionManager管理会话

  • 支持JSON或原始格式响应(json_response参数控制)

  • 提供生命周期管理(lifespan)确保资源正确释放

  • 所有请求通过/mcp端点处理

适用场景:

  • 需要双向通信的Web应用

  • 需要持久化会话管理的应用--使用有状态的实现方式

三种传输方式对比

特性stdio模式SSE模式可流式HTTP模式
通信方向双向服务器→客户端双向
协议支持标准输入输出HTTP/1.1HTTP/1.1+
实时性
复杂度
适用场景CLI工具/调试服务器推送通知复杂双向通信

最佳实践建议

  1. 开发阶段:建议从stdio模式开始,便于调试和快速验证逻辑

  2. 简单推送场景:选择SSE模式,实现简单且资源消耗低

  3. 复杂交互应用:采用可流式HTTP模式,提供完整的双向通信能力

  4. 性能考量

    • stdio模式性能最高但适用场景有限

    • SSE适合大量客户端连接但只需服务器推送的场景

    • 可流式HTTP提供最灵活的功能但资源消耗相对较高

结语

        MCP提供的三种传输方式各具特色,能够满足不同场景下的通信需求。目前小编已经编写好三种方式的demo,大家可以开箱即用啦!觉得不错的话,也帮小编点个star支持一下~

  •        纯净的简易框架

使开发者快速的搭建一个支持mcp协议所有传输方式(STDIO、SSE、StreamableHttp)的mcp server服务框架Enable developers to quickly build an MCP server service framework that supports stdio and sse。使开发者快速的搭建一个支持stdio与sse的mcp server服务框架 - wenb1n-dev/easyMcphttps://github.com/wenb1n-dev/easyMcp

  •         具体实现的案例

mysql的增删改查功能; 还包含了数据库异常分析能力;且便于开发者们进行个性化的工具扩展https://github.com/wenb1n-dev/mysql_mcp_server_pro.git       详情可看: 【LLM】使用MySQL MCP Server让大模型轻松操作本地数据库-CSDN博客

http://www.xdnf.cn/news/13492.html

相关文章:

  • 《C++ 继承》
  • 2024年12月6级第一套
  • 【HarmonyOS 5.0】开发实战:从UI到Native全解析
  • 鸿蒙多语言开发实战:3 步实现中英文动态切换(无需重启 App)附完整代码 + 避坑指南
  • CentOS7下的集群化部署
  • 电子接口与微控制器核心知识:串口、并口、USB、UART、RS232/RS485、ESP32与STM32详解
  • 零基础学前端-传统前端开发(第二期-HTML介绍与应用)(XSS防御)
  • C# StringBuilder代码中预分配容量的作用
  • 企业中使用 MCP Server 实现业务打通
  • (二)TensorRT-LLM | 模型导出(v0.20.0rc3)
  • 第一讲:认识C++程序
  • 《网络世界的“隐形窥探者”:深度剖析网络监听》
  • 系统设计 --- MongoDB亿级数据查询优化策略
  • MMaDA: Multimodal Large Diffusion Language Models
  • Vue3实现键盘字母筛选功能
  • Java 中高级开发岗技能与面试要点梳理
  • LLM基础6_在未标记数据上进行预训练
  • HTML盒子模型
  • 1.一起学习仓颉-编译环境,ide,输出hello,world
  • GitLab Web 界面创建分支后pathspec ... did not match any file(s)
  • MNIST数据集上朴素贝叶斯分类器(MATLAB)
  • 扁平表+递归拼树思想
  • cf2117E
  • 【Pandas】pandas DataFrame interpolate
  • echarts 数据大屏(无UI设计 极简洁版)
  • [2025CVPR]DeepVideo-R1:基于难度感知回归GRPO的视频强化微调框架详解
  • 黄晓军所长:造血干细胞移植后晚期效应及患者健康相关生存质量
  • SQL进阶之旅 Day 23:事务隔离级别与性能优化
  • CentOS 安装Python 3教程
  • 38 C 语言字符串搜索与分割函数详解:strchr、strrchr、strpbrk、strstr、strcspn、strtok