Lifespan

Gerçek sunucuların çoğu, ömürleri boyunca bir şeyi elde tutar: bir veritabanı havuzu, bir HTTP istemcisi, yüklenmiş bir model.

Bunu her çağrıda yeniden kurmak istemezsiniz, ama düzgünce kapatmak istersiniz. İşte lifespan (yaşam döngüsü) bunun için var.

Türü belirli bir lifespan

Lifespan, sunucuyu alan ve tek bir nesne yield eden bir @asynccontextmanager'dır. Yield ettiğiniz şey, sunucu çalıştığı sürece her işleyicinin erişimindedir.

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

Aşağıdan yukarıya okuyun:

• app_lifespan, Database'i yield'den önce bağlar, sonra da bir finally içinde bağlantısını keser. İşte başlatma ve kapatma.
• Bir AppContext yield eder: kurduğunuz şeyleri tutan düz bir dataclass. Bugün bir alan, yarın on.
• Bağlamanın tamamı MCPServer("Bookshop", lifespan=app_lifespan) satırından ibaret.
• Aracın içinde, yield edilen nesne ctx.request_context.lifespan_context'tir.

Lifespan bir kez çalışır. Sunucu başladığında (ilk istekten önce) içine girilir, sunucu durduğunda içinden çıkılır. Aradaki her istek aynı AppContext'i paylaşır.

Modelin gördüğü

Yeni bir şey yok. ctx bir Context parametresidir; bu yüzden SDK onu enjekte eder ve girdi şemasına hiç ulaşmaz:

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

Modelin geçirebileceği tek argüman genre. Lifespan sunucunuzun kendi işidir.

@mcp.resource() ve @mcp.prompt() fonksiyonları da ctx parametresi alabilir; bir sonraki bölümün açıklayacağı bir nedenle bu parametre yalın Context olarak yazılır. ctx'in taşıdığı her şey Context nesnesi sayfasında.

Gerçekten türü belirli

Tür açıklamasına bir daha bakın: ctx: Context[AppContext].

Tür denetleyiciniz için ctx.request_context.lifespan_context'in bir AppContext olmasını sağlayan işte bu tek tür parametresidir. .db otomatik tamamlanır; .dbb ise daha sunucuyu çalıştırmadan hata verir.

Bunun yerine yalın Context yazarsanız lifespan_context'in türü dict[str, Any] olur: tür denetleyicisinin, lifespan'inizin ne yield ettiğini bilmesinin yolu yoktur. Nesne çalışma zamanında yine oradadır; yalnızca yardımı kaybedersiniz.

Context is not available outside of a request

İşleyişi gözlemleme

"Başlatma ilk istekten önce çalışır" cümlesi, körü körüne inanmak zorunda kalmamanız gereken türden bir cümle.

Sunucuyu yaşam döngüsüne kadar sadeleştirin: Database'e bir connected bayrağı verin, connect() ve disconnect() içinde değiştirin ve onu bildiren bir araç ekleyin.

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'in modül düzeyinde durmasının tek bir nedeni var: ona sunucunun dışından bakabilmeniz.

Özet

• lifespan=, sunucuyu alan ve tek bir nesne yield eden bir @asynccontextmanager alır.
• yield'den önceki kod başlatmadır. Sonrasındaki finally kapatmadır.
• İstek başına değil, sunucunun tüm ömrü boyunca bir kez çalışır.
• yield ettiğiniz şey her araçta, kaynakta ve prompt'ta ctx.request_context.lifespan_context olur.
• ctx: Context[AppContext] bu erişimi araçlarda tam tür bilgisiyle donatır. Kaynaklar ve prompt'lar yalın Context alır.
• lifespan= yoksa boş bir dict gelir, asla None değil.

Çağrının ortasında durup kullanıcıya yalnızca onun bildiği bir şeyi soran işleyici, Elicitation (kullanıcıdan bilgi isteme) sayfasının konusu.