Lifespan
Tradução automática
Esta página foi traduzida automaticamente a partir da documentação em inglês, e a página em inglês é a versão de referência. Se algo parecer errado, Traduções explica como avisar.
A maioria dos servidores reais mantém alguma coisa durante a vida inteira: um pool de banco de dados, um cliente HTTP, um modelo carregado.
Você não quer construir isso a cada chamada, e quer fechar tudo de forma limpa. É para isso que serve o lifespan (ciclo de vida do servidor).
Um lifespan tipado
Um lifespan é um @asynccontextmanager que recebe o servidor e faz yield de um único objeto. Seja qual for o objeto que você entregar, ele fica disponível para todos os handlers enquanto o servidor estiver rodando.
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}."
Leia de baixo para cima:
app_lifespanconecta oDatabaseantes doyielde o desconecta depois, dentro de umfinally. Isso é a inicialização e o encerramento.- Ele entrega um
AppContext, uma dataclass comum que agrupa o que você configurou. Um campo hoje, dez amanhã. MCPServer("Bookshop", lifespan=app_lifespan)é toda a ligação necessária.- Dentro da ferramenta (tool), o objeto entregue é
ctx.request_context.lifespan_context.
O lifespan executa uma vez. O servidor entra nele ao iniciar (antes da primeira requisição) e sai dele ao parar. Todas as requisições nesse intervalo compartilham o mesmo AppContext.
Info
Se você já escreveu um lifespan do FastAPI, já conhece isso. Mesmo decorador, mesmo yield, mesmo finally.
O que o modelo vê
Nada de novo. ctx é um parâmetro Context, então o SDK o injeta e ele nunca chega ao schema de entrada:
{
"type": "object",
"properties": {
"genre": {"title": "Genre", "type": "string"}
},
"required": ["genre"],
"title": "count_booksArguments"
}
genre é o único argumento que o modelo pode passar. O lifespan é assunto do seu servidor.
Funções @mcp.resource() e @mcp.prompt() também podem receber um parâmetro ctx, escrito como um Context puro por um motivo que a próxima seção explica. Tudo o que ctx carrega está em O Context.
É tipado de verdade
Olhe de novo a anotação: ctx: Context[AppContext].
Esse único parâmetro de tipo é o motivo pelo qual ctx.request_context.lifespan_context é um AppContext para o seu verificador de tipos. .db autocompleta; .dbb é um erro antes mesmo de você executar o servidor.
Escreva um Context puro no lugar e lifespan_context passa a ser tipado como dict[str, Any]: o verificador de tipos não tem como saber o que o seu lifespan entregou. O objeto continua lá em tempo de execução; o que você perdeu foi a ajuda.
Warning
Context[AppContext] é uma grafia só para ferramentas. Coloque-a em uma função
@mcp.resource() ou @mcp.prompt() e toda chamada a esse handler falha. O cliente recebe um
erro de volta, e o log do servidor mostra o porquê:
Context is not available outside of a request
Em recursos e prompts, escreva o ctx: Context puro. O objeto que o seu lifespan entregou
continua sendo ctx.request_context.lifespan_context em tempo de execução; você abre mão do
parâmetro de tipo, não do objeto.
Tip
Sempre existe um lifespan. Se você não passar um, o padrão do SDK entrega um dict vazio,
então ctx.request_context.lifespan_context é {}, nunca None. Esse padrão também é o
motivo de um Context puro tipá-lo como dict[str, Any].
Veja acontecer
"A inicialização roda antes da primeira requisição" é o tipo de afirmação em que você não deveria ter que acreditar sem ver.
Reduza o servidor só ao ciclo de vida: dê ao Database uma flag connected, inverta-a em connect() e disconnect(), e adicione uma ferramenta que informe o valor dela.
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 fica no nível do módulo por um único motivo: para que você possa observá-lo de fora do servidor.
Check
Três momentos, três valores:
- Antes de o servidor iniciar,
database.connectedéFalse. Importar o módulo não conectou nada. - Enquanto ele está rodando, chame
database_statuse o resultado é"connected". - Pare o servidor e o bloco
finallyexecuta:database.connectedéFalsede novo.
O trabalho aconteceu exatamente onde você o colocou: em volta do yield, não na importação e nem a cada requisição.
Recapitulando
lifespan=aceita um@asynccontextmanagerque recebe o servidor e fazyieldde um único objeto.- O código antes do
yieldé a inicialização. Ofinallydepois dele é o encerramento. - Ele executa uma vez, em torno da vida inteira do servidor, não a cada requisição.
- O que você entregar no
yieldéctx.request_context.lifespan_contextem toda ferramenta, recurso e prompt. ctx: Context[AppContext]deixa esse acesso totalmente tipado em ferramentas. Recursos e prompts recebem oContextpuro.- Não passar
lifespan=significa umdictvazio, nuncaNone.
Um handler que para no meio da chamada para perguntar ao usuário algo que só ele sabe é Elicitação (elicitation).