diff --git a/README.md b/README.md index 8de0d79..370b4f3 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,9 @@ The [Model Context Protocol (MCP)](https://modelcontextprotocol.io) lets you bui The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing: ```python +# Add lifespan support for startup/shutdown with strong typing +from dataclasses import dataclass +from typing import AsyncIterator from mcp.server.fastmcp import FastMCP # Create a named server @@ -136,14 +139,17 @@ mcp = FastMCP("My App") # Specify dependencies for deployment and development mcp = FastMCP("My App", dependencies=["pandas", "numpy"]) -# Add lifespan support for startup/shutdown +@dataclass +class AppContext: + db: Database # Replace with your actual DB type + @asynccontextmanager -async def app_lifespan(server: FastMCP) -> AsyncIterator[dict]: - """Manage application lifecycle""" +async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: + """Manage application lifecycle with type-safe context""" try: # Initialize on startup await db.connect() - yield {"db": db} + yield AppContext(db=db) finally: # Cleanup on shutdown await db.disconnect() @@ -151,7 +157,7 @@ async def app_lifespan(server: FastMCP) -> AsyncIterator[dict]: # Pass lifespan to server mcp = FastMCP("My App", lifespan=app_lifespan) -# Access lifespan context in tools +# Access type-safe lifespan context in tools @mcp.tool() def query_db(ctx: Context) -> str: """Tool that uses initialized resources""" @@ -387,7 +393,6 @@ async def query_db(name: str, arguments: dict) -> list: The lifespan API provides: - A way to initialize resources when the server starts and clean them up when it stops - Access to initialized resources through the request context in handlers -- Support for both low-level Server and FastMCP classes - Type-safe context passing between lifespan and request handlers ```python diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index bc341b4..5ae30a5 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -100,7 +100,7 @@ class Settings(BaseSettings, Generic[LifespanResultT]): lifespan: ( Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]] | None - ) = Field(None, description="Lifespan contexte manager") + ) = Field(None, description="Lifespan context manager") def lifespan_wrapper( diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index a4a8510..643e1a2 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -85,7 +85,10 @@ from mcp.shared.session import RequestResponder logger = logging.getLogger(__name__) -request_ctx: contextvars.ContextVar[RequestContext[ServerSession]] = ( +LifespanResultT = TypeVar("LifespanResultT") + +# This will be properly typed in each Server instance's context +request_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any]] = ( contextvars.ContextVar("request_ctx") ) @@ -102,9 +105,6 @@ class NotificationOptions: self.tools_changed = tools_changed -LifespanResultT = TypeVar("LifespanResultT") - - @asynccontextmanager async def lifespan(server: "Server") -> AsyncIterator[object]: """Default lifespan context manager that does nothing. @@ -212,7 +212,7 @@ class Server(Generic[LifespanResultT]): ) @property - def request_context(self) -> RequestContext[ServerSession]: + def request_context(self) -> RequestContext[ServerSession, LifespanResultT]: """If called outside of a request context, this will raise a LookupError.""" return request_ctx.get() @@ -510,7 +510,7 @@ class Server(Generic[LifespanResultT]): message: RequestResponder, req: Any, session: ServerSession, - lifespan_context: object, + lifespan_context: LifespanResultT, raise_exceptions: bool, ): logger.info(f"Processing request of type {type(req).__name__}") diff --git a/src/mcp/shared/context.py b/src/mcp/shared/context.py index 50e5d51..a45fdac 100644 --- a/src/mcp/shared/context.py +++ b/src/mcp/shared/context.py @@ -1,15 +1,16 @@ from dataclasses import dataclass -from typing import Any, Generic, TypeVar +from typing import Generic, TypeVar from mcp.shared.session import BaseSession from mcp.types import RequestId, RequestParams SessionT = TypeVar("SessionT", bound=BaseSession) +LifespanContextT = TypeVar("LifespanContextT") @dataclass -class RequestContext(Generic[SessionT]): +class RequestContext(Generic[SessionT, LifespanContextT]): request_id: RequestId meta: RequestParams.Meta | None session: SessionT - lifespan_context: Any + lifespan_context: LifespanContextT