diff --git a/src/gumloop/_gumcp_transport.py b/src/gumloop/_gumcp_transport.py index f3c2fc7..b6d1a5f 100644 --- a/src/gumloop/_gumcp_transport.py +++ b/src/gumloop/_gumcp_transport.py @@ -32,6 +32,8 @@ _MAX_BATCH = 5 _HTTP_STATUS_RE = re.compile(r"HTTP\s+(\d{3})") _LIVENESS_POLL_SECONDS = 1.0 +# Above the backend's 300 s per-tool-call limit, so it only ends a call whose answer was lost. +_CALL_TIMEOUT_SECONDS = 330.0 def gumcp_env_ready() -> bool: @@ -252,6 +254,20 @@ def _map_exception(exc: BaseException, *, ref: str, server_id: str, tool_name: s ) +class _CancelScopeError(Exception): + """anyio's "cancel scope" CancelledError from a broken session, as an ordinary error.""" + + +async def _call_tool(client: Any, name: str, arguments: dict[str, Any]) -> Any: + # On 3.10/3.11 wait_for drops a CancelledError's message, so "cancel scope" is converted in here. + try: + return await client.call_tool(name, arguments) + except asyncio.CancelledError as exc: + if "cancel scope" in str(exc): + raise _CancelScopeError(str(exc)) from exc + raise + + class GumcpTransport: """Owns one multi-server ``AsyncClient`` keyed by the current GUMCP env fingerprint.""" @@ -325,9 +341,9 @@ async def call_one( result_ref = ref or fallback_ref try: client = await self._ensure_client() - content = await client.call_tool( - f"{server_id}__{tool_name}", - arguments or {}, + content = await asyncio.wait_for( + _call_tool(client, f"{server_id}__{tool_name}", arguments or {}), + timeout=_CALL_TIMEOUT_SECONDS, ) return McpToolCallResult( ref=result_ref, @@ -338,6 +354,18 @@ async def call_one( ) except GumloopError: raise + except asyncio.TimeoutError: + logger.warning("MCP tool call %s__%s got no answer within %gs", server_id, tool_name, _CALL_TIMEOUT_SECONDS) + return _error_result( + ref=result_ref, + server_id=server_id, + tool_name=tool_name, + status="error", + code="tool_execution_failed", + message=f"Tool call timed out after {_CALL_TIMEOUT_SECONDS:g}s.", + error_type="api_error", + details={"server_id": server_id, "tool_name": tool_name}, + ) except asyncio.CancelledError as exc: # Real cancellation propagates; only the anyio cancel-scope failure maps. if "cancel scope" not in str(exc): diff --git a/src/gumloop/_version.py b/src/gumloop/_version.py index 3cf9ae0..ba2fca4 100644 --- a/src/gumloop/_version.py +++ b/src/gumloop/_version.py @@ -1,3 +1,3 @@ from __future__ import annotations -__version__ = "0.5.9" +__version__ = "0.5.10" diff --git a/tests/sdk/test_gumcp_transport.py b/tests/sdk/test_gumcp_transport.py index 470179c..f556dbb 100644 --- a/tests/sdk/test_gumcp_transport.py +++ b/tests/sdk/test_gumcp_transport.py @@ -5,12 +5,14 @@ import logging import os import sys +from contextlib import AsyncExitStack from pathlib import Path from typing import Any from unittest.mock import AsyncMock from unittest.mock import MagicMock from unittest.mock import patch +import anyio import httpx import pytest import respx @@ -482,6 +484,90 @@ async def close(self) -> None: assert len(set(seen_loops)) == 1 # everything ran on the transport loop +class _HangingClient: + hung_call_cancelled = False + + def __init__(self, **_kwargs: Any) -> None: + pass + + async def call_tool(self, tool: str, _args: dict[str, Any]) -> list[str]: + if tool == "gmail__hang": + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + _HangingClient.hung_call_cancelled = True + raise + await asyncio.sleep(0.01) + return [tool] + + async def close(self) -> None: + pass + + +def test_a_call_that_never_answers_times_out_and_the_next_call_succeeds( + gumcp_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + import gumloop._gumcp_transport as transport_mod + + monkeypatch.setattr(transport_mod, "_CALL_TIMEOUT_SECONDS", 0.1) + monkeypatch.setattr(_HangingClient, "hung_call_cancelled", False) + + with patch("gumloop._gumcp_transport._import_async_client", return_value=_HangingClient): + client = Gumloop(access_token="http-token") + hung = client.mcp.execute("gmail", "hang", {}).results[0] + after = client.mcp.execute("gmail", "read_emails", {}).results[0] + client.close() + + assert hung.status == "error" + assert hung.error is not None + assert hung.error["code"] == "tool_execution_failed" + assert "timed out after 0.1s" in hung.error["message"] + assert _HangingClient.hung_call_cancelled + assert after.status == "success" + + +class _LazyConnectClient: + def __init__(self, **_kwargs: Any) -> None: + self._connection: AsyncExitStack | None = None + + async def call_tool(self, tool: str, _args: dict[str, Any]) -> list[str]: + if self._connection is None: + self._connection = AsyncExitStack() + await self._connection.enter_async_context(anyio.create_task_group()) + return [tool] + + async def close(self) -> None: + pass + + +def test_the_call_that_opens_the_connection_succeeds(gumcp_env: None) -> None: + with patch("gumloop._gumcp_transport._import_async_client", return_value=_LazyConnectClient): + client = Gumloop(access_token="http-token") + results = [client.mcp.execute("gmail", f"tool_{i}", {}).results[0] for i in range(2)] + client.close() + + assert [r.status for r in results] == ["success", "success"] + + +def test_threaded_fanout_survives_one_call_that_never_answers(gumcp_env: None, monkeypatch: pytest.MonkeyPatch) -> None: + from concurrent.futures import ThreadPoolExecutor + + import gumloop._gumcp_transport as transport_mod + + monkeypatch.setattr(transport_mod, "_CALL_TIMEOUT_SECONDS", 0.2) + tools = [f"tool_{i}" for i in range(23)] + ["hang"] + + with patch("gumloop._gumcp_transport._import_async_client", return_value=_HangingClient): + client = Gumloop(access_token="http-token") + with ThreadPoolExecutor(max_workers=6) as ex: + futures = {tool: ex.submit(client.mcp.execute, "gmail", tool, {}) for tool in tools} + results = {tool: future.result(timeout=10).results[0] for tool, future in futures.items()} + client.close() + + assert results.pop("hang").status == "error" + assert all(r.status == "success" for r in results.values()) + + def test_close_on_unused_transport_is_a_noop(gumcp_env: None) -> None: transport = GumcpTransport() transport.close()