fix: shield cleanup operations from cancel scope conflicts

Apply CancelScope(shield=True) to prevent RuntimeError when cleaning up
HTTP MCP clients. This fixes the 'Attempted to exit cancel scope in a
different task' error that occurs with asyncio.

Changes:
- session.py: Add _closing flag, shield cleanup in _receive_loop
- streamable_http.py: Shield terminate_session and stream cleanup

Related to: modelcontextprotocol#577
Based on: modelcontextprotocol#1817

Author: Egor Egerev

src/mcp/client/streamable_http.py

@@ -563,9 +563,13 @@ def start_get_stream() -> None:
                 try:
                     yield (read_stream, write_stream, transport.get_session_id)
                 finally:
-                    if transport.session_id and terminate_on_close:
-                        await transport.terminate_session(client)
+                    # Shield cleanup from cancellation to prevent cancel scope conflicts
+                    with anyio.CancelScope(shield=True):
+                        if transport.session_id and terminate_on_close:
+                            await transport.terminate_session(client)
                     tg.cancel_scope.cancel()
         finally:
-            await read_stream_writer.aclose()
-            await write_stream.aclose()
+            # Shield stream cleanup from cancellation
+            with anyio.CancelScope(shield=True):
+                await read_stream_writer.aclose()
+                await write_stream.aclose()

src/mcp/shared/session.py

@@ -175,6 +175,7 @@ class BaseSession(
     _in_flight: dict[RequestId, RequestResponder[ReceiveRequestT, SendResultT]]
     _progress_callbacks: dict[RequestId, ProgressFnT]
     _response_routers: list[ResponseRouter]
+    _closing: bool = False
 
     def __init__(
         self,
@@ -242,6 +243,9 @@ async def send_request(
 
         Do not use this method to emit notifications! Use send_notification() instead.
         """
+        if self._closing:
+            raise McpError(ErrorData(code=CONNECTION_CLOSED, message="Connection closed"))
+
         request_id = self._request_id
         self._request_id = request_id + 1
 
@@ -287,7 +291,8 @@ async def send_request(
                 return result_type.model_validate(response_or_error.result, by_name=False)
 
         finally:
-            self._response_streams.pop(request_id, None)
+            if not self._closing:
+                self._response_streams.pop(request_id, None)
             self._progress_callbacks.pop(request_id, None)
             await response_stream.aclose()
             await response_stream_reader.aclose()
@@ -421,15 +426,18 @@ async def _receive_loop(self) -> None:
             finally:
                 # after the read stream is closed, we need to send errors
                 # to any pending requests
-                for id, stream in self._response_streams.items():
-                    error = ErrorData(code=CONNECTION_CLOSED, message="Connection closed")
-                    try:
-                        await stream.send(JSONRPCError(jsonrpc="2.0", id=id, error=error))
-                        await stream.aclose()
-                    except Exception:  # pragma: no cover
-                        # Stream might already be closed
-                        pass
-                self._response_streams.clear()
+                # Shield cleanup from cancellation to prevent cancel scope conflicts
+                self._closing = True
+                with anyio.CancelScope(shield=True):
+                    for id, stream in self._response_streams.items():
+                        error = ErrorData(code=CONNECTION_CLOSED, message="Connection closed")
+                        try:
+                            await stream.send(JSONRPCError(jsonrpc="2.0", id=id, error=error))
+                            await stream.aclose()
+                        except Exception:  # pragma: no cover
+                            # Stream might already be closed
+                            pass
+                    self._response_streams.clear()
 
     def _normalize_request_id(self, response_id: RequestId) -> RequestId:
         """Normalize a response ID to match how request IDs are stored.
@@ -481,7 +489,7 @@ async def _handle_response(self, message: SessionMessage) -> None:
                     return  # Handled
 
         # Fall back to normal response streams
-        stream = self._response_streams.pop(response_id, None)
+        stream = self._response_streams.pop(response_id, None) if not self._closing else None
         if stream:
             await stream.send(message.message)
         else: