Ce face Context, progres și lifespan în SDK-ul Python (MCPServer)
Un parametru adnotat cu Context (din mcp.server.mcpserver) este injectat de SDK în unelte, resurse și prompturi; numele parametrului nu contează și nu apare în schema văzută de model. Contextul oferă ctx.request_id, await ctx.read_resource(uri) pentru resursele proprii ale serverului, await ctx.report_progress(progress, total, message) (care trimite notifications/progress), ctx.elicit() și ctx.elicit_url(), ctx.headers (pe HTTP) și ctx.request_context.lifespan_context. Pe conexiunile 2026-07-28, notificările de schimbare se trimit cu await ctx.notify_tools_changed() și metodele înrudite, către stream-urile subscriptions/listen.
Lifespan: un @asynccontextmanager care primește serverul și face yield unui singur obiect (de exemplu un dataclass cu pool-ul de conexiuni). Codul dinainte de yield rulează la pornire, cel din finally la oprire, o singură dată pentru toată viața serverului; în v2, și pe Streamable HTTP lifespan-ul rulează o singură dată, partajat de toate cererile. Obiectul ajunge în ctx.request_context.lifespan_context, iar Context[AppContext] îl face tipizat. Fără lifespan=, valoarea este un dict gol, nu None.
Logging și sampling. Logging-ul prin protocol este depreciat de specificația 2026-07-28 și nu are înlocuitor: ctx.log(), ctx.debug(), ctx.info(), ctx.warning(), ctx.error() și ctx.session.send_log_message() emit MCPDeprecationWarning. Documentația SDK recomandă modulul standard logging, care scrie pe stderr; nivelul se setează cu MCPServer(..., log_level="DEBUG"). La fel, ctx.session.create_message() (sampling) funcționează doar pe conexiuni din era 2025 cu canal înapoi; forma suportată este un parametru Resolve(...) al cărui resolver întoarce Sample(...), sau un InputRequiredResult întors direct.
Sintaxă și opțiuni
| Opțiune / Parametru | Descriere |
|---|---|
ctx: Context |
Injectat automat; invizibil pentru model. |
ctx.request_id |
Id-ul cererii servite. |
await ctx.read_resource(uri) |
Citește o resursă a serverului prin același registru ca resources/read. |
await ctx.report_progress(progress, total, message) |
Progres către client; progress trebuie să crească la fiecare apel. |
ctx.request_context.lifespan_context |
Obiectul întors de lifespan la pornire. |
ctx.notify_tools_changed() / notify_resource_updated(uri) |
Notificări de schimbare către abonamentele subscriptions/listen. |
ctx.info() / ctx.error() / ctx.log() |
Depreciate (logging prin protocol); folosește logging. |
Resolve(...) + Sample(...) |
Forma suportată pentru sampling în v2; clientul trebuie să declare capabilitatea sampling. |
Exemple practice
Exemplul 1: Lifespan tipizat cu o conexiune la baza de date
Database este o clasă de exemplu din documentație.
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server import MCPServer
from mcp.server.mcpserver import Context
@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}."Exemplul 2: Raportarea progresului
@mcp.tool()
async def import_catalog(urls: list[str], ctx: Context) -> str:
"""Import book records from a list of catalog URLs."""
for done, url in enumerate(urls, start=1):
await ctx.report_progress(done, total=len(urls), message=f"Imported {url}")
return f"Imported {len(urls)} records."Exemplul 3: Logging recomandat în v2
import logging
from mcp.server import MCPServer
logger = logging.getLogger(__name__)
mcp = MCPServer("Bookshop")
@mcp.tool()
def search_books(query: str) -> str:
"""Search the catalog by title or author."""
logger.info("Searching for %r", query)
return f"Found 3 books matching {query!r}."Greșeli frecvente și bune practici
- Folosești
ctx.info()ca să trimiți loguri clientului. Este depreciat odată cu logging-ul din protocol; logul ajunge oricum doar la operator, nu la model. - Deschizi o conexiune la baza de date în fiecare apel de unealtă. Pune-o în lifespan, care rulează o singură dată.
- Apelezi
ctx.session.create_message()pe o conexiune 2026-07-28. Nu există canal înapoi; foloseșteResolvecuSamplesauInputRequiredResult.