企业官网建设流程全解析

将 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-sdk

mcp.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.py）：

from mcp.server import MCPServer mcp = MCPServer("Notes") @mcp.tool() def add_note(text: str) -> str: """Save a note.""" return f"Saved: {text}" app = mcp.streamable_http_app()

app就是普通的 ASGI 应用，交给任何 ASGI 服务器即可：

uvicorn server:app

MCP 端点位于/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, endpoint=streamable_http_app)）；
• 一个 lifespan：Starlette(..., lifespan=lambda 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）：

完整的参数语义（如json_response、stateless_http、重试与超时策略）请参考 docs/run/index.md；生产环境的安全配置详见 docs/run/deploy.md。

旧版 SSE 传输对应的方法为mcp.sse_app()，行为一致（默认端点/sse、消息端点/messages/），但它承载的是已被 Streamable HTTP 取代的旧传输。

默认只应答 localhost：DNS 重绑定防护

开箱即用状态下，该应用只应答发往 localhost 的请求。原因在于streamable_http_app()无法预知自己会被部署在哪个主机名下，于是以最安全的允许列表启用了 DNS 重绑定防护；在本机开发时这恰好是正确行为。

从源码可以看到具体逻辑（src/mcp/server/lowlevel/server.py）：当未显式传入transport_security且host为"127.0.0.1"、"localhost"或"::1"时，自动构造：

TransportSecuritySettings( enable_dns_rebinding_protection=True, 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 f"Saved: {text}" @asynccontextmanager async def lifespan(app: Starlette) -> AsyncIterator[None]: async with mcp.session_manager.run(): yield app = Starlette( routes=[Mount("/", app=mcp.streamable_http_app())], lifespan=lifespan, )

这里有三条容易踩坑的规则：

1. 路径叠加：Mount("/", ...)加上默认的/mcp路径，端点仍保持在/mcp。Starlette 按顺序匹配路由，Mount("/")会匹配任意路径，所以你自己的路由必须放在它之前，其后的一切都将无法访问。
2. 生命周期是核心：lifespan函数在宿主应用的整个生命周期内进入mcp.session_manager.run()。这是最容易被遗忘的一行。
3. 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，会抛出RuntimeError（src/mcp/server/lowlevel/server.py）。

警告：宿主应用拥有 lifespan

streamable_http_app()把session_manager.run()接入了它返回的 Starlette 的 lifespan，但被挂载的子应用的 lifespan 永远不会运行。挂载之后，内置的 lifespan 就成了死代码。无论谁位于 ASGI 栈的顶层，都必须在自己的 lifespan 中进入mcp.session_manager.run()。

一个快速验证方法：删掉lifespan=lifespan这一行再启动服务器。服务器能正常启动，路由也能解析，但第一个发往/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.py）：

from 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 f"Saved: {text}" @tasks.tool() def add_task(title: str) -> str: """Create a task.""" return f"Created: {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", app=notes.streamable_http_app()), Mount("/tasks", app=tasks.streamable_http_app()), ], lifespan=lifespan, )

要点：

• AsyncExitStack同时进入两个 manager：它们一起启动，并按逆序优雅关闭；
• 端点为/notes/mcp和/tasks/mcp：挂载前缀加上默认路径。

修改端点路径

结尾的/mcp来自streamable_http_path参数。把它设为"/"，挂载前缀本身就成为完整的对外路径（取自 docs_src/asgi/tutorial004.py）：

app = Starlette( routes=[Mount("/notes", app=mcp.streamable_http_app(streamable_http_path="/"))], lifespan=lifespan, )

此时客户端连接到/notes/而不是/notes/mcp。

浏览器客户端的 CORS 配置

基于浏览器的客户端需要你授予两类权限：发送MCP 请求头，以及读取MCP 返回的响应头。两者都是宿主应用上的 CORS 配置，并且要与上文的传输安全白名单保持一致（取自 docs_src/asgi/tutorial005.py）：

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 f"Saved: {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("/", app=mcp.streamable_http_app(transport_security=security))], 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"], ) ], lifespan=lifespan, )

四个要点：

• 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.py）：

from 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 f"Saved: {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和/health；
• GET /health返回{"status": "ok"}，与 MCP 完全无关。

自定义路由的其他参数（来自 src/mcp/server/mcpserver/server.py）：

警告：自定义路由永不鉴权。即使服务器其余部分启用了认证，自定义路由也不会被保护。这是有意为之：健康检查和 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（可用AsyncExitStack）；
• streamable_http_path="/"把端点移到挂载前缀本身；
• 浏览器客户端需要 CORS：allow_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