diff --git a/tests/tools/test_mcp_stability.py b/tests/tools/test_mcp_stability.py
index f204615aca0..b843439fff7 100644
--- a/tests/tools/test_mcp_stability.py
+++ b/tests/tools/test_mcp_stability.py
@@ -736,3 +736,75 @@ class TestMCPInitialConnectionRetry:
                 await task
 
         asyncio.get_event_loop().run_until_complete(_run())
+
+
+# ---------------------------------------------------------------------------
+# Fix: drain pending tasks before closing the MCP loop
+# ---------------------------------------------------------------------------
+
+class TestMCPLoopDrainOnStop:
+    """_stop_mcp_loop reaps pending tasks while the loop is still open."""
+
+    def test_pending_task_cleanup_runs_before_close(self):
+        """A task left on the loop must finish its cleanup before close.
+
+        Regression for #60197: a task still suspended when the loop closes is
+        resumed later by the GC, and its ``finally`` then drives ``cancel()``
+        -> ``call_soon()`` against the closed loop, surfacing as an ignored
+        ``RuntimeError: Event loop is closed``. Servers absent from
+        ``_servers`` (e.g. parked after an initial-connect failure) never get
+        ``shutdown()``, so this drain is their only reaper.
+        """
+        import time
+        import tools.mcp_tool as mcp_mod
+
+        state = {
+            "started": False,
+            "cleanup_ran": False,
+            "cleanup_error": None,
+            "task": None,
+        }
+
+        async def _parked():
+            state["task"] = asyncio.current_task()
+            state["started"] = True
+            try:
+                await asyncio.sleep(3600)
+            finally:
+                # Mirrors _wait_for_reconnect_or_shutdown's finally: cancelling
+                # a helper task needs call_soon(), which a closed loop rejects.
+                try:
+                    helper = asyncio.ensure_future(asyncio.sleep(0))
+                    helper.cancel()
+                    state["cleanup_ran"] = True
+                except BaseException as exc:
+                    state["cleanup_error"] = exc
+
+        with mcp_mod._lock:
+            mcp_mod._servers.clear()
+            mcp_mod._server_connecting.clear()
+        try:
+            mcp_mod._ensure_mcp_loop()
+            with mcp_mod._lock:
+                loop = mcp_mod._mcp_loop
+            assert loop is not None
+            asyncio.run_coroutine_threadsafe(_parked(), loop)
+
+            deadline = time.monotonic() + 5
+            while not state["started"] and time.monotonic() < deadline:
+                time.sleep(0.01)
+            assert state["started"], "task never started on the MCP loop"
+
+            mcp_mod._stop_mcp_loop()
+
+            assert state["task"] is not None
+            assert state["task"].done(), "task left pending when the loop closed"
+            assert state["cleanup_error"] is None, (
+                f"cleanup ran against a closed loop: {state['cleanup_error']!r}"
+            )
+            assert state["cleanup_ran"], "task cleanup never ran"
+        finally:
+            with mcp_mod._lock:
+                mcp_mod._servers.clear()
+                mcp_mod._server_connecting.clear()
+            mcp_mod._stop_mcp_loop()
diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py
index af7fc6b20ec..bc5a8d7f273 100644
--- a/tools/mcp_tool.py
+++ b/tools/mcp_tool.py
@@ -6618,6 +6618,23 @@ def _stop_mcp_loop_if_idle() -> bool:
     return _stop_mcp_loop(only_if_idle=True)
 
 
+async def _drain_mcp_loop_tasks() -> None:
+    """Cancel every task still pending on the MCP loop and reap it.
+
+    Cancelling is not enough on its own: ``Task.cancel()`` only schedules the
+    throw, so the task must be awaited before the loop goes away. Gather them
+    here — while the loop is still open — so each one runs its own ``finally``.
+    """
+    current = asyncio.current_task()
+    pending = [t for t in asyncio.all_tasks() if t is not current and not t.done()]
+    if not pending:
+        return
+    logger.debug("Draining %d pending task(s) from the MCP loop", len(pending))
+    for task in pending:
+        task.cancel()
+    await asyncio.gather(*pending, return_exceptions=True)
+
+
 def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool:
     """Stop the background event loop and join its thread."""
     global _mcp_loop, _mcp_thread
@@ -6630,6 +6647,24 @@ def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool:
         _mcp_loop = None
         _mcp_thread = None
     if loop is not None:
+        # Drain before stopping: closing the loop with tasks still suspended
+        # leaves their coroutines for the GC, whose finalizer then resumes them
+        # to run cleanup against a loop that is already closed -> "Event loop
+        # is closed" (#60197). ``shutdown_mcp_servers`` only reaps servers held
+        # in ``_servers``, so anything else left on this loop ends up here.
+        if loop.is_running():
+            from agent.async_utils import safe_schedule_threadsafe
+
+            future = safe_schedule_threadsafe(
+                _drain_mcp_loop_tasks(), loop,
+                logger=logger,
+                log_message="MCP loop drain: failed to schedule",
+            )
+            if future is not None:
+                try:
+                    future.result(timeout=5)
+                except BaseException as exc:
+                    logger.debug("Error draining MCP loop tasks: %s", exc)
         loop.call_soon_threadsafe(loop.stop)
         if thread is not None:
             thread.join(timeout=5)