server: add max_body_bytes guard for HTTP request bodies

README.md

@@ -1429,6 +1429,8 @@ app = Starlette(
 )
 ```
 
+Security note: StreamableHTTP enforces a default `max_body_bytes=1_000_000` limit for incoming `application/json` POST bodies (413 on oversized payloads). Override via `mcp.streamable_http_app(max_body_bytes=...)` or `mcp.run("streamable-http", ..., max_body_bytes=...)`. Set to `None` to disable (not recommended).
+
 _Full example: [examples/snippets/servers/streamable_http_basic_mounting.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_http_basic_mounting.py)_
 <!-- /snippet-source -->
 
@@ -1601,6 +1603,8 @@ app = Starlette(
 app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))
 ```
 
+Security note: SSE message endpoints enforce a default `max_body_bytes=1_000_000` limit for incoming `application/json` POST bodies (413 on oversized payloads). Override via `mcp.sse_app(max_body_bytes=...)` or `mcp.run("sse", ..., max_body_bytes=...)`. Set to `None` to disable (not recommended).
+
 You can also mount multiple MCP servers at different sub-paths. The SSE transport automatically detects the mount path via ASGI's `root_path` mechanism, so message endpoints are correctly routed:
 
 ```python

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,57 @@
+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
@@ -276,7 +277,7 @@ def session_manager(self) -> StreamableHTTPSessionManager:
                 "Session manager can only be accessed after calling streamable_http_app(). "
                 "The session manager is created lazily to avoid unnecessary initialization."
             )
-        return self._session_manager  # pragma: no cover
+        return self._session_manager
 
     def list_prompts(self):
         def decorator(
@@ -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/mcpserver/server.py

@@ -28,6 +28,7 @@
 from mcp.server.context import LifespanContextT, RequestT, ServerRequestContext
 from mcp.server.elicitation import ElicitationResult, ElicitSchemaModelT, UrlElicitationResult, elicit_with_validation
 from mcp.server.elicitation import elicit_url as _elicit_url
+from mcp.server.http_body import DEFAULT_MAX_BODY_BYTES
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import LifespanResultT, Server
 from mcp.server.lowlevel.server import lifespan as default_lifespan
@@ -208,7 +209,7 @@ def session_manager(self) -> StreamableHTTPSessionManager:
         Raises:
             RuntimeError: If called before streamable_http_app() has been called.
         """
-        return self._lowlevel_server.session_manager  # pragma: no cover
+        return self._lowlevel_server.session_manager
 
     @overload
     def run(self, transport: Literal["stdio"] = ...) -> None: ...
@@ -223,6 +224,7 @@ def run(
         sse_path: str = ...,
         message_path: str = ...,
         transport_security: TransportSecuritySettings | None = ...,
+        max_body_bytes: int | None = ...,
     ) -> None: ...
 
     @overload
@@ -238,6 +240,7 @@ def run(
         event_store: EventStore | None = ...,
         retry_interval: int | None = ...,
         transport_security: TransportSecuritySettings | None = ...,
+        max_body_bytes: int | None = ...,
     ) -> None: ...
 
     def run(
@@ -725,6 +728,7 @@ async def run_sse_async(  # pragma: no cover
         sse_path: str = "/sse",
         message_path: str = "/messages/",
         transport_security: TransportSecuritySettings | None = None,
+        max_body_bytes: int | None = DEFAULT_MAX_BODY_BYTES,
     ) -> None:
         """Run the server using SSE transport."""
         import uvicorn
@@ -734,6 +738,7 @@ async def run_sse_async(  # pragma: no cover
             message_path=message_path,
             transport_security=transport_security,
             host=host,
+            max_body_bytes=max_body_bytes,
         )
 
         config = uvicorn.Config(
@@ -756,6 +761,7 @@ async def run_streamable_http_async(  # pragma: no cover
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
         transport_security: TransportSecuritySettings | None = None,
+        max_body_bytes: int | None = DEFAULT_MAX_BODY_BYTES,
     ) -> None:
         """Run the server using StreamableHTTP transport."""
         import uvicorn
@@ -768,6 +774,7 @@ async def run_streamable_http_async(  # pragma: no cover
             retry_interval=retry_interval,
             transport_security=transport_security,
             host=host,
+            max_body_bytes=max_body_bytes,
         )
 
         config = uvicorn.Config(
@@ -785,6 +792,7 @@ def sse_app(
         sse_path: str = "/sse",
         message_path: str = "/messages/",
         transport_security: TransportSecuritySettings | None = None,
+        max_body_bytes: int | None = DEFAULT_MAX_BODY_BYTES,
         host: str = "127.0.0.1",
     ) -> Starlette:
         """Return an instance of the SSE server app."""
@@ -796,7 +804,11 @@ def sse_app(
                 allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
             )
 
-        sse = SseServerTransport(message_path, security_settings=transport_security)
+        sse = SseServerTransport(
+            message_path,
+            security_settings=transport_security,
+            max_body_bytes=max_body_bytes,
+        )
 
         async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no cover
             # Add client ID from auth context into request context if available
@@ -914,6 +926,7 @@ 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",
     ) -> Starlette:
         """Return an instance of the StreamableHTTP server app."""
@@ -924,6 +937,7 @@ def streamable_http_app(
             event_store=event_store,
             retry_interval=retry_interval,
             transport_security=transport_security,
+            max_body_bytes=max_body_bytes,
             host=host,
             auth=self.settings.auth,
             token_verifier=self._token_verifier,

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
@@ -194,7 +205,7 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
             logger.debug("Yielding read and write streams")
             yield (read_stream, write_stream)
 
-    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:  # pragma: no cover
+    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:
         logger.debug("Handling POST message")
         request = Request(scope, receive)
 
@@ -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: