Move RequestBodyLimitMiddleware out of the Streamable HTTP manager module

The middleware and DEFAULT_MAX_REQUEST_BODY_SIZE are now used by the SSE
transport and the OAuth routes as well, so they move next to the other
shared HTTP request checks in mcp.server.transport_security. Both names
remain importable from mcp.server.streamable_http_manager. The middleware's
own unit tests move with it; no behaviour change.

Author: maxisbey

src/mcp/server/auth/routes.py

@@ -17,7 +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.server.transport_security 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
 

src/mcp/server/lowlevel/server.py

@@ -65,12 +65,8 @@ 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 (
-    DEFAULT_MAX_REQUEST_BODY_SIZE,
-    StreamableHTTPASGIApp,
-    StreamableHTTPSessionManager,
-)
-from mcp.server.transport_security import TransportSecuritySettings
+from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
+from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPDeprecationWarning
 from mcp.shared.message import SessionMessage

src/mcp/server/mcpserver/server.py

@@ -87,9 +87,9 @@
 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 DEFAULT_MAX_REQUEST_BODY_SIZE, StreamableHTTPSessionManager
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus
-from mcp.server.transport_security import TransportSecuritySettings
+from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
 from mcp.shared.uri_template import UriTemplate
 

src/mcp/server/sse.py

@@ -51,8 +51,9 @@ 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 (
+    DEFAULT_MAX_REQUEST_BODY_SIZE,
+    RequestBodyLimitMiddleware,
     TransportSecurityMiddleware,
     TransportSecuritySettings,
 )

src/mcp/server/streamable_http_manager.py

@@ -4,25 +4,25 @@
 
 import contextlib
 import logging
-from collections import deque
 from collections.abc import AsyncIterator
-from typing import TYPE_CHECKING, Any, Final
+from typing import TYPE_CHECKING, Any
 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 ASGIApp, Message, Receive, Scope, Send
+from starlette.types import 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.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE as DEFAULT_MAX_REQUEST_BODY_SIZE
+from mcp.server.transport_security import RequestBodyLimitMiddleware as RequestBodyLimitMiddleware
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._compat import resync_tracer
 from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER
@@ -34,9 +34,6 @@
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024
-"""Default maximum HTTP request body size in bytes (4 MiB)."""
-
 
 class StreamableHTTPSessionManager:
     """Manages StreamableHTTP sessions with optional resumability via event store.
@@ -371,66 +368,6 @@ 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":
-            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."""
 

src/mcp/server/transport_security.py

@@ -1,13 +1,20 @@
-"""DNS rebinding protection for MCP server transports."""
+"""Request checks shared by the HTTP server transports: Host/Origin header validation and body size limits."""
 
 import logging
+from collections import deque
+from typing import Final
 
 from pydantic import BaseModel, Field
+from starlette.datastructures import Headers
 from starlette.requests import Request
 from starlette.responses import Response
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 logger = logging.getLogger(__name__)
 
+DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024
+"""Default maximum HTTP request body size in bytes (4 MiB)."""
+
 
 # TODO(Marcelo): We should flatten these settings. To be fair, I don't think we should even have this middleware.
 class TransportSecuritySettings(BaseModel):
@@ -114,3 +121,63 @@ async def validate_request(self, request: Request, is_post: bool = False) -> Res
             return Response("Invalid Origin header", status_code=403)
 
         return None
+
+
+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":
+            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)

tests/server/auth/test_error_handling.py

@@ -16,7 +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 mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE
 from tests.server.mcpserver.auth.test_auth_integration import MockOAuthProvider
 
 

tests/server/test_sse_security.py

@@ -18,8 +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.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared._stream_protocols import WriteStream
 from mcp.shared.message import SessionMessage
 from tests.interaction.transports import StreamingASGITransport

tests/server/test_streamable_http_manager.py

@@ -10,19 +10,15 @@
 import httpx2
 import pytest
 from mcp_types import INVALID_REQUEST, ListToolsResult, PaginatedRequestParams
-from starlette.types import Message, Receive, Scope, Send
+from starlette.types import Message, Scope
 
 from mcp import Client
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server, ServerRequestContext, streamable_http_manager
 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 (
-    DEFAULT_MAX_REQUEST_BODY_SIZE,
-    RequestBodyLimitMiddleware,
-    StreamableHTTPSessionManager,
-)
+from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, StreamableHTTPSessionManager
 
 
 @pytest.mark.anyio
@@ -146,114 +142,6 @@ async def send(message: Message) -> None:
     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}]
-
-
-@pytest.mark.anyio
-@pytest.mark.parametrize("method", ["GET", "PUT", "OPTIONS", "HEAD", "DELETE"])
-async def test_request_body_limit_applies_to_every_method(method: str) -> None:
-    """SDK-defined: the limit is a property of the request body, not of the method that carries it."""
-    app = AsyncMock()
-    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": method, "path": "/mcp", "headers": []}
-    middleware = RequestBodyLimitMiddleware(app, max_body_size=8)
-
-    await middleware(scope, receive, send)
-
-    assert [message["status"] for message in sent_messages if message["type"] == "http.response.start"] == [413]
-    app.assert_not_awaited()
-
-
-@pytest.mark.anyio
-async def test_request_body_limit_leaves_non_http_scopes_alone() -> None:
-    """SDK-defined: only HTTP requests carry a body to limit; other ASGI scopes go straight to the app."""
-    app = AsyncMock()
-    receive = AsyncMock()
-    send = AsyncMock()
-    scope: Scope = {"type": "lifespan"}
-    middleware = RequestBodyLimitMiddleware(app, max_body_size=8)
-
-    await middleware(scope, receive, send)
-
-    app.assert_awaited_once_with(scope, receive, send)
-    receive.assert_not_awaited()
-
-
 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"))