From ee0b8689ca12bb82da92a0c69f5c683c14de55d0 Mon Sep 17 00:00:00 2001 From: codesphereug Date: Sat, 7 Nov 2026 15:50:27 +0300 Subject: [PATCH] fix(sse): add configurable max_sse_event_size to sse_client and SseServerParameters (fixes #3803) --- src/mcp/client/session_group.py | 4 +++ src/mcp/client/sse.py | 8 ++++- tests/client/test_session_group.py | 5 +-- tests/shared/test_sse.py | 51 ++++++++++++++++++++++++++++++ 4 files changed, 65 insertions(+), 3 deletions(-) diff --git a/src/mcp/client/session_group.py b/src/mcp/client/session_group.py index 13d8928ed4..23b5813467 100644 --- a/src/mcp/client/session_group.py +++ b/src/mcp/client/session_group.py @@ -43,6 +43,9 @@ class SseServerParameters(BaseModel): # Timeout for SSE read operations (in seconds). sse_read_timeout: float = 300.0 + # Maximum bytes in one server-sent event. None disables the limit. + max_sse_event_size: int | None = Field(default=DEFAULT_MAX_SSE_EVENT_SIZE, gt=0) + class StreamableHttpParameters(BaseModel): """Parameters for initializing a streamable_http_client.""" @@ -321,6 +324,7 @@ async def _establish_session( headers=server_params.headers, timeout=server_params.timeout, sse_read_timeout=server_params.sse_read_timeout, + max_sse_event_size=server_params.max_sse_event_size, ) read, write = await session_stack.enter_async_context(client) else: diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py index f72ff273a9..657f81ce2a 100644 --- a/src/mcp/client/sse.py +++ b/src/mcp/client/sse.py @@ -22,6 +22,8 @@ logger = logging.getLogger(__name__) +DEFAULT_MAX_SSE_EVENT_SIZE = 1024 * 1024 + def remove_request_params(url: str) -> str: return urljoin(url, urlparse(url).path) @@ -41,6 +43,7 @@ async def sse_client( httpx_client_factory: McpHttpClientFactory = create_mcp_http_client, auth: httpx2.Auth | None = None, on_session_created: Callable[[str], None] | None = None, + max_sse_event_size: int | None = DEFAULT_MAX_SSE_EVENT_SIZE, ): """Client transport for SSE. @@ -62,12 +65,15 @@ async def sse_client( they make. auth: Optional httpx2 authentication handler. on_session_created: Optional callback invoked with the session ID when received. + max_sse_event_size: Maximum bytes buffered for one SSE event. None disables the limit. """ + if max_sse_event_size is not None and max_sse_event_size <= 0: + raise ValueError("max_sse_event_size must be positive or None") logger.debug(f"Connecting to SSE endpoint: {remove_request_params(url)}") async with httpx_client_factory( headers=headers, auth=auth, timeout=httpx2.Timeout(timeout, read=sse_read_timeout) ) as client: - async with sse_within_origin(client, url) as event_source: + async with sse_within_origin(client, url, max_event_size=max_sse_event_size) as event_source: event_source.response.raise_for_status() logger.debug("SSE connection established") diff --git a/tests/client/test_session_group.py b/tests/client/test_session_group.py index 922e1eebcb..d890dccf8f 100644 --- a/tests/client/test_session_group.py +++ b/tests/client/test_session_group.py @@ -306,10 +306,10 @@ async def test_client_session_group_disconnect_non_existent_server(): "mcp.client.session_group.mcp.stdio_client", ), ( - SseServerParameters(url="http://test.com/sse", timeout=10.0), + SseServerParameters(url="http://test.com/sse", timeout=10.0, max_sse_event_size=32 * 1024 * 1024), "sse", "mcp.client.session_group.sse_client", - ), # url, headers, timeout, sse_read_timeout + ), # url, headers, timeout, sse_read_timeout, max_sse_event_size ( StreamableHttpParameters( url="http://test.com/stream", terminate_on_close=False, max_sse_event_size=32 * 1024 * 1024 @@ -374,6 +374,7 @@ async def test_client_session_group_establish_session_parameterized( headers=server_params_instance.headers, timeout=server_params_instance.timeout, sse_read_timeout=server_params_instance.sse_read_timeout, + max_sse_event_size=server_params_instance.max_sse_event_size, ) elif client_type_name == "streamablehttp": # pragma: no branch assert isinstance(server_params_instance, StreamableHttpParameters) diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index 77d1b28a0a..b24a3fa66e 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -523,3 +523,54 @@ async def test_sse_session_cleanup_on_disconnect() -> None: headers={"Content-Type": "application/json"}, ) assert response.status_code == 404 + + +@pytest.mark.anyio +async def test_sse_client_none_disables_the_sse_event_limit() -> None: + """Regression test for https://github.com/modelcontextprotocol/python-sdk/issues/3803 + + Passing max_sse_event_size=None to sse_client disables the 1 MiB default SSE + event size limit so tool results over 1 MiB can be received over SSE. + """ + payload = "x" * (2 * 1024 * 1024) + response = JSONRPCResponse( + jsonrpc="2.0", + id=1, + result={"text": payload}, + ) + response_json = response.model_dump_json(by_alias=True, exclude_none=True) + event_stream = ( + "event: endpoint\ndata: /messages/?session_id=abc123\n\n" + f"event: message\ndata: {response_json}\n\n" + ) + + def serve(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, headers={"content-type": "text/event-stream"}, text=event_stream) + + def factory( + headers: dict[str, str] | None = None, + timeout: httpx2.Timeout | None = None, + auth: httpx2.Auth | None = None, + ) -> httpx2.AsyncClient: + return httpx2.AsyncClient(transport=httpx2.MockTransport(serve)) + + with anyio.fail_after(5): + async with sse_client("http://test/sse", httpx_client_factory=factory, max_sse_event_size=None) as ( + read_stream, + _, + ): + msg = await read_stream.receive() + + assert isinstance(msg, SessionMessage) + assert isinstance(msg.message, types.JSONRPCResponse) + assert msg.message.result == {"text": payload} + + +@pytest.mark.anyio +@pytest.mark.parametrize("invalid_size", [0, -1]) +async def test_sse_client_rejects_non_positive_max_sse_event_size(invalid_size: int) -> None: + """sse_client rejects zero or negative max_sse_event_size values.""" + with pytest.raises(ValueError, match="max_sse_event_size must be positive or None"): + async with sse_client("http://test/sse", max_sse_event_size=invalid_size): + pytest.fail("should not connect") # pragma: no cover +