Fix multiple close_sse_stream support and add reconnection logging (SEP-1699)

- Register SSE writer in _replay_events() so subsequent close_sse_stream() calls work
- Send priming event on each reconnection
- Handle ClosedResourceError gracefully in both POST and GET SSE writers
- Add disconnect/reconnect logging at INFO level for visibility
- Add test for multiple reconnections during long-running tool calls
- Remove pragma from store_event (now covered by tests)

Author: felixweinberger

src/mcp/client/streamable_http.py

@@ -363,6 +363,7 @@ async def _handle_sse_response(
 
         # Stream ended without response - reconnect if we received an event with ID
         if last_event_id is not None:
+            logger.info("SSE stream disconnected, reconnecting...")
             await self._handle_reconnection(ctx, last_event_id, retry_interval_ms)
 
     async def _handle_reconnection(
@@ -399,7 +400,7 @@ async def _handle_reconnection(
                 timeout=httpx.Timeout(self.timeout, read=self.sse_read_timeout),
             ) as event_source:
                 event_source.response.raise_for_status()
-                logger.debug("Reconnection GET SSE connection established")
+                logger.info("Reconnected to SSE stream")
 
                 # Track for potential further reconnection
                 reconnect_last_event_id: str | None = last_event_id
@@ -423,6 +424,7 @@ async def _handle_reconnection(
 
                 # Stream ended again without response - reconnect again (reset attempt counter)
                 if reconnect_last_event_id is not None:
+                    logger.info("SSE stream disconnected, reconnecting...")
                     await self._handle_reconnection(
                         ctx, reconnect_last_event_id, reconnect_retry_ms, 0
                     )

src/mcp/server/streamable_http.py

@@ -208,6 +208,12 @@ def close_sse_stream(self, request_id: RequestId) -> None:
         if writer:
             writer.close()
 
+        # Also close and remove request streams
+        if request_id in self._request_streams:
+            send_stream, receive_stream = self._request_streams.pop(request_id)
+            send_stream.close()
+            receive_stream.close()
+
     def _create_session_message(
         self,
         message: JSONRPCMessage,
@@ -545,6 +551,9 @@ async def sse_writer():
                                     JSONRPCResponse | JSONRPCError,
                                 ):
                                     break
+                    except anyio.ClosedResourceError:
+                        # Expected when close_sse_stream() is called
+                        logger.debug("SSE stream closed by close_sse_stream()")
                     except Exception:
                         logger.exception("Error in SSE writer")
                     finally:
@@ -848,6 +857,13 @@ async def send_event(event_message: EventMessage) -> None:
 
                         # If stream ID not in mapping, create it
                         if stream_id and stream_id not in self._request_streams:
+                            # Register SSE writer so close_sse_stream() can close it
+                            self._sse_stream_writers[stream_id] = sse_stream_writer
+
+                            # Send priming event for this new connection
+                            await self._send_priming_event(stream_id, sse_stream_writer)
+
+                            # Create new request streams for this connection
                             self._request_streams[stream_id] = anyio.create_memory_object_stream[EventMessage](0)
                             msg_reader = self._request_streams[stream_id][1]
 
@@ -857,6 +873,9 @@ async def send_event(event_message: EventMessage) -> None:
                                     event_data = self._create_event_data(event_message)
 
                                     await sse_stream_writer.send(event_data)
+                except anyio.ClosedResourceError:
+                    # Expected when close_sse_stream() is called
+                    logger.debug("Replay SSE stream closed by close_sse_stream()")
                 except Exception:
                     logger.exception("Error in replay sender")
 

tests/shared/test_streamable_http.py

@@ -80,7 +80,7 @@ def __init__(self):
         self._events: list[tuple[StreamId, EventId, types.JSONRPCMessage | None]] = []
         self._event_id_counter = 0
 
-    async def store_event(self, stream_id: StreamId, message: types.JSONRPCMessage | None) -> EventId:  # pragma: no cover
+    async def store_event(self, stream_id: StreamId, message: types.JSONRPCMessage | None) -> EventId:
         """Store an event and return its ID."""
         self._event_id_counter += 1
         event_id = str(self._event_id_counter)
@@ -175,6 +175,17 @@ async def handle_list_tools() -> list[Tool]:
                     description="Tool that sends notification1, closes stream, sends notification2, notification3",
                     inputSchema={"type": "object", "properties": {}},
                 ),
+                Tool(
+                    name="tool_with_multiple_stream_closes",
+                    description="Tool that closes SSE stream multiple times during execution",
+                    inputSchema={
+                        "type": "object",
+                        "properties": {
+                            "checkpoints": {"type": "integer", "default": 3},
+                            "sleep_time": {"type": "number", "default": 0.2},
+                        },
+                    },
+                ),
             ]
 
         @self.call_tool()
@@ -314,6 +325,25 @@ async def handle_call_tool(name: str, args: dict[str, Any]) -> list[TextContent]
                 )
                 return [TextContent(type="text", text="All notifications sent")]
 
