fix(sse): add configurable max_sse_event_size to sse_client and SseServerParameters (fixes #3803)

Author: codesphere-u

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:

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")
 

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)

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
+