Lifespan

Die meisten echten Server halten etwas für ihre gesamte Lebensdauer: einen Datenbank-Pool, einen HTTP-Client, ein geladenes Modell.

Du willst das nicht bei jedem Aufruf neu aufbauen, und du willst es sauber schließen. Genau dafür gibt es den Lifespan (Start- und Stopp-Phase des Servers).

Ein typisierter Lifespan

Ein Lifespan ist ein @asynccontextmanager, der den Server erhält und per yield ein Objekt liefert. Was immer du dabei lieferst, steht jedem Handler zur Verfügung, solange der Server läuft.

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

Lies es von unten nach oben:

• app_lifespan verbindet die Database vor dem yield und trennt sie danach, in einem finally. Das sind Start und Stopp.
• Es liefert einen AppContext, eine schlichte Dataclass, die die eingerichteten Dinge hält. Heute ein Feld, morgen zehn.
• MCPServer("Bookshop", lifespan=app_lifespan) ist die ganze Verdrahtung.
• Im Tool ist das gelieferte Objekt ctx.request_context.lifespan_context.

Der Lifespan läuft einmal. Er wird betreten, wenn der Server startet (vor dem ersten Request), und verlassen, wenn der Server stoppt. Alle Requests dazwischen teilen sich denselben AppContext.

Was das Modell sieht

Nichts Neues. ctx ist ein Context-Parameter, also injiziert das SDK ihn, und er landet nie im Eingabeschema:

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

genre ist das einzige Argument, das das Modell übergeben kann. Der Lifespan ist Sache deines Servers.

Auch @mcp.resource()- und @mcp.prompt()-Funktionen können einen ctx-Parameter annehmen, geschrieben als bloßer Context – aus einem Grund, zu dem der nächste Abschnitt kommt. Alles, was ctx mitbringt, steht in Der Context.

Es ist wirklich typisiert

Sieh dir die Annotation noch einmal an: ctx: Context[AppContext].

Dieser eine Typparameter ist der Grund, warum ctx.request_context.lifespan_context für deinen Type Checker ein AppContext ist. .db wird automatisch vervollständigt; .dbb ist ein Fehler, bevor du den Server überhaupt startest.

Schreibst du stattdessen einen bloßen Context, ist lifespan_context als dict[str, Any] typisiert: Der Type Checker kann nicht wissen, was dein Lifespan geliefert hat. Das Objekt ist zur Laufzeit immer noch da; du hast nur die Hilfe verloren.

Context is not available outside of a request

Zusehen, wie es passiert

„Der Start läuft vor dem ersten Request“ ist die Art von Satz, die du nicht einfach glauben müssen solltest.

Reduziere den Server auf den Lebenszyklus: Gib Database ein connected-Flag, schalte es in connect() und disconnect() um und füge ein Tool hinzu, das es meldet.

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 lebt aus einem Grund auf Modulebene: damit du es von außerhalb des Servers betrachten kannst.

Zusammenfassung

• lifespan= nimmt einen @asynccontextmanager, der den Server erhält und per yield ein Objekt liefert.
• Code vor dem yield ist der Start. Das finally danach ist der Stopp.
• Er läuft einmal, rund um die gesamte Lebensdauer des Servers, nicht pro Request.
• Was immer du per yield lieferst, ist ctx.request_context.lifespan_context in jedem Tool, jeder Ressource und jedem Prompt.
• ctx: Context[AppContext] macht diesen Zugriff in Tools vollständig typisiert. Ressourcen und Prompts nehmen den bloßen Context.
• Kein lifespan= bedeutet ein leeres dict, nie None.

Ein Handler, der mitten im Aufruf anhält, um die Person am Host nach etwas zu fragen, das nur sie weiß, ist Elicitation (Rückfrage bei der Person am Host).