[v1.x] Add Streamable HTTP request body limits (#3101)

Author: Kludex

docs/server.md

@@ -1253,6 +1253,7 @@ The FastMCP server instance accessible via `ctx.fastmcp` provides access to serv
   - `host` and `port` - Server network configuration
   - `mount_path`, `sse_path`, `streamable_http_path` - Transport paths
   - `stateless_http` - Whether the server operates in stateless mode
+  - `max_request_body_size` - Maximum Streamable HTTP POST body size in bytes
   - And other configuration options
 
 ```python
@@ -1417,6 +1418,14 @@ Note that `uv run mcp run` or `uv run mcp dev` only supports server using FastMC
 
 > **Note**: Streamable HTTP transport is the recommended transport for production deployments. Use `stateless_http=True` and `json_response=True` for optimal scalability.
 
+Streamable HTTP POST bodies are limited to 4 MiB by default. Larger requests receive HTTP 413
+before parsing or session creation. If your server intentionally accepts larger MCP messages,
+configure the smallest suitable byte limit:
+
+```python
+mcp = FastMCP("Large messages", max_request_body_size=8 * 1024 * 1024)
+```
+
 <!-- snippet-source examples/snippets/servers/streamable_config.py -->
 ```python
 """

src/mcp/server/fastmcp/server.py

@@ -62,7 +62,7 @@
 from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
 from mcp.server.streamable_http import EventStore
-from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import Annotations, AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
@@ -106,6 +106,7 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     json_response: bool
     stateless_http: bool
     """Define if the server should create a new transport per request."""
+    max_request_body_size: int
 
     # resource settings
     warn_on_duplicate_resources: bool
@@ -166,6 +167,7 @@ def __init__(  # noqa: PLR0913
         streamable_http_path: str = "/mcp",
         json_response: bool = False,
         stateless_http: bool = False,
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
         warn_on_duplicate_resources: bool = True,
         warn_on_duplicate_tools: bool = True,
         warn_on_duplicate_prompts: bool = True,
@@ -193,6 +195,7 @@ def __init__(  # noqa: PLR0913
             streamable_http_path=streamable_http_path,
             json_response=json_response,
             stateless_http=stateless_http,
+            max_request_body_size=max_request_body_size,
             warn_on_duplicate_resources=warn_on_duplicate_resources,
             warn_on_duplicate_tools=warn_on_duplicate_tools,
             warn_on_duplicate_prompts=warn_on_duplicate_prompts,
@@ -960,6 +963,7 @@ def streamable_http_app(self) -> Starlette:
                 json_response=self.settings.json_response,
                 stateless=self.settings.stateless_http,  # Use the stateless setting
                 security_settings=self.settings.transport_security,
+                max_request_body_size=self.settings.max_request_body_size,
             )
 
         # Create the ASGI handler

src/mcp/server/streamable_http_manager.py

@@ -4,15 +4,17 @@
 
 import contextlib
 import logging
+from collections import deque
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, Final
 from uuid import uuid4
 
 import anyio
 from anyio.abc import TaskStatus
+from starlette.datastructures import Headers
 from starlette.requests import Request
 from starlette.responses import Response
-from starlette.types import Receive, Scope, Send
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
 from mcp.server.lowlevel.server import Server as MCPServer
@@ -26,6 +28,9 @@
 
 logger = logging.getLogger(__name__)
 
+DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024
+"""Default maximum Streamable HTTP request body size in bytes (4 MiB)."""
+
 
 class StreamableHTTPSessionManager:
     """
@@ -60,6 +65,8 @@ class StreamableHTTPSessionManager:
             retry_interval is also configured, ensure the idle timeout comfortably exceeds the retry interval to
             avoid reaping sessions during normal SSE polling gaps. Default is None (no timeout). A value of 1800
             (30 minutes) is recommended for most deployments.
+        max_request_body_size: Maximum size in bytes for Streamable HTTP POST request bodies. Requests that
+            exceed this limit receive a 413 response before parsing or session creation. Defaults to 4 MiB.
     """
 
     def __init__(
@@ -71,11 +78,14 @@ def __init__(
         security_settings: TransportSecuritySettings | None = None,
         retry_interval: int | None = None,
         session_idle_timeout: float | None = None,
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
     ):
         if session_idle_timeout is not None and session_idle_timeout <= 0:
             raise ValueError("session_idle_timeout must be a positive number of seconds")
         if stateless and session_idle_timeout is not None:
             raise RuntimeError("session_idle_timeout is not supported in stateless mode")
+        if max_request_body_size <= 0:
+            raise ValueError("max_request_body_size must be a positive number of bytes")
 
         self.app = app
         self.event_store = event_store
@@ -84,6 +94,8 @@ def __init__(
         self.security_settings = security_settings
         self.retry_interval = retry_interval
         self.session_idle_timeout = session_idle_timeout
+        self.max_request_body_size = max_request_body_size
+        self.asgi_app = RequestBodyLimitMiddleware(self._handle_request, max_request_body_size)
 
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
@@ -156,6 +168,14 @@ async def handle_request(
             receive: ASGI receive function
             send: ASGI send function
         """
+        await self.asgi_app(scope, receive, send)
+
+    async def _handle_request(
+        self,
+        scope: Scope,
+        receive: Receive,
+        send: Send,
+    ) -> None:
         if self._task_group is None:
             raise RuntimeError("Task group is not initialized. Make sure to use run().")
 
@@ -341,3 +361,63 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                 body.model_dump_json(by_alias=True, exclude_none=True), status_code=404, media_type="application/json"
             )
             await response(scope, receive, send)
