Fixed typing issues. Reverted Context generic type parameters.

Author: Yassine Lassoued

src/mcp/server/fastmcp/server.py

@@ -36,7 +36,7 @@
 from mcp.server.session import ServerSession, ServerSessionT
 from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
-from mcp.shared.context import LifespanContextT, RequestContext, RequestT
+from mcp.shared.context import LifespanContextT, RequestContext
 from mcp.types import (
     AnyFunction,
     EmbeddedResource,
@@ -98,9 +98,11 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
 def lifespan_wrapper(
     app: FastMCP,
     lifespan: Callable[[FastMCP], AbstractAsyncContextManager[LifespanResultT]],
-) -> Callable[[MCPServer[LifespanResultT]], AbstractAsyncContextManager[object]]:
+) -> Callable[
+    [MCPServer[LifespanResultT, Request]], AbstractAsyncContextManager[object]
+]:
     @asynccontextmanager
-    async def wrap(s: MCPServer[LifespanResultT]) -> AsyncIterator[object]:
+    async def wrap(s: MCPServer[LifespanResultT, Request]) -> AsyncIterator[object]:
         async with lifespan(app) as context:
             yield context
 
@@ -182,7 +184,7 @@ async def list_tools(self) -> list[MCPTool]:
             for info in tools
         ]
 
-    def get_context(self) -> Context[ServerSession, object, Request]:
+    def get_context(self) -> Context[ServerSession, object]:
         """
         Returns a Context object. Note that the context will only be valid
         during a request; outside a request, most methods will error.
@@ -559,7 +561,7 @@ def _convert_to_content(
     return [TextContent(type="text", text=result)]
 
 
-class Context(BaseModel, Generic[ServerSessionT, LifespanContextT, RequestT]):
+class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
     """Context object providing access to MCP capabilities.
 
     This provides a cleaner interface to MCP's RequestContext functionality.
@@ -593,13 +595,14 @@ def my_tool(x: int, ctx: Context) -> str:
     The context is optional - tools that don't need it can omit the parameter.
     """
 
-    _request_context: RequestContext[ServerSessionT, LifespanContextT, RequestT] | None
+    _request_context: RequestContext[ServerSessionT, LifespanContextT, Request] | None
     _fastmcp: FastMCP | None
 
     def __init__(
         self,
         *,
-        request_context: RequestContext[ServerSessionT, LifespanContextT, RequestT] | None = None,
+        request_context: RequestContext[ServerSessionT, LifespanContextT, Request]
+        | None = None,
         fastmcp: FastMCP | None = None,
         **kwargs: Any,
     ):
@@ -615,7 +618,9 @@ def fastmcp(self) -> FastMCP:
         return self._fastmcp
 
     @property
-    def request_context(self) -> RequestContext[ServerSessionT, LifespanContextT]:
+    def request_context(
+        self,
+    ) -> RequestContext[ServerSessionT, LifespanContextT, Request]:
         """Access to the underlying request context."""
         if self._request_context is None:
             raise ValueError("Context is not available outside of a request")

src/mcp/server/fastmcp/tools/base.py

@@ -76,7 +76,7 @@ def from_function(
     async def run(
         self,
         arguments: dict[str, Any],
-        context: Context[ServerSessionT, LifespanContextT, Any] | None = None,
+        context: Context[ServerSessionT, LifespanContextT] | None = None,
     ) -> Any:
         """Run the tool with arguments."""
         try:

src/mcp/server/fastmcp/tools/tool_manager.py

@@ -50,7 +50,7 @@ async def call_tool(
         self,
         name: str,
         arguments: dict[str, Any],
-        context: Context[ServerSessionT, LifespanContextT, Any] | None = None,
+        context: Context[ServerSessionT, LifespanContextT] | None = None,
     ) -> Any:
         """Call a tool by name with arguments."""
         tool = self.get_tool(name)

src/mcp/server/lowlevel/server.py

@@ -91,7 +91,7 @@ async def main():
 LifespanResultT = TypeVar("LifespanResultT")
 
 # This will be properly typed in each Server instance's context
-request_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any, RequestT]] = (
+request_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any, Any]] = (
     contextvars.ContextVar("request_ctx")
 )
 
@@ -128,7 +128,8 @@ def __init__(
         version: str | None = None,
         instructions: str | None = None,
         lifespan: Callable[
-            [Server[LifespanResultT, RequestT]], AbstractAsyncContextManager[LifespanResultT]
+            [Server[LifespanResultT, RequestT]],
+            AbstractAsyncContextManager[LifespanResultT],
         ] = lifespan,
     ):
         self.name = name
@@ -213,7 +214,9 @@ def get_capabilities(
         )
 
     @property
-    def request_context(self) -> RequestContext[ServerSession, LifespanResultT, RequestT]:
+    def request_context(
+        self,
+    ) -> RequestContext[ServerSession, LifespanResultT, RequestT]:
         """If called outside of a request context, this will raise a LookupError."""
         return request_ctx.get()
 
@@ -518,7 +521,12 @@ async def _handle_message(
                 ):
                     with responder:
                         await self._handle_request(
-                            message, req, session, lifespan_context, raise_exceptions, request
+                            message,
+                            req,
+                            session,
+                            lifespan_context,
+                            raise_exceptions,
+                            request,
                         )
                 case types.ClientNotification(root=notify):
                     await self._handle_notification(notify)

src/mcp/shared/progress.py

@@ -1,7 +1,7 @@
 from collections.abc import Generator
 from contextlib import contextmanager
 from dataclasses import dataclass, field
-from typing import Generic, Any
+from typing import Any, Generic
 
 from pydantic import BaseModel
 

tests/client/test_list_roots_callback.py

@@ -1,5 +1,3 @@
-from types import NoneType
-
 import pytest
 from pydantic import FileUrl