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

[ylassoued]

Thank you @panz2018 and @mconflitti-pbc for sharing your code.
The question below is now irrelevant :-). You may now skip this comment.
I am struggling to get the authentication headers in the MCP endpoints. If I understood it correctly, you are suggesting to use function handle_sse to somehow inject the authentication headers. It is not clear to me though how to do so. is there a way to inject the headers into the request context or as request parameters?

[ylassoued]

I ended up implementing a solution similar to that suggested by @mconflitti-pbc. But rather than injecting the "raw" request into Request, I injected the request scope (which contains the headers) into RequestContext, which is available from the application context from within any MCP endpoint. I did not fork the repo to modify the code. Instead I extended the classes, with the hope that request headers get supported sometime soon in the mcp Python SDK.

Anyway, for those who are interested in the solution, attached is the source code.

mcp.zip

Now, in the mcp router, you can access the request scope from any type of endpoint as follows:

from .mcp.server import FastMCP

mcp = FastMCP("MCP")

@mcp.tool(name="echo_tool")
def echo_tool(message: str) -> str:
 ctx = mcp.get_context()
 request_context = ctx.request_context
 scope= request_context.scope
 headers ={k: v for k, v in scope.get("headers",{})} if scope is not None else{}
 """Echo a message as a tool"""
 return f"Tool echo: {message}. These are your headers: {headers}"

To include the MCP router in your main, assuming that app is a FastAPI instance:

app.mount("/", mcp.sse_app())

I hope this will be helpful.

[ylassoued]

@dsp-ant, any interest in implementing the approach described above directly in the MCP SDK classes? If so, I would be happy to contribute this "feature", unless it violates the specifications of course or somebody else is already working on a solution to support request headers.

[mconflitti-pbc]

That seems like a reasonable way of going about it! One reason to pass the entire request would be to allow access to other things than just the headers, but this would be sufficient for my use case. Would definitely encourage you to open a PR!

[ylassoued]

Thanks @mconflitti-pbc ! Yes, I can pass the whole request along instead of the scope. It's the exact same flow :-). It's actually preferable as it allows you to get anything you need from the request. I'll open a PR then.

[mconflitti-pbc]

Will be happy to take a look when it is up!

[ryaneggz]

@ylassoued this seems feasible to me, nice work!

---

Update:

Was able to get this working, created a commit with my working code for handling APIToken auth over here
ruska-ai/mcp-sse@5951778

nice solution @ylassoued Close?

[ylassoued]

Thank you @ryaneggz! Actually, I have just implemented this (differently though) in my fork: https://github.com/ylassoued/python-sdk/tree/ylassoued/feat-request. Following @mconflitti-pbc's suggestion, I went for injecting the whole request (instead of the request scope) into RequestContext. Besides, since RequestContext is generic (not Scarlette- or HTTP-bound), I had to make the request type generic, and had to propagate this in all the code. Please refer to my fork above. I will create a PR.

[felixweinberger]

This looks addressed via #816