将 MCP Streamable HTTP 服务器嵌入现有 ASGI 应用python-sdk 集成实战指南【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdkmcp.run(streamable-http)会替你启动一个完整的 Web 服务器但在真实工程里MCP 服务器往往只是大型 Web 应用的一个组成部分——你可能已经有一套 ASGI 部署体系uvicorn、Hypercorn、FastAPI或希望把工具服务挂到现有域名和网关后面。本文基于 python-sdk 的 docs/run/asgi.md 与对应源码系统讲解如何用mcp.streamable_http_app()把 MCP 服务器包装成标准的 Starlette/ASGI 应用并覆盖挂载Mount/Host、生命周期lifespan、路径改写、CORS 与自定义路由等全部实战细节。读完你可以把任意数量的 MCP 服务器安全地嵌进现有 Starlette/FastAPI 应用中并准确处理浏览器客户端的跨域与会话问题。从一个独立的 ASGI 应用开始MCPServer.streamable_http_app()返回一个Starlette 应用源码见 src/mcp/server/mcpserver/server.py底层实现在 src/mcp/server/lowlevel/server.py。Starlette 应用本身就是 ASGI 应用因此任何能承载 ASGI 的服务器——uvicorn、Hypercorn、另一个 Starlette、FastAPI——都能直接托管你的 MCP 服务器。最小的完整示例取自 docs_src/asgi/tutorial001.pyfrom mcp.server import MCPServer mcp MCPServer(Notes) mcp.tool() def add_note(text: str) - str: Save a note. return fSaved: {text} app mcp.streamable_http_app()app就是普通的 ASGI 应用交给任何 ASGI 服务器即可uvicorn server:appMCP 端点位于/mcp所以客户端连接地址为http://127.0.0.1:8000/mcp。从源码看这个应用自带了两个关键结构src/mcp/server/lowlevel/server.py一条路由/mcp注册的是StreamableHTTPASGIApp(session_manager)即 Streamable HTTP 传输的 ASGI 处理器源码中为Route(streamable_http_path, endpointstreamable_http_app)一个 lifespanStarlette(..., lifespanlambda app: session_manager.run())它启动mcp.session_manager——管理所有存活会话后台工作的对象。直接运行uvicorn server:app时两者都被自动处理你完全不用关心。参数与mcp.run(streamable-http, ...)的关系streamable_http_app()接受与mcp.run(streamable-http, ...)相同的关键字参数唯独没有port——端口属于承载应用的服务器。host参数仍然接受但在这里并不真正绑定任何地址它只参与决定默认传输安全策略详见下文。完整签名来自 src/mcp/server/lowlevel/server.py参数默认值作用streamable_http_path/mcpStreamable HTTP 端点路径json_responseFalse是否以 JSON 而非 SSE 流返回响应stateless_httpFalse是否启用无状态 HTTP 模式event_storeNone服务端推送事件存储retry_intervalNone客户端重连建议间隔max_request_body_sizeDEFAULT_MAX_REQUEST_BODY_SIZE请求体大小上限字节session_idle_timeoutDEFAULT_SESSION_IDLE_TIMEOUT会话空闲超时max_sessionsDEFAULT_MAX_SESSIONS最大并发会话数transport_securityNone传输层安全主机/来源白名单设置host127.0.0.1参与默认安全策略判定不绑定端口完整的参数语义如json_response、stateless_http、重试与超时策略请参考 docs/run/index.md生产环境的安全配置详见 docs/run/deploy.md。旧版 SSE 传输对应的方法为mcp.sse_app()行为一致默认端点/sse、消息端点/messages/但它承载的是已被 Streamable HTTP 取代的旧传输。默认只应答 localhostDNS 重绑定防护开箱即用状态下该应用只应答发往 localhost 的请求。原因在于streamable_http_app()无法预知自己会被部署在哪个主机名下于是以最安全的允许列表启用了 DNS 重绑定防护在本机开发时这恰好是正确行为。从源码可以看到具体逻辑src/mcp/server/lowlevel/server.py当未显式传入transport_security且host为127.0.0.1、localhost或::1时自动构造TransportSecuritySettings( enable_dns_rebinding_protectionTrue, allowed_hosts[127.0.0.1:*, localhost:*, [::1]:*], allowed_origins[http://127.0.0.1:*, http://localhost:*, http://[::1]:*], )部署到真实主机名后面时这意味着每个请求都会被421 Misdirected Request拒绝直到你通过transport_security传入与实际提供服务的主机对应的白名单。注意应用自身构建的任何路由都不会被先行检查——安全策略在路由匹配之前就已生效。白名单配置以及从能跑的应用到真实主机名之间的所有步骤都由 docs/run/deploy.md 负责讲解。挂载到更大的应用中Mount、Host 与丢失的 lifespan一旦 MCP 服务器成为更大应用的一部分你就需要把streamable_http_app()放进 Starlette 的Mount中。而挂载的那一刻生命周期就成了你的责任from collections.abc import AsyncIterator from contextlib import asynccontextmanager from starlette.applications import Starlette from starlette.routing import Mount from mcp.server import MCPServer mcp MCPServer(Notes) mcp.tool() def add_note(text: str) - str: Save a note. return fSaved: {text} asynccontextmanager async def lifespan(app: Starlette) - AsyncIterator[None]: async with mcp.session_manager.run(): yield app Starlette( routes[Mount(/, appmcp.streamable_http_app())], lifespanlifespan, )这里有三条容易踩坑的规则路径叠加Mount(/, ...)加上默认的/mcp路径端点仍保持在/mcp。Starlette 按顺序匹配路由Mount(/)会匹配任意路径所以你自己的路由必须放在它之前其后的一切都将无法访问。生命周期是核心lifespan函数在宿主应用的整个生命周期内进入mcp.session_manager.run()。这是最容易被遗忘的一行。session_manager的惰性创建mcp.session_manager只在streamable_http_app()被调用之后才存在源码中它在streamable_http_app()内被赋值给self._session_manager见 src/mcp/server/lowlevel/server.py。因此路由必须在模块级构建而 manager 只能在 lifespan 内部被触碰。若在调用streamable_http_app()之前访问session_manager会抛出RuntimeErrorsrc/mcp/server/lowlevel/server.py。警告宿主应用拥有 lifespanstreamable_http_app()把session_manager.run()接入了它返回的 Starlette 的 lifespan但被挂载的子应用的 lifespan 永远不会运行。挂载之后内置的 lifespan 就成了死代码。无论谁位于 ASGI 栈的顶层都必须在自己的 lifespan 中进入mcp.session_manager.run()。一个快速验证方法删掉lifespanlifespan这一行再启动服务器。服务器能正常启动路由也能解析但第一个发往/mcp的请求就会失败RuntimeError: Task group is not initialized. Make sure to use run().从 src/mcp/server/streamable_http_manager.py 可以看到run()创建了支撑所有会话操作的 task group并保证每个实例只能调用一次重复调用会抛出RuntimeError需要重启进程并新建实例。没有任何其他路径能启动 session manager——除了它的run()。按主机名路由Host 路由Starlette 的Host路由工作方式相同把Mount(/, ...)换成Host(mcp.example.com, ...)即可按主机名而非路径路由。lifespan 规则不变传输安全规则也不变。Host(mcp.example.com, ...)路由只会收到发往该主机名的请求但传输层自身的 Host 白名单见 docs/run/deploy.md仍然先行执行——如果白名单里没有mcp.example.com该路由会对每个请求都回以421。一个应用挂载多个服务器每个MCPServer都是拥有独立 session manager 的独立应用。你可以按需挂载任意多个并在唯一的宿主 lifespan 中进入每一个 manager取自 docs_src/asgi/tutorial003.pyfrom collections.abc import AsyncIterator from contextlib import AsyncExitStack, asynccontextmanager from starlette.applications import Starlette from starlette.routing import Mount from mcp.server import MCPServer notes MCPServer(Notes) tasks MCPServer(Tasks) notes.tool() def add_note(text: str) - str: Save a note. return fSaved: {text} tasks.tool() def add_task(title: str) - str: Create a task. return fCreated: {title} asynccontextmanager async def lifespan(app: Starlette) - AsyncIterator[None]: async with AsyncExitStack() as stack: await stack.enter_async_context(notes.session_manager.run()) await stack.enter_async_context(tasks.session_manager.run()) yield app Starlette( routes[ Mount(/notes, appnotes.streamable_http_app()), Mount(/tasks, apptasks.streamable_http_app()), ], lifespanlifespan, )要点AsyncExitStack同时进入两个 manager它们一起启动并按逆序优雅关闭端点为/notes/mcp和/tasks/mcp挂载前缀加上默认路径。修改端点路径结尾的/mcp来自streamable_http_path参数。把它设为/挂载前缀本身就成为完整的对外路径取自 docs_src/asgi/tutorial004.pyapp Starlette( routes[Mount(/notes, appmcp.streamable_http_app(streamable_http_path/))], lifespanlifespan, )此时客户端连接到/notes/而不是/notes/mcp。浏览器客户端的 CORS 配置基于浏览器的客户端需要你授予两类权限发送MCP 请求头以及读取MCP 返回的响应头。两者都是宿主应用上的 CORS 配置并且要与上文的传输安全白名单保持一致取自 docs_src/asgi/tutorial005.pyfrom starlette.applications import Starlette from starlette.middleware import Middleware from starlette.middleware.cors import CORSMiddleware from starlette.routing import Mount from mcp.server import MCPServer from mcp.server.transport_security import TransportSecuritySettings mcp MCPServer(Notes) mcp.tool() def add_note(text: str) - str: Save a note. return fSaved: {text} asynccontextmanager async def lifespan(app: Starlette) - AsyncIterator[None]: async with mcp.session_manager.run(): yield security TransportSecuritySettings( allowed_hosts[mcp.example.com, mcp.example.com:*], allowed_origins[https://app.example.com], ) app Starlette( routes[Mount(/, appmcp.streamable_http_app(transport_securitysecurity))], middleware[ Middleware( CORSMiddleware, allow_origins[https://app.example.com], allow_methods[GET, POST, DELETE], allow_headers[ Authorization, Content-Type, Last-Event-ID, Mcp-Method, Mcp-Name, Mcp-Protocol-Version, Mcp-Session-Id, ], expose_headers[Mcp-Session-Id], ) ], lifespanlifespan, )四个要点allow_headers是最容易被遗忘的一半。浏览器会对每个 MCP 请求执行preflight预检因为Content-Type: application/json和Mcp-*请求头不在 CORS 的默认安全名单上预检未授权的请求头浏览器永远不会真正发出。allow_headers[*]也有效Starlette 会按预检请求所问的内容作答。expose_headers[Mcp-Session-Id]是读取的一半。Streamable HTTP 会在该响应头中返回会话 ID而浏览器默认不向 JavaScript 暴露响应头除非 CORS 按名称显式暴露。缺少这一项客户端将永远无法发出第二个请求。allow_origins是你的决定不是 MCP 的。务必写精确并在上文的allowed_origins中保持一致浏览器强制执行 CORS但服务器自己也检查Origin——即使预检顺利通过传输层不信任的来源仍会得到403。allow_methods列出 Streamable HTTP 使用的三种方法POST发送消息、GET打开服务端到客户端的流、DELETE结束会话。自定义路由健康检查与 OAuth 回调mcp.custom_route()在同一应用上注册一个普通 HTTP 端点用于承载每个部署服务都需要但与 MCP 无关的事情——健康检查、OAuth 回调等取自 docs_src/asgi/tutorial006.pyfrom starlette.requests import Request from starlette.responses import JSONResponse, Response from mcp.server import MCPServer mcp MCPServer(Notes) mcp.tool() def add_note(text: str) - str: Save a note. return fSaved: {text} mcp.custom_route(/health, methods[GET]) async def health(request: Request) - Response: return JSONResponse({status: ok}) app mcp.streamable_http_app()行为要点处理器是纯 Starlette 风格一个接收Request、返回Response的async函数streamable_http_app()会收集所有自定义路由源码中custom_starlette_routes被追加到路由列表末尾见 src/mcp/server/mcpserver/server.py 与 src/mcp/server/lowlevel/server.py。此时app.routes为/mcp和/healthGET /health返回{status: ok}与 MCP 完全无关。自定义路由的其他参数来自 src/mcp/server/mcpserver/server.py参数默认值说明path必填路由路径如/oauth/callbackmethods必填支持的 HTTP 方法列表如[GET, POST]nameNone路由名称供 Starlette 反向 URL 查找使用include_in_schemaTrue是否包含进 OpenAPI schema警告自定义路由永不鉴权。即使服务器其余部分启用了认证自定义路由也不会被保护。这是有意为之健康检查和 OAuth 回调在任何 token 存在之前就必须可达。不要把任何私有内容放在这类路由后面。小结mcp.streamable_http_app()返回一个带/mcp路由的 Starlette 应用任何 ASGI 服务器都能运行它开箱即用时应用只应答 localhost 请求部署到真实主机名后所有请求会被421拒绝直到通过transport_security传入白名单——这部分及通往生产的其余环节由 docs/run/deploy.md 负责Mount或Host把应用放进更大的 Starlette/FastAPI 应用挂载会禁用内置 lifespan宿主应用的 lifespan 必须进入mcp.session_manager.run()否则第一个请求就会失败一个应用承载多个服务器 多个挂载 一个进入所有 session manager 的 lifespan可用AsyncExitStackstreamable_http_path/把端点移到挂载前缀本身浏览器客户端需要 CORSallow_headers放行Mcp-*请求头expose_headers[Mcp-Session-Id]暴露响应头mcp.custom_route()在/mcp旁边添加普通、未鉴权的 HTTP 端点健康检查、OAuth 回调。一旦服务器在真实 URL 上可达客户端即可用该 URL 连接参见 docs/client/index.md 中的客户端文档。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考