第四章 FastMCP 服务器框架之SSE 传输集成
目录
- 引言
- SSE 传输协议核心实现
- 双向通信机制分析
- 会话关联与路由机制
- DNS 重绑定防护安全中间件
- Web 服务器环境部署配置
- 结论
引言
SSE（Server-Sent Events）传输协议在FastMCP框架中实现了高效的双向通信机制。通过SseServerTransport类提供的两个ASGI应用，系统能够建立持久化的服务器到客户端消息推送通道，并接收客户端的指令请求。本文档深入解析该实现的核心机制，包括会话关联、路由配置和安全防护等关键方面。
SSE 传输协议核心实现
SseServerTransport类是SSE传输协议的核心实现，提供了两个ASGI应用来支持双向通信。该类通过内存流（memory object stream）机制管理客户端会话，并利用UUID作为会话标识符。
本节来源
- SseServerTransport
双向通信机制分析
连接建立与消息推送
connect_sse方法负责建立SSE连接并推送服务器消息。当客户端发起GET请求时，该方法会创建一个新的会话，并通过EventSourceResponse向客户端发送事件流。
图源
- SseServerTransport
客户端指令接收
handle_post_message方法处理客户端通过POST请求发送的指令。该方法会验证会话ID，并将客户端消息传递给相应的处理流。
图源
- SseServerTransport
本节来源
- SseServerTransport
会话关联与路由机制
Session ID 机制
SseServerTransport使用UUID作为会话标识符，确保每个SSE连接都有唯一的会话ID。该ID通过查询参数session_id与客户端POST请求关联，实现双向通信的会话绑定。
session_id = uuid4()
self._read_stream_writers[session_id] = read_stream_writer
client_post_uri_data = f"{quote(full_message_path_for_client)}?session_id={session_id.hex}"
Mount Path 路由影响
mount_path参数影响SSE端点的最终路由。通过_normalize_path方法，系统将挂载路径与消息端点路径组合，生成客户端使用的完整路径。
图源
- SseServerTransport
本节来源
- SseServerTransport
DNS 重绑定防护安全中间件
TransportSecurityMiddleware实现了DNS重绑定攻击防护，通过验证请求头确保安全性。
安全验证流程
图源
- TransportSecurityMiddleware
配置选项
安全中间件支持以下配置：
enable_dns_rebinding_protection: 启用DNS重绑定防护allowed_hosts: 允许的Host头值列表allowed_origins: 允许的Origin头值列表
本节来源
- TransportSecurityMiddleware
Web 服务器环境部署配置
Starlette 集成示例
在Starlette应用中集成SSE传输的典型配置如下：
# 创建SSE传输实例
sse = SseServerTransport("/messages/")
# 配置路由
routes = [
Route("/sse", endpoint=handle_sse, methods=["GET"]),
Mount("/messages/", app=sse.handle_post_message),
]
# 处理SSE连接
async def handle_sse(request):
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await app.run(
streams[0], streams[1], app.create_initialization_options()
)
return Response() # 必须返回Response避免TypeError
FastMCP 集成示例
使用FastMCP框架的简化集成：
from mcp.server.fastmcp import FastMCP
# 创建MCP服务器
mcp = FastMCP("My App")
# 添加工具
@mcp.tool()
def hello() -> str:
return "Hello from MCP!"
# 挂载StreamableHTTP服务器
app = Starlette(
routes=[
Mount("/", app=mcp.streamable_http_app()),
]
)
本节来源
- streamable_http_basic_mounting.py
- echo.py
结论
SSE传输协议在FastMCP中的实现提供了一套完整的双向通信解决方案。通过SseServerTransport类的两个ASGI应用，系统实现了服务器消息推送和客户端指令接收的分离。会话ID机制确保了通信的关联性，而安全中间件则提供了必要的防护。在Web服务器环境中，通过简单的路由配置即可完成集成，为实时通信应用提供了高效可靠的基础设施。
