Життєвий цикл

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

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

Типізований життєвий цикл

Життєвий цикл — це @asynccontextmanager, який отримує сервер і через 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 під'єднує Database до yield і від'єднує її після, у блоці finally. Це і є запуск та зупинка.
• Він віддає AppContext — звичайний dataclass, що містить усе налаштоване під час запуску. Сьогодні одне поле, завтра десять.
• MCPServer("Bookshop", lifespan=app_lifespan) — оце й усе підключення.
• Усередині інструмента відданий об'єкт доступний як ctx.request_context.lifespan_context.

Життєвий цикл виконується один раз. Вхід у нього відбувається, коли сервер стартує (до першого запиту), а вихід — коли сервер зупиняється. Усі запити між цими моментами спільно користуються тим самим AppContext.

Що бачить модель

Нічого нового. 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]: засіб перевірки типів ніяк не може дізнатися, що віддав ваш життєвий цикл. Під час виконання об'єкт нікуди не зникає; ви лише втрачаєте допомогу.

Context is not available outside of a request

Подивіться, як це працює

«Запуск виконується до першого запиту» — речення з тих, які не варто брати на віру.

Скоротіть сервер до самого життєвого циклу: додайте до 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 живе на рівні модуля з однієї причини: щоб на неї можна було глянути ззовні сервера.

Підсумки

• lifespan= приймає @asynccontextmanager, який отримує сервер і через yield віддає один об'єкт.
• Код до yield — це запуск. Блок finally після нього — зупинка.
• Він виконується один раз, навколо всього життя сервера, а не на кожен запит.
• Усе, що ви віддаєте через yield, стає ctx.request_context.lifespan_context у кожному інструменті, ресурсі та промпті.
• ctx: Context[AppContext] робить цей доступ повністю типізованим в інструментах. Ресурси й промпти приймають простий Context.
• Без lifespan= буде порожній dict, і ніколи не None.

Обробник, що зупиняється посеред виклику, аби запитати в користувача щось відоме лише йому, — це Еліцитація.