René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Pass entire request object to handlers; add raw request to MCP base Request","articleBody":"**Is your feature request related to a problem? Please describe.**\nI have an MCP server that sits in front of my API to allow LLMs to interact with it. My API requires an authorization header.\n\nI 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.\n\n**Describe the solution you'd like**\n```python\n# src/mcp/types.py\nclass Request(BaseModel, Generic[RequestParamsT, MethodT]):\n    \"\"\"Base class for JSON-RPC requests.\"\"\"\n\n    method: MethodT\n    params: RequestParamsT\n    headers: dict[str, Any] | None = None # \u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\n    model_config = ConfigDict(extra=\"allow\")\n\n---------------------\n# src/mcp/server/fastmcp/server.py\n    def call_tool(self):\n        def decorator(\n            func: Callable[\n                ...,\n                Awaitable[\n                    Sequence[\n                        types.TextContent | types.ImageContent | types.EmbeddedResource\n                    ]\n                ],\n            ],\n        ):\n            logger.debug(\"Registering handler for CallToolRequest\")\n\n            async def handler(req: types.CallToolRequest):\n                try:\n                    results = await func(req)  # \u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\n                    return types.ServerResult(\n                        types.CallToolResult(content=list(results), isError=False)\n                    )\n                except Exception as e:\n                    return types.ServerResult(\n                        types.CallToolResult(\n                            content=[types.TextContent(type=\"text\", text=str(e))],\n                            isError=True,\n                        )\n                    )\n\n            self.request_handlers[types.CallToolRequest] = handler\n            return func\n\n        return decorator\n\n-----------------------------\n# src/mcp/server/fastmcp/server.py\n    async def run_sse_async(self, middleware: list[type] = []) -\u003e None:\n        \"\"\"Run the server using SSE transport.\"\"\"\n        from starlette.applications import Starlette\n        from starlette.routing import Mount, Route\n\n        sse = SseServerTransport(\"/messages/\")\n\n        async def handle_sse(request):\n            async with sse.connect_sse(\n                request.scope, request.receive, request._send\n            ) as streams:\n                await self._mcp_server.run(\n                    streams[0],\n                    streams[1],\n                    self._mcp_server.create_initialization_options(),\n                    raw_request=request, # \u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\n                )\n\n        starlette_app = Starlette(\n            debug=self.settings.debug,\n            routes=[\n                Route(\"/sse\", endpoint=handle_sse),\n                Mount(\"/messages/\", app=sse.handle_post_message),\n            ],\n        )\n\n        config = uvicorn.Config(\n            starlette_app,\n            host=self.settings.host,\n            port=self.settings.port,\n            log_level=self.settings.log_level.lower(),\n        )\n        server = uvicorn.Server(config)\n        await server.serve()\n\n---------------------------\n# src/mcp/server/lowlevel/server.py\n    async def run(\n        self,\n        read_stream: MemoryObjectReceiveStream[types.JSONRPCMessage | Exception],\n        write_stream: MemoryObjectSendStream[types.JSONRPCMessage],\n        initialization_options: InitializationOptions,\n        raw_request: Any | None = None, # \u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\n        # When False, exceptions are returned as messages to the client.\n        # When True, exceptions are raised, which will cause the server to shut down\n        # but also make tracing exceptions much easier during testing and when using\n        # in-process servers.\n        raise_exceptions: bool = False,\n    ):\n        with warnings.catch_warnings(record=True) as w:\n            async with ServerSession(\n                read_stream, write_stream, initialization_options\n            ) as session:\n                async for message in session.incoming_messages:\n                    logger.debug(f\"Received message: {message}\")\n\n                    match message:\n                        case (\n                            RequestResponder(\n                                request=types.ClientRequest(root=req)\n                            ) as responder\n                        ):\n                            with responder:\n                                if raw_request is not None:\n                                    req.headers = raw_request.headers # \u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\n                                await self._handle_request(\n                                    message, req, session, raise_exceptions\n                                )\n                        case types.ClientNotification(root=notify):\n                            await self._handle_notification(notify)\n\n                    for warning in w:\n                        logger.info(\n                            f\"Warning: {warning.category.__name__}: {warning.message}\"\n                        )\n```\n\nand then use this like:\n\n```python\n# already supported on client\ntransport = await exit_stack.enter_async_context(\n    sse_client(url, headers={\"authorization\": \"...\"})\n)\n```\n\n```python\n# on server\nmcp_server = FastMCP(\"example\", transport=\"sse\")\n\nasync def handle_call_tool(\n    self: FastMCP, req: types.CallToolRequest # \u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\u003c\n) -\u003e Sequence[types.TextContent | types.ImageContent | types.EmbeddedResource]:\n    headers = {}\n    if \"authorization\" in req.headers:\n        headers = {\"Authorization\": req.headers[\"authorization\"]}\n    # ...http client call to api or if MCP is served from the app itself, check the key\n```\n\nI 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.","author":{"url":"https://github.com/mconflitti-pbc","@type":"Person","name":"mconflitti-pbc"},"datePublished":"2025-02-07T18:37:51.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":10},"url":"https://github.com/195/python-sdk/issues/195"}