fix(streamable-http): tolerate closed read streams during SSE cleanup

Author: Jianke LIN

src/mcp/client/streamable_http.py

@@ -44,6 +44,16 @@
 StreamWriter = ContextSendStream[SessionMessageOrError]
 StreamReader = ContextReceiveStream[SessionMessage]
 
+
+async def _send_or_ignore_closed(read_stream_writer: StreamWriter, message: SessionMessageOrError) -> bool:
+    try:
+        await read_stream_writer.send(message)
+    except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+        logger.debug("Read stream closed before Streamable HTTP message could be delivered", exc_info=True)
+        return False
+    return True
+
+
 MCP_SESSION_ID = "mcp-session-id"
 LAST_EVENT_ID = "last-event-id"
 
@@ -157,17 +167,17 @@ async def _handle_sse_event(
                 # Otherwise, return False to continue listening
                 return isinstance(message, JSONRPCResponse | JSONRPCError)
 
-            # Forwarding to a closed read stream lands here when the caller cancels mid-SSE
-            # (BrokenResourceError, not a parse failure); coverage is timing-dependent in the
-            # streaming story's modern HTTP cancellation leg.
+            except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+                logger.debug("Read stream closed while forwarding SSE message", exc_info=True)
+                return True
             except Exception as exc:  # pragma: lax no cover
                 logger.exception("Error parsing SSE message")
                 if original_request_id is not None:
                     error_data = ErrorData(code=PARSE_ERROR, message=f"Failed to parse SSE message: {exc}")
                     error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
-                    await read_stream_writer.send(error_msg)
+                    await _send_or_ignore_closed(read_stream_writer, error_msg)
                     return True
-                await read_stream_writer.send(exc)
+                await _send_or_ignore_closed(read_stream_writer, exc)
                 return False
         else:  # pragma: no cover
             logger.warning(f"Unknown SSE event: {sse.event}")
@@ -383,7 +393,7 @@ async def _handle_sse_response(
         if last_event_id is None:
             error_data = ErrorData(code=CONNECTION_CLOSED, message="SSE stream disconnected before response completed")
             error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
-            await ctx.read_stream_writer.send(error_msg)
+            await _send_or_ignore_closed(ctx.read_stream_writer, error_msg)
             return
 
         logger.info("SSE stream disconnected, reconnecting...")
@@ -407,7 +417,7 @@ async def _handle_reconnection(
                 data={"last_event_id": last_event_id},
             )
             error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
-            await ctx.read_stream_writer.send(error_msg)
+            await _send_or_ignore_closed(ctx.read_stream_writer, error_msg)
             logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
             return
 

tests/client/test_streamable_http.py

@@ -13,7 +13,14 @@
 import httpx
 import pytest
 from inline_snapshot import snapshot
-from mcp_types import CONNECTION_CLOSED, METHOD_NOT_FOUND, JSONRPCError, JSONRPCNotification, JSONRPCRequest, JSONRPCResponse
+from mcp_types import (
+    CONNECTION_CLOSED,
+    METHOD_NOT_FOUND,
+    JSONRPCError,
+    JSONRPCNotification,
+    JSONRPCRequest,
+    JSONRPCResponse,
+)
 
 from mcp.client.streamable_http import RequestContext, StreamableHTTPTransport, streamable_http_client
 from mcp.shared._context_streams import create_context_streams