use request obeject

Author: ihrpr

src/mcp/server/lowlevel/server.py

@@ -215,9 +215,7 @@ def get_capabilities(
         )
 
     @property
-    def request_context(
-        self,
-    ) -> RequestContext[ServerSession, LifespanResultT]:
+    def request_context(self) -> RequestContext[ServerSession, LifespanResultT]:
         """If called outside of a request context, this will raise a LookupError."""
         return request_ctx.get()
 

src/mcp/server/sse.py

@@ -53,7 +53,6 @@ async def handle_sse(request):
 
 import mcp.types as types
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
-from mcp.types import RequestData
 
 logger = logging.getLogger(__name__)
 
@@ -204,18 +203,9 @@ async def handle_post_message(
             await writer.send(err)
             return
 
-        # Extract request headers and other context
-        request_context: RequestData = {
-            "headers": dict(request.headers),
-            "method": request.method,
-            "url": str(request.url),
-            "client": request.client,
-            "path_params": request.path_params,
-            "query_params": dict(request.query_params),
-        }
-
-        # Create session message with request context
-        metadata = ServerMessageMetadata(request_context=request_context)
+        # Pass the raw request object for maximum flexibility
+        # Consumers can use isinstance to check the request type
+        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/types.py

@@ -36,8 +36,8 @@
 Role = Literal["user", "assistant"]
 RequestId = str | int
 AnyFunction: TypeAlias = Callable[..., Any]
-# Http request-specific data (e.g., headers, auth info)
-RequestData = dict[str, Any]
+# HTTP request object from web framework (e.g., starlette.requests.Request)
+RequestData: TypeAlias = object
 
 
 class RequestParams(BaseModel):

tests/server/fastmcp/test_integration.py

@@ -15,6 +15,7 @@
 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
@@ -89,7 +90,7 @@ def echo(message: str) -> str:
         return f"Echo: {message}"
 
     # Create the SSE app
-    app: Starlette = mcp.sse_app()
+    app = mcp.sse_app()
 
     return mcp, app
 
@@ -201,7 +202,7 @@ def make_everything_fastmcp_app():
     """Create a comprehensive FastMCP server with SSE transport."""
     mcp = make_everything_fastmcp()
     # Create the SSE app
-    app: Starlette = mcp.sse_app()
+    app = mcp.sse_app()
     return mcp, app
 
 
@@ -440,11 +441,10 @@ def make_fastmcp_with_context_app():
     def echo_headers(ctx: Context) -> str:
         """Returns the request headers as JSON."""
         headers_info = {}
-        try:
-            if ctx.request_context.request:
-                headers_info = ctx.request_context.request.get("headers", {})
-        except Exception:
-            pass
+        if ctx.request_context.request and isinstance(
+            ctx.request_context.request, Request
+        ):
+            headers_info = dict(ctx.request_context.request.headers)
         return json.dumps(headers_info)
 
     # Tool that returns full request context
@@ -457,17 +457,15 @@ def echo_context(custom_request_id: str, ctx: Context) -> str:
             "method": None,
             "url": None,
         }
-        try:
-            if ctx.request_context.request:
-                context_data["headers"] = ctx.request_context.request.get("headers", {})
-                context_data["method"] = ctx.request_context.request.get("method")
-                context_data["url"] = ctx.request_context.request.get("url")
-        except Exception:
-            pass
+        if ctx.request_context.request:
+            if isinstance(ctx.request_context.request, Request):
+                context_data["headers"] = dict(ctx.request_context.request.headers)
+                context_data["method"] = ctx.request_context.request.method
+                context_data["url"] = str(ctx.request_context.request.url)
         return json.dumps(context_data)
 
     # Create the SSE app
-    app: Starlette = mcp.sse_app()
+    app = mcp.sse_app()
     return mcp, app
 
 

tests/shared/test_sse.py

@@ -333,7 +333,11 @@ async def handle_call_tool(name: str, args: dict) -> list[TextContent]:
             try:
                 context = self.request_context
                 if context.request:
-                    headers_info = context.request.get("headers", {})
+                    # Check if it's a Starlette Request object
+                    from starlette.requests import Request
+
+                    if isinstance(context.request, Request):
+                        headers_info = dict(context.request.headers)
             except LookupError:
                 pass  # No request context available