use request.scope

Author: ihrpr

src/mcp/server/sse.py

@@ -203,9 +203,8 @@ async def handle_post_message(
             await writer.send(err)
             return
 
-        # Pass the raw request object for maximum flexibility
-        # Consumers can use isinstance to check the request type
-        metadata = ServerMessageMetadata(request_context=request)
+        # Pass the ASGI scope for framework-agnostic access to request data
+        metadata = ServerMessageMetadata(request_context=dict(request.scope))
         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,9 @@
 Role = Literal["user", "assistant"]
 RequestId = str | int
 AnyFunction: TypeAlias = Callable[..., Any]
-# HTTP request object from web framework (e.g., starlette.requests.Request)
-RequestData: TypeAlias = object
+# 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

@@ -441,10 +441,12 @@ def make_fastmcp_with_context_app():
     def echo_headers(ctx: Context) -> str:
         """Returns the request headers as JSON."""
         headers_info = {}
-        if ctx.request_context.request and isinstance(
-            ctx.request_context.request, Request
-        ):
-            headers_info = dict(ctx.request_context.request.headers)
+        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
+            }
         return json.dumps(headers_info)
 
     # Tool that returns full request context
@@ -455,13 +457,15 @@ def echo_context(custom_request_id: str, ctx: Context) -> str:
             "custom_request_id": custom_request_id,
             "headers": {},
             "method": None,
-            "url": None,
+            "path": None,
         }
         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)
+            # 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")
         return json.dumps(context_data)
 
     # Create the SSE app
@@ -558,10 +562,7 @@ async def test_fast_mcp_with_request_context(
                 context_data["headers"].get("authorization")
                 == "Bearer fastmcp-test-token"
             )
-            assert context_data["method"] == "POST"  # SSE messages are POSTed
-            assert (
-                "/messages/" in context_data["url"]
-            )  # Should contain the messages endpoint
+            assert context_data["method"] == "POST"  #
 
 
 @pytest.mark.anyio

tests/shared/test_sse.py

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