fix: add exception handling in StreamableHTTP shutdown

Update the shutdown logic to wrap terminate() calls in try-except blocks,
preventing one transport's termination error from affecting others.

Changes:
- Use logger.exception() instead of logger.debug() for better error visibility
- Simplify SSE writer closing by iterating values() directly instead of pop()
- Improve code comments to explain why graceful termination is needed

This follows the same approach as PR #2259 with minor improvements in
error handling.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: weiguangli-io

src/mcp/server/streamable_http.py

@@ -784,12 +784,11 @@ 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()
+        # Close all SSE stream writers so that active EventSourceResponse
+        # coroutines complete gracefully instead of being cancelled mid-stream.
+        for writer in list(self._sse_stream_writers.values()):
+            writer.close()
+        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,9 +130,14 @@ 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
+                # Gracefully terminate all active sessions before cancelling
+                # tasks so that EventSourceResponse coroutines can complete
+                # and Uvicorn does not log ASGI-incomplete-response errors.
                 for transport in list(self._server_instances.values()):
-                    await transport.terminate()
+                    try:
+                        await transport.terminate()
+                    except Exception:
+                        logger.exception("Error terminating transport during shutdown")
                 # Cancel task group to stop all spawned tasks
                 tg.cancel_scope.cancel()
                 self._task_group = None