feat(community): migrate GroundRoute tools to httpx.AsyncClient (#7183)

* feat(community): migrate GroundRoute tools to async HTTP

The GroundRoute provider sent its search and fetch requests through a sync
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, and several providers queried in parallel queued
behind each other.

Make both tools coroutines over httpx.AsyncClient, resolve the API key and
the configured max_results in a single config read, and keep that read off
the event loop with asyncio.to_thread.

* refactor(community): drop the dead GroundRoute API key wrapper

The async migration resolved the key and max_results through
_resolve_tool_settings, which left _get_api_key as a second seam around
the same config read: only TestGetApiKey still exercised it. web_fetch
now reads its key through the same resolver, and the tests point at
_resolve_tool_settings.

Also record which httpx client factory the provider reaches for in the
runtime anchor. MockTransport runs its handler on the calling thread
without real blocking I/O, so a provider that stayed async but sent the
request through a sync client would satisfy the thread assertion; the
recorded factory is what catches that partial regression.

---------

Co-authored-by: treasuresyq <309743290+treasuresyq@users.noreply.github.com>
This commit is contained in:
treasuresyq 2026-10-21 23:08:27 +08:00 committed by GitHub
parent d649cc73d2
commit 8f9409bccc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 291 additions and 115 deletions

View File

@ -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

View File

@ -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}"

View File

@ -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"

View File

@ -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"

View File

@ -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()