server: add max_body_bytes guard for HTTP request bodies

Author: Theodor N. Engøy

src/mcp/server/auth/handlers/register.py

@@ -12,6 +12,7 @@
 from mcp.server.auth.json_response import PydanticJSONResponse
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, RegistrationError, RegistrationErrorCode
 from mcp.server.auth.settings import ClientRegistrationOptions
+from mcp.server.http_body import BodyTooLargeError, read_request_body
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
 
 # this alias is a no-op; it's just to separate out the types exposed to the
@@ -32,10 +33,12 @@ class RegistrationHandler:
     async def handle(self, request: Request) -> Response:
         # Implements dynamic client registration as defined in https://datatracker.ietf.org/doc/html/rfc7591#section-3.1
         try:
-            body = await request.body()
+            body = await read_request_body(request, max_body_bytes=self.options.max_body_bytes)
             client_metadata = OAuthClientMetadata.model_validate_json(body)
 
             # Scope validation is handled below
+        except BodyTooLargeError:
+            return Response("Payload too large", status_code=413, headers={"Connection": "close"})
         except ValidationError as validation_error:
             return PydanticJSONResponse(
                 content=RegistrationErrorResponse(

src/mcp/server/auth/settings.py

@@ -1,11 +1,15 @@
 from pydantic import AnyHttpUrl, BaseModel, Field
 
+from mcp.server.http_body import DEFAULT_MAX_BODY_BYTES
+
 
 class ClientRegistrationOptions(BaseModel):
     enabled: bool = False
     client_secret_expiry_seconds: int | None = None
     valid_scopes: list[str] | None = None
     default_scopes: list[str] | None = None
+    # Limit the size of incoming /register request bodies to avoid DoS via unbounded reads.
+    max_body_bytes: int = DEFAULT_MAX_BODY_BYTES
 
 
 class RevocationOptions(BaseModel):

src/mcp/server/http_body.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from starlette.requests import Request
+
+DEFAULT_MAX_BODY_BYTES = 1_000_000
+
+
+@dataclass(frozen=True)
+class BodyTooLargeError(Exception):
+    max_body_bytes: int
+
+    def __str__(self) -> str:
+        return f"Request body exceeds max_body_bytes={self.max_body_bytes}"
+
+
+async def read_request_body(request: Request, *, max_body_bytes: int | None = DEFAULT_MAX_BODY_BYTES) -> bytes:
+    """Read an HTTP request body with a hard cap.
+
+    Notes:
+    - This avoids unbounded buffering of the request body in Python.
+    - If the body exceeds max_body_bytes, this raises BodyTooLargeError as soon
+      as possible.
+    """
+    if max_body_bytes is None:
+        return await request.body()
+
+    if max_body_bytes <= 0:
+        raise ValueError("max_body_bytes must be positive or None")
+
+    # Fast-path: reject based on Content-Length when provided.
+    content_length = request.headers.get("content-length")
+    if content_length is not None:
+        try:
+            if int(content_length) > max_body_bytes:
+                raise BodyTooLargeError(max_body_bytes)
+        except ValueError:
+            # Ignore invalid Content-Length; we'll enforce while streaming.
+            pass
+
+    body = bytearray()
+    async for chunk in request.stream():
+        if not chunk:
+            continue
+
+        # Never buffer more than max_body_bytes bytes.
+        remaining = max_body_bytes - len(body)
+        if remaining <= 0:
+            raise BodyTooLargeError(max_body_bytes)
+        if len(chunk) > remaining:
+            body.extend(chunk[:remaining])
+            raise BodyTooLargeError(max_body_bytes)
+
+        body.extend(chunk)
+
+    return bytes(body)
+

src/mcp/server/lowlevel/server.py

@@ -93,6 +93,7 @@ async def main():
 from mcp.server.auth.settings import AuthSettings
 from mcp.server.context import ServerRequestContext
 from mcp.server.experimental.request_context import Experimental
+from mcp.server.http_body import DEFAULT_MAX_BODY_BYTES
 from mcp.server.lowlevel.experimental import ExperimentalHandlers
 from mcp.server.lowlevel.func_inspection import create_call_wrapper
 from mcp.server.lowlevel.helper_types import ReadResourceContents
@@ -810,14 +811,20 @@ def streamable_http_app(
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
         transport_security: TransportSecuritySettings | None = None,
+        max_body_bytes: int | None = DEFAULT_MAX_BODY_BYTES,
         host: str = "127.0.0.1",
         auth: AuthSettings | None = None,
         token_verifier: TokenVerifier | None = None,
         auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
         custom_starlette_routes: list[Route] | None = None,
         debug: bool = False,
     ) -> Starlette:
-        """Return an instance of the StreamableHTTP server app."""
+        """Return an instance of the StreamableHTTP server app.
+
+        Args:
+            max_body_bytes: Maximum size (in bytes) for JSON POST request bodies. Defaults
+                            to 1_000_000. Set to None to disable this guard.
+        """
         # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
         if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
             transport_security = TransportSecuritySettings(
@@ -833,6 +840,7 @@ def streamable_http_app(
             json_response=json_response,
             stateless=stateless_http,
             security_settings=transport_security,
+            max_body_bytes=max_body_bytes,
         )
         self._session_manager = session_manager
 

src/mcp/server/sse.py

@@ -51,6 +51,7 @@ async def handle_sse(request):
 from starlette.types import Receive, Scope, Send
 
 from mcp import types
+from mcp.server.http_body import DEFAULT_MAX_BODY_BYTES, BodyTooLargeError, read_request_body
 from mcp.server.transport_security import (
     TransportSecurityMiddleware,
     TransportSecuritySettings,
@@ -75,14 +76,21 @@ class SseServerTransport:
     _read_stream_writers: dict[UUID, MemoryObjectSendStream[SessionMessage | Exception]]
     _security: TransportSecurityMiddleware
 
-    def __init__(self, endpoint: str, security_settings: TransportSecuritySettings | None = None) -> None:
+    def __init__(
+        self,
+        endpoint: str,
+        security_settings: TransportSecuritySettings | None = None,
+        max_body_bytes: int | None = DEFAULT_MAX_BODY_BYTES,
+    ) -> None:
         """Creates a new SSE server transport, which will direct the client to POST
         messages to the relative path given.
 
         Args:
             endpoint: A relative path where messages should be posted
                     (e.g., "/messages/").
             security_settings: Optional security settings for DNS rebinding protection.
+            max_body_bytes: Maximum size (in bytes) for JSON POST request bodies. Defaults
+                            to 1_000_000. Set to None to disable this guard.
 
         Note:
             We use relative paths instead of full URLs for several reasons:
@@ -98,6 +106,8 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
         """
 
         super().__init__()
+        if max_body_bytes is not None and max_body_bytes <= 0:
+            raise ValueError("max_body_bytes must be positive or None")
 
         # Validate that endpoint is a relative path and not a full URL
         if "://" in endpoint or endpoint.startswith("//") or "?" in endpoint or "#" in endpoint:
@@ -113,6 +123,7 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
         self._endpoint = endpoint
         self._read_stream_writers = {}
         self._security = TransportSecurityMiddleware(security_settings)
+        self._max_body_bytes = max_body_bytes
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
     @asynccontextmanager
@@ -223,7 +234,12 @@ async def handle_post_message(self, scope: Scope, receive: Receive, send: Send)
             response = Response("Could not find session", status_code=404)
             return await response(scope, receive, send)
 
-        body = await request.body()
+        try:
+            body = await read_request_body(request, max_body_bytes=self._max_body_bytes)
+        except BodyTooLargeError as e:
+            response = Response("Payload too large", status_code=413, headers={"Connection": "close"})
+            logger.warning(f"Received payload too large: {e}")
+            return await response(scope, receive, send)
         logger.debug(f"Received JSON: {body}")
 
         try:

src/mcp/server/streamable_http.py

@@ -24,6 +24,7 @@
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
+from mcp.server.http_body import DEFAULT_MAX_BODY_BYTES, BodyTooLargeError, read_request_body
 from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
@@ -132,6 +133,7 @@ def __init__(
         event_store: EventStore | None = None,
         security_settings: TransportSecuritySettings | None = None,
         retry_interval: int | None = None,
+        max_body_bytes: int | None = DEFAULT_MAX_BODY_BYTES,
     ) -> None:
         """Initialize a new StreamableHTTP server transport.
 
@@ -148,18 +150,23 @@ def __init__(
                            retry field. When set, the server will send a retry field in
                            SSE priming events to control client reconnection timing for
                            polling behavior. Only used when event_store is provided.
+            max_body_bytes: Maximum size (in bytes) for JSON POST request bodies. Defaults
+                           to 1_000_000. Set to None to disable this guard.
 
         Raises:
             ValueError: If the session ID contains invalid characters.
         """
         if mcp_session_id is not None and not SESSION_ID_PATTERN.fullmatch(mcp_session_id):
             raise ValueError("Session ID must only contain visible ASCII characters (0x21-0x7E)")
+        if max_body_bytes is not None and max_body_bytes <= 0:
+            raise ValueError("max_body_bytes must be positive or None")
 
         self.mcp_session_id = mcp_session_id
         self.is_json_response_enabled = is_json_response_enabled
         self._event_store = event_store
         self._security = TransportSecurityMiddleware(security_settings)
         self._retry_interval = retry_interval
+        self._max_body_bytes = max_body_bytes
         self._request_streams: dict[
             RequestId,
             tuple[
@@ -427,6 +434,43 @@ async def _validate_accept_header(self, request: Request, scope: Scope, send: Se
             return False
         return True
 
+    async def _parse_jsonrpc_message(
+        self,
+        request: Request,
+        scope: Scope,
+        receive: Receive,
+        send: Send,
+    ) -> JSONRPCMessage | None:
+        """Read + parse a JSON-RPC message from an HTTP request body."""
+        try:
+            body = await read_request_body(request, max_body_bytes=self._max_body_bytes)
+        except BodyTooLargeError as e:
+            response = self._create_error_response(
+                f"Payload too large: {e}",
+                HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
+                headers={"Connection": "close"},
+            )
+            await response(scope, receive, send)
+            return None
+
+        try:
+            raw_message = pydantic_core.from_json(body)
+        except ValueError as e:
+            response = self._create_error_response(f"Parse error: {str(e)}", HTTPStatus.BAD_REQUEST, PARSE_ERROR)
+            await response(scope, receive, send)
+            return None
+
+        try:
+            return jsonrpc_message_adapter.validate_python(raw_message, by_name=False)
+        except ValidationError as e:  # pragma: no cover
+            response = self._create_error_response(
+                f"Validation error: {str(e)}",
+                HTTPStatus.BAD_REQUEST,
+                INVALID_PARAMS,
+            )
+            await response(scope, receive, send)
+            return None
+
     async def _handle_post_request(self, scope: Scope, request: Request, receive: Receive, send: Send) -> None:
         """Handle POST requests containing JSON-RPC messages."""
         writer = self._read_stream_writer
@@ -446,25 +490,8 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 await response(scope, receive, send)
                 return
 
-            # Parse the body - only read it once
-            body = await request.body()
-
-            try:
-                raw_message = pydantic_core.from_json(body)
-            except ValueError as e:
-                response = self._create_error_response(f"Parse error: {str(e)}", HTTPStatus.BAD_REQUEST, PARSE_ERROR)
-                await response(scope, receive, send)
-                return
-
-            try:
-                message = jsonrpc_message_adapter.validate_python(raw_message, by_name=False)
-            except ValidationError as e:  # pragma: no cover
-                response = self._create_error_response(
-                    f"Validation error: {str(e)}",
-                    HTTPStatus.BAD_REQUEST,
-                    INVALID_PARAMS,
-                )
-                await response(scope, receive, send)
+            message = await self._parse_jsonrpc_message(request, scope, receive, send)
+            if message is None:
                 return
 
             # Check if this is an initialization request

src/mcp/server/streamable_http_manager.py

@@ -15,6 +15,7 @@
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
+from mcp.server.http_body import DEFAULT_MAX_BODY_BYTES
 from mcp.server.streamable_http import (
     MCP_SESSION_ID_HEADER,
     EventStore,
@@ -56,6 +57,8 @@ class StreamableHTTPSessionManager:
         security_settings: Optional transport security settings.
         retry_interval: Retry interval in milliseconds to suggest to clients in SSE
                        retry field. Used for SSE polling behavior.
+        max_body_bytes: Maximum size (in bytes) for JSON POST request bodies. Defaults
+                        to 1_000_000. Set to None to disable this guard.
     """
 
     def __init__(
@@ -66,13 +69,15 @@ def __init__(
         stateless: bool = False,
         security_settings: TransportSecuritySettings | None = None,
         retry_interval: int | None = None,
+        max_body_bytes: int | None = DEFAULT_MAX_BODY_BYTES,
     ):
         self.app = app
         self.event_store = event_store
         self.json_response = json_response
         self.stateless = stateless
         self.security_settings = security_settings
         self.retry_interval = retry_interval
+        self.max_body_bytes = max_body_bytes
 
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
@@ -147,6 +152,7 @@ async def _handle_stateless_request(self, scope: Scope, receive: Receive, send:
             is_json_response_enabled=self.json_response,
             event_store=None,  # No event store in stateless mode
             security_settings=self.security_settings,
+            max_body_bytes=self.max_body_bytes,
         )
 
         # Start server in a new task
@@ -198,6 +204,7 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
                     event_store=self.event_store,  # May be None (no resumability)
                     security_settings=self.security_settings,
                     retry_interval=self.retry_interval,
+                    max_body_bytes=self.max_body_bytes,
                 )
 
                 assert http_transport.mcp_session_id is not None

tests/server/auth/test_error_handling.py

@@ -116,6 +116,31 @@ async def test_registration_error_handling(client: httpx.AsyncClient, oauth_prov
         assert data["error_description"] == "The redirect URI is invalid"
 
 
+@pytest.mark.anyio
+async def test_registration_rejects_payload_too_large(oauth_provider: MockOAuthProvider):
+    client_registration_options = ClientRegistrationOptions(enabled=True, max_body_bytes=10)
+    revocation_options = RevocationOptions(enabled=False)
+
+    auth_routes = create_auth_routes(
+        oauth_provider,
+        issuer_url=AnyHttpUrl("http://localhost"),
+        client_registration_options=client_registration_options,
+        revocation_options=revocation_options,
+    )
+    app = Starlette(routes=auth_routes)
+
+    transport = ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
+        body = b'{"a":"' + (b"x" * 20) + b'"}'
+        response = await client.post(
+            "/register",
+            content=body,
+            headers={"Content-Type": "application/json"},
+        )
+        assert response.status_code == 413, response.content
+        assert response.text == "Payload too large"
+
+
 @pytest.mark.anyio
 async def test_authorize_error_handling(
     client: httpx.AsyncClient,