Added support for raw request injection in RequestContext.

Author: Yassine Lassoued

.idea/.gitignore

@@ -14,14 +14,14 @@
 class SamplingFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext["ClientSession", Any, Any],
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.ErrorData: ...
 
 
 class ListRootsFnT(Protocol):
     async def __call__(
-        self, context: RequestContext["ClientSession", Any]
+        self, context: RequestContext["ClientSession", Any, Any]
     ) -> types.ListRootsResult | types.ErrorData: ...
 
 
@@ -50,7 +50,7 @@ async def _default_message_handler(
 
 
 async def _default_sampling_callback(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext["ClientSession", Any, Any],
     params: types.CreateMessageRequestParams,
 ) -> types.CreateMessageResult | types.ErrorData:
     return types.ErrorData(
@@ -60,7 +60,7 @@ async def _default_sampling_callback(
 
 
 async def _default_list_roots_callback(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext["ClientSession", Any, Any],
 ) -> types.ListRootsResult | types.ErrorData:
     return types.ErrorData(
         code=types.INVALID_REQUEST,
@@ -331,7 +331,7 @@ async def send_roots_list_changed(self) -> None:
     async def _received_request(
         self, responder: RequestResponder[types.ServerRequest, types.ClientResult]
     ) -> None:
-        ctx = RequestContext[ClientSession, Any](
+        ctx = RequestContext[ClientSession, Any, Any](
             request_id=responder.request_id,
             meta=responder.request_meta,
             session=self,

.idea/deployment.xml

@@ -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
+from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import (
     AnyFunction,
     EmbeddedResource,
@@ -182,7 +182,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.
@@ -491,6 +491,7 @@ async def handle_sse(request: Request) -> None:
                     streams[0],
                     streams[1],
                     self._mcp_server.create_initialization_options(),
+                    request=request,
                 )
 
         return Starlette(
@@ -558,7 +559,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.
@@ -592,13 +593,13 @@ 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,
     ):

.idea/inspectionProfiles/profiles_settings.xml

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