Fix #355: Fix type error with lifespan context (#368)

Co-authored-by: Marcelo Trylesinski <[email protected]>

Author: dsp-ant

src/mcp/server/fastmcp/server.py

@@ -652,9 +652,9 @@ async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContent
         Returns:
             The resource content as either text or bytes
         """
-        assert self._fastmcp is not None, (
-            "Context is not available outside of a request"
-        )
+        assert (
+            self._fastmcp is not None
+        ), "Context is not available outside of a request"
         return await self._fastmcp.read_resource(uri)
 
     async def log(

src/mcp/shared/context.py

@@ -7,7 +7,7 @@
 from mcp.types import RequestId, RequestParams
 
 SessionT = TypeVar("SessionT", bound=BaseSession[Any, Any, Any, Any, Any])
-LifespanContextT = TypeVar("LifespanContextT", default=None)
+LifespanContextT = TypeVar("LifespanContextT")
 
 
 @dataclass

src/mcp/shared/progress.py

@@ -5,7 +5,7 @@
 
 from pydantic import BaseModel
 
-from mcp.shared.context import RequestContext
+from mcp.shared.context import LifespanContextT, RequestContext
 from mcp.shared.session import (
     BaseSession,
     ReceiveNotificationT,
@@ -60,7 +60,8 @@ def progress(
             SendResultT,
             ReceiveRequestT,
             ReceiveNotificationT,
-        ]
+        ],
+        LifespanContextT,
     ],
     total: float | None = None,
 ) -> Generator[

tests/issues/test_355_type_error.py

@@ -0,0 +1,50 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+
+from mcp.server.fastmcp import Context, FastMCP
+
+
+class Database:  # Replace with your actual DB type
+    @classmethod
+    async def connect(cls):
+        return cls()
+
+    async def disconnect(self):
+        pass
+
+    def query(self):
+        return "Hello, World!"
+
+
+# Create a named server
+mcp = FastMCP("My App")
+
+
+@dataclass
+class AppContext:
+    db: Database
+
+
+@asynccontextmanager
+async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
+    """Manage application lifecycle with type-safe context"""
+    # Initialize on startup
+    db = await Database.connect()
+    try:
+        yield AppContext(db=db)
+    finally:
+        # Cleanup on shutdown
+        await db.disconnect()
+
+
+# Pass lifespan to server
+mcp = FastMCP("My App", lifespan=app_lifespan)
+
+
+# Access type-safe lifespan context in tools
+@mcp.tool()
+def query_db(ctx: Context) -> str:
+    """Tool that uses initialized resources"""
+    db = ctx.request_context.lifespan_context.db
+    return db.query()