refactor: improve lifespan context typing and documentation

- Add proper generic parameter for lifespan context type
- Update README with TypedDict example for strong typing
- Fix context variable initialization in server
- Improve property return type safety
- Remove redundant documentation
- Ensure compatibility with existing tests
This commit is contained in:
David Soria Parra
2025-02-12 22:12:09 +00:00
parent fddba00723
commit 4d3e05f6f6
4 changed files with 22 additions and 16 deletions

View File

@@ -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: The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing:
```python ```python
# Add lifespan support for startup/shutdown with strong typing
from dataclasses import dataclass
from typing import AsyncIterator
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
# Create a named server # Create a named server
@@ -136,14 +139,17 @@ mcp = FastMCP("My App")
# Specify dependencies for deployment and development # Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"]) 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 @asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[dict]: async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""Manage application lifecycle""" """Manage application lifecycle with type-safe context"""
try: try:
# Initialize on startup # Initialize on startup
await db.connect() await db.connect()
yield {"db": db} yield AppContext(db=db)
finally: finally:
# Cleanup on shutdown # Cleanup on shutdown
await db.disconnect() await db.disconnect()
@@ -151,7 +157,7 @@ async def app_lifespan(server: FastMCP) -> AsyncIterator[dict]:
# Pass lifespan to server # Pass lifespan to server
mcp = FastMCP("My App", lifespan=app_lifespan) mcp = FastMCP("My App", lifespan=app_lifespan)
# Access lifespan context in tools # Access type-safe lifespan context in tools
@mcp.tool() @mcp.tool()
def query_db(ctx: Context) -> str: def query_db(ctx: Context) -> str:
"""Tool that uses initialized resources""" """Tool that uses initialized resources"""
@@ -387,7 +393,6 @@ async def query_db(name: str, arguments: dict) -> list:
The lifespan API provides: The lifespan API provides:
- A way to initialize resources when the server starts and clean them up when it stops - 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 - 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 - Type-safe context passing between lifespan and request handlers
```python ```python

View File

@@ -100,7 +100,7 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
lifespan: ( lifespan: (
Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]] | None Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]] | None
) = Field(None, description="Lifespan contexte manager") ) = Field(None, description="Lifespan context manager")
def lifespan_wrapper( def lifespan_wrapper(

View File

@@ -85,7 +85,10 @@ from mcp.shared.session import RequestResponder
logger = logging.getLogger(__name__) 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") contextvars.ContextVar("request_ctx")
) )
@@ -102,9 +105,6 @@ class NotificationOptions:
self.tools_changed = tools_changed self.tools_changed = tools_changed
LifespanResultT = TypeVar("LifespanResultT")
@asynccontextmanager @asynccontextmanager
async def lifespan(server: "Server") -> AsyncIterator[object]: async def lifespan(server: "Server") -> AsyncIterator[object]:
"""Default lifespan context manager that does nothing. """Default lifespan context manager that does nothing.
@@ -212,7 +212,7 @@ class Server(Generic[LifespanResultT]):
) )
@property @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.""" """If called outside of a request context, this will raise a LookupError."""
return request_ctx.get() return request_ctx.get()
@@ -510,7 +510,7 @@ class Server(Generic[LifespanResultT]):
message: RequestResponder, message: RequestResponder,
req: Any, req: Any,
session: ServerSession, session: ServerSession,
lifespan_context: object, lifespan_context: LifespanResultT,
raise_exceptions: bool, raise_exceptions: bool,
): ):
logger.info(f"Processing request of type {type(req).__name__}") logger.info(f"Processing request of type {type(req).__name__}")

View File

@@ -1,15 +1,16 @@
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, Generic, TypeVar from typing import Generic, TypeVar
from mcp.shared.session import BaseSession from mcp.shared.session import BaseSession
from mcp.types import RequestId, RequestParams from mcp.types import RequestId, RequestParams
SessionT = TypeVar("SessionT", bound=BaseSession) SessionT = TypeVar("SessionT", bound=BaseSession)
LifespanContextT = TypeVar("LifespanContextT")
@dataclass @dataclass
class RequestContext(Generic[SessionT]): class RequestContext(Generic[SessionT, LifespanContextT]):
request_id: RequestId request_id: RequestId
meta: RequestParams.Meta | None meta: RequestParams.Meta | None
session: SessionT session: SessionT
lifespan_context: Any lifespan_context: LifespanContextT