Lifespan

ज़्यादातर असली servers पूरी ज़िंदगी भर कुछ न कुछ संभाले रखते हैं: database pool, HTTP client, load किया हुआ model।

इसे हर call पर दोबारा बनाना कोई नहीं चाहता, और इसे साफ़-सुथरे ढंग से बंद करना ज़रूर चाहिए। lifespan इसी के लिए है।

Typed lifespan

lifespan एक @asynccontextmanager है जिसे server मिलता है और जो एक object yield करता है। आप जो भी yield करते हैं, वह server के चलते रहने तक हर handler को उपलब्ध रहता है।

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 yield से पहले Database को connect करता है और उसके बाद, finally में, disconnect करता है। यही startup और shutdown है।
• यह AppContext yield करता है, एक सादा dataclass जिसमें वे चीज़ें हैं जो आपने set up कीं। आज एक field, कल दस।
• MCPServer("Bookshop", lifespan=app_lifespan) ही पूरी wiring है।
• tool के अंदर, yield किया गया object ctx.request_context.lifespan_context है।

lifespan एक बार चलता है। server शुरू होने पर (पहली request से पहले) इसमें प्रवेश होता है और server रुकने पर इससे बाहर निकला जाता है। बीच की हर request वही AppContext साझा करती है।

model को क्या दिखता है

कुछ नया नहीं। ctx एक Context parameter है, इसलिए SDK इसे inject करता है और यह input schema तक कभी नहीं पहुँचता:

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

genre ही एकमात्र argument है जो model दे सकता है। lifespan आपके server का अपना मामला है।

@mcp.resource() और @mcp.prompt() functions भी ctx parameter ले सकते हैं, जिसे सिर्फ़ Context लिखा जाता है; इसकी वजह अगला section बताता है। ctx में जो कुछ भी है, वह सब Context में है।

यह सच में typed है

annotation को फिर से देखें: ctx: Context[AppContext]।

इसी एक type parameter की वजह से आपके type checker के लिए ctx.request_context.lifespan_context एक AppContext है। .db autocomplete होता है; .dbb server चलाने से पहले ही error है।

इसकी जगह सिर्फ़ Context लिखें तो lifespan_context का type dict[str, Any] हो जाता है: type checker के पास यह जानने का कोई तरीका नहीं कि आपके lifespan ने क्या yield किया। runtime पर object फिर भी मौजूद रहता है; बस मदद चली जाती है।

Context is not available outside of a request

इसे होते हुए देखें

"startup पहली request से पहले चलता है" ऐसा वाक्य है जिस पर आपको बिना देखे भरोसा नहीं करना पड़ना चाहिए।

server को सिर्फ़ lifecycle तक सीमित कर दें: Database को एक connected flag दें, connect() और disconnect() में उसे पलटें, और एक tool जोड़ें जो उसकी स्थिति बताए।

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 module level पर एक ही वजह से है: ताकि आप इसे server के बाहर से देख सकें।

सारांश

• lifespan= एक @asynccontextmanager लेता है जिसे server मिलता है और जो एक object yield करता है।
• yield से पहले का code startup है। उसके बाद का finally shutdown है।
• यह एक बार चलता है, server की पूरी ज़िंदगी के इर्द-गिर्द, हर request पर नहीं।
• आप जो भी yield करते हैं, वह हर tool, resource और prompt में ctx.request_context.lifespan_context है।
• ctx: Context[AppContext] tools में इस access को पूरी तरह typed बना देता है। resources और prompts सिर्फ़ Context लेते हैं।
• lifespan= न हो तो खाली dict मिलता है, कभी None नहीं।

जो handler call के बीच रुककर user से वह पूछता है जो सिर्फ़ user ही जानता है, वह Elicitation है।