# 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}"
                        )

# 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