fix: terminate active StreamableHTTP sessions during shutdown (#2150)

This commit addresses issue #2150 where StreamableHTTP sessions were not
properly terminating active HTTP sessions during shutdown.

Root causes fixed:
1. StreamableHTTPSessionManager.run() was directly canceling the task group
   without first calling terminate() on active transport instances
2. StreamableHTTPServerTransport.terminate() was only closing _request_streams,
   leaving _sse_stream_writers unclosed

Changes made:
1. In streamable_http_manager.py: Added explicit terminate() calls for all
   active server instances before canceling the task group in the finally
   block of run()
2. In streamable_http.py: Enhanced terminate() to close all SSE stream writers
   by iterating through _sse_stream_writers and calling close() on each writer
   before clearing the dictionary

This ensures all active HTTP connections and streams are properly closed during
shutdown, preventing resource leaks and ensuring clean termination.

Github-Issue: #2150

Author: weiguangli-io

src/mcp/server/streamable_http.py

@@ -784,6 +784,16 @@ async def terminate(self) -> None:
         self._terminated = True
         logger.info(f"Terminating session: {self.mcp_session_id}")
 
+        # Close all SSE stream writers
+        sse_stream_writer_keys = list(self._sse_stream_writers.keys())
+        for key in sse_stream_writer_keys:  # pragma: no cover
+            writer = self._sse_stream_writers.pop(key, None)
+            if writer:
+                writer.close()
+
+        # Clear the SSE stream writers dictionary
+        self._sse_stream_writers.clear()
+
         # We need a copy of the keys to avoid modification during iteration
         request_stream_keys = list(self._request_streams.keys())
 

src/mcp/server/streamable_http_manager.py

@@ -130,6 +130,9 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]:
                 yield  # Let the application run
             finally:
                 logger.info("StreamableHTTP session manager shutting down")
+                # Terminate all active server instances before cancelling tasks
+                for transport in list(self._server_instances.values()):
+                    await transport.terminate()
                 # Cancel task group to stop all spawned tasks
                 tg.cancel_scope.cancel()
                 self._task_group = None