MCP Python SDK 服务端 lifespan 全解:从连接池管理到生命周期验证实战
【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk
导读
本篇文章基于官方 Python SDK for Model Context Protocol(仓库 pythonsd/python-sdk)的文档展开,系统讲解服务端lifespan(生命周期)机制的完整用法。真实 MCP 服务器几乎都需要在存活期间持有某个"常驻"资源——数据库连接池、HTTP 客户端、加载好的模型——你不希望每次请求都重新创建,又希望在服务器退出时干净地关闭。lifespan 正是为此设计的官方机制。读完本文,你将掌握:如何用@asynccontextmanager编写类型化 lifespan、如何让yield出的对象被所有 handler 共享、类型参数Context[AppContext]的威力与使用边界,以及如何通过一个最小实验亲眼验证"启动先于首请求、结束于 finally"的生命周期时序。
为什么需要 lifespan:服务器级别的常驻资源
绝大多数真实服务器都会持有某些存活期与服务器本身一致的资源:数据库连接池、HTTP 客户端、加载到内存中的模型等。如果每次调用都新建,性能与连接数都不可接受;如果从不关闭,又会泄漏连接与句柄。lifespan 解决的就是这个"创建一次、干净关闭"的问题。
lifespan 的本质是一个@asynccontextmanager异步上下文管理器:它接收服务器实例,yield出一个对象,这个对象在服务器运行的整个期间对所有 handler 可见。yield之前的代码是启动逻辑,yield之后的代码(通常放在finally中)是关闭逻辑。
如果你写过 FastAPI 的
lifespan,这里的知识是相通的:同一个装饰器、同一个yield、同一个finally。
类型化 lifespan:完整接线示例
以下是最小但完整的示例(完整源码见 docs_src/lifespan/tutorial001.py),建议自下而上阅读:
from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass from mcp.server import MCPServer from mcp.server.mcpserver import Context class Database: @classmethod async def connect(cls) -> "Database": return cls() async def disconnect(self) -> None: ... def query(self) -> int: return 3 @dataclass class AppContext: db: Database @asynccontextmanager async def app_lifespan(server: MCPServer) -> AsyncIterator[AppContext]: db = await Database.connect() try: yield AppContext(db=db) finally: await db.disconnect() mcp = MCPServer("Bookshop", lifespan=app_lifespan) @mcp.tool() def count_books(genre: str, ctx: Context[AppContext]) -> str: """Count the books in a genre.""" db = ctx.request_context.lifespan_context.db return f"{db.query()} books in {genre!r}."逐层拆解这段代码:
app_lifespan是启动与关闭的全部:yield之前连接Database,yield之后在finally中断开连接。异步上下文管理器保证无论期间发生什么,finally都会执行,关闭逻辑绝不遗漏。AppContext是一个普通 dataclass:它只是"承载你设置好的一堆东西"的容器。今天放一个字段db,明天可以扩展成十个字段——工具函数依然只需要通过ctx.request_context.lifespan_context一处入口访问。MCPServer("Bookshop", lifespan=app_lifespan)就是全部接线工作:把 lifespan 作为构造参数传入即可,SDK 负责在正确时机进入和退出。- 工具内部通过
ctx.request_context.lifespan_context拿到 yield 出的对象:ctx是 SDK 注入的Context参数,不参与工具的输入 schema。
生命周期时序:一次执行,全程共享
lifespan恰好执行一次:服务器启动时(在第一个请求之前)进入,服务器停止时退出。期间的所有请求共享同一个AppContext实例——这正是"连接池/客户端/模型只建一次"语义的来源。
从源码可以印证这一点。在底层服务器实现中,Server.run用async with self.lifespan(self) as lifespan_context:包裹整个消息循环(见 src/mcp/server/lowlevel/server.py),也就是说 lifespan 上下文以with方式包住整个连接生命周期,yield 出的对象随后通过lifespan_state传递给每个请求(src/mcp/server/lowlevel/server.py)。
如果你不传lifespan=,SDK 会使用默认实现——一个什么都不做、直接yield {}的异步上下文管理器(见 src/mcp/server/lowlevel/server.py)。这解释了文档中的关键保证:lifespan 永远存在,ctx.request_context.lifespan_context至少是{},绝不会是None。这也是为什么裸Context会把lifespan_context类型标注为dict[str, Any]。
模型视角:ctx 是 SDK 注入的,不进 schema
对调用方(LLM)来说,lifespan 是完全透明的。ctx是一个Context 参数,由 SDK 在调用时注入,绝不会出现在工具的输入 schema 里。以count_books为例,模型能看到的输入 schema 只有genre一个字段:
{ "type": "object", "properties": { "genre": {"title": "Genre", "type": "string"} }, "required": ["genre"], "title": "count_booksArguments" }模型唯一能传的参数是genre。lifespan 是你的服务器内部事务,与协议无关。
这一点在 SDK 实现中同样成立:Context.request_context属性在无活动请求时会直接抛出ValueError("Context is not available outside of a request")(见 src/mcp/server/mcpserver/context.py),而每个请求的request_context都携带lifespan_context字段(定义见 src/mcp/server/context.py)。@mcp.resource()与@mcp.prompt()函数同样可以接收ctx参数,但它们应按下一节的原因写成不带类型参数的裸Context。ctx携带的全部内容,可进一步查阅文档 Context。
它真的是类型安全的:Context[AppContext] 的威力
再看一次注解:ctx: Context[AppContext]。
正是这一个类型参数,让类型检查器(如 mypy / pyright)确信ctx.request_context.lifespan_context就是AppContext类型。于是.db能自动补全,而敲出.dbb会在服务器运行之前就成为类型错误——IDE 里直接标红。
反过来,如果写成不带类型参数的裸Context,lifespan_context的类型就是dict[str, Any]:类型检查器无法得知你的 lifespan yield 了什么。对象在运行时依然存在,但你失去了编译期的全部帮助。
从源码看,这一设计的根基在于Context的泛型声明与LifespanContextT类型变量:ServerRequestContext的lifespan_context字段是泛型的(src/mcp/server/context.py),Context类自身也声明为Generic[LifespanT_co]且协变(src/mcp/server/context.py),因此Context[AppContext]可以安全地向下兼容为Context[object]等更宽类型。
重要警告:Context[AppContext] 是工具专用写法
警告:
Context[AppContext]只适用于工具(@mcp.tool())函数。如果把它写到@mcp.resource()或@mcp.prompt()函数上,该 handler 的每次调用都会失败。客户端会收到错误,服务器日志中会显示原因:Context is not available outside of a request在资源与提示词中,请写成裸
ctx: Context。你的 lifespan yield 出的对象在运行时仍然位于ctx.request_context.lifespan_context中——你放弃的只是类型参数,不是对象本身。
产生这一限制的原因与实现细节一致:Context.request_context属性在请求上下文尚未建立时(如资源/提示词 handler 的某些调用路径)会抛出上述ValueError(见 src/mcp/server/mcpserver/context.py)。
提示:lifespan 永远存在
lifespan永远存在。即使你不传lifespan=,SDK 的默认 lifespan 也会 yield 一个空dict,因此ctx.request_context.lifespan_context是{},绝不会是None。裸Context将其类型标为dict[str, Any],正是因为这个默认值。你的代码可以放心地直接访问lifespan_context而无需判空——当然,若你依赖自定义对象,仍需通过类型参数来恢复精确类型。
亲眼验证:启动先于首请求,关闭落于 finally
"启动代码在第一个请求之前运行"这类论断,不该靠直觉接受,值得亲手验证。做法是把服务器精简到只剩生命周期本身:
- 给
Database加一个connected布尔标志; - 在
connect()与disconnect()中翻转该标志; - 添加一个报告该标志状态的工具。
完整示例见 docs_src/lifespan/tutorial002.py:
from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass from mcp.server import MCPServer from mcp.server.mcpserver import Context class Database: def __init__(self) -> None: self.connected = False async def connect(self) -> None: self.connected = True async def disconnect(self) -> None: self.connected = False @dataclass class AppContext: db: Database database = Database() @asynccontextmanager async def app_lifespan(server: MCPServer) -> AsyncIterator[AppContext]: await database.connect() try: yield AppContext(db=database) finally: await database.disconnect() mcp = MCPServer("Bookshop", lifespan=app_lifespan) @mcp.tool() def database_status(ctx: Context[AppContext]) -> str: """Report whether the database connection is up.""" db = ctx.request_context.lifespan_context.db return "connected" if db.connected else "disconnected"注意:database放在模块级别,唯一的原因是从服务器"外部"观察它——你可以在自己的测试或调试代码里直接读取database.connected,而不需要穿过 MCP 协议。
三个时间点,三个值
按文档的验证清单,在三个时刻观察:
| 时刻 | database.connected | 说明 |
|---|---|---|
| 服务器启动前 | False | 导入模块不会连接任何东西;连接只发生在 lifespan 的yield之前 |
| 服务器运行中 | True(调用database_status返回"connected") | 启动代码已在首个请求之前执行完毕 |
| 服务器停止后 | False | finally块运行,disconnect()被调用 |
结论很清晰:工作恰好发生在你放置它的位置——yield的周围。既不在模块导入时,也不在每个请求时。这正印证了 lifespan 与"每次请求都初始化"或"导入时初始化"两种模式的本质区别。
总结:lifespan 核心要点
lifespan=参数接收一个@asynccontextmanager:它接收服务器实例并yield出一个对象。yield之前的代码是启动;其后的finally是关闭。- 它在服务器的整个生命周期内只执行一次,而非每个请求一次。
- 你 yield 出的对象,在所有工具、资源与提示词中都可以通过
ctx.request_context.lifespan_context访问。 ctx: Context[AppContext]让工具中的该访问获得完整类型。资源与提示词请使用裸Context(Context[AppContext]写在资源/提示词上会导致每次调用失败)。- 不传
lifespan=时,默认值是一个空dict,绝不会是None。
接下来可以继续阅读:在调用中途停下来向用户询问只有用户才知道的信息的 handler,属于Elicitation(询问)机制,参见文档 Elicitation。
【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考