Жизненный цикл

Машинный перевод

Эта страница переведена с английской документации автоматически, и основной версией остаётся английская страница. Если что-то читается неправильно, на странице Переводы объясняется, как об этом сообщить.

Большинство настоящих серверов держат что-то на протяжении всей своей работы: пул соединений с базой данных, HTTP-клиент, загруженную модель.

Создавать это при каждом вызове не хочется, а вот закрыть аккуратно — нужно. Для этого и служит жизненный цикл (lifespan).

Типизированный жизненный цикл

Жизненный цикл — это @asynccontextmanager, который получает сервер и отдаёт через 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_lifespan подключает Database до yield и отключает её после, в блоке finally. Это запуск и остановка.
Он отдаёт AppContext — обычный dataclass с тем, что вы подготовили. Сегодня одно поле, завтра десять.
MCPServer("Bookshop", lifespan=app_lifespan) — вот и вся связка.
Внутри инструмента отданный объект — это ctx.request_context.lifespan_context.

Жизненный цикл выполняется один раз. Вход в него происходит при запуске сервера (до первого запроса), выход — при остановке. Все запросы между этими моментами разделяют один и тот же AppContext.

Info

Если вы писали lifespan для FastAPI, вы это уже знаете. Тот же декоратор, тот же 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]: анализатору типов неоткуда узнать, что отдал ваш жизненный цикл. Во время выполнения объект по-прежнему на месте; вы лишь теряете подсказки.

Warning

Context[AppContext] — запись только для инструментов. Поставьте её на функцию @mcp.resource() или @mcp.prompt() — и каждый вызов этого обработчика завершится ошибкой. Клиент получит ошибку в ответ, а в логе сервера будет видна причина:

Context is not available outside of a request

В ресурсах и промптах пишите просто ctx: Context. Объект, который отдал ваш жизненный цикл, во время выполнения по-прежнему лежит в ctx.request_context.lifespan_context; вы отказываетесь от параметра типа, а не от объекта.

Tip

Жизненный цикл есть всегда. Если не передать свой, вариант SDK по умолчанию отдаёт пустой 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.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).