Pass entire request object to handlers; add raw request to MCP base Request

[mconflitti-pbc]

Is your feature request related to a problem? Please describe.
I have an MCP server that sits in front of my API to allow LLMs to interact with it. My API requires an authorization header.

I have hacked a way to do this in my fork, but essentially the MCP client is able to pass through headers. Only need to use this for the /sse request. Currently, the handlers extract the arguments they need in the decorator. We could instead add a field to the base Request class called raw_request or headers if we just need that and then ensure this is added to the request object before passing it to the handler.

Describe the solution you'd like

# src/mcp/types.py
class Request(BaseModel, Generic[RequestParamsT, MethodT]):
    """Base class for JSON-RPC requests."""

    method: MethodT
    params: RequestParamsT
    headers: dict[str, Any] | None = None # <<<<<<<<
    model_config = ConfigDict(extra="allow")

---------------------
# src/mcp/server/fastmcp/server.py
    def call_tool(self):
        def decorator(
            func: Callable[
                ...,
                Awaitable[
                    Sequence[
                        types.TextContent | types.ImageContent | types.EmbeddedResource
                    ]
                ],
            ],
        ):
            logger.debug("Registering handler for CallToolRequest")

            async def handler(req: types.CallToolRequest):
                try:
                    results = await func(req)  # <<<<<<<<<
                    return types.ServerResult(
                        types.CallToolResult(content=list(results), isError=False)
                    )
                except Exception as e:
                    return types.ServerResult(
                        types.CallToolResult(
                            content=[types.TextContent(type="text", text=str(e))],
                            isError=True,
                        )
                    )

            self.request_handlers[types.CallToolRequest] = handler
            return func

        return decorator

-----------------------------
# src/mcp/server/fastmcp/server.py
    async def run_sse_async(self, middleware: list[type] = []) -> None:
        """Run the server using SSE transport."""
        from starlette.applications import Starlette
        from starlette.routing import Mount, Route

        sse = SseServerTransport("/messages/")

        async def handle_sse(request):
            async with sse.connect_sse(
                request.scope, request.receive, request._send
            ) as streams:
                await self._mcp_server.run(
                    streams[0],
                    streams[1],
                    self._mcp_server.create_initialization_options(),
                    raw_request=request, # <<<<<<<<<<<<<<
                )

        starlette_app = Starlette(
            debug=self.settings.debug,
            routes=[
                Route("/sse", endpoint=handle_sse),
                Mount("/messages/", app=sse.handle_post_message),
            ],
        )

        config = uvicorn.Config(
            starlette_app,
            host=self.settings.host,
            port=self.settings.port,
            log_level=self.settings.log_level.lower(),
        )
        server = uvicorn.Server(config)
        await server.serve()

---------------------------
# src/mcp/server/lowlevel/server.py
    async def run(
        self,
        read_stream: MemoryObjectReceiveStream[types.JSONRPCMessage | Exception],
        write_stream: MemoryObjectSendStream[types.JSONRPCMessage],
        initialization_options: InitializationOptions,
        raw_request: Any | None = None, # <<<<<<<<<<<<<<<<<<<<<<<<<
        # When False, exceptions are returned as messages to the client.
        # When True, exceptions are raised, which will cause the server to shut down
        # but also make tracing exceptions much easier during testing and when using
        # in-process servers.
        raise_exceptions: bool = False,
    ):
        with warnings.catch_warnings(record=True) as w:
            async with ServerSession(
                read_stream, write_stream, initialization_options
            ) as session:
                async for message in session.incoming_messages:
                    logger.debug(f"Received message: {message}")

                    match message:
                        case (
                            RequestResponder(
                                request=types.ClientRequest(root=req)
                            ) as responder
                        ):
                            with responder:
                                if raw_request is not None:
                                    req.headers = raw_request.headers # <<<<<<<<<<<<<<<<
                                await self._handle_request(
                                    message, req, session, raise_exceptions
                                )
                        case types.ClientNotification(root=notify):
                            await self._handle_notification(notify)

                    for warning in w:
                        logger.info(
                            f"Warning: {warning.category.__name__}: {warning.message}"
                        )

and then use this like:

# already supported on client
transport = await exit_stack.enter_async_context(
    sse_client(url, headers={"authorization": "..."})
)

# on server
mcp_server = FastMCP("example", transport="sse")

async def handle_call_tool(
    self: FastMCP, req: types.CallToolRequest # <<<<<<<<<<<<<<<<<<
) -> Sequence[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    headers = {}
    if "authorization" in req.headers:
        headers = {"Authorization": req.headers["authorization"]}
    # ...http client call to api or if MCP is served from the app itself, check the key

I know auth is a part of the 2025 H1 roadmap so this may be usurped already in terms of how things will be supported. This goes beyond auth headers though since it could be useful to have access to the raw request in total instead within the tool execution context.

[panz2018]

I have developed web servers that integrate MCP SSE functionality:

• Using FastAPI: fastapi_mcp_sse
• Using Starlette: starlette_mcp_sse

These servers can be extended with custom routes while retaining full MCP SSE capabilities. Since these example servers are fully developed platforms, it is possible to add the auth in FastAPI or Starlette. In /sse route and handle_sse function, add the auth part there: https://github.com/panz2018/fastapi_mcp_sse/blob/main/src/app.py#L35