Fix close_sse_stream to close reader and improve demo output (SEP-1699)

- Close request stream receiver in close_sse_stream() so sse_writer loop
  exits cleanly instead of hitting ClosedResourceError
- Log reconnections at INFO level so they're visible without DEBUG
- Remove misleading "Watch DEBUG logs" text from demo client
- Fix type annotation (str instead of str | None)

Author: felixweinberger

examples/clients/sse-polling-client/mcp_sse_polling_client/main.py

@@ -32,7 +32,6 @@ async def run_demo(url: str, items: int, checkpoint_every: int) -> None:
     print(f"{'=' * 60}")
     print(f"Server URL: {url}")
     print(f"Processing {items} items with checkpoints every {checkpoint_every}")
-    print("Watch DEBUG logs for reconnection messages")
     print(f"{'=' * 60}\n")
 
     async with streamablehttp_client(url) as (read_stream, write_stream, _):
@@ -47,8 +46,7 @@ async def run_demo(url: str, items: int, checkpoint_every: int) -> None:
             print(f"Available tools: {[t.name for t in tools.tools]}\n")
 
             # Call the process_batch tool
-            print(f"Calling process_batch(items={items}, checkpoint_every={checkpoint_every})...")
-            print("Watch for reconnections in DEBUG logs\n")
+            print(f"Calling process_batch(items={items}, checkpoint_every={checkpoint_every})...\n")
             print("-" * 40)
 
             result = await session.call_tool(

src/mcp/client/streamable_http.py

@@ -399,10 +399,10 @@ 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
+                reconnect_last_event_id: str = last_event_id
                 reconnect_retry_ms = retry_interval_ms
 
                 async for sse in event_source.aiter_sse():
@@ -422,10 +422,7 @@ async def _handle_reconnection(
                         return
 
                 # Stream ended again without response - reconnect again (reset attempt counter)
-                if reconnect_last_event_id is not None:
-                    await self._handle_reconnection(
-                        ctx, reconnect_last_event_id, reconnect_retry_ms, 0
-                    )
+                await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0)
         except Exception as e:
             logger.debug(f"Reconnection failed: {e}")
             # Try to reconnect again if we still have an event ID

src/mcp/server/streamable_http.py

@@ -208,6 +208,10 @@ def close_sse_stream(self, request_id: RequestId) -> None:
         if writer:
             writer.close()
 
+        # Close request stream receiver so sse_writer loop exits cleanly
+        if request_id in self._request_streams:
+            self._request_streams[request_id][1].close()
+
     def _create_session_message(
         self,
         message: JSONRPCMessage,