将 MCP Python SDK 服务器挂载到现有 ASGI 应用streamable_http_app 完整实战指南【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk本指南基于官方 Python SDKMCP Python SDK的 run/asgi 文档讲解如何放弃mcp.run(streamable-http)自带的独立 Web 服务器改用mcp.streamable_http_app()返回的 Starlette 应用把 MCP 服务器无缝嵌入 uvicorn、Hypercorn、FastAPI 等任意 ASGI 宿主。读完本文你将掌握独立挂载与Mount/Host子应用挂载的区别、session manager 生命周期归属的坑、多服务器单应用编排、端点路径定制、浏览器客户端的 CORS 配置以及自定义 HTTP 路由。为什么需要把 MCP 服务器交给 ASGI 宿主mcp.run(streamable-http)会直接为你启动一个 Web 服务器内部使用 uvicorn见 src/mcp/server/mcpserver/server.py 附近的uvicorn.Server启动逻辑。但很多场景下你并不想要一个独立进程MCP 服务器只是更大 Web 应用的一个组成部分例如一个同时提供 REST API 和 MCP 能力的应用你已经有一套 ASGI 部署体系uvicorn 启动脚本、Hypercorn、容器编排、进程管理器不想为 MCP 单独再开一个端口和管理进程。此时 SDK 提供了mcp.streamable_http_app()它返回一个Starlette 应用。而 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。streamable_http_app 返回的应用内部携带了什么streamable_http_app()返回的应用自带两样东西使你独立运行时完全不用操心一条路由/mcp即 Streamable HTTP 端点。测试 tests/docs_src/test_asgi.py 断言tutorial001.app.routes恰好只有一条Route路径为/mcp。一个 lifespan生命周期它负责启动mcp.session_manager——管理每个活跃会话后台任务的对象。从源码 src/mcp/server/streamable_http_manager.py 可以看到session_manager.run()会创建 anyio task group 并进入应用 lifespan退出时取消任务组、清空会话实例与归属映射。单独运行uvicorn server:app时这两者都被自动管理你无需任何额外代码。与 mcp.run(streamable-http) 的参数关系streamable_http_app()接受与mcp.run(streamable-http, ...)相同的关键字参数唯独没有port端口属于承载应用的宿主服务器不归 MCP 管。host参数仍然接受但在这里并不绑定任何地址默认值为127.0.0.1见 src/mcp/server/mcpserver/server.py它真正控制的内容在《部署与扩展》docs/run/deploy.md中说明。测试 tests/docs_src/test_asgi.py 通过inspect.signature精确验证了参数集合streamable_http_path, json_response, stateless_http, event_store, retry_interval, max_request_body_size, session_idle_timeout, max_sessions, transport_security, host这些参数的行为均可通过测试验证例如max_request_body_size超限返回 HTTP 413tests/docs_src/test_asgi.pysession_idle_timeout与max_sessions会传入 session managertests/docs_src/test_asgi.py。此外mcp.sse_app()为已被取代的 SSE 传输提供同样的能力用法一致。默认只响应 localhost直到你显式放行开箱即用的应用只响应发往 localhost 的请求。streamable_http_app()无法预知自己会被部署在哪个主机名之后因此它默认启用 DNS rebinding 防护并配上最安全的主机白名单——在你本机开发时这完全正确。一旦部署到真实主机名之后在未向transport_security传入白名单之前每个请求都会被以421 Misdirected Request拒绝你构建的任何业务代码甚至都不会被执行到。这一点由TransportSecuritySettingssrc/mcp/server/transport_security.py中的enable_dns_rebinding_protection、allowed_hosts、allowed_origins字段控制。测试 tests/docs_src/test_asgi.py 证实默认应用对真实主机名返回421 Invalid Host header对外来 Origin 返回403 Invalid Origin header。白名单配置以及从能跑的应用到真实主机名之间的一切反向代理、TLS、端口绑定等都属于《部署与扩展》docs/run/deploy.md的范畴。把应用挂载进更大的应用Mount 与生命周期接管一旦 MCP 服务器成为更大应用的一部分你需要把它放进Mount。而做这件事的瞬间lifespan 就变成你自己的责任了。参考教程 docs_src/asgi/tutorial002.pyfrom 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(/)匹配所有路径因此你自己的路由必须放在它之前其后的一切都不可达。测试 tests/docs_src/test_asgi.py 实证了这一点Route(/about)放在Mount(/)之后访问返回 404放在之前则返回 200。lifespan 函数在宿主应用整个生命周期内进入mcp.session_manager.run()。这是所有人都会忘掉的一行。mcp.session_manager只有在调用streamable_http_app()之后才存在见 src/mcp/server/mcpserver/server.py此时才会创建StreamableHTTPSessionManager。这就是为什么路由在模块级构建而 session manager 只在 lifespan 内部被触碰。Host 路由按主机名而非路径路由Starlette 的Host路由工作原理相同把Mount(/, ...)换成Host(mcp.example.com, ...)即可按主机名而非路径路由。生命周期规则不变传输安全规则也不变。Host(mcp.example.com, ...)只接收发往该主机名的请求但传输层自身的 Host 白名单仍然先行执行——白名单里没有mcp.example.com时该路由对每个请求都回复421。警告宿主应用拥有生命周期streamable_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().除了run()方法没有任何东西能启动 session manager。测试 tests/docs_src/test_asgi.py 使用 httpx2 的 ASGITransport 直接复现了这一错误。多服务器单应用一个生命周期进入多个 session manager每个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挂载前缀 默认路径。测试 tests/docs_src/test_asgi.py 验证了挂载结构与两端服务器都能正常调用工具。自定义端点路径streamable_http_path末尾那个/mcp是streamable_http_path参数决定的。把它设为/挂载前缀本身就成为完整公开路径。参考教程 docs_src/asgi/tutorial004.pyfrom 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(/notes, appmcp.streamable_http_app(streamable_http_path/))], lifespanlifespan, )现在客户端连接到/notes/而不是/notes/mcp。测试 tests/docs_src/test_asgi.py 断言内部路由路径变为了/。浏览器客户端的 CORS 配置运行在浏览器里的 MCP 客户端需要你授予两项权限发送它的 MCP 请求头以及读取MCP 返回的响应头。两者都属于宿主应用的 CORS 配置并且要与上面的传输安全白名单保持一致。参考教程 docs_src/asgi/tutorial005.pyfrom collections.abc import AsyncIterator from contextlib import asynccontextmanager from 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 安全名单safelist中——预检没授予的头浏览器永远不会发出对应请求。allow_headers[*]也有效Starlette 会按预检所请求的内容应答。测试 tests/docs_src/test_asgi.py 用带Mcp-*头的真实预检验证了放行结果。expose_headers[Mcp-Session-Id]是读取的那一半。Streamable HTTP 通过这个响应头返回会话 ID而浏览器默认对 JavaScript 隐藏响应头除非 CORS 显式点名暴露。少了它客户端永远无法发出它的第二次请求例如携带会话 ID 的后续 initialize 之后的请求。allow_origins是你的决定不是 MCP 的。要写精确并把它镜像到上面的allowed_origins浏览器负责执行 CORS但服务器自身也会检查Origin头传输层不信任的 Origin 即使在干净的预检之后也会收到403。allow_methods列出 Streamable HTTP 使用的三种方法POST发送消息、GET打开服务器到客户端的流、DELETE结束会话。端到端测试 tests/docs_src/test_asgi.py 完整跑通了文档场景真实主机名 浏览器 Origin 带Mcp-*头的预检 实际请求最终mcp-session-id响应头存在且 CORS 头正确。自定义路由在同名应用上加普通 HTTP 端点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()会收集每一条自定义路由app.routes现在同时包含/mcp和/health见 tests/docs_src/test_asgi.py。GET /health返回{status: ok}完全没有任何 MCP 参与tests/docs_src/test_asgi.py。从源码 src/mcp/server/mcpserver/server.py 可以看到装饰器把路由追加到_custom_starlette_routes列表最终由streamable_http_app()组装进 Starlette 的routes。它还支持name参数用于 Starlette 的 URL 反向解析和include_in_schema是否进入 OpenAPI schema默认True。警告自定义路由永不鉴权自定义路由永远不会被认证即使服务器其余部分都启用了认证。这是刻意设计健康检查和 OAuth 回调必须在任何 token 存在之前就能被访问。因此不要把任何私有内容放在自定义路由后面。要点速查mcp.streamable_http_app()返回带单条/mcp路由的 Starlette 应用任何 ASGI 服务器都能运行它。默认应用只响应发往 localhost 的请求部署到真实主机名后在向transport_security传入白名单之前一切请求都会被421拒绝。这属于《部署与扩展》docs/run/deploy.md的职责范围也是通往生产环境的其余道路。Mount或Host把它放进更大的 Starlette 或 FastAPI 应用。挂载会禁用内置 lifespan。宿主应用的 lifespan 必须进入mcp.session_manager.run()否则第一次请求就失败。一个应用里放多个服务器 多个挂载 一个进入每个 session manager 的 lifespan。streamable_http_path/把端点移到挂载前缀本身。浏览器客户端需要 CORSallow_headers放行Mcp-*请求头expose_headers[Mcp-Session-Id]暴露响应头。mcp.custom_route()在/mcp旁边添加普通、不鉴权的 HTTP 端点。服务器在真实 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），仅供参考