diff --git a/navi/mcp/client.py b/navi/mcp/client.py index feabb82..718175f 100644 --- a/navi/mcp/client.py +++ b/navi/mcp/client.py @@ -309,10 +309,24 @@ else: self._fail(cmd.fut, ValueError(f"unknown op: {cmd.op}")) except asyncio.CancelledError: - # Runner is being cancelled (e.g. app teardown). Release - # the in-flight caller, then let close_all run in finally. - self._cancel(cmd.fut) - raise + task = asyncio.current_task() + if task is None or task.cancelling(): + # Runner is really being cancelled (app teardown). + # Release the in-flight caller, then let close_all run + # in finally. + self._cancel(cmd.fut) + raise + # A failed transport connect inside the MCP SDK's anyio + # cancel scopes can surface as CancelledError without an + # actual task.cancel() (seen with streamable_http against a + # dead endpoint). Treat it as an ordinary connect failure: + # one unreachable MCP server must not take down the whole + # startup (McpManager.load_all only catches Exception). + self._connected = connected + self._instructions = instructions + self._fail(cmd.fut, RuntimeError( + f"MCP server {self.name!r} connect failed (anyio CancelledError)" + )) except Exception as exc: # On a connect/ensure failure, mirror the disconnected # state before surfacing the error to the caller. @@ -321,6 +335,14 @@ self._fail(cmd.fut, exc) finally: await close_all() + # Runner is gone — fail anything still queued so callers waiting + # in _send don't hang forever. + while not queue.empty(): + try: + cmd = queue.get_nowait() + except Exception: + break + self._fail(cmd.fut, RuntimeError(f"MCP runner for {self.name!r} exited")) @staticmethod def _resolve(fut: asyncio.Future | None, value: Any) -> None: diff --git a/tests/unit/test_mcp.py b/tests/unit/test_mcp.py index ae94189..9efd011 100644 --- a/tests/unit/test_mcp.py +++ b/tests/unit/test_mcp.py @@ -241,3 +241,51 @@ assert args["session_id"] == "real-session-id" finally: current_session_id.reset(token) + + +class TestLoadAllTolerance: + async def test_load_all_survives_dead_streamable_http_endpoint(self): + """Regression: a streamable_http transport connecting to a dead + endpoint raises CancelledError from inside the MCP SDK's anyio + cancel scopes — without any task.cancel(). The runner used to treat + that as "runner is being cancelled", cancelled the caller's future, + and McpManager.load_all (which only catches Exception) blew up the + whole server startup. Now the connect failure surfaces as a normal + exception: the server starts, the server is retried by the + health-check loop.""" + manager = McpManager() + cfg = McpServerConfig( + transport="streamable_http", + url="http://127.0.0.1:9/mcp", # port 9 (discard) — nothing listens + ) + # Before the fix this raised CancelledError out of load_all. + await manager.load_all({"dead": cfg}) + assert "dead" in manager.clients + assert manager._connected_status["dead"] is False + await manager.disconnect_all() + + async def test_runner_death_fails_pending_callers(self): + """If the runner task dies unexpectedly, callers waiting in _send + get an error instead of hanging forever.""" + import asyncio + + client = McpClient( + "dead", + McpServerConfig(transport="streamable_http", url="http://127.0.0.1:9/mcp"), + ) + # Poison the open path so the runner exits through the CancelledError + # branch with a real task cancellation: connect once (fails normally), + # then cancel the runner mid-connect and re-issue a connect that + # would outlive it. + with pytest.raises(Exception): + await client.connect() + # The first connect failed; the runner is still alive (it keeps the + # client for health-check retries). Kill it hard and make sure a + # subsequent command surfaces an error, not a hang. + runner = client._runner_task + assert runner is not None + runner.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(asyncio.shield(runner), timeout=5) + with pytest.raises(RuntimeError): + await asyncio.wait_for(client.connect(), timeout=5)