-
Notifications
You must be signed in to change notification settings - Fork 3.6k
Add Streamable HTTP request body limits #3095
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,27 +4,25 @@ | |
|
|
||
| import contextlib | ||
| import logging | ||
| from collections import deque | ||
| from collections.abc import AsyncIterator | ||
| from typing import TYPE_CHECKING, Any | ||
| from typing import TYPE_CHECKING, Any, Final | ||
| 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 Receive, Scope, Send | ||
| from starlette.types import ASGIApp, Message, 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.streamable_http import MCP_SESSION_ID_HEADER, EventStore, StreamableHTTPServerTransport | ||
| from mcp.server.transport_security import TransportSecuritySettings | ||
| from mcp.shared._compat import resync_tracer | ||
| from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER | ||
|
|
@@ -36,6 +34,9 @@ | |
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024 | ||
| """Default maximum Streamable HTTP request body size in bytes (4 MiB).""" | ||
|
|
||
|
|
||
| class StreamableHTTPSessionManager: | ||
| """Manages StreamableHTTP sessions with optional resumability via event store. | ||
|
|
@@ -69,6 +70,8 @@ | |
| retry_interval is also configured, ensure the idle timeout comfortably exceeds the retry interval to | ||
| avoid reaping sessions during normal SSE polling gaps. Default is None (no timeout). A value of 1800 | ||
| (30 minutes) is recommended for most deployments. | ||
| max_request_body_size: Maximum size in bytes for Streamable HTTP POST request bodies. Requests that | ||
| exceed this limit receive a 413 response before parsing or session creation. Defaults to 4 MiB. | ||
| """ | ||
|
|
||
| def __init__( | ||
|
|
@@ -80,11 +83,14 @@ | |
| security_settings: TransportSecuritySettings | None = None, | ||
| retry_interval: int | None = None, | ||
| session_idle_timeout: float | None = None, | ||
| max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE, | ||
| ): | ||
| if session_idle_timeout is not None and session_idle_timeout <= 0: | ||
| raise ValueError("session_idle_timeout must be a positive number of seconds") | ||
| if stateless and session_idle_timeout is not None: | ||
| raise RuntimeError("session_idle_timeout is not supported in stateless mode") | ||
| if max_request_body_size <= 0: | ||
| raise ValueError("max_request_body_size must be a positive number of bytes") | ||
|
|
||
| self.app = app | ||
| self.event_store = event_store | ||
|
|
@@ -93,6 +99,8 @@ | |
| self.security_settings = security_settings | ||
| self.retry_interval = retry_interval | ||
| self.session_idle_timeout = session_idle_timeout | ||
| self.max_request_body_size = max_request_body_size | ||
| self.asgi_app = RequestBodyLimitMiddleware(self._handle_request, max_request_body_size) | ||
|
|
||
| # Session tracking (only used if not stateless) | ||
| self._session_creation_lock = anyio.Lock() | ||
|
|
@@ -159,6 +167,9 @@ | |
|
|
||
| Dispatches to the appropriate handler based on stateless mode. | ||
| """ | ||
| await self.asgi_app(scope, receive, send) | ||
|
|
||
| async def _handle_request(self, scope: Scope, receive: Receive, send: Send) -> None: | ||
| if self._task_group is None: | ||
| raise RuntimeError("Task group is not initialized. Make sure to use run().") | ||
|
|
||
|
|
@@ -360,11 +371,71 @@ | |
| await response(scope, receive, send) | ||
|
|
||
|
|
||
| class RequestBodyLimitMiddleware: | ||
| """Reject oversized HTTP request bodies before invoking an ASGI application.""" | ||
|
Check notice on line 375 in src/mcp/server/streamable_http_manager.py
|
||
|
Comment on lines
+374
to
+375
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟣 Pre-existing issue, not introduced by this PR: the legacy SSE transport's POST endpoint ( Extended reasoning...What the issue isThis PR adds The code pathThe SSE transport is still shipped and mountable — Step-by-step proof
Why this is pre-existing and non-blockingThe PR does not touch How to fix (follow-up)
|
||
|
|
||
| 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" or scope["method"] != "POST": | ||
| 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.""" | ||
|
|
||
| def __init__(self, session_manager: StreamableHTTPSessionManager): | ||
| self.session_manager = session_manager | ||
|
|
||
| async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: | ||
| await self.session_manager.handle_request(scope, receive, send) | ||
| await self.session_manager.asgi_app(scope, receive, send) | ||
Uh oh!
There was an error while loading. Please reload this page.