Use mcp_server.request_context.request to retrieve http request info

Author: Edison-A-N

fastapi_mcp/server.py

@@ -35,13 +35,7 @@ def decorator(
 
             async def handler(req: types.CallToolRequest):
                 try:
-                    # Pull the original HTTP request info from the MCP message. It was injected in
-                    # `FastApiSseTransport.handle_fastapi_post_message()`
-                    if hasattr(req.params, "_http_request_info") and req.params._http_request_info is not None:
-                        http_request_info = HTTPRequestInfo.model_validate(req.params._http_request_info)
-                        results = await func(req.params.name, (req.params.arguments or {}), http_request_info)
-                    else:
-                        results = await func(req.params.name, (req.params.arguments or {}))
+                    results = await func(req.params.name, (req.params.arguments or {}))
                     return types.ServerResult(types.CallToolResult(content=list(results), isError=False))
                 except Exception as e:
                     return types.ServerResult(
@@ -188,6 +182,20 @@ async def handle_list_tools() -> List[types.Tool]:
         async def handle_call_tool(
             name: str, arguments: Dict[str, Any], http_request_info: Optional[HTTPRequestInfo] = None
         ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
+            request = self.server.request_context.request
+
+            http_request_info = None
+            if request:
+                body = await request.body()
+                http_request_info = HTTPRequestInfo(
+                    method=request.method,
+                    path=request.url.path,
+                    headers=dict(request.headers),
+                    cookies=request.cookies,
+                    query_params=dict(request.query_params),
+                    body=body.decode(),
+                )
+
             return await self._execute_api_tool(
                 client=self._http_client,
                 tool_name=name,

fastapi_mcp/transport/sse.py

@@ -9,7 +9,6 @@
 from pydantic import ValidationError
 from mcp.server.sse import SseServerTransport
 from mcp.types import JSONRPCMessage, JSONRPCError, ErrorData
-from fastapi_mcp.types import HTTPRequestInfo
 
 
 logger = logging.getLogger(__name__)
@@ -59,20 +58,6 @@ async def handle_fastapi_post_message(self, request: Request) -> Response:
 
         try:
             message = JSONRPCMessage.model_validate_json(body)
-
-            # HACK to inject the HTTP request info into the MCP message,
-            # so we can use it for auth.
-            # It is then used in our custom `LowlevelMCPServer.call_tool()` decorator.
-            if hasattr(message.root, "params") and message.root.params is not None:
-                message.root.params["_http_request_info"] = HTTPRequestInfo(
-                    method=request.method,
-                    path=request.url.path,
-                    headers=dict(request.headers),
-                    cookies=request.cookies,
-                    query_params=dict(request.query_params),
-                    body=body.decode(),
-                ).model_dump(mode="json")
-
             logger.debug(f"Validated client message: {message}")
         except ValidationError as err:
             logger.error(f"Failed to parse message: {err}")