support streamable http transport

shahar4499 · shahar4499 · commit fbe06f822b71 · 2025-07-23T15:14:17.000+03:00
diff --git a/fastapi_mcp/server.py b/fastapi_mcp/server.py
@@ -10,6 +10,7 @@
 
 from fastapi_mcp.openapi.convert import convert_openapi_to_mcp_tools
 from fastapi_mcp.transport.sse import FastApiSseTransport
+from fastapi_mcp.transport.http import FastApiStreamableHttpTransport
 from fastapi_mcp.types import HTTPRequestInfo, AuthConfig
 
 import logging
@@ -35,7 +36,7 @@ def decorator(
 
             async def handler(req: types.CallToolRequest):
                 try:
-                    # Pull the original HTTP request info from the MCP message. It was injected in
+                    # HACK: 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)
@@ -241,6 +242,32 @@ def _register_mcp_endpoints_sse(
         self._register_mcp_connection_endpoint_sse(router, transport, mount_path, dependencies)
         self._register_mcp_messages_endpoint_sse(router, transport, mount_path, dependencies)
 
+    def _register_mcp_http_endpoint(
+        self,
+        router: FastAPI | APIRouter,
+        transport: FastApiStreamableHttpTransport,
+        mount_path: str,
+        dependencies: Optional[Sequence[params.Depends]],
+    ):
+        @router.api_route(
+            mount_path,
+            methods=["GET", "POST", "DELETE"],
+            include_in_schema=False,
+            operation_id="mcp_http",
+            dependencies=dependencies,
+        )
+        async def handle_mcp_streamable_http(request: Request):
+            return await transport.handle_fastapi_request(request)
+
+    def _register_mcp_endpoints_http(
+        self,
+        router: FastAPI | APIRouter,
+        transport: FastApiStreamableHttpTransport,
+        mount_path: str,
+        dependencies: Optional[Sequence[params.Depends]],
+    ):
+        self._register_mcp_http_endpoint(router, transport, mount_path, dependencies)
+
     def _setup_auth_2025_03_26(self):
         from fastapi_mcp.auth.proxy import (
             setup_oauth_custom_metadata,
@@ -296,7 +323,7 @@ def _setup_auth(self):
         else:
             logger.info("No auth config provided, skipping auth setup")
 
-    def mount(
+    def mount_http(
         self,
         router: Annotated[
             Optional[FastAPI | APIRouter],
@@ -317,17 +344,64 @@ def mount(
                 """
             ),
         ] = "/mcp",
-        transport: Annotated[
-            Literal["sse"],
+    ) -> None:
+        """
+        Mount the MCP server with HTTP transport to **any** FastAPI app or APIRouter.
+
+        There is no requirement that the FastAPI app or APIRouter is the same as the one that the MCP
+        server was created from.
+        """
+        # Normalize mount path
+        if not mount_path.startswith("/"):
+            mount_path = f"/{mount_path}"
+        if mount_path.endswith("/"):
+            mount_path = mount_path[:-1]
+
+        if not router:
+            router = self.fastapi
+
+        assert isinstance(router, (FastAPI, APIRouter)), f"Invalid router type: {type(router)}"
+
+        http_transport = FastApiStreamableHttpTransport()
+        dependencies = self._auth_config.dependencies if self._auth_config else None
+
+        self._register_mcp_endpoints_http(router, http_transport, mount_path, dependencies)
+        self._setup_auth()
+
+        # HACK: If we got a router and not a FastAPI instance, we need to re-include the router so that
+        # FastAPI will pick up the new routes we added. The problem with this approach is that we assume
+        # that the router is a sub-router of self.fastapi, which may not always be the case.
+        #
+        # TODO: Find a better way to do this.
+        if isinstance(router, APIRouter):
+            self.fastapi.include_router(router)
+
+        logger.info(f"MCP HTTP server listening at {mount_path}")
+
+    def mount_sse(
+        self,
+        router: Annotated[
+            Optional[FastAPI | APIRouter],
             Doc(
                 """
-                The transport type for the MCP server. Currently only 'sse' is supported.
+                The FastAPI app or APIRouter to mount the MCP server to. If not provided, the MCP
+                server will be mounted to the FastAPI app.
                 """
             ),
-        ] = "sse",
+        ] = None,
+        mount_path: Annotated[
+            str,
+            Doc(
+                """
+                Path where the MCP server will be mounted.
+                Mount path is appended to the root path of FastAPI router, or to the prefix of APIRouter.
+                Defaults to '/sse'.
+                """
+            ),
+        ] = "/sse",
     ) -> None:
         """
-        Mount the MCP server to **any** FastAPI app or APIRouter.
+        Mount the MCP server with SSE transport to **any** FastAPI app or APIRouter.
 
         There is no requirement that the FastAPI app or APIRouter is the same as the one that the MCP
         server was created from.
@@ -347,14 +421,9 @@ def mount(
         messages_path = f"{base_path}/messages/"
 
         sse_transport = FastApiSseTransport(messages_path)
-
         dependencies = self._auth_config.dependencies if self._auth_config else None
 
-        if transport == "sse":
-            self._register_mcp_endpoints_sse(router, sse_transport, mount_path, dependencies)
-        else:  # pragma: no cover
-            raise ValueError(f"Invalid transport: {transport}")  # pragma: no cover
-
+        self._register_mcp_endpoints_sse(router, sse_transport, mount_path, dependencies)
         self._setup_auth()
 
         # HACK: If we got a router and not a FastAPI instance, we need to re-include the router so that
@@ -365,7 +434,65 @@ def mount(
         if isinstance(router, APIRouter):
             self.fastapi.include_router(router)
 
-        logger.info(f"MCP server listening at {mount_path}")
+        logger.info(f"MCP SSE server listening at {mount_path}")
+
+    def mount(
+        self,
+        router: Annotated[
+            Optional[FastAPI | APIRouter],
+            Doc(
+                """
+                The FastAPI app or APIRouter to mount the MCP server to. If not provided, the MCP
+                server will be mounted to the FastAPI app.
+                """
+            ),
+        ] = None,
+        mount_path: Annotated[
+            str,
+            Doc(
+                """
+                Path where the MCP server will be mounted.
+                Mount path is appended to the root path of FastAPI router, or to the prefix of APIRouter.
+                Defaults to '/mcp'.
+                """
+            ),
+        ] = "/mcp",
+        transport: Annotated[
+            Literal["sse"],
+            Doc(
+                """
+                The transport type for the MCP server. Currently only 'sse' is supported.
+                This parameter is deprecated.
+                """
+            ),
+        ] = "sse",
+    ) -> None:
+        """
+        [DEPRECATED] Mount the MCP server to **any** FastAPI app or APIRouter.
+
+        This method is deprecated and will be removed in a future version.
+        Use mount_http() for HTTP transport (recommended) or mount_sse() for SSE transport instead.
+
+        For backwards compatibility, this method defaults to SSE transport.
+
+        There is no requirement that the FastAPI app or APIRouter is the same as the one that the MCP
+        server was created from.
+        """
+        import warnings
+
+        warnings.warn(
+            "mount() is deprecated and will be removed in a future version. "
+            "Use mount_http() for HTTP transport (recommended) or mount_sse() for SSE transport instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+
+        if transport == "sse":
+            self.mount_sse(router, mount_path)
+        else:  # pragma: no cover
+            raise ValueError(  # pragma: no cover
+                f"Unsupported transport: {transport}. Use mount_sse() or mount_http() instead."
+            )
 
     async def _execute_api_tool(
         self,
diff --git a/fastapi_mcp/transport/http.py b/fastapi_mcp/transport/http.py
@@ -0,0 +1,164 @@
+import logging
+import json
+
+from fastapi import Request, Response, HTTPException
+from mcp.server.streamable_http import StreamableHTTPServerTransport
+from mcp.server.transport_security import TransportSecuritySettings
+from mcp.types import JSONRPCMessage
+from pydantic import ValidationError
+from fastapi_mcp.types import HTTPRequestInfo
+
+logger = logging.getLogger(__name__)
+
+
+class FastApiStreamableHttpTransport(StreamableHTTPServerTransport):
+    def __init__(
+        self,
+        mcp_session_id: str | None = None,
+        is_json_response_enabled: bool = True,  # Default to JSON for HTTP transport
+        event_store=None,
+        security_settings: TransportSecuritySettings | None = None,
+    ):
+        super().__init__(
+            mcp_session_id=mcp_session_id,
+            is_json_response_enabled=is_json_response_enabled,
+            event_store=event_store,
+            security_settings=security_settings,
+        )
+        logger.debug(f"FastApiStreamableHttpTransport initialized with session_id: {mcp_session_id}")
+
+    async def handle_fastapi_request(self, request: Request) -> Response:
+        """
+        FastAPI-native request handler that adapts the SDK's handle_request method.
+
+        The approach here is necessarily different from FastApiSseTransport.
+        In FastApiSseTransport, we reimplement the SSE transport logic to have a more FastAPI-native transport.
+        It proved to be less bug-prone since it avoids deconstructing and reconstructing raw ASGI objects.
+
+        But, we took a different approach here because StreamableHTTPServerTransport handles more complexity,
+        and multiple request methods (GET/POST/DELETE), so we want to leverage that logic and avoid reimplementing.
+
+        So we use an enhanced adapter pattern: intercept and enhance POST requests for HTTPRequestInfo injection,
+        while delegating the complex protocol handling to the SDK.
+        """
+        logger.debug(f"Handling FastAPI request: {request.method} {request.url.path}")
+
+        if request.method == "POST":
+            return await self._handle_post_with_injection(request)
+        else:
+            # For GET and DELETE requests, delegate directly to SDK since they don't need injection
+            return await self._delegate_to_sdk(request)
+
+    async def _handle_post_with_injection(self, request: Request) -> Response:
+        """
+        Handle POST requests with HTTPRequestInfo injection.
+
+        This mirrors the approach in FastApiSseTransport.handle_fastapi_post_message()
+        to ensure consistency in how we handle authentication context and header forwarding.
+
+        The injection happens at the JSON-RPC message level, just like in SSE transport,
+        so that the downstream tool handlers receive the same request context regardless
+        of transport type.
+        """
+        try:
+            # Read and parse the request body first, just like SSE transport does
+            body = await request.body()
+            logger.debug(f"Received JSON: {body.decode()}")
+
+            try:
+                raw_message = json.loads(body)
+            except json.JSONDecodeError as e:
+                logger.error(f"Failed to parse JSON: {e}")
+                raise HTTPException(status_code=400, detail=f"Parse error: {str(e)}")
+
+            try:
+                message = JSONRPCMessage.model_validate(raw_message)
+            except ValidationError as e:
+                logger.error(f"Failed to validate message: {e}")
+                raise HTTPException(status_code=400, detail=f"Validation error: {str(e)}")
+
+            # 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("Injected HTTPRequestInfo into message for auth context")
+
+            modified_body = message.model_dump_json(by_alias=True, exclude_none=True).encode()
+            modified_request = self._create_modified_request(request, modified_body)
+
+            # Delegate to SDK with the modified request
+            return await self._delegate_to_sdk(modified_request)
+
+        except HTTPException:
+            # Re-raise FastAPI HTTPExceptions directly for proper error handling
+            raise
+        except Exception:
+            logger.exception("Error processing POST request")
+            raise HTTPException(status_code=500, detail="Internal server error")
+
+    def _create_modified_request(self, original_request: Request, modified_body: bytes) -> Request:
+        """
+        Create a new Request object with modified body content.
+
+        This is necessary because we need to inject HTTPRequestInfo into the JSON-RPC message
+        before passing it to the SDK, but Request objects are immutable.
+        """
+
+        # Create a new receive callable that returns our modified body
+        async def modified_receive():
+            return {
+                "type": "http.request",
+                "body": modified_body,
+                "more_body": False,
+            }
+
+        # Create new request with modified receive
+        return Request(original_request.scope, modified_receive)
+
+    async def _delegate_to_sdk(self, request: Request) -> Response:
+        """
+        Delegate request handling to the underlying StreamableHTTPServerTransport.
+
+        This captures the ASGI response from the SDK and converts it to a FastAPI Response,
+        maintaining the adapter pattern while providing FastAPI-native integration.
+        """
+        # Capture the response from the SDK's handle_request method
+        response_started = False
+        response_status = 200
+        response_headers = []
+        response_body = b""
+
+        async def capture_send(message):
+            nonlocal response_started, response_status, response_headers, response_body
+
+            if message["type"] == "http.response.start":
+                response_started = True
+                response_status = message["status"]
+                response_headers = message.get("headers", [])
+            elif message["type"] == "http.response.body":
+                response_body += message.get("body", b"")
+
+        try:
+            # Delegate to the SDK's handle_request method with ASGI interface
+            await self.handle_request(request.scope, request.receive, capture_send)
+
+            # Convert the captured ASGI response to a FastAPI Response
+            headers_dict = {name.decode(): value.decode() for name, value in response_headers}
+
+            return Response(
+                content=response_body,
+                status_code=response_status,
+                headers=headers_dict,
+            )
+
+        except Exception as e:
+            logger.exception(f"Error in StreamableHTTPServerTransport: {e}")
+            raise HTTPException(status_code=500, detail="Internal server error")
diff --git a/pyproject.toml b/pyproject.toml
@@ -29,7 +29,7 @@ dependencies = [
     "fastapi>=0.100.0",
     "typer>=0.9.0",
     "rich>=13.0.0",
-    "mcp>=1.8.1",
+    "mcp>=1.12.0",
     "pydantic>=2.0.0",
     "pydantic-settings>=2.5.2",
     "uvicorn>=0.20.0",
@@ -49,6 +49,7 @@ dev = [
     "pre-commit>=4.2.0",
     "pyjwt>=2.10.1",
     "cryptography>=44.0.2",
+    "types-jsonschema>=4.25.0.20250720",
 ]
 
 [project.urls]
diff --git a/uv.lock b/uv.lock
