generic

Author: ihrpr

src/mcp/server/fastmcp/server.py

@@ -49,7 +49,7 @@
 from mcp.server.stdio import stdio_server
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-from mcp.shared.context import LifespanContextT, RequestContext
+from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import (
     AnyFunction,
     EmbeddedResource,
@@ -124,9 +124,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
 
@@ -147,7 +149,7 @@ def __init__(
     ):
         self.settings = Settings(**settings)
 
-        self._mcp_server = MCPServer(
+        self._mcp_server: MCPServer[object, Request] = MCPServer(
             name=name or "FastMCP",
             instructions=instructions,
             lifespan=(
@@ -260,7 +262,7 @@ async def list_tools(self) -> list[MCPTool]:
             for info in tools
         ]
 
-    def get_context(self) -> Context[ServerSession, object]:
+    def get_context(self) -> Context[ServerSession, object, Request]:
         """
         Returns a Context object. Note that the context will only be valid
         during a request; outside a request, most methods will error.
@@ -893,7 +895,7 @@ def _convert_to_content(
     return [TextContent(type="text", text=result)]
 
 
-class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
+class Context(BaseModel, Generic[ServerSessionT, LifespanContextT, RequestT]):
     """Context object providing access to MCP capabilities.
 
     This provides a cleaner interface to MCP's RequestContext functionality.
@@ -927,13 +929,15 @@ 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] | None
+    _request_context: RequestContext[ServerSessionT, LifespanContextT, RequestT] | None
     _fastmcp: FastMCP | None
 
     def __init__(
         self,
         *,
-        request_context: RequestContext[ServerSessionT, LifespanContextT] | None = None,
+        request_context: (
+            RequestContext[ServerSessionT, LifespanContextT, RequestT] | None
+        ) = None,
         fastmcp: FastMCP | None = None,
         **kwargs: Any,
     ):
@@ -949,7 +953,9 @@ def fastmcp(self) -> FastMCP:
         return self._fastmcp
 
     @property
-    def request_context(self) -> RequestContext[ServerSessionT, LifespanContextT]:
+    def request_context(
+        self,
+    ) -> RequestContext[ServerSessionT, LifespanContextT, RequestT]:
         """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

@@ -14,7 +14,7 @@
 if TYPE_CHECKING:
     from mcp.server.fastmcp.server import Context
     from mcp.server.session import ServerSessionT
-    from mcp.shared.context import LifespanContextT
+    from mcp.shared.context import LifespanContextT, RequestT
 
 
 class Tool(BaseModel):
@@ -85,7 +85,7 @@ def from_function(
     async def run(
         self,
         arguments: dict[str, Any],
-        context: Context[ServerSessionT, LifespanContextT] | None = None,
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
     ) -> Any:
         """Run the tool with arguments."""
         try:

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

@@ -6,7 +6,7 @@
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.tools.base import Tool
 from mcp.server.fastmcp.utilities.logging import get_logger
-from mcp.shared.context import LifespanContextT
+from mcp.shared.context import LifespanContextT, RequestT
 from mcp.types import ToolAnnotations
 
 if TYPE_CHECKING:
@@ -65,7 +65,7 @@ async def call_tool(
         self,
         name: str,
         arguments: dict[str, Any],
-        context: Context[ServerSessionT, LifespanContextT] | None = None,
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
     ) -> Any:
         """Call a tool by name with arguments."""
         tool = self.get_tool(name)

src/mcp/server/lowlevel/server.py

@@ -72,11 +72,12 @@ async def main():
 import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
-from typing import Any, Generic, TypeVar
+from typing import Any, Generic
 
 import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from pydantic import AnyUrl
+from typing_extensions import TypeVar
 
 import mcp.types as types
 from mcp.server.lowlevel.helper_types import ReadResourceContents
@@ -91,9 +92,10 @@ async def main():
 logger = logging.getLogger(__name__)
 
 LifespanResultT = TypeVar("LifespanResultT")
+RequestT = TypeVar("RequestT", default=Any)
 
 # This will be properly typed in each Server instance's context
-request_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any]] = (
+request_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any, Any]] = (
     contextvars.ContextVar("request_ctx")
 )
 
@@ -111,7 +113,7 @@ def __init__(
 
 
 @asynccontextmanager
-async def lifespan(server: Server[LifespanResultT]) -> AsyncIterator[object]:
+async def lifespan(server: Server[LifespanResultT, RequestT]) -> AsyncIterator[object]:
     """Default lifespan context manager that does nothing.
 
     Args:
@@ -123,14 +125,15 @@ async def lifespan(server: Server[LifespanResultT]) -> AsyncIterator[object]:
     yield {}
 
 
-class Server(Generic[LifespanResultT]):
+class Server(Generic[LifespanResultT, RequestT]):
     def __init__(
         self,
         name: str,
         version: str | None = None,
         instructions: str | None = None,
         lifespan: Callable[
-            [Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]
+            [Server[LifespanResultT, RequestT]],
+            AbstractAsyncContextManager[LifespanResultT],
         ] = lifespan,
     ):
         self.name = name
@@ -215,7 +218,9 @@ def get_capabilities(
         )
 
     @property
-    def request_context(self) -> RequestContext[ServerSession, LifespanResultT]:
+    def request_context(
+        self,
+    ) -> RequestContext[ServerSession, LifespanResultT, RequestT]:
         """If called outside of a request context, this will raise a LookupError."""
         return request_ctx.get()
 

src/mcp/server/sse.py

@@ -204,7 +204,7 @@ async def handle_post_message(
             return
 
         # Pass the ASGI scope for framework-agnostic access to request data
-        metadata = ServerMessageMetadata(request_context=dict(request.scope))
+        metadata = ServerMessageMetadata(request_context=request)
         session_message = SessionMessage(message, metadata=metadata)
         logger.debug(f"Sending session message to writer: {session_message}")
         response = Response("Accepted", status_code=202)

src/mcp/server/streamable_http_manager.py

@@ -56,7 +56,7 @@ class StreamableHTTPSessionManager:
 
     def __init__(
         self,
-        app: MCPServer[Any],
+        app: MCPServer[Any, Any],
         event_store: EventStore | None = None,
         json_response: bool = False,
         stateless: bool = False,

src/mcp/shared/context.py

@@ -4,16 +4,17 @@
 from typing_extensions import TypeVar
 
 from mcp.shared.session import BaseSession
-from mcp.types import RequestData, RequestId, RequestParams
+from mcp.types import RequestId, RequestParams
 
 SessionT = TypeVar("SessionT", bound=BaseSession[Any, Any, Any, Any, Any])
 LifespanContextT = TypeVar("LifespanContextT")
+RequestT = TypeVar("RequestT", default=Any)
 
 
 @dataclass
-class RequestContext(Generic[SessionT, LifespanContextT]):
+class RequestContext(Generic[SessionT, LifespanContextT, RequestT]):
     request_id: RequestId
     meta: RequestParams.Meta | None
     session: SessionT
     lifespan_context: LifespanContextT
-    request: RequestData | None = None
+    request: RequestT | None = None

src/mcp/shared/message.py

@@ -8,7 +8,7 @@
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 
-from mcp.types import JSONRPCMessage, RequestData, RequestId
+from mcp.types import JSONRPCMessage, RequestId
 
 ResumptionToken = str
 
@@ -31,7 +31,7 @@ class ServerMessageMetadata:
 
     related_request_id: RequestId | None = None
     # Request-specific context (e.g., headers, auth info)
-    request_context: RequestData | None = None
+    request_context: object | None = None
 
 
 MessageMetadata = ClientMessageMetadata | ServerMessageMetadata | None

src/mcp/types.py

@@ -36,9 +36,6 @@
 Role = Literal["user", "assistant"]
 RequestId = str | int
 AnyFunction: TypeAlias = Callable[..., Any]
-# Dictionary containing request metadata (headers, path, method, etc.
-# based on  ASGI scope for most of the trasport implementations)
-RequestData: TypeAlias = dict[str, Any]
 
 
 class RequestParams(BaseModel):

tests/server/fastmcp/test_integration.py

@@ -10,11 +10,13 @@
 import socket
 import time
 from collections.abc import Generator
+from typing import Any
 
 import pytest
 import uvicorn
 from pydantic import AnyUrl
 from starlette.applications import Starlette
+from starlette.requests import Request
 
 import mcp.types as types
 from mcp.client.session import ClientSession
@@ -437,20 +439,17 @@ def make_fastmcp_with_context_app():
 
     # Tool that echoes request headers
     @mcp.tool(description="Echo request headers from context")
-    def echo_headers(ctx: Context) -> str:
+    def echo_headers(ctx: Context[Any, Any, Request]) -> str:
         """Returns the request headers as JSON."""
         headers_info = {}
         if ctx.request_context.request:
-            # Extract headers from ASGI scope
-            headers_list = ctx.request_context.request.get("headers", [])
-            headers_info = {
-                k.decode("latin-1"): v.decode("latin-1") for k, v in headers_list
-            }
+            # Now the type system knows request is a Starlette Request object
+            headers_info = dict(ctx.request_context.request.headers)
         return json.dumps(headers_info)
 
     # Tool that returns full request context
     @mcp.tool(description="Echo request context with custom data")
-    def echo_context(custom_request_id: str, ctx: Context) -> str:
+    def echo_context(custom_request_id: str, ctx: Context[Any, Any, Request]) -> str:
         """Returns request context including headers and custom data."""
         context_data = {
             "custom_request_id": custom_request_id,
@@ -459,12 +458,11 @@ def echo_context(custom_request_id: str, ctx: Context) -> str:
             "path": None,
         }
         if ctx.request_context.request:
-            # Extract data from ASGI scope
-            headers_list = ctx.request_context.request.get("headers", [])
-            context_data["headers"] = {
-                k.decode("latin-1"): v.decode("latin-1") for k, v in headers_list
-            }
-            context_data["method"] = ctx.request_context.request.get("method")
+            # Now we can access Request attributes directly
+            request = ctx.request_context.request
+            context_data["headers"] = dict(request.headers)
+            context_data["method"] = request.method
+            context_data["path"] = request.url.path
         return json.dumps(context_data)
 
     # Create the SSE app

tests/server/fastmcp/test_tool_manager.py

@@ -9,7 +9,7 @@
 from mcp.server.fastmcp.tools import Tool, ToolManager
 from mcp.server.fastmcp.utilities.func_metadata import ArgModelBase, FuncMetadata
 from mcp.server.session import ServerSessionT
-from mcp.shared.context import LifespanContextT
+from mcp.shared.context import LifespanContextT, RequestT
 from mcp.types import ToolAnnotations
 
 
@@ -347,7 +347,7 @@ def tool_without_context(x: int) -> str:
         assert tool.context_kwarg is None
 
         def tool_with_parametrized_context(
-            x: int, ctx: Context[ServerSessionT, LifespanContextT]
+            x: int, ctx: Context[ServerSessionT, LifespanContextT, RequestT]
         ) -> str:
             return str(x)
 

tests/shared/test_sse.py

@@ -322,7 +322,7 @@ async def test_sse_client_basic_connection_mounted_app(
 
 
 # Test server with request context that returns headers in the response
-class RequestContextServer(Server):
+class RequestContextServer(Server[object, Request]):
     def __init__(self):
         super().__init__("request_context_server")
 
@@ -333,12 +333,8 @@ async def handle_call_tool(name: str, args: dict) -> list[TextContent]:
             try:
                 context = self.request_context
                 if context.request:
-                    # Extract headers from ASGI scope
-                    headers_list = context.request.get("headers", [])
-                    headers_info = {
-                        k.decode("latin-1"): v.decode("latin-1")
-                        for k, v in headers_list
-                    }
+                    # The request is a Starlette Request object
+                    headers_info = dict(context.request.headers)
             except LookupError:
                 pass  # No request context available