+            elif name == "tool_with_multiple_stream_closes":
+                num_checkpoints = args.get("checkpoints", 3)
+                sleep_time = args.get("sleep_time", 0.2)
+
+                for i in range(num_checkpoints):
+                    await ctx.session.send_log_message(
+                        level="info",
+                        data=f"checkpoint_{i}",
+                        logger="multi_close_tool",
+                        related_request_id=ctx.request_id,
+                    )
+
+                    if ctx.close_sse_stream:
+                        await ctx.close_sse_stream()
+
+                    await anyio.sleep(sleep_time)
+
+                return [TextContent(type="text", text=f"Completed {num_checkpoints} checkpoints")]
+
             return [TextContent(type="text", text=f"Called {name}")]
 
 
@@ -950,7 +980,7 @@ async def test_streamablehttp_client_tool_invocation(initialized_client_session:
     """Test client tool invocation."""
     # First list tools
     tools = await initialized_client_session.list_tools()
-    assert len(tools.tools) == 8
+    assert len(tools.tools) == 9
     assert tools.tools[0].name == "test_tool"
 
     # Call the tool
@@ -987,7 +1017,7 @@ async def test_streamablehttp_client_session_persistence(basic_server: None, bas
 
             # Make multiple requests to verify session persistence
             tools = await session.list_tools()
-            assert len(tools.tools) == 8
+            assert len(tools.tools) == 9
 
             # Read a resource
             resource = await session.read_resource(uri=AnyUrl("foobar://test-persist"))
@@ -1016,7 +1046,7 @@ async def test_streamablehttp_client_json_response(json_response_server: None, j
 
             # Check tool listing
             tools = await session.list_tools()
-            assert len(tools.tools) == 8
+            assert len(tools.tools) == 9
 
             # Call a tool and verify JSON response handling
             result = await session.call_tool("test_tool", {})
@@ -1086,7 +1116,7 @@ async def test_streamablehttp_client_session_termination(basic_server: None, bas
 
             # Make a request to confirm session is working
             tools = await session.list_tools()
-            assert len(tools.tools) == 8
+            assert len(tools.tools) == 9
 
     headers: dict[str, str] = {}  # pragma: no cover
     if captured_session_id:  # pragma: no cover
@@ -1152,7 +1182,7 @@ async def mock_delete(self: httpx.AsyncClient, *args: Any, **kwargs: Any) -> htt
 
             # Make a request to confirm session is working
             tools = await session.list_tools()
-            assert len(tools.tools) == 8
+            assert len(tools.tools) == 9
 
     headers: dict[str, str] = {}  # pragma: no cover
     if captured_session_id:  # pragma: no cover
@@ -1944,3 +1974,56 @@ async def message_handler(
             assert result.content[0].type == "text"
             assert isinstance(result.content[0], TextContent)
             assert result.content[0].text == "All notifications sent"
+
+
+@pytest.mark.anyio
+async def test_streamablehttp_multiple_reconnections(
+    event_server: tuple[SimpleEventStore, str],
+):
+    """Verify multiple close_sse_stream() calls each trigger a client reconnect.
+
+    Server uses retry_interval=500ms, tool sleeps 600ms after each close to ensure
+    client has time to reconnect before the next checkpoint.
+
+    With 3 checkpoints, we expect 8 resumption tokens:
+    - 1 priming (initial POST connection)
+    - 3 notifications (checkpoint_0, checkpoint_1, checkpoint_2)
+    - 3 priming (one per reconnect after each close)
+    - 1 response
+    """
+    _, server_url = event_server
+    resumption_tokens: list[str] = []
+
+    async def on_resumption_token(token: str) -> None:
+        resumption_tokens.append(token)
+
+    async with streamablehttp_client(f"{server_url}/mcp") as (read_stream, write_stream, _):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+
+            # Use send_request with metadata to track resumption tokens
+            metadata = ClientMessageMetadata(on_resumption_token_update=on_resumption_token)
+            result = await session.send_request(
+                types.ClientRequest(
+                    types.CallToolRequest(
+                        method="tools/call",
+                        params=types.CallToolRequestParams(
+                            name="tool_with_multiple_stream_closes",
+                            # retry_interval=500ms, so sleep 600ms to ensure reconnect completes
+                            arguments={"checkpoints": 3, "sleep_time": 0.6},
+                        ),
+                    )
+                ),
+                types.CallToolResult,
+                metadata=metadata,
+            )
+
+            assert result.content[0].type == "text"
+            assert isinstance(result.content[0], TextContent)
+            assert "Completed 3 checkpoints" in result.content[0].text
+
+    # 4 priming + 3 notifications + 1 response = 8 tokens
+    assert len(resumption_tokens) == 8, (
+        f"Expected 8 resumption tokens (4 priming + 3 notifs + 1 response), "
+        f"got {len(resumption_tokens)}: {resumption_tokens}"
+    )