+
+
+class RequestBodyLimitMiddleware:
+    """Reject oversized HTTP request bodies before invoking an ASGI application."""
+
+    def __init__(self, app: ASGIApp, max_body_size: int) -> None:
+        self.app = app
+        self.max_body_size = max_body_size
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http" or scope["method"] != "POST":
+            await self.app(scope, receive, send)
+            return
+
+        headers = Headers(scope=scope)
+        content_length = headers.get("content-length")
+        if content_length is not None:
+            try:
+                declared_size = int(content_length)
+            except ValueError:
+                pass
+            else:
+                if declared_size > self.max_body_size:
+                    response = Response("Request body too large", status_code=413)
+                    return await response(scope, receive, send)
+
+        received_body = bytearray()
+        received_request = False
+        body_complete = False
+        trailing_message: Message | None = None
+        while True:
+            message = await receive()
+            if message["type"] != "http.request":
+                trailing_message = message
+                break
+
+            received_request = True
+            body = message.get("body", b"")
+            if len(received_body) + len(body) > self.max_body_size:
+                response = Response("Request body too large", status_code=413)
+                return await response(scope, receive, send)
+            received_body.extend(body)
+            if not message.get("more_body", False):
+                body_complete = True
+                break
+
+        cached_messages: deque[Message] = deque()
+        if received_request:
+            cached_messages.append(
+                {"type": "http.request", "body": bytes(received_body), "more_body": not body_complete}
+            )
+        if trailing_message is not None:
+            cached_messages.append(trailing_message)
+
+        async def replay() -> Message:
+            if cached_messages:
+                return cached_messages.popleft()
+            return await receive()
+
+        await self.app(scope, replay, send)

tests/server/fastmcp/test_server.py

@@ -1490,3 +1490,12 @@ def test_streamable_http_no_redirect() -> None:
 
     # Verify path values
     assert streamable_routes[0].path == "/mcp", "Streamable route path should be /mcp"
+
+
+def test_streamable_http_app_passes_the_configured_request_body_limit_to_its_manager() -> None:
+    """SDK-defined: FastMCP forwards its public request-body setting to the Streamable HTTP manager."""
+    mcp = FastMCP(max_request_body_size=8)
+
+    mcp.streamable_http_app()
+
+    assert mcp.session_manager.max_request_body_size == 8