fix(mcp): detect server death, fail loudly instead of raw ClosedResourceError

livekit-agents/livekit/agents/llm/mcp.py

@@ -13,6 +13,7 @@
 from typing import Any, Literal
 from urllib.parse import urlparse
 
+import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from typing_extensions import Self
 
@@ -26,6 +27,7 @@
     from mcp.client.sse import sse_client
     from mcp.client.stdio import StdioServerParameters
     from mcp.client.streamable_http import GetSessionIdCallback, streamable_http_client
+    from mcp.shared.exceptions import McpError
     from mcp.shared.message import SessionMessage
 except ImportError as e:
     raise ImportError(
@@ -47,6 +49,18 @@
 MCPTool = RawFunctionTool
 
 
+def _is_connection_dead(exc: BaseException) -> bool:
+    """Whether *exc* indicates the MCP transport is gone (e.g. the server process died).
+
+    Once the underlying stdio/HTTP streams close, ``ClientSession`` raises a raw
+    ``anyio.ClosedResourceError`` (with an empty message) on every subsequent call, and an
+    in-flight request surfaces as ``McpError(CONNECTION_CLOSED)``.
+    """
+    if isinstance(exc, (anyio.ClosedResourceError, anyio.BrokenResourceError)):
+        return True
+    return isinstance(exc, McpError) and exc.error.code == mcp.types.CONNECTION_CLOSED
+
+
 @dataclass
 class MCPToolResultContext:
     """Context passed to an MCPToolResultResolver callback."""
@@ -101,10 +115,19 @@ def invalidate_cache(self) -> None:
 
     async def initialize(self) -> None:
         if self._client_task and not self._client_task.done():
-            logger.warning("MCPServer is already initializing")
-            if self._ready_fut:
-                await self._ready_fut
-            return
+            if self._client is None and self._ready_fut is not None and self._ready_fut.done():
+                logger.debug(
+                    "MCPServer client task is unwinding after a dead connection; "
+                    "waiting for cleanup before reinitializing"
+                )
+                await self._client_task
+                self._client_task = None
+                self._ready_fut = None
+            else:
+                logger.warning("MCPServer is already initializing")
+                if self._ready_fut:
+                    await self._ready_fut
+                return
 
         self._ready_fut = ready_fut = asyncio.Future[None]()
         self._client_task = asyncio.create_task(
@@ -131,6 +154,10 @@ async def _run_client(self, ready_fut: asyncio.Future[None]) -> None:
         except BaseException as e:
             if not ready_fut.done():
                 ready_fut.set_exception(e)  # raising from `await initialize()`
+            elif _is_connection_dead(e):
+                # the transport died after startup (e.g. the server process exited);
+                # unwind quietly — the `finally` below tears the dead client down.
+                logger.debug("MCP client connection closed")
             else:
                 if isinstance(e, Exception):
                     logger.exception("MCP client connection failed with unexpected error")
@@ -140,14 +167,32 @@ async def _run_client(self, ready_fut: asyncio.Future[None]) -> None:
             self._lk_tools = None
             self._closing_ev.clear()
 
+    def _handle_dead_connection(self) -> None:
+        # the server process/transport is gone: flip `initialized` back to False and wake
+        # the parked `_run_client` so it unwinds and runs its cleanup.
+        self._client = None
+        self._lk_tools = None
+        self._cache_dirty = True
+        self._closing_ev.set()
+
     async def list_tools(self) -> list[MCPTool]:
         if self._client is None:
             raise RuntimeError("MCPServer isn't initialized")
 
         if not self._cache_dirty and self._lk_tools is not None:
             return self._lk_tools
 
-        tools = await self._client.list_tools()
+        try:
+            tools = await self._client.list_tools()
+        except Exception as e:
+            if _is_connection_dead(e):
+                self._handle_dead_connection()
+                raise RuntimeError(
+                    "MCPServer connection failed: internal service is unavailable. "
+                    "Please check that the MCPServer is still running."
+                ) from None
+            raise
+
         lk_tools = [
             self._make_function_tool(tool.name, tool.description, tool.inputSchema, tool.meta)
             for tool in tools.tools
@@ -172,7 +217,20 @@ async def _tool_called(raw_arguments: dict[str, Any]) -> Any:
                     "Please check that the MCPServer is still running."
                 )
 
-            tool_result = await self._client.call_tool(name, raw_arguments)
+            try:
+                tool_result = await self._client.call_tool(name, raw_arguments)
+            except Exception as e:
+                # the MCP server process may have died mid-session; the transport then
+                # raises a raw anyio.ClosedResourceError (empty message) on every call.
+                # Fail loudly with a clear ToolError and tear the dead client down so
+                # `initialized` reports False instead of silently staying True.
+                if _is_connection_dead(e):
+                    self._handle_dead_connection()
+                    raise ToolError(
+                        "Tool invocation failed: internal service is unavailable. "
+                        "Please check that the MCPServer is still running."
+                    ) from None
+                raise
 
             if tool_result.isError:
                 error_str = "\n".join(

tests/test_mcp.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import asyncio
+
+import anyio
+import pytest
+
+pytest.importorskip("mcp")
+
+import mcp.types
+from mcp.shared.exceptions import McpError
+from mcp.types import CONNECTION_CLOSED, METHOD_NOT_FOUND, ErrorData
+
+from livekit.agents.llm import mcp as lk_mcp
+from livekit.agents.llm.tool_context import ToolError
+
+pytestmark = pytest.mark.unit
+
+
+def test_is_connection_dead():
+    assert lk_mcp._is_connection_dead(anyio.ClosedResourceError())
+    assert lk_mcp._is_connection_dead(anyio.BrokenResourceError())
+    assert lk_mcp._is_connection_dead(
+        McpError(ErrorData(code=CONNECTION_CLOSED, message="Connection closed"))
+    )
+    # a normal protocol error is not a dead connection
+    assert not lk_mcp._is_connection_dead(
+        McpError(ErrorData(code=METHOD_NOT_FOUND, message="Method not found"))
+    )
+    assert not lk_mcp._is_connection_dead(ValueError("boom"))
+
+
+class _DeadClient:
+    """Fake ClientSession whose transport has died (server process gone)."""
+
+    async def call_tool(self, name: str, arguments: dict) -> object:
+        raise anyio.ClosedResourceError
+
+
+class _ErrorResultClient:
+    """Fake ClientSession that returns a normal tool error result."""
+
+    async def call_tool(self, name: str, arguments: dict) -> mcp.types.CallToolResult:
+        return mcp.types.CallToolResult(
+            content=[mcp.types.TextContent(type="text", text="tool blew up")],
+            isError=True,
+        )
+
+
+class _DeadListToolsClient:
+    """Fake ClientSession whose transport has died while listing tools."""
+
+    async def list_tools(self) -> object:
+        raise anyio.ClosedResourceError
+
+
+def _tool(server: lk_mcp.MCPServer):
+    return server._make_function_tool("ping", None, {"type": "object", "properties": {}}, None)
+
+
+async def test_dead_connection_fails_loudly_and_flips_initialized():
+    # a session that connected and then died underneath us must not keep raising raw
+    # anyio.ClosedResourceError forever: it should fail loudly and stop reporting as live.
+    server = lk_mcp.MCPServerStdio(command="unused", args=[])
+    server._client = _DeadClient()  # type: ignore[assignment]
+    assert server.initialized is True
+
+    with pytest.raises(ToolError):
+        await _tool(server)(raw_arguments={})
+
+    assert server.initialized is False
+
+
+async def test_list_tools_dead_connection_tears_down_and_raises():
+    server = lk_mcp.MCPServerStdio(command="unused", args=[])
+    server._client = _DeadListToolsClient()  # type: ignore[assignment]
+    server._cache_dirty = True
+
+    with pytest.raises(RuntimeError, match="MCPServer connection failed"):
+        await server.list_tools()
+
+    assert server.initialized is False
+
+
+async def test_initialize_waits_for_previous_dead_connection_cleanup():
+    server = lk_mcp.MCPServerStdio(command="unused", args=[])
+
+    task_started = asyncio.Event()
+    task_continue = asyncio.Event()
+
+    async def old_task() -> None:
+        task_started.set()
+        await task_continue.wait()
+
+    server._client = None
+    server._ready_fut = asyncio.get_running_loop().create_future()
+    server._ready_fut.set_result(None)
+    server._client_task = asyncio.create_task(old_task())
+
+    await task_started.wait()
+
+    async def fake_run_client(self, ready_fut: asyncio.Future[None]) -> None:
+        ready_fut.set_result(None)
+
+    server._run_client = fake_run_client.__get__(server, type(server))  # type: ignore[assignment]
+
+    initialize_task = asyncio.create_task(server.initialize())
+    await asyncio.sleep(0.01)
+    assert not initialize_task.done()
+
+    task_continue.set()
+    await initialize_task
+
+    assert server._client_task is not None
+    assert server._client_task.done()
+
+
+async def test_tool_error_result_still_propagates():
+    # a normal isError result must still surface as a ToolError and must NOT tear the
+    # connection down (i.e. it is not mistaken for a dead transport).
+    server = lk_mcp.MCPServerStdio(command="unused", args=[])
+    server._client = _ErrorResultClient()  # type: ignore[assignment]
+
+    with pytest.raises(ToolError) as exc_info:
+        await _tool(server)(raw_arguments={})
+
+    assert "tool blew up" in str(exc_info.value)
+    assert server.initialized is True