diff --git a/backend/packages/harness/deerflow/community/groundroute/AGENTS.md b/backend/packages/harness/deerflow/community/groundroute/AGENTS.md index 55f6f0f2de..ae74289dc9 100644 --- a/backend/packages/harness/deerflow/community/groundroute/AGENTS.md +++ b/backend/packages/harness/deerflow/community/groundroute/AGENTS.md @@ -11,6 +11,23 @@ truncating to the snippet limit, and renders a missing or falsy title as empty, well-formed result object whose leaf fields are not strings still returns text rather than raising. +Both tools are coroutines and send their request through `httpx.AsyncClient`. A +sync client would run the whole tool on the default executor and hold one of that +pool's threads for the network round-trip, so several providers queried in parallel +queued behind each other. Resolve the config in the same coroutine, but keep the +config file read off the event loop: the tool body now runs on the loop, so a file +read left in it would block it. `backend/tests/blocking_io/test_groundroute_search_async.py` +pins all three — the request is awaited on the loop thread, the config lookup does +not run there, and the request goes out through the async client. That last one +matters because a `MockTransport` runs its handler on the calling thread without +real blocking I/O, so a provider that stayed async but fell back to a sync client +would satisfy the thread assertion; the anchor records which client factory the +provider reached for. + +Both tools resolve the key and `max_results` through `_resolve_tool_settings`, one +config read per call. Do not add a second seam around that read: a wrapper with no +production caller can drift from the real call path. + `web_search` returns the structured `{"error": ..., "query": ...}` response; `web_fetch` returns the `Error: ...` string. Log the malformed container's type, or that no usable result objects remain, before returning — never the query, URL, credentials, or payload diff --git a/backend/packages/harness/deerflow/community/groundroute/tools.py b/backend/packages/harness/deerflow/community/groundroute/tools.py index bbb3f9fa20..d7f158c5d5 100644 --- a/backend/packages/harness/deerflow/community/groundroute/tools.py +++ b/backend/packages/harness/deerflow/community/groundroute/tools.py @@ -16,6 +16,7 @@ Langflow component: `web_fetch` reads one URL via GroundRoute mode=page and returns its extracted text. """ +import asyncio import json import logging import os @@ -37,19 +38,29 @@ _FETCH_SNIPPET_LIMIT = 4096 _api_key_warned: set[str] = set() -def _get_api_key(tool_name: str) -> str | None: - """Resolve the GroundRoute key from a given tool's config block, then the env var. +def _resolve_tool_settings(tool_name: str) -> tuple[str | None, object]: + """Resolve the API key and the configured `max_results` in one config read. - `tool_name` is the config section to read (web_search vs web_fetch) so a flow that - runs GroundRoute for fetch but a different engine for search still reads the right - key. Mirrors serper/exa/firecrawl, which all take the tool name. + Config resolution stats the config file, so callers must run this off the + agent event loop. Reading the key and the result count from a single + resolution keeps the tool from touching the config file twice per call. + + `tool_name` is the config section to read (web_search vs web_fetch) so a + flow that runs GroundRoute for fetch but a different engine for search + still reads the right key. """ - config = get_app_config().get_tool_config(tool_name) - if config is not None: - api_key = (config.model_extra or {}).get("api_key") - if isinstance(api_key, str) and api_key.strip(): - return api_key.strip() - return os.getenv("GROUNDROUTE_API_KEY") + tool_config = get_app_config().get_tool_config(tool_name) + max_results = None + api_key = None + if tool_config is not None: + extra = tool_config.model_extra or {} + max_results = extra.get("max_results") + candidate = extra.get("api_key") + if isinstance(candidate, str) and candidate.strip(): + api_key = candidate.strip() + if api_key is None: + api_key = os.getenv("GROUNDROUTE_API_KEY") + return api_key, max_results def _coerce_max_results(value: object, *, default: int = _DEFAULT_MAX_RESULTS) -> int: @@ -99,9 +110,9 @@ def _response_results(data: object) -> list[dict] | None: return usable -def _post_search(api_key: str, body: dict) -> dict: - with httpx.Client(timeout=_TIMEOUT_S) as client: - response = client.post( +async def _post_search(api_key: str, body: dict) -> dict: + async with httpx.AsyncClient(timeout=_TIMEOUT_S) as client: + response = await client.post( _GROUNDROUTE_ENDPOINT, json=body, headers={"Authorization": f"Bearer {api_key}"}, @@ -111,7 +122,7 @@ def _post_search(api_key: str, body: dict) -> dict: @tool("web_search", parse_docstring=True) -def web_search_tool(query: str, max_results: int | None = None) -> str: +async def web_search_tool(query: str, max_results: int | None = None) -> str: """Search the web for information using GroundRoute. GroundRoute routes the query across six search engines and returns the result @@ -121,19 +132,18 @@ def web_search_tool(query: str, max_results: int | None = None) -> str: query: Search keywords describing what you want to find. Be specific for better results. max_results: Maximum number of search results to return. If omitted, uses the configured value (default 5). Clamped to 1-50. """ + # Config resolution can read files; keep it off the event loop. + api_key, configured_max_results = await asyncio.to_thread(_resolve_tool_settings, "web_search") # Honor the caller-supplied max_results; fall back to config only when omitted. if max_results is None: - config = get_app_config().get_tool_config("web_search") - if config is not None: - max_results = (config.model_extra or {}).get("max_results") + max_results = configured_max_results count = _DEFAULT_MAX_RESULTS if max_results is None else _coerce_max_results(max_results) - api_key = _get_api_key("web_search") if not api_key: return _missing_key_error("web_search", query=query) try: - data = _post_search(api_key, {"query": query, "max_results": count}) + data = await _post_search(api_key, {"query": query, "max_results": count}) except httpx.HTTPStatusError as e: logger.error("GroundRoute API returned HTTP %s: %s", e.response.status_code, e.response.text) return json.dumps( @@ -163,7 +173,7 @@ def web_search_tool(query: str, max_results: int | None = None) -> str: @tool("web_fetch", parse_docstring=True) -def web_fetch_tool(url: str) -> str: +async def web_fetch_tool(url: str) -> str: """Fetch the contents of a web page at a given URL via GroundRoute. Only fetch EXACT URLs that have been provided directly by the user or have been returned in results from the web_search and web_fetch tools. This tool can NOT access content that requires authentication, such as private Google Docs or pages behind login walls. @@ -173,12 +183,12 @@ def web_fetch_tool(url: str) -> str: Args: url: The URL to fetch the contents of. """ - api_key = _get_api_key("web_fetch") + api_key, _ = await asyncio.to_thread(_resolve_tool_settings, "web_fetch") if not api_key: return _missing_key_error("web_fetch", url=url) try: - data = _post_search(api_key, {"query": url, "mode": "page", "max_results": 1}) + data = await _post_search(api_key, {"query": url, "mode": "page", "max_results": 1}) except httpx.HTTPStatusError as e: logger.error("GroundRoute fetch returned HTTP %s: %s", e.response.status_code, e.response.text) return f"Error: GroundRoute API error: HTTP {e.response.status_code}" diff --git a/backend/tests/blocking_io/test_groundroute_search_async.py b/backend/tests/blocking_io/test_groundroute_search_async.py new file mode 100644 index 0000000000..9c597e4b4c --- /dev/null +++ b/backend/tests/blocking_io/test_groundroute_search_async.py @@ -0,0 +1,133 @@ +"""GroundRoute community tools must not occupy an executor thread or block the loop. + +The GroundRoute provider resolved its config and sent its search request +through a synchronous ``httpx.Client``. A sync LangChain tool invoked through +``ainvoke`` runs on the default executor, so the request held one of that +pool's threads for the whole network round-trip: with several providers +queried in parallel the default pool ran out of threads and the tool calls +queued behind each other. + +The probes below assert the whole fix: + +* the request is awaited on the event loop thread instead of an executor + thread, so no pool thread is held for the round-trip; +* the config lookup, which reads the config file, still runs off the loop. + The tool body is now a coroutine that executes on the loop, so a file read + left in it would block the loop; the strict blocking-I/O gate fails if the + lookup regresses onto the loop. +* the request goes out through the async client. A `MockTransport` runs its + handler on the calling thread and performs no real blocking I/O, so a + provider that stayed async but sent the request through a sync + `httpx.Client` would satisfy the thread assertion above; the recorded + client factory is what catches that partial regression. +""" + +import threading +from pathlib import Path +from types import SimpleNamespace + +import httpx +import pytest + +from deerflow.community.groundroute import tools as groundroute_tools + +pytestmark = pytest.mark.asyncio + + +def _patch_transport(monkeypatch, request_threads: list[int], clients: list[str]) -> None: + """Answer every request locally and record the thread that served it. + + The provider's module holds the shared ``httpx`` module, so the reference + itself is replaced with a namespace exposing both client factories. That + keeps the test hermetic — no request leaves the machine — whichever + factory the provider uses. Each factory records its name in ``clients`` + so a test can assert which one the provider reached for. + """ + + def respond(request: httpx.Request) -> httpx.Response: + request_threads.append(threading.get_ident()) + return httpx.Response(200, json={"results": []}) + + transport = httpx.MockTransport(respond) + + def async_client(**kwargs): + clients.append("AsyncClient") + return httpx.AsyncClient(transport=transport, **kwargs) + + def sync_client(**kwargs): + clients.append("Client") + return httpx.Client(transport=transport, **kwargs) + + monkeypatch.setattr( + groundroute_tools, + "httpx", + SimpleNamespace( + AsyncClient=async_client, + Client=sync_client, + HTTPStatusError=httpx.HTTPStatusError, + ), + ) + + +def _patch_config(monkeypatch, load_config=None) -> None: + def config(): + return SimpleNamespace(get_tool_config=lambda _name: SimpleNamespace(model_extra={"api_key": "synthetic-key"})) + + monkeypatch.setattr(groundroute_tools, "get_app_config", load_config or config) + + +async def test_groundroute_search_request_runs_on_the_event_loop(monkeypatch) -> None: + """The request is awaited on the loop, not dispatched to an executor thread.""" + request_threads: list[int] = [] + clients: list[str] = [] + loop_thread = threading.get_ident() + _patch_transport(monkeypatch, request_threads, clients) + _patch_config(monkeypatch) + + result = await groundroute_tools.web_search_tool.ainvoke({"query": "vector databases"}) + + assert '"error": "No results found"' in result + assert request_threads, "the request must reach the transport" + assert all(thread == loop_thread for thread in request_threads) + assert clients == ["AsyncClient"], "the request must go out through the async client" + + +async def test_groundroute_fetch_request_runs_on_the_event_loop(monkeypatch) -> None: + """web_fetch shares the same client, so it must stay off the executor too.""" + request_threads: list[int] = [] + clients: list[str] = [] + loop_thread = threading.get_ident() + _patch_transport(monkeypatch, request_threads, clients) + _patch_config(monkeypatch) + + result = await groundroute_tools.web_fetch_tool.ainvoke({"url": "https://example.com/report"}) + + assert "Error: No results found" == result + assert request_threads, "the request must reach the transport" + assert all(thread == loop_thread for thread in request_threads) + assert clients == ["AsyncClient"], "the request must go out through the async client" + + +async def test_groundroute_config_lookup_does_not_block_loop(monkeypatch) -> None: + """The tool body runs on the loop, so the config file read must not.""" + config_threads: list[int] = [] + loop_thread = threading.get_ident() + + def load_config(): + # Keep a real file read in the lookup so the strict blocking-I/O gate + # fails if the tool resolves config on the event loop. + Path(__file__).read_text(encoding="utf-8") + config_threads.append(threading.get_ident()) + return SimpleNamespace(get_tool_config=lambda _name: SimpleNamespace(model_extra={"api_key": "synthetic-key"})) + + request_threads: list[int] = [] + clients: list[str] = [] + _patch_transport(monkeypatch, request_threads, clients) + _patch_config(monkeypatch, load_config) + + result = await groundroute_tools.web_search_tool.ainvoke({"query": "vector databases"}) + + assert '"error": "No results found"' in result + assert config_threads, "config lookup must run" + assert all(thread != loop_thread for thread in config_threads) + assert clients == ["AsyncClient"], "the request must go out through the async client" diff --git a/backend/tests/test_groundroute_response_shapes.py b/backend/tests/test_groundroute_response_shapes.py index ed9f940cbd..59f9fba7e2 100644 --- a/backend/tests/test_groundroute_response_shapes.py +++ b/backend/tests/test_groundroute_response_shapes.py @@ -11,6 +11,7 @@ _LOGGER_NAME = "deerflow.community.groundroute.tools" _FORMAT_ERROR = "GroundRoute returned an unexpected response format" _SEARCH_QUERY = "文档 search" _FETCH_URL = "https://example.com/文档" +pytestmark = pytest.mark.anyio @pytest.fixture(autouse=True) @@ -28,9 +29,9 @@ def _invoker(monkeypatch, tool_name): config = SimpleNamespace(get_tool_config=lambda _: SimpleNamespace(model_extra={"api_key": "synthetic-test-key"})) monkeypatch.setattr(tools, "get_app_config", lambda: config) - real_client = httpx.Client + real_client = httpx.AsyncClient - def invoke(payload, *, status_code=200): + async def invoke(payload, *, status_code=200): requests = [] def respond(request): @@ -41,12 +42,12 @@ def _invoker(monkeypatch, tool_name): monkeypatch.setattr( tools, "httpx", - SimpleNamespace(Client=lambda **kwargs: real_client(transport=transport, **kwargs), HTTPStatusError=httpx.HTTPStatusError), + SimpleNamespace(AsyncClient=lambda **kwargs: real_client(transport=transport, **kwargs), HTTPStatusError=httpx.HTTPStatusError), ) if tool_name == "web_search": - result = tools.web_search_tool.invoke({"query": _SEARCH_QUERY}) + result = await tools.web_search_tool.ainvoke({"query": _SEARCH_QUERY}) else: - result = tools.web_fetch_tool.invoke({"url": _FETCH_URL}) + result = await tools.web_fetch_tool.ainvoke({"url": _FETCH_URL}) assert len(requests) == 1 return result @@ -63,8 +64,8 @@ def invoke_fetch(monkeypatch): return _invoker(monkeypatch, "web_fetch") -def _search(invoke_search, payload, *, status_code=200): - return json.loads(invoke_search(payload, status_code=status_code)) +async def _search(invoke_search, payload, *, status_code=200): + return json.loads(await invoke_search(payload, status_code=status_code)) def _error_logs(caplog): @@ -79,29 +80,29 @@ def _assert_no_payload_values_leaked(caplog): @pytest.mark.parametrize("payload", [None, False, 0, "invalid", [], ["provider-private-payload"]]) -def test_malformed_top_level_payload_returns_format_error(invoke_search, payload, caplog): - assert _search(invoke_search, payload) == {"error": _FORMAT_ERROR, "query": "文档 search"} +async def test_malformed_top_level_payload_returns_format_error(invoke_search, payload, caplog): + assert await _search(invoke_search, payload) == {"error": _FORMAT_ERROR, "query": "文档 search"} assert _error_logs(caplog) == [f"GroundRoute returned unexpected payload type: {type(payload).__name__}"] _assert_no_payload_values_leaked(caplog) @pytest.mark.parametrize("results", [False, 0, "", "provider-private-payload", {}, {"title": "provider-private-payload"}]) -def test_malformed_results_container_returns_format_error(invoke_search, results, caplog): - assert _search(invoke_search, {"results": results}) == {"error": _FORMAT_ERROR, "query": "文档 search"} +async def test_malformed_results_container_returns_format_error(invoke_search, results, caplog): + assert await _search(invoke_search, {"results": results}) == {"error": _FORMAT_ERROR, "query": "文档 search"} assert _error_logs(caplog) == [f"GroundRoute returned unexpected 'results' payload type: {type(results).__name__}"] _assert_no_payload_values_leaked(caplog) @pytest.mark.parametrize("results", [[None], [False, 0, "provider-private-payload", []]]) -def test_all_malformed_entries_return_format_error(invoke_search, results, caplog): - assert _search(invoke_search, {"results": results}) == {"error": _FORMAT_ERROR, "query": "文档 search"} +async def test_all_malformed_entries_return_format_error(invoke_search, results, caplog): + assert await _search(invoke_search, {"results": results}) == {"error": _FORMAT_ERROR, "query": "文档 search"} assert _error_logs(caplog) == ["GroundRoute returned 'results' with no usable result objects"] _assert_no_payload_values_leaked(caplog) -def test_mixed_entries_preserve_valid_results_in_order(invoke_search, caplog): +async def test_mixed_entries_preserve_valid_results_in_order(invoke_search, caplog): payload = {"results": [None, {"title": "文档", "url": "https://other.example/one", "snippet": "概述", "source_engine": "serper"}, "invalid", {}, [], {"title": "Second"}]} - assert _search(invoke_search, payload) == [ + assert await _search(invoke_search, payload) == [ {"title": "文档", "url": "https://other.example/one", "snippet": "概述", "source_engine": "serper"}, {"title": "", "url": "", "snippet": "", "source_engine": ""}, {"title": "Second", "url": "", "snippet": "", "source_engine": ""}, @@ -110,14 +111,14 @@ def test_mixed_entries_preserve_valid_results_in_order(invoke_search, caplog): @pytest.mark.parametrize("payload", [{}, {"results": None}, {"request_id": "r1"}, {"results": []}]) -def test_missing_or_empty_results_keep_no_results_response(invoke_search, payload, caplog): - assert _search(invoke_search, payload) == {"error": "No results found", "query": "文档 search"} +async def test_missing_or_empty_results_keep_no_results_response(invoke_search, payload, caplog): + assert await _search(invoke_search, payload) == {"error": "No results found", "query": "文档 search"} assert _error_logs(caplog) == [] -def test_valid_results_keep_existing_normalization(invoke_search, caplog): +async def test_valid_results_keep_existing_normalization(invoke_search, caplog): payload = {"results": [{"title": "T", "url": "https://other.example", "snippet": "s", "source_engine": "exa"}, {}]} - assert _search(invoke_search, payload) == [ + assert await _search(invoke_search, payload) == [ {"title": "T", "url": "https://other.example", "snippet": "s", "source_engine": "exa"}, {"title": "", "url": "", "snippet": "", "source_engine": ""}, ] @@ -125,24 +126,24 @@ def test_valid_results_keep_existing_normalization(invoke_search, caplog): @pytest.mark.parametrize("payload", [None, False, [], "provider-private-payload", {"results": {"0": "provider-private-payload"}}, {"results": [None]}]) -def test_fetch_malformed_payload_returns_format_error(invoke_fetch, payload, caplog): - assert invoke_fetch(payload) == "Error: GroundRoute returned an unexpected response format" +async def test_fetch_malformed_payload_returns_format_error(invoke_fetch, payload, caplog): + assert await invoke_fetch(payload) == "Error: GroundRoute returned an unexpected response format" assert len(_error_logs(caplog)) == 1 _assert_no_payload_values_leaked(caplog) -def test_fetch_missing_or_empty_results_keep_no_results_error(invoke_fetch, caplog): - assert invoke_fetch({"results": []}) == "Error: No results found" - assert invoke_fetch({}) == "Error: No results found" +async def test_fetch_missing_or_empty_results_keep_no_results_error(invoke_fetch, caplog): + assert await invoke_fetch({"results": []}) == "Error: No results found" + assert await invoke_fetch({}) == "Error: No results found" assert _error_logs(caplog) == [] -def test_fetch_skips_malformed_entries_and_returns_the_first_usable_result(invoke_fetch): - assert invoke_fetch({"results": [None, {"title": "Page", "content": "Body text", "url": "https://other.example"}]}) == "# Page\n\nBody text" +async def test_fetch_skips_malformed_entries_and_returns_the_first_usable_result(invoke_fetch): + assert await invoke_fetch({"results": [None, {"title": "Page", "content": "Body text", "url": "https://other.example"}]}) == "# Page\n\nBody text" -def test_valid_fetch_keeps_existing_response(invoke_fetch, caplog): - assert invoke_fetch({"results": [{"title": "Page", "content": "Body text"}]}) == "# Page\n\nBody text" +async def test_valid_fetch_keeps_existing_response(invoke_fetch, caplog): + assert await invoke_fetch({"results": [{"title": "Page", "content": "Body text"}]}) == "# Page\n\nBody text" assert _error_logs(caplog) == [] @@ -156,23 +157,23 @@ def test_valid_fetch_keeps_existing_response(invoke_fetch, caplog): ({"results": [{"title": None, "content": "Body text"}]}, "# \n\nBody text"), ], ) -def test_fetch_coerces_non_string_fields(invoke_fetch, payload, expected, caplog): - assert invoke_fetch(payload) == expected +async def test_fetch_coerces_non_string_fields(invoke_fetch, payload, expected, caplog): + assert await invoke_fetch(payload) == expected assert _error_logs(caplog) == [] -def test_http_transport_patch_does_not_replace_global_client(invoke_search): - real_client = httpx.Client - invoke_search({"results": []}) - assert httpx.Client is real_client - with httpx.Client(transport=httpx.MockTransport(lambda request: httpx.Response(204))) as client: - assert client.get("https://unrelated.invalid/probe").status_code == 204 +async def test_http_transport_patch_does_not_replace_global_client(invoke_search): + real_client = httpx.AsyncClient + await invoke_search({"results": []}) + assert httpx.AsyncClient is real_client + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(204))) as client: + assert (await client.get("https://unrelated.invalid/probe")).status_code == 204 @pytest.mark.parametrize("status_code", [402, 503]) -def test_search_http_error_keeps_existing_structured_response(invoke_search, status_code): - assert _search(invoke_search, {"error": "synthetic-denial"}, status_code=status_code) == {"error": f"GroundRoute API error: HTTP {status_code}", "query": "文档 search"} +async def test_search_http_error_keeps_existing_structured_response(invoke_search, status_code): + assert await _search(invoke_search, {"error": "synthetic-denial"}, status_code=status_code) == {"error": f"GroundRoute API error: HTTP {status_code}", "query": "文档 search"} -def test_fetch_http_error_keeps_existing_error_string(invoke_fetch): - assert invoke_fetch({"error": "synthetic-denial"}, status_code=402) == "Error: GroundRoute API error: HTTP 402" +async def test_fetch_http_error_keeps_existing_error_string(invoke_fetch): + assert await invoke_fetch({"error": "synthetic-denial"}, status_code=402) == "Error: GroundRoute API error: HTTP 402" diff --git a/backend/tests/test_groundroute_tools.py b/backend/tests/test_groundroute_tools.py index 95de840fae..0dbdf78541 100644 --- a/backend/tests/test_groundroute_tools.py +++ b/backend/tests/test_groundroute_tools.py @@ -1,7 +1,7 @@ """Unit tests for the GroundRoute community web search + fetch tools.""" import json -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest @@ -43,10 +43,10 @@ def _make_search_response(payload: dict) -> MagicMock: def _patch_post(mock_resp: MagicMock): - """Patch httpx.Client so the context-managed .post returns mock_resp.""" - patcher = patch("deerflow.community.groundroute.tools.httpx.Client") + """Patch httpx.AsyncClient so the context-managed .post returns mock_resp.""" + patcher = patch("deerflow.community.groundroute.tools.httpx.AsyncClient") mock_client_cls = patcher.start() - mock_client_cls.return_value.__enter__.return_value.post.return_value = mock_resp + mock_client_cls.return_value.__aenter__.return_value.post = AsyncMock(return_value=mock_resp) return patcher, mock_client_cls @@ -62,16 +62,16 @@ def _per_tool_config(**by_tool): return app -class TestGetApiKey: +class TestResolveToolSettings: def test_returns_config_key_when_present(self): with patch("deerflow.community.groundroute.tools.get_app_config") as mock: tool_config = MagicMock() tool_config.model_extra = {"api_key": "from-config"} mock.return_value.get_tool_config.return_value = tool_config - from deerflow.community.groundroute.tools import _get_api_key + from deerflow.community.groundroute.tools import _resolve_tool_settings - assert _get_api_key("web_search") == "from-config" + assert _resolve_tool_settings("web_search")[0] == "from-config" def test_falls_back_to_env_when_config_key_empty(self): with patch("deerflow.community.groundroute.tools.get_app_config") as mock: @@ -79,17 +79,17 @@ class TestGetApiKey: tool_config.model_extra = {"api_key": " "} mock.return_value.get_tool_config.return_value = tool_config with patch.dict("os.environ", {"GROUNDROUTE_API_KEY": "env-key"}, clear=True): - from deerflow.community.groundroute.tools import _get_api_key + from deerflow.community.groundroute.tools import _resolve_tool_settings - assert _get_api_key("web_search") == "env-key" + assert _resolve_tool_settings("web_search")[0] == "env-key" def test_returns_none_when_no_key_anywhere(self): with patch("deerflow.community.groundroute.tools.get_app_config") as mock: mock.return_value.get_tool_config.return_value = None with patch.dict("os.environ", {}, clear=True): - from deerflow.community.groundroute.tools import _get_api_key + from deerflow.community.groundroute.tools import _resolve_tool_settings - assert _get_api_key("web_search") is None + assert _resolve_tool_settings("web_search")[0] is None def test_reads_the_named_tools_config_block(self): """web_fetch must read the web_fetch block, not web_search (multi-engine flows).""" @@ -98,14 +98,15 @@ class TestGetApiKey: web_search={"api_key": "search-key"}, web_fetch={"api_key": "fetch-key"}, ) - from deerflow.community.groundroute.tools import _get_api_key + from deerflow.community.groundroute.tools import _resolve_tool_settings - assert _get_api_key("web_search") == "search-key" - assert _get_api_key("web_fetch") == "fetch-key" + assert _resolve_tool_settings("web_search")[0] == "search-key" + assert _resolve_tool_settings("web_fetch")[0] == "fetch-key" class TestWebSearchTool: - def test_basic_search_returns_normalized_list_with_source_engine(self, mock_config_with_key): + @pytest.mark.anyio + async def test_basic_search_returns_normalized_list_with_source_engine(self, mock_config_with_key): payload = { "request_id": "r1", "results": [ @@ -117,7 +118,7 @@ class TestWebSearchTool: try: from deerflow.community.groundroute.tools import web_search_tool - result = web_search_tool.invoke({"query": "vector databases"}) + result = await web_search_tool.ainvoke({"query": "vector databases"}) parsed = json.loads(result) finally: patcher.stop() @@ -132,7 +133,8 @@ class TestWebSearchTool: } assert {r["source_engine"] for r in parsed} == {"serper", "exa"} - def test_uses_web_search_config_key(self): + @pytest.mark.anyio + async def test_uses_web_search_config_key(self): """web_search authenticates with the web_search config block's key.""" with patch("deerflow.community.groundroute.tools.get_app_config") as mock: mock.return_value = _per_tool_config( @@ -144,15 +146,16 @@ class TestWebSearchTool: try: from deerflow.community.groundroute.tools import web_search_tool - web_search_tool.invoke({"query": "hello world"}) - call = mock_client_cls.return_value.__enter__.return_value.post.call_args + await web_search_tool.ainvoke({"query": "hello world"}) + call = mock_client_cls.return_value.__aenter__.return_value.post.call_args finally: patcher.stop() assert call.kwargs["headers"]["Authorization"] == "Bearer search-key" assert call.kwargs["json"]["query"] == "hello world" - def test_agent_max_results_is_honored_over_config(self): + @pytest.mark.anyio + async def test_agent_max_results_is_honored_over_config(self): """A caller-supplied max_results wins over the config value (not silently discarded).""" with patch("deerflow.community.groundroute.tools.get_app_config") as mock: tool_config = MagicMock() @@ -163,28 +166,30 @@ class TestWebSearchTool: try: from deerflow.community.groundroute.tools import web_search_tool - web_search_tool.invoke({"query": "test", "max_results": 20}) - body = mock_client_cls.return_value.__enter__.return_value.post.call_args.kwargs["json"] + await web_search_tool.ainvoke({"query": "test", "max_results": 20}) + body = mock_client_cls.return_value.__aenter__.return_value.post.call_args.kwargs["json"] finally: patcher.stop() assert body["max_results"] == 20 - def test_config_max_results_used_when_caller_omits(self, mock_config_with_key): + @pytest.mark.anyio + async def test_config_max_results_used_when_caller_omits(self, mock_config_with_key): """When the caller omits max_results, the configured value is used.""" payload = {"results": [{"url": "u", "title": "t", "snippet": "s", "source_engine": "exa"}]} patcher, mock_client_cls = _patch_post(_make_search_response(payload)) try: from deerflow.community.groundroute.tools import web_search_tool - web_search_tool.invoke({"query": "test"}) - body = mock_client_cls.return_value.__enter__.return_value.post.call_args.kwargs["json"] + await web_search_tool.ainvoke({"query": "test"}) + body = mock_client_cls.return_value.__aenter__.return_value.post.call_args.kwargs["json"] finally: patcher.stop() assert body["max_results"] == 5 - def test_max_results_clamped_to_cap(self): + @pytest.mark.anyio + async def test_max_results_clamped_to_cap(self): with patch("deerflow.community.groundroute.tools.get_app_config") as mock: tool_config = MagicMock() tool_config.model_extra = {"api_key": "k", "max_results": "500"} @@ -194,79 +199,85 @@ class TestWebSearchTool: try: from deerflow.community.groundroute.tools import web_search_tool - web_search_tool.invoke({"query": "test"}) - body = mock_client_cls.return_value.__enter__.return_value.post.call_args.kwargs["json"] + await web_search_tool.ainvoke({"query": "test"}) + body = mock_client_cls.return_value.__aenter__.return_value.post.call_args.kwargs["json"] finally: patcher.stop() assert body["max_results"] == 50 - def test_empty_results_returns_error_json(self, mock_config_with_key): + @pytest.mark.anyio + async def test_empty_results_returns_error_json(self, mock_config_with_key): patcher, _ = _patch_post(_make_search_response({"results": []})) try: from deerflow.community.groundroute.tools import web_search_tool - parsed = json.loads(web_search_tool.invoke({"query": "no results"})) + parsed = json.loads(await web_search_tool.ainvoke({"query": "no results"})) finally: patcher.stop() assert parsed["error"] == "No results found" assert parsed["query"] == "no results" - def test_missing_api_key_returns_error_json(self, mock_config_no_key): + @pytest.mark.anyio + async def test_missing_api_key_returns_error_json(self, mock_config_no_key): with patch.dict("os.environ", {}, clear=True): from deerflow.community.groundroute.tools import web_search_tool - parsed = json.loads(web_search_tool.invoke({"query": "test"})) + parsed = json.loads(await web_search_tool.ainvoke({"query": "test"})) assert "error" in parsed assert "GROUNDROUTE_API_KEY" in parsed["error"] - def test_missing_api_key_logs_warning_once(self, mock_config_no_key, caplog): + @pytest.mark.anyio + async def test_missing_api_key_logs_warning_once(self, mock_config_no_key, caplog): import logging with patch.dict("os.environ", {}, clear=True): from deerflow.community.groundroute.tools import web_search_tool with caplog.at_level(logging.WARNING, logger="deerflow.community.groundroute.tools"): - web_search_tool.invoke({"query": "q1"}) - web_search_tool.invoke({"query": "q2"}) + await web_search_tool.ainvoke({"query": "q1"}) + await web_search_tool.ainvoke({"query": "q2"}) warnings = [r for r in caplog.records if r.levelno == logging.WARNING] assert len(warnings) == 1 - def test_http_error_returns_structured_error(self, mock_config_with_key): + @pytest.mark.anyio + async def test_http_error_returns_structured_error(self, mock_config_with_key): mock_resp = MagicMock() mock_resp.raise_for_status.side_effect = httpx.HTTPStatusError("402", request=MagicMock(), response=MagicMock(status_code=402, text="Payment Required")) patcher, _ = _patch_post(mock_resp) try: from deerflow.community.groundroute.tools import web_search_tool - parsed = json.loads(web_search_tool.invoke({"query": "test"})) + parsed = json.loads(await web_search_tool.ainvoke({"query": "test"})) finally: patcher.stop() assert "error" in parsed assert "402" in parsed["error"] - def test_network_exception_returns_error_json(self, mock_config_with_key): + @pytest.mark.anyio + async def test_network_exception_returns_error_json(self, mock_config_with_key): patcher, mock_client_cls = _patch_post(MagicMock()) - mock_client_cls.return_value.__enter__.return_value.post.side_effect = Exception("timeout") + mock_client_cls.return_value.__aenter__.return_value.post.side_effect = Exception("timeout") try: from deerflow.community.groundroute.tools import web_search_tool - parsed = json.loads(web_search_tool.invoke({"query": "test"})) + parsed = json.loads(await web_search_tool.ainvoke({"query": "test"})) finally: patcher.stop() assert "error" in parsed - def test_partial_fields_default_to_empty_string(self, mock_config_with_key): + @pytest.mark.anyio + async def test_partial_fields_default_to_empty_string(self, mock_config_with_key): patcher, _ = _patch_post(_make_search_response({"results": [{}]})) try: from deerflow.community.groundroute.tools import web_search_tool - parsed = json.loads(web_search_tool.invoke({"query": "test"})) + parsed = json.loads(await web_search_tool.ainvoke({"query": "test"})) finally: patcher.stop() @@ -274,14 +285,15 @@ class TestWebSearchTool: class TestWebFetchTool: - def test_fetch_returns_titled_content(self, mock_config_with_key): + @pytest.mark.anyio + async def test_fetch_returns_titled_content(self, mock_config_with_key): payload = {"results": [{"title": "Page", "content": "Body text", "url": "https://ex.com"}]} patcher, mock_client_cls = _patch_post(_make_search_response(payload)) try: from deerflow.community.groundroute.tools import web_fetch_tool - result = web_fetch_tool.invoke({"url": "https://ex.com"}) - body = mock_client_cls.return_value.__enter__.return_value.post.call_args.kwargs["json"] + result = await web_fetch_tool.ainvoke({"url": "https://ex.com"}) + body = mock_client_cls.return_value.__aenter__.return_value.post.call_args.kwargs["json"] finally: patcher.stop() @@ -290,7 +302,8 @@ class TestWebFetchTool: assert body["mode"] == "page" assert body["query"] == "https://ex.com" - def test_fetch_uses_web_fetch_config_key(self): + @pytest.mark.anyio + async def test_fetch_uses_web_fetch_config_key(self): """web_fetch must authenticate with the web_fetch config block's key, not web_search's.""" with patch("deerflow.community.groundroute.tools.get_app_config") as mock: mock.return_value = _per_tool_config( @@ -302,28 +315,30 @@ class TestWebFetchTool: try: from deerflow.community.groundroute.tools import web_fetch_tool - web_fetch_tool.invoke({"url": "https://ex.com"}) - call = mock_client_cls.return_value.__enter__.return_value.post.call_args + await web_fetch_tool.ainvoke({"url": "https://ex.com"}) + call = mock_client_cls.return_value.__aenter__.return_value.post.call_args finally: patcher.stop() assert call.kwargs["headers"]["Authorization"] == "Bearer fetch-key" - def test_fetch_missing_key_returns_error(self, mock_config_no_key): + @pytest.mark.anyio + async def test_fetch_missing_key_returns_error(self, mock_config_no_key): with patch.dict("os.environ", {}, clear=True): from deerflow.community.groundroute.tools import web_fetch_tool - parsed = json.loads(web_fetch_tool.invoke({"url": "https://ex.com"})) + parsed = json.loads(await web_fetch_tool.ainvoke({"url": "https://ex.com"})) assert "error" in parsed assert "GROUNDROUTE_API_KEY" in parsed["error"] - def test_fetch_no_results_returns_error_string(self, mock_config_with_key): + @pytest.mark.anyio + async def test_fetch_no_results_returns_error_string(self, mock_config_with_key): patcher, _ = _patch_post(_make_search_response({"results": []})) try: from deerflow.community.groundroute.tools import web_fetch_tool - result = web_fetch_tool.invoke({"url": "https://ex.com"}) + result = await web_fetch_tool.ainvoke({"url": "https://ex.com"}) finally: patcher.stop()