生命周期
大多数真实的服务器在整个运行期间都会持有某样东西:数据库连接池、HTTP 客户端、加载好的模型。
你不想每次调用都重新构建它,又希望能干净地关闭它。这就是生命周期(lifespan)的用途。
带类型的生命周期
生命周期是一个 @asynccontextmanager,它接收服务器并 yield 一个对象。无论 yield 出什么,只要服务器在运行,每个处理函数都能用到它。
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,并在之后的finally里断开连接。这就是启动和关闭。- 它 yield 一个
AppContext,一个普通的 dataclass,装着你准备好的东西。今天是一个字段,明天可能是十个。 MCPServer("Bookshop", lifespan=app_lifespan)就是全部的接线。- 在工具内部,yield 出的对象是
ctx.request_context.lifespan_context。
生命周期只运行一次。服务器启动时(第一个请求之前)进入,服务器停止时退出。其间的每个请求共享同一个 AppContext。
Info
如果你写过 FastAPI 的 lifespan,这些你已经会了。同样的装饰器,同样的 yield,同样的 finally。
模型看到什么
没有新东西。ctx 是一个 Context 参数,所以 SDK 会注入它,它永远不会进入输入模式:
{
"type": "object",
"properties": {
"genre": {"title": "Genre", "type": "string"}
},
"required": ["genre"],
"title": "count_booksArguments"
}
genre 是模型唯一能传入的参数。生命周期是服务器自己的事。
@mcp.resource() 和 @mcp.prompt() 函数也可以接收 ctx 参数,只是要写成裸的 Context,原因下一节会讲到。ctx 携带的所有内容详见 Context。
它确实带类型
再看一眼那个注解:ctx: Context[AppContext]。
正是这一个类型参数,让 ctx.request_context.lifespan_context 在类型检查器眼里就是一个 AppContext。.db 能自动补全;.dbb 在你运行服务器之前就会报错。
如果改写成裸的 Context,lifespan_context 的类型就是 dict[str, Any]:类型检查器无从知道你的生命周期 yield 了什么。运行时对象还在,只是失去了类型上的帮助。
Warning
Context[AppContext] 是仅限工具的写法。把它放在 @mcp.resource() 或 @mcp.prompt() 函数上,对该处理函数的每次调用都会失败。客户端会收到一个错误,服务器日志会说明原因:
Context is not available outside of a request
在资源和提示词里,写裸的 ctx: Context。生命周期 yield 出的对象在运行时仍然是 ctx.request_context.lifespan_context;你放弃的是类型参数,不是对象。
Tip
生命周期总是存在。如果你不传,SDK 的默认实现会 yield 一个空 dict,所以 ctx.request_context.lifespan_context 是 {},绝不会是 None。也正是因为这个默认值,裸的 Context 才把它的类型定为 dict[str, Any]。
亲眼看它发生
“启动在第一个请求之前运行”这种话,不该只凭信任接受。
把服务器精简到只剩生命周期:给 Database 加一个 connected 标志,在 connect() 和 disconnect() 里翻转它,再加一个报告它的工具。
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 放在模块级别只有一个原因:这样就能从服务器外部观察它。
Check
三个时刻,三个值:
- 服务器启动前,
database.connected是False。导入模块什么也没连接。 - 运行期间,调用
database_status,结果是"connected"。 - 停止服务器,
finally块运行:database.connected又变回False。
工作恰好发生在你放的位置:围绕 yield,不在导入时,也不是每个请求一次。
回顾
lifespan=接收一个@asynccontextmanager,它接收服务器并yield一个对象。yield之前的代码是启动。之后的finally是关闭。- 它只运行一次,围绕服务器的整个生命,而不是每个请求一次。
- 无论
yield出什么,它在每个工具、资源和提示词里都是ctx.request_context.lifespan_context。 ctx: Context[AppContext]让这种访问在工具里完全带类型。资源和提示词用裸的Context。- 不传
lifespan=意味着一个空dict,绝不会是None。
在调用中途停下来,向用户询问只有他们知道的事情的处理函数,详见 征询(elicitation)。