Cycle de vie

La plupart des vrais serveurs conservent quelque chose pendant toute leur durée de vie : un pool de connexions à la base de données, un client HTTP, un modèle chargé en mémoire.

Vous ne voulez pas le reconstruire à chaque appel, et vous voulez le fermer proprement. C’est à cela que sert le cycle de vie (lifespan).

Un cycle de vie typé

Un cycle de vie est un @asynccontextmanager qui reçoit le serveur et produit avec yield un seul objet. Ce que vous produisez ainsi reste accessible à chaque gestionnaire (handler) aussi longtemps que le serveur tourne.

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}."

Lisez-le de bas en haut :

• app_lifespan connecte la Database avant le yield et la déconnecte après, dans un finally. C’est le démarrage et l’arrêt.
• Il produit un AppContext, une simple dataclass qui contient ce que vous avez initialisé. Un champ aujourd’hui, dix demain.
• MCPServer("Bookshop", lifespan=app_lifespan) est tout le câblage nécessaire.
• Dans l’outil, l’objet produit est ctx.request_context.lifespan_context.

Le cycle de vie s’exécute une seule fois. On y entre au démarrage du serveur (avant la première requête) et on en sort à l’arrêt du serveur. Toutes les requêtes entre les deux partagent le même AppContext.

Ce que voit le modèle

Rien de nouveau. ctx est un paramètre Context : le SDK l’injecte et il n’atteint jamais le schéma d’entrée :

{
  "type": "object",
  "properties": {
    "genre": {"title": "Genre", "type": "string"}
  },
  "required": ["genre"],
  "title": "count_booksArguments"
}

genre est le seul argument que le modèle peut passer. Le cycle de vie, c’est l’affaire de votre serveur.

Les fonctions @mcp.resource() et @mcp.prompt() peuvent elles aussi prendre un paramètre ctx, annoté d’un simple Context pour une raison que la section suivante explique. Tout ce que transporte ctx est décrit dans L’objet Context.

C’est réellement typé

Regardez de nouveau l’annotation : ctx: Context[AppContext].

Ce seul paramètre de type est la raison pour laquelle ctx.request_context.lifespan_context est un AppContext pour votre vérificateur de types. .db s’autocomplète ; .dbb est une erreur avant même que vous n’ayez lancé le serveur.

Écrivez un simple Context à la place et lifespan_context est typé dict[str, Any] : le vérificateur de types n’a aucun moyen de savoir ce que votre cycle de vie a produit. L’objet est toujours là à l’exécution ; vous avez perdu l’assistance.

Context is not available outside of a request

Le voir se produire

« Le démarrage s’exécute avant la première requête » est le genre de phrase que vous ne devriez pas avoir à croire sur parole.

Réduisez le serveur à son cycle de vie : donnez à Database un indicateur connected, basculez-le dans connect() et disconnect(), et ajoutez un outil qui en rend compte.

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 est défini au niveau du module pour une seule raison : pouvoir l’observer depuis l’extérieur du serveur.

Récapitulatif

• lifespan= prend un @asynccontextmanager qui reçoit le serveur et produit avec yield un seul objet.
• Le code avant le yield est le démarrage. Le finally qui suit est l’arrêt.
• Il s’exécute une seule fois, autour de toute la vie du serveur, pas à chaque requête.
• Ce que vous produisez avec yield est ctx.request_context.lifespan_context dans chaque outil, ressource et prompt.
• ctx: Context[AppContext] rend cet accès entièrement typé dans les outils. Les ressources et les prompts prennent le simple Context.
• Pas de lifespan= signifie un dict vide, jamais None.

Un gestionnaire qui s’interrompt en plein appel pour demander à l’utilisateur quelque chose que lui seul connaît, c’est l’Élicitation.