Add Streamable HTTP request body limits (#3095)

docs/migration.md

@@ -622,6 +622,7 @@ Transport-specific parameters have been moved from the `MCPServer` constructor t
 - `sse_path`, `message_path` - SSE transport paths
 - `streamable_http_path` - StreamableHTTP endpoint path
 - `json_response`, `stateless_http` - StreamableHTTP behavior
+- `max_request_body_size` - StreamableHTTP request-body limit
 - `event_store`, `retry_interval` - StreamableHTTP event handling
 - `transport_security` - DNS rebinding protection
 
@@ -675,6 +676,21 @@ app = Starlette(routes=[Mount("/", app=mcp.streamable_http_app(json_response=Tru
 
 If you were mutating these via `mcp.settings` after construction (e.g., `mcp.settings.port = 9000`), pass them to `run()` / `sse_app()` / `streamable_http_app()` instead — these fields no longer exist on `Settings`. The `debug` and `log_level` parameters remain on the constructor.
 
+### Streamable HTTP request bodies are limited to 4 MiB
+
+V2 applies a 4 MiB default limit to Streamable HTTP POST bodies and returns HTTP 413 before parsing
+the JSON or creating a session when that limit is exceeded.
+
+Most servers need no migration. If your application intentionally accepts larger MCP messages,
+set an explicit byte limit on `run()` or `streamable_http_app()`:
+
+```python
+mcp.run(transport="streamable-http", max_request_body_size=8 * 1024 * 1024)
+```
+
+The limit must be positive and applies to both legacy session-based requests and V2's modern
+single-exchange requests. Keep the smallest value your application actually needs.
+
 ### Streamable HTTP: lifespan now entered once at manager startup
 
 When serving streamable HTTP (stateful or `stateless_http=True`), the server's `lifespan` context manager is now entered once when `StreamableHTTPSessionManager.run()` starts, and the resulting state is shared across all sessions and requests. Previously each session (stateful) or each request (stateless) entered and exited `lifespan` independently.

docs/run/index.md

@@ -67,6 +67,9 @@ Each transport has its own keyword arguments, all on `run()`:
 * `streamable_http_path`: where the MCP endpoint lives. Default `/mcp`.
 * `json_response=True`: answer with plain JSON instead of an SSE stream.
 * `stateless_http=True`: a fresh transport per request, no session tracking.
+* `max_request_body_size`: largest accepted POST body in bytes. Defaults to 4 MiB; larger requests
+  receive HTTP 413 before parsing or session creation. Raise it only when legitimate MCP messages
+  exceed that size.
 * `event_store`, `retry_interval`, `transport_security`: resumability and DNS-rebinding protection. They can wait, until you deploy somewhere other than localhost; **[Deploy & scale](deploy.md)** covers `transport_security`.
 
 !!! warning

examples/stories/legacy_routing/README.md

@@ -68,7 +68,7 @@ async def mcp_endpoint(scope, receive, send):
         case "legacy":
             await my_existing_v1_manager.handle_request(scope, replay, send)
         case "modern":
-            await modern_manager.handle_request(scope, replay, send)
+            await modern_manager.asgi_app(scope, replay, send)
         case rejection:
             await send_jsonrpc_error(send, rejection)  # map via ERROR_CODE_HTTP_STATUS
 ```

src/mcp/server/lowlevel/server.py

@@ -64,7 +64,11 @@ from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestCon
 from mcp.server.models import InitializationOptions
 from mcp.server.runner import serve_dual_era_loop
 from mcp.server.streamable_http import EventStore
-from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
+from mcp.server.streamable_http_manager import (
+    DEFAULT_MAX_REQUEST_BODY_SIZE,
+    StreamableHTTPASGIApp,
+    StreamableHTTPSessionManager,
+)
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPDeprecationWarning
@@ -713,6 +717,7 @@ class Server(Generic[LifespanResultT]):
         stateless_http: bool = False,
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
         transport_security: TransportSecuritySettings | None = None,
         host: str = "127.0.0.1",
         auth: AuthSettings | None = None,
@@ -737,6 +742,7 @@ class Server(Generic[LifespanResultT]):
             json_response=json_response,
             stateless=stateless_http,
             security_settings=transport_security,
+            max_request_body_size=max_request_body_size,
         )
         self._session_manager = session_manager
 

src/mcp/server/mcpserver/server.py

@@ -87,7 +87,7 @@ from mcp.server.request_state import RequestStateBoundary, RequestStateSecurity
 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.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
@@ -369,6 +369,7 @@ class MCPServer(Generic[LifespanResultT]):
         stateless_http: bool = ...,
         event_store: EventStore | None = ...,
         retry_interval: int | None = ...,
+        max_request_body_size: int = ...,
         transport_security: TransportSecuritySettings | None = ...,
     ) -> None: ...
 
@@ -1050,6 +1051,7 @@ class MCPServer(Generic[LifespanResultT]):
         stateless_http: bool = False,
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
         transport_security: TransportSecuritySettings | None = None,
     ) -> None:
         """Run the server using StreamableHTTP transport."""
@@ -1061,6 +1063,7 @@ class MCPServer(Generic[LifespanResultT]):
             stateless_http=stateless_http,
             event_store=event_store,
             retry_interval=retry_interval,
+            max_request_body_size=max_request_body_size,
             transport_security=transport_security,
             host=host,
         )
@@ -1209,6 +1212,7 @@ class MCPServer(Generic[LifespanResultT]):
         stateless_http: bool = False,
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
         transport_security: TransportSecuritySettings | None = None,
         host: str = "127.0.0.1",
     ) -> Starlette:
@@ -1219,6 +1223,7 @@ class MCPServer(Generic[LifespanResultT]):
             stateless_http=stateless_http,
             event_store=event_store,
             retry_interval=retry_interval,
+            max_request_body_size=max_request_body_size,
             transport_security=transport_security,
             host=host,
             auth=self.settings.auth,

src/mcp/server/streamable_http_manager.py

@@ -4,27 +4,25 @@ from __future__ import annotations
 
 import contextlib
 import logging
+from collections import deque
 from collections.abc import AsyncIterator
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Final
 from uuid import uuid4
 
 import anyio
 from anyio.abc import TaskStatus
 from mcp_types import DEFAULT_NEGOTIATED_VERSION, INVALID_REQUEST, ErrorData, JSONRPCError
 from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS
+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._streamable_http_modern import handle_modern_request
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
 from mcp.server.connection import Connection
 from mcp.server.runner import serve_connection, serve_loop
-from mcp.server.streamable_http import (
+from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, EventStore, StreamableHTTPServerTransport
-    MCP_SESSION_ID_HEADER,
-    EventStore,
-    StreamableHTTPServerTransport,
-)
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._compat import resync_tracer
 from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER
@@ -36,6 +34,9 @@ if TYPE_CHECKING:
 
 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:
     """Manages StreamableHTTP sessions with optional resumability via event store.
@@ -69,6 +70,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__(
@@ -80,11 +83,14 @@ class StreamableHTTPSessionManager:
         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
@@ -93,6 +99,8 @@ class StreamableHTTPSessionManager:
         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()
@@ -159,6 +167,9 @@ class StreamableHTTPSessionManager:
 
         Dispatches to the appropriate handler based on stateless mode.
         """
+        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().")
 
@@ -360,6 +371,66 @@ class StreamableHTTPSessionManager:
             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)
+
+
 class StreamableHTTPASGIApp:
     """ASGI application for Streamable HTTP server transport."""
 
@@ -367,4 +438,4 @@ class StreamableHTTPASGIApp:
         self.session_manager = session_manager
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        await self.session_manager.handle_request(scope, receive, send)
+        await self.session_manager.asgi_app(scope, receive, send)

tests/docs_src/test_asgi.py

@@ -43,6 +43,7 @@ async def test_streamable_http_app_takes_runs_options_except_port() -> None:
         "stateless_http",
         "event_store",
         "retry_interval",
+        "max_request_body_size",
         "transport_security",
         "host",
     }
@@ -56,6 +57,17 @@ async def test_a_request_before_the_session_manager_runs_is_rejected() -> None:
             await http.post("/mcp")
 
 
+async def test_streamable_http_app_applies_the_configured_request_body_limit() -> None:
+    """The documented `max_request_body_size` option rejects larger requests with HTTP 413."""
+    server = MCPServer("Notes")
+    app = server.streamable_http_app(max_request_body_size=8)
+    transport = httpx2.ASGITransport(app=app)
+    async with server.session_manager.run():
+        async with httpx2.AsyncClient(transport=transport, base_url="http://localhost") as http:
+            response = await http.post("/mcp", content=b"123456789")
+    assert response.status_code == 413
+
+
 async def test_mounting_at_the_root_keeps_the_default_path() -> None:
     """tutorial002: `Mount("/")` plus the default `streamable_http_path` leaves the endpoint at `/mcp`."""
     (mount,) = tutorial002.app.routes

tests/docs_src/test_legacy_clients.py

@@ -52,6 +52,7 @@ def test_streamable_http_app_has_no_era_knob() -> None:
         "stateless_http",
         "event_store",
         "retry_interval",
+        "max_request_body_size",
         "transport_security",
         "host",
     }

tests/server/test_streamable_http_manager.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+from collections.abc import Iterator
 from typing import Any
 from unittest.mock import AsyncMock, patch
 
@@ -9,7 +10,7 @@ import anyio
 import httpx2
 import pytest
 from mcp_types import INVALID_REQUEST, ListToolsResult, PaginatedRequestParams
-from starlette.types import Message, Scope
+from starlette.types import Message, Receive, Scope, Send
 
 from mcp import Client
 from mcp.client.streamable_http import streamable_http_client
@@ -17,7 +18,11 @@ from mcp.server import Server, ServerRequestContext, streamable_http_manager
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
 from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, StreamableHTTPServerTransport
-from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.streamable_http_manager import (
+    DEFAULT_MAX_REQUEST_BODY_SIZE,
+    RequestBodyLimitMiddleware,
+    StreamableHTTPSessionManager,
+)
 
 
 @pytest.mark.anyio
@@ -71,9 +76,9 @@ async def test_handle_request_without_run_raises_error():
     manager = StreamableHTTPSessionManager(app=app)
 
     # Mock ASGI parameters
-    scope = {"type": "http", "method": "POST", "path": "/test"}
+    scope: Scope = {"type": "http", "method": "POST", "path": "/test", "headers": []}
 
-    async def receive():  # pragma: no cover
+    async def receive() -> Message:
         return {"type": "http.request", "body": b""}
 
     async def send(message: Message):  # pragma: no cover
@@ -86,6 +91,148 @@ async def test_handle_request_without_run_raises_error():
     assert "Task group is not initialized. Make sure to use run()." in str(excinfo.value)
 
 
+@pytest.mark.anyio
+async def test_oversized_content_length_is_rejected_before_body_read_or_session_creation() -> None:
+    """SDK-defined: an oversized declared body gets HTTP 413 before the server reads it or creates a session."""
+    manager = StreamableHTTPSessionManager(app=Server("test-size-limit"), max_request_body_size=8)
+    sent_messages: list[Message] = []
+    receive = AsyncMock(return_value={"type": "http.request", "body": b"123456789", "more_body": False})
+
+    async def send(message: Message) -> None:
+        sent_messages.append(message)
+
+    scope: Scope = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "headers": [(b"content-length", b"9")],
+    }
+    async with manager.run():
+        await manager.handle_request(scope, receive, send)
+        assert manager._server_instances == {}
+
+    response_start = next(message for message in sent_messages if message["type"] == "http.response.start")
+    assert response_start["status"] == 413
+    receive.assert_not_awaited()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("headers", [[], [(b"content-length", b"invalid")], [(b"content-length", b"8")]])
+async def test_oversized_streamed_body_is_rejected_before_session_creation(
+    headers: list[tuple[bytes, bytes]],
+) -> None:
+    """SDK-defined: streamed bodies enforce the limit with missing, invalid, or understated length."""
+    manager = StreamableHTTPSessionManager(app=Server("test-streamed-size-limit"), max_request_body_size=8)
+    sent_messages: list[Message] = []
+    request_messages: Iterator[Message] = iter(
+        [
+            {"type": "http.request", "body": b"1234", "more_body": True},
+            {"type": "http.request", "body": b"56789", "more_body": False},
+        ]
+    )
+
+    async def receive() -> Message:
+        return next(request_messages)
+
+    async def send(message: Message) -> None:
+        sent_messages.append(message)
+
+    scope: Scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": headers}
+    async with manager.run():
+        await manager.asgi_app(scope, receive, send)
+        assert manager._server_instances == {}
+
+    response_start = next(message for message in sent_messages if message["type"] == "http.response.start")
+    assert response_start["status"] == 413
+
+
+@pytest.mark.anyio
+async def test_client_disconnect_while_streaming_request_body_is_replayed() -> None:
+    """SDK-defined: raw ASGI is required to prove a disconnect before body completion reaches the transport."""
+    disconnect: Message = {"type": "http.disconnect"}
+    request_messages: Iterator[Message] = iter(
+        [{"type": "http.request", "body": b"1234", "more_body": True}, disconnect]
+    )
+    received_messages: list[Message] = []
+
+    async def receive() -> Message:
+        return next(request_messages)
+
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        received_messages.append(await receive())
+        received_messages.append(await receive())
+
+    scope: Scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": []}
+    middleware = RequestBodyLimitMiddleware(app, max_body_size=8)
+
+    await middleware(scope, receive, AsyncMock())
+
+    assert received_messages == [
+        {"type": "http.request", "body": b"1234", "more_body": True},
+        disconnect,
+    ]
+
+
+@pytest.mark.anyio
+async def test_client_disconnect_before_request_body_is_replayed() -> None:
+    """SDK-defined: raw ASGI proves a disconnect before the first body message reaches the transport."""
+    disconnect: Message = {"type": "http.disconnect"}
+    received_messages: list[Message] = []
+
+    async def receive() -> Message:
+        return disconnect
+
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        received_messages.append(await receive())
+
+    scope: Scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": []}
+    middleware = RequestBodyLimitMiddleware(app, max_body_size=8)
+
+    await middleware(scope, receive, AsyncMock())
+
+    assert received_messages == [disconnect]
+
+
+@pytest.mark.anyio
+async def test_request_body_chunks_are_replayed_as_one_message() -> None:
+    """SDK-defined: raw ASGI proves chunk overhead is discarded before the body reaches the transport."""
+    request_messages: Iterator[Message] = iter(
+        [
+            {"type": "http.request", "body": b"12", "more_body": True},
+            {"type": "http.request", "body": b"34", "more_body": True},
+            {"type": "http.request", "body": b"56", "more_body": False},
+        ]
+    )
+    received_messages: list[Message] = []
+
+    async def receive() -> Message:
+        return next(request_messages)
+
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        received_messages.append(await receive())
+
+    scope: Scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": []}
+    middleware = RequestBodyLimitMiddleware(app, max_body_size=8)
+
+    await middleware(scope, receive, AsyncMock())
+
+    assert received_messages == [{"type": "http.request", "body": b"123456", "more_body": False}]
+
+
+def test_request_body_limit_defaults_to_four_mib() -> None:
+    """SDK-defined: Streamable HTTP request bodies are limited to 4 MiB by default."""
+    manager = StreamableHTTPSessionManager(app=Server("test-default-size-limit"))
+    assert manager.max_request_body_size == DEFAULT_MAX_REQUEST_BODY_SIZE == 4 * 1024 * 1024
+
+
+@pytest.mark.parametrize("max_request_body_size", [0, -1])
+def test_request_body_limit_rejects_non_positive_values(max_request_body_size: int) -> None:
+    """SDK-defined: callers cannot disable request-size protection with a non-positive value."""
+    with pytest.raises(ValueError) as exc_info:
+        StreamableHTTPSessionManager(app=Server("test-invalid-size-limit"), max_request_body_size=max_request_body_size)
+    assert str(exc_info.value) == "max_request_body_size must be a positive number of bytes"
+
+
 class TestException(Exception):
     __test__ = False  # Prevent pytest from collecting this as a test class
     pass
@@ -122,7 +269,7 @@ async def test_stateful_session_cleanup_on_graceful_exit(running_manager: tuple[
         "headers": [(b"content-type", b"application/json")],
     }
 
-    async def mock_receive():  # pragma: no cover
+    async def mock_receive():
         return {"type": "http.request", "body": b"", "more_body": False}
 
     # Trigger session creation
@@ -180,7 +327,7 @@ async def test_stateful_session_cleanup_on_exception(running_manager: tuple[Stre
         "headers": [(b"content-type", b"application/json")],
     }
 
-    async def mock_receive():  # pragma: no cover
+    async def mock_receive():
         return {"type": "http.request", "body": b"", "more_body": False}
 
     # Trigger session creation
@@ -298,7 +445,7 @@ async def test_unknown_session_id_returns_404(caplog: pytest.LogCaptureFixture):
         }
 
         async def mock_receive():
-            return {"type": "http.request", "body": b"{}", "more_body": False}  # pragma: no cover
+            return {"type": "http.request", "body": b"{}", "more_body": False}
 
         with caplog.at_level(logging.INFO):
             await manager.handle_request(scope, mock_receive, mock_send)
@@ -378,7 +525,7 @@ async def test_idle_session_is_reaped(caplog: pytest.LogCaptureFixture, request:
             "headers": [(b"content-type", b"application/json")],
         }
 
-        async def mock_receive():  # pragma: no cover
+        async def mock_receive():
             return {"type": "http.request", "body": b"", "more_body": False}
 
         await manager.handle_request(scope, mock_receive, mock_send)