Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/mcp/client/session_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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:
Expand Down
8 changes: 7 additions & 1 deletion src/mcp/client/sse.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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.

Expand All @@ -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")

Expand Down
5 changes: 3 additions & 2 deletions tests/client/test_session_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down
51 changes: 51 additions & 0 deletions tests/shared/test_sse.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Loading