Apply the request body limit to the SSE and OAuth endpoints

docs/migration.md

@@ -755,7 +755,7 @@ Transport-specific parameters have been moved off the `MCPServer` constructor an
 - `sse_path`, `message_path` - SSE transport paths, on `run(transport="sse", ...)` and `sse_app()`
 - `streamable_http_path` - StreamableHTTP endpoint path, on `run(transport="streamable-http", ...)` and `streamable_http_app()`
 - `json_response`, `stateless_http` - StreamableHTTP behavior, same two places; each also removes a server-to-client channel, see [Server-initiated sampling, elicitation, and roots raise `NoBackChannelError`](#server-initiated-sampling-elicitation-and-roots-raise-nobackchannelerror)
-- `max_request_body_size` - StreamableHTTP request-body limit, same two places
+- `max_request_body_size` - HTTP request-body limit, on `run()` for both HTTP transports and on both app methods
 - `event_store`, `retry_interval` - StreamableHTTP event handling, same two places
 - `transport_security` - DNS rebinding protection, on `run()` for both HTTP transports and on both app methods
 

src/mcp/server/auth/routes.py

@@ -17,6 +17,7 @@
 from mcp.server.auth.middleware.client_auth import ClientAuthenticator
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
+from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, RequestBodyLimitMiddleware
 from mcp.shared.auth import JWT_BEARER_GRANT_TYPE, OAuthMetadata, ProtectedResourceMetadata
 from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER
 
@@ -51,12 +52,17 @@ def validate_issuer_url(url: AnyHttpUrl):
 ID_JAG_GRANT_PROFILE = "urn:ietf:params:oauth:grant-profile:id-jag"
 
 
+def _body_limited(handler: Callable[[Request], Response | Awaitable[Response]]) -> ASGIApp:
+    """Wrap an endpoint so POST bodies over the default limit are answered with 413 before it runs."""
+    return RequestBodyLimitMiddleware(request_response(handler), DEFAULT_MAX_REQUEST_BODY_SIZE)
+
+
 def cors_middleware(
     handler: Callable[[Request], Response | Awaitable[Response]],
     allow_methods: list[str],
 ) -> ASGIApp:
     cors_app = CORSMiddleware(
-        app=request_response(handler),
+        app=_body_limited(handler),
         allow_origins="*",
         allow_methods=allow_methods,
         allow_headers=[MCP_PROTOCOL_VERSION_HEADER],
@@ -102,7 +108,7 @@ def create_auth_routes(
             AUTHORIZATION_PATH,
             # do not allow CORS for authorization endpoint;
             # clients should just redirect to this
-            endpoint=AuthorizationHandler(provider).handle,
+            endpoint=_body_limited(AuthorizationHandler(provider).handle),
             methods=["GET", "POST"],
         ),
         Route(

src/mcp/server/mcpserver/server.py

@@ -365,6 +365,7 @@ def run(
         port: int = ...,
         sse_path: str = ...,
         message_path: str = ...,
+        max_request_body_size: int = ...,
         transport_security: TransportSecuritySettings | None = ...,
     ) -> None: ...
 
@@ -1031,6 +1032,7 @@ async def run_sse_async(  # pragma: no cover
         port: int = 8000,
         sse_path: str = "/sse",
         message_path: str = "/messages/",
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
         transport_security: TransportSecuritySettings | None = None,
     ) -> None:
         """Run the server using SSE transport."""
@@ -1039,6 +1041,7 @@ async def run_sse_async(  # pragma: no cover
         starlette_app = self.sse_app(
             sse_path=sse_path,
             message_path=message_path,
+            max_request_body_size=max_request_body_size,
             transport_security=transport_security,
             host=host,
         )
@@ -1093,6 +1096,7 @@ def sse_app(
         *,
         sse_path: str = "/sse",
         message_path: str = "/messages/",
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
         transport_security: TransportSecuritySettings | None = None,
         host: str = "127.0.0.1",
     ) -> Starlette:
@@ -1105,7 +1109,9 @@ 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_request_body_size=max_request_body_size
+        )
 
         async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no cover
             # Add client ID from auth context into request context if available

src/mcp/server/sse.py

@@ -51,6 +51,7 @@ async def handle_sse(request):
 from starlette.types import Receive, Scope, Send
 
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
+from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, RequestBodyLimitMiddleware
 from mcp.server.transport_security import (
     TransportSecurityMiddleware,
     TransportSecuritySettings,
@@ -79,14 +80,22 @@ class SseServerTransport:
     _session_owners: dict[UUID, AuthorizationContext]
     _security: TransportSecurityMiddleware
 
-    def __init__(self, endpoint: str, security_settings: TransportSecuritySettings | None = None) -> None:
+    def __init__(
+        self,
+        endpoint: str,
+        security_settings: TransportSecuritySettings | None = None,
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
+    ) -> 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_request_body_size: Maximum size in bytes for POSTed message bodies. Requests that
+                declare or stream a larger body receive HTTP 413. Defaults to 4 MiB, matching
+                `StreamableHTTPSessionManager`.
 
         Note:
             We use relative paths instead of full URLs for several reasons:
@@ -103,6 +112,9 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
 
         super().__init__()
 
+        if max_request_body_size <= 0:
+            raise ValueError("max_request_body_size must be a positive number of bytes")
+
         # Validate that endpoint is a relative path and not a full URL
         if "://" in endpoint or endpoint.startswith("//") or "?" in endpoint or "#" in endpoint:
             raise ValueError(
@@ -118,6 +130,7 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
         self._read_stream_writers = {}
         self._session_owners = {}
         self._security = TransportSecurityMiddleware(security_settings)
+        self._post_message_app = RequestBodyLimitMiddleware(self._handle_post_message, max_request_body_size)
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
     @asynccontextmanager
@@ -203,6 +216,17 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
             self._session_owners.pop(session_id, None)
 
     async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """ASGI application for the message endpoint.
+
+        Only POST is accepted (other methods get 405), and bodies larger than
+        `max_request_body_size` are answered with 413 before the message is handled.
+        """
+        if scope["method"] != "POST":
+            response = Response(status_code=405, headers={"Allow": "POST"})
+            return await response(scope, receive, send)
+        await self._post_message_app(scope, receive, send)
+
+    async def _handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:
         logger.debug("Handling POST message")
         request = Request(scope, receive)
 

src/mcp/server/streamable_http_manager.py

@@ -35,7 +35,7 @@
 logger = logging.getLogger(__name__)
 
 DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024
-"""Default maximum Streamable HTTP request body size in bytes (4 MiB)."""
+"""Default maximum HTTP request body size in bytes (4 MiB)."""
 
 
 class StreamableHTTPSessionManager:

tests/server/auth/test_error_handling.py

@@ -16,6 +16,7 @@
 from mcp.server.auth.provider import AuthorizeError, RegistrationError, TokenError
 from mcp.server.auth.routes import create_auth_routes
 from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
+from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE
 from tests.server.mcpserver.auth.test_auth_integration import MockOAuthProvider
 
 
@@ -288,3 +289,37 @@ async def test_token_error_handling_refresh_token(
         data = refresh_response.json()
         assert data["error"] == "invalid_scope"
         assert data["error_description"] == "The requested scope is invalid"
+
+
+_FORM = "application/x-www-form-urlencoded"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("path", "content_type"),
+    [("/token", _FORM), ("/revoke", _FORM), ("/register", "application/json"), ("/authorize", _FORM)],
+)
+async def test_oversized_request_body_returns_413(client: httpx2.AsyncClient, path: str, content_type: str):
+    """Each endpoint that reads a request body rejects one over 4 MiB before parsing it."""
+    response = await client.post(
+        path, content=b"x" * (DEFAULT_MAX_REQUEST_BODY_SIZE + 1), headers={"Content-Type": content_type}
+    )
+    assert response.status_code == 413
+
+
+@pytest.mark.anyio
+async def test_request_body_within_the_limit_is_still_parsed(client: httpx2.AsyncClient):
+    """A small body is passed through to the handler intact: the form is parsed and its fields validated."""
+    response = await client.post("/token", data={"grant_type": "authorization_code"})
+    assert response.status_code == 401
+    assert response.json() == {"error": "invalid_client", "error_description": "Missing client_id"}
+
+
+@pytest.mark.anyio
+async def test_options_preflight_is_not_body_limited(client: httpx2.AsyncClient):
+    """CORS preflight requests still get their CORS answer; only POST bodies are limited."""
+    response = await client.options(
+        "/token", headers={"Origin": "https://client.example.com", "Access-Control-Request-Method": "POST"}
+    )
+    assert response.status_code == 200
+    assert response.headers["access-control-allow-origin"] == "*"

tests/server/mcpserver/test_server.py

@@ -5,6 +5,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import anyio
+import httpx2
 import pytest
 from inline_snapshot import snapshot
 from mcp_types import (
@@ -1785,6 +1786,19 @@ def test_streamable_http_no_redirect() -> None:
     assert streamable_routes[0].path == "/mcp", "Streamable route path should be /mcp"
 
 
+async def test_sse_app_applies_the_configured_request_body_limit() -> None:
+    """`sse_app(max_request_body_size=...)` rejects larger POSTs to the message endpoint with HTTP 413."""
+    app = MCPServer("test").sse_app(max_request_body_size=8, host="0.0.0.0")
+    transport = httpx2.ASGITransport(app=app)
+    async with httpx2.AsyncClient(transport=transport, base_url="http://localhost") as http:
+        response = await http.post(
+            "/messages/?session_id=12345678123456781234567812345678",
+            content=b"123456789",
+            headers={"Content-Type": "application/json"},
+        )
+    assert response.status_code == 413
+
+
 async def test_report_progress_delegates_to_session_report_progress():
     """Context.report_progress delegates to ServerSession.report_progress unconditionally.
 

tests/server/test_sse_security.py

@@ -18,6 +18,7 @@
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
 from mcp.server.sse import SseServerTransport
+from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._stream_protocols import WriteStream
 from mcp.shared.message import SessionMessage
@@ -204,9 +205,18 @@ def _authenticated_user(client_id: str, subject: str | None = None, issuer: str
 
 
 def _sse_scope(
-    method: str, path: str, user: AuthenticatedUser | None, *, query_string: bytes = b"", body: bytes = b""
+    method: str,
+    path: str,
+    user: AuthenticatedUser | None,
+    *,
+    query_string: bytes = b"",
+    body: bytes | list[bytes] = b"",
 ) -> tuple[Scope, Receive, Send, list[Message]]:
-    """Build an ASGI scope/receive/send triple for a request to the SSE transport."""
+    """Build an ASGI scope/receive/send triple for a request to the SSE transport.
+
+    `body` may be a list of chunks to deliver the request body over several `http.request` messages;
+    no Content-Length header is set either way.
+    """
     scope: Scope = {
         "type": "http",
         "method": method,
@@ -218,9 +228,11 @@ def _sse_scope(
     if user is not None:
         scope["user"] = user
     sent: list[Message] = []
+    chunks = list(body) if isinstance(body, list) else [body]
 
     async def receive() -> Message:
-        return {"type": "http.request", "body": body, "more_body": False}
+        chunk = chunks.pop(0)
+        return {"type": "http.request", "body": chunk, "more_body": bool(chunks)}
 
     async def send(message: Message) -> None:
         sent.append(message)
@@ -233,6 +245,10 @@ def _response_status(sent: list[Message]) -> int:
     return response_start["status"]
 
 
+def _response_body(sent: list[Message]) -> bytes:
+    return b"".join(msg.get("body", b"") for msg in sent if msg["type"] == "http.response.body")
+
+
 async def _post_message(transport: SseServerTransport, session_id: str, user: AuthenticatedUser | None) -> int:
     """POST a message to an SSE session as `user` and return the response status."""
     body = b'{"jsonrpc": "2.0", "id": 1, "method": "ping", "params": null}'
@@ -368,6 +384,80 @@ async def test_sse_post_with_a_disallowed_host_is_rejected_before_session_lookup
     assert _response_status(sent) == 421
 
 
+# A well-formed session ID that no live session owns.
+_UNKNOWN_SESSION = b"session_id=12345678123456781234567812345678"
+
+
+@pytest.mark.anyio
+async def test_sse_post_body_over_the_limit_returns_413():
+    """A POST body larger than max_request_body_size is answered with 413 before any session handling."""
+    transport = SseServerTransport("/messages/", max_request_body_size=8)
+    scope, receive, send, sent = _sse_scope(
+        "POST", "/messages/", None, query_string=_UNKNOWN_SESSION, body=b"123456789"
+    )
+
+    await transport.handle_post_message(scope, receive, send)
+    assert _response_status(sent) == 413
+    assert _response_body(sent) == b"Request body too large"
+
+
+@pytest.mark.anyio
+async def test_sse_post_body_limit_defaults_to_four_mib():
+    """Without an explicit limit, a body one byte over 4 MiB (and no Content-Length) is answered with 413."""
+    transport = SseServerTransport("/messages/")
+    body = b"x" * (DEFAULT_MAX_REQUEST_BODY_SIZE + 1)
+    scope, receive, send, sent = _sse_scope("POST", "/messages/", None, query_string=_UNKNOWN_SESSION, body=body)
+
+    await transport.handle_post_message(scope, receive, send)
+    assert _response_status(sent) == 413
+
+
+@pytest.mark.anyio
+async def test_sse_post_streamed_body_over_the_limit_returns_413():
+    """The limit counts bytes across body chunks, not just a declared Content-Length."""
+    transport = SseServerTransport("/messages/", max_request_body_size=8)
+    scope, receive, send, sent = _sse_scope(
+        "POST", "/messages/", None, query_string=_UNKNOWN_SESSION, body=[b"1234", b"56789"]
+    )
+
+    await transport.handle_post_message(scope, receive, send)
+    assert _response_status(sent) == 413
+
+
+@pytest.mark.anyio
+async def test_sse_post_within_the_limit_reaches_session_lookup():
+    """A body within the limit is passed on intact: an unknown session still gets its 404."""
+    transport = SseServerTransport("/messages/", max_request_body_size=64)
+    scope, receive, send, sent = _sse_scope(
+        "POST", "/messages/", None, query_string=_UNKNOWN_SESSION, body=[b'{"jsonrpc": ', b'"2.0"}']
+    )
+
+    await transport.handle_post_message(scope, receive, send)
+    assert _response_status(sent) == 404
+    assert _response_body(sent) == b"Could not find session"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("method", ["GET", "PUT"])
+async def test_sse_message_endpoint_answers_405_to_non_post(method: str):
+    """The message endpoint only accepts POST; other methods get 405 with an Allow header."""
+    transport = SseServerTransport("/messages/")
+    scope, receive, send, sent = _sse_scope(method, "/messages/", None, query_string=_UNKNOWN_SESSION, body=b"{}")
+
+    await transport.handle_post_message(scope, receive, send)
+    assert _response_status(sent) == 405
+    response_start = next(msg for msg in sent if msg["type"] == "http.response.start")
+    assert (b"allow", b"POST") in response_start["headers"]
+
+
+@pytest.mark.parametrize("max_request_body_size", [0, -1])
+def test_sse_transport_rejects_a_non_positive_body_limit(max_request_body_size: int):
+    """The body limit must be a positive number of bytes, matching StreamableHTTPSessionManager."""
+    with pytest.raises(ValueError) as exc_info:
+        SseServerTransport("/messages/", max_request_body_size=max_request_body_size)
+    assert str(exc_info.value) == "max_request_body_size must be a positive number of bytes"
+
+
 @pytest.mark.anyio
 async def test_sse_round_trip_delivers_posted_messages_and_streams_responses():
     """A POSTed JSON-RPC message reaches the server's read stream, and a message