diff --git a/docs/server.md b/docs/server.md index 6340687c31..6402596a57 100644 --- a/docs/server.md +++ b/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) +``` + ```python """ diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 7b0de09836..e915a12bfd 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/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 diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py index 1a1a85721d..0ee6d362b5 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/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) diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index 7f7d27e8c3..b134489bc5 100644 --- a/tests/server/fastmcp/test_server.py +++ b/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 diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py index 0ae07c43ad..9deeeeb37a 100644 --- a/tests/server/test_streamable_http_manager.py +++ b/tests/server/test_streamable_http_manager.py @@ -1,19 +1,24 @@ """Tests for StreamableHTTPSessionManager.""" import json +from collections.abc import Iterator from typing import Any from unittest.mock import AsyncMock, patch import anyio import pytest -from starlette.types import Message, Scope +from starlette.types import Message, Receive, Scope, Send from mcp.server import streamable_http_manager from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser from mcp.server.auth.provider import AccessToken from mcp.server.lowlevel import Server 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, +) from mcp.types import INVALID_REQUEST @@ -68,9 +73,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 @@ -83,6 +88,126 @@ 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.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 + + +@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}, + {"type": "http.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"123456", "more_body": False}, + {"type": "http.disconnect"}, + ] + + +@pytest.mark.anyio +async def test_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] + + +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 @@ -118,7 +243,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 @@ -177,7 +302,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 @@ -297,7 +422,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} await manager.handle_request(scope, mock_receive, mock_send) @@ -336,7 +461,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)