跳转至

生命周期

机器翻译

本页由英文文档自动翻译而来,以英文页面为准。如果有读起来不对的地方,翻译页面说明了如何反馈。

大多数真实的服务器在整个运行期间都会持有某样东西:数据库连接池、HTTP 客户端、加载好的模型。

你不想每次调用都重新构建它,又希望能干净地关闭它。这就是生命周期(lifespan)的用途。

带类型的生命周期

生命周期是一个 @asynccontextmanager,它接收服务器并 yield 一个对象。无论 yield 出什么,只要服务器在运行,每个处理函数都能用到它。

server.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_lifespanyield 之前连接 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 在你运行服务器之前就会报错。

如果改写成裸的 Contextlifespan_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() 里翻转它,再加一个报告它的工具。

server.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 放在模块级别只有一个原因:这样就能从服务器外部观察它。

Check

三个时刻,三个值:

  • 服务器启动前,database.connectedFalse。导入模块什么也没连接。
  • 运行期间,调用 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)