[v1.x] 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 (listed in its
__all__). The middleware's own unit tests move with it, plus the
mid-stream disconnect replay case main already carries; no behaviour
change.

Author: maxisbey

src/mcp/server/auth/routes.py

@@ -18,7 +18,7 @@
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
 from mcp.server.streamable_http import MCP_PROTOCOL_VERSION_HEADER
-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 OAuthMetadata
 
 

src/mcp/server/fastmcp/server.py

@@ -62,8 +62,8 @@
 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.transport_security import TransportSecuritySettings
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import Annotations, AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
 from mcp.types import Prompt as MCPPrompt

src/mcp/server/sse.py

@@ -53,8 +53,9 @@ async def handle_sse(request):
 
 import mcp.types as types
 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,17 +4,15 @@
 
 import contextlib
 import logging
-from collections import deque
 from collections.abc import AsyncIterator
-from typing import Any, Final
+from typing import Any
 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 ASGIApp, Message, Receive, Scope, Send
+from starlette.types import Receive, Scope, Send
 
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
 from mcp.server.lowlevel.server import Server as MCPServer
@@ -23,13 +21,16 @@
     EventStore,
     StreamableHTTPServerTransport,
 )
-from mcp.server.transport_security import TransportSecuritySettings
+from mcp.server.transport_security import (
+    DEFAULT_MAX_REQUEST_BODY_SIZE,
+    RequestBodyLimitMiddleware,
+    TransportSecuritySettings,
+)
 from mcp.types import INVALID_REQUEST, ErrorData, JSONRPCError
 
-logger = logging.getLogger(__name__)
+__all__ = ["DEFAULT_MAX_REQUEST_BODY_SIZE", "RequestBodyLimitMiddleware", "StreamableHTTPSessionManager"]
 
-DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024
-"""Default maximum HTTP request body size in bytes (4 MiB)."""
+logger = logging.getLogger(__name__)
 
 
 class StreamableHTTPSessionManager:
@@ -361,63 +362,3 @@ 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":
-            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)

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 HTTPConnection
 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)."""
+
 
 class TransportSecuritySettings(BaseModel):
     """Settings for MCP transport security features.
@@ -125,3 +132,63 @@ async def validate_request(self, request: HTTPConnection, is_post: bool = False)
             return Response("Invalid Origin header", status_code=403)  # pragma: no cover
 
         return None  # pragma: no cover
+
+
+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

@@ -14,7 +14,7 @@
 
 from mcp.server.auth.provider import AuthorizeError, RegistrationError, TokenError
 from mcp.server.auth.routes import create_auth_routes
-from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE
+from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE
 
 # TODO(Marcelo): This TYPE_CHECKING shouldn't be here, but pytest doesn't seem to get the module correctly.
 if TYPE_CHECKING:

tests/server/test_sse_security.py

@@ -21,8 +21,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.types import Tool
 from tests.test_helpers import wait_for_server
 

tests/server/test_streamable_http_manager.py

@@ -7,7 +7,7 @@
 
 import anyio
 import pytest
-from starlette.types import Message, Receive, Scope, Send
+from starlette.types import Message, Scope
 
 from mcp.server import streamable_http_manager
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
@@ -16,7 +16,6 @@
 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.types import INVALID_REQUEST
@@ -143,92 +142,6 @@ async def send(message: Message) -> None:
     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]
-
-
-@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"))