diff --git a/docs/migration.md b/docs/migration.md index 73329622cb..876608db70 100644 --- a/docs/migration.md +++ b/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. diff --git a/docs/run/index.md b/docs/run/index.md index b3cea554cc..4ff8f53deb 100644 --- a/docs/run/index.md +++ b/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 diff --git a/examples/stories/legacy_routing/README.md b/examples/stories/legacy_routing/README.md index 857b3270d1..84a4528c96 100644 --- a/examples/stories/legacy_routing/README.md +++ b/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 ``` diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index dc83563967..abf98c6fdb 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -64,7 +64,11 @@ async def main(): 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 @@ def streamable_http_app( 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 @@ def streamable_http_app( json_response=json_response, stateless=stateless_http, security_settings=transport_security, + max_request_body_size=max_request_body_size, ) self._session_manager = session_manager diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py index d933e82d55..7fc1cd1948 100644 --- a/src/mcp/server/mcpserver/server.py +++ b/src/mcp/server/mcpserver/server.py @@ -87,7 +87,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.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus from mcp.server.transport_security import TransportSecuritySettings from mcp.shared.exceptions import MCPError @@ -369,6 +369,7 @@ def run( 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 @@ async def run_streamable_http_async( # pragma: no cover 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 @@ async def run_streamable_http_async( # pragma: no cover 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 @@ def streamable_http_app( 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 @@ def streamable_http_app( 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, diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py index 60b0989611..31f587ee66 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/src/mcp/server/streamable_http_manager.py @@ -4,27 +4,25 @@ 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 ( - MCP_SESSION_ID_HEADER, - EventStore, - StreamableHTTPServerTransport, -) +from mcp.server.streamable_http import 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 @@ 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 @@ 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 @@ -93,6 +99,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() @@ -159,6 +167,9 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No 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 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE 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 @@ def __init__(self, session_manager: StreamableHTTPSessionManager): 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) diff --git a/tests/docs_src/test_asgi.py b/tests/docs_src/test_asgi.py index f10ee2e22c..ef237bc212 100644 --- a/tests/docs_src/test_asgi.py +++ b/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 diff --git a/tests/docs_src/test_legacy_clients.py b/tests/docs_src/test_legacy_clients.py index c052c6940d..3018e4fcaa 100644 --- a/tests/docs_src/test_legacy_clients.py +++ b/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", } diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py index c825ffb207..70440d9d03 100644 --- a/tests/server/test_streamable_http_manager.py +++ b/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 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.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 send(message: Message): # pragma: no cover 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 mock_send(message: Message): "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 mock_send(message: Message): "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 mock_send(message: Message): } 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 mock_send(message: Message): "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)