feat(community): offload image search to a thread via asyncio.to_thread (#5788)

* feat(community): offload image search to a thread via asyncio.to_thread

Make image_search_tool async and run the blocking ddgs call in a worker
thread with asyncio.to_thread, continuing the TODO.md planned feature
('Replace sync requests with httpx.AsyncClient in community tools').

The ddgs SDK exposes no async API, so instead of swapping the HTTP
client the blocking DDGS.images() call is offloaded to a thread — the
same pattern the infoquest/jina_ai tools use for readability
extraction. Callers see no change in behavior.

Tests invoke the async tool via ainvoke under @pytest.mark.anyio.

* fix(image-search): offload config lookup with DDGS search

* test(image-search): pin DDGS search off the event loop

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/community/image_search/tools.py

@@ -2,6 +2,7 @@
 Image Search Tool - Search images using DuckDuckGo for reference in image generation.
 """
 
+import asyncio
 import json
 import logging
 
@@ -93,7 +94,7 @@ def _search_images(
 
 
 @tool("image_search", parse_docstring=True)
-def image_search_tool(
+async def image_search_tool(
     query: str,
     max_results: int = 5,
     size: str | None = None,
@@ -124,22 +125,26 @@ def image_search_tool(
         license_image: License filter. Options: "any", "Public", "Share", "ShareCommercially", "Modify", "ModifyCommercially".
             Use this when the reference image will be redistributed, so the results are already license-cleared.
     """
-    config = get_app_config().get_tool_config("image_search")
 
-    # Override max_results from config if set
-    if config is not None and "max_results" in config.model_extra:
-        max_results = config.model_extra.get("max_results", max_results)
-    max_results = _coerce_max_results(max_results)
+    def search_with_config() -> list[dict]:
+        # Both config loading and DDGS perform blocking I/O. Keep the entire
+        # search setup in one worker so neither can stall the agent event loop.
+        config = get_app_config().get_tool_config("image_search")
+        resolved_max_results = max_results
+        if config is not None and "max_results" in config.model_extra:
+            resolved_max_results = config.model_extra.get("max_results", resolved_max_results)
 
-    results = _search_images(
-        query=query,
-        max_results=max_results,
-        size=size,
-        color=color,
-        type_image=type_image,
-        layout=layout,
-        license_image=license_image,
-    )
+        return _search_images(
+            query=query,
+            max_results=_coerce_max_results(resolved_max_results),
+            size=size,
+            color=color,
+            type_image=type_image,
+            layout=layout,
+            license_image=license_image,
+        )
+
+    results = await asyncio.to_thread(search_with_config)
 
     if not results:
         return json.dumps({"error": "No images found", "query": query}, ensure_ascii=False)

backend/tests/blocking_io/test_image_search.py

@@ -0,0 +1,35 @@
+"""Image search must keep config file reads off the agent event loop."""
+
+import threading
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from deerflow.community.image_search import tools
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_image_search_config_lookup_does_not_block_loop(monkeypatch) -> None:
+    observed_threads: list[threading.Thread] = []
+
+    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")
+        return SimpleNamespace(get_tool_config=lambda _name: None)
+
+    def probe_search(**_kwargs):
+        Path(__file__).read_text(encoding="utf-8")
+        observed_threads.append(threading.current_thread())
+        return []
+
+    monkeypatch.setattr(tools, "get_app_config", load_config)
+    monkeypatch.setattr(tools, "_search_images", probe_search)
+
+    result = await tools.image_search_tool.ainvoke({"query": "a cat"})
+
+    assert '"error": "No images found"' in result
+    assert observed_threads, "search must run"
+    assert all(thread is not threading.main_thread() for thread in observed_threads)

backend/tests/test_image_search.py

@@ -9,7 +9,8 @@ from deerflow.community.image_search import tools
 from deerflow.community.image_search.tools import image_search_tool
 
 
-def test_image_search_uses_full_image_url_not_thumbnail():
+@pytest.mark.anyio
+async def test_image_search_uses_full_image_url_not_thumbnail():
     # Regression: `image_url` must expose the full-resolution `image` from the DDGS result,
     # not the low-res `thumbnail` (both fields were previously set to `thumbnail`).
     fake_results = [
@@ -26,14 +27,14 @@ def test_image_search_uses_full_image_url_not_thumbnail():
         patch("deerflow.community.image_search.tools._search_images", return_value=fake_results),
         patch("deerflow.community.image_search.tools.get_app_config", return_value=cfg),
     ):
-        output = json.loads(image_search_tool.invoke({"query": "a cat"}))
+        output = json.loads(await image_search_tool.ainvoke({"query": "a cat"}))
 
     result = output["results"][0]
     assert result["image_url"] == "https://example.com/full.jpg"
     assert result["thumbnail_url"] == "https://example.com/thumb.jpg"
 
 
-def _search_kwargs(query="a cat", **kwargs):
+async def _search_kwargs(query="a cat", **kwargs):
     """Invoke image_search_tool and return the kwargs handed to _search_images."""
     fake_results = [{"title": "a cat", "image": "https://example.com/full.jpg", "thumbnail": "https://example.com/thumb.jpg"}]
     cfg = MagicMock()
@@ -43,12 +44,13 @@ def _search_kwargs(query="a cat", **kwargs):
         patch("deerflow.community.image_search.tools._search_images", return_value=fake_results) as mock_search,
         patch("deerflow.community.image_search.tools.get_app_config", return_value=cfg),
     ):
-        image_search_tool.invoke({"query": query, **kwargs})
+        await image_search_tool.ainvoke({"query": query, **kwargs})
 
     return mock_search.call_args.kwargs
 
 
-def test_image_search_passes_color_and_license_filters():
+@pytest.mark.anyio
+async def test_image_search_passes_color_and_license_filters():
     """`color` / `license_image` reach the DDGS call.
 
     `_search_images` accepts both and forwards them into the `f` filter payload
@@ -56,19 +58,20 @@ def test_image_search_passes_color_and_license_filters():
     passed them, so they were unreachable: DDGS supports them and the provider
     already declared support for them, yet no caller could set one.
     """
-    kwargs = _search_kwargs(color="Monochrome", license_image="ModifyCommercially")
+    kwargs = await _search_kwargs(color="Monochrome", license_image="ModifyCommercially")
 
     assert kwargs["color"] == "Monochrome"
     assert kwargs["license_image"] == "ModifyCommercially"
 
 
-def test_image_search_omits_unset_filters():
+@pytest.mark.anyio
+async def test_image_search_omits_unset_filters():
     """An unset filter is absent/None, so _search_images leaves it out of the DDGS call.
 
     `_search_images` only forwards truthy filters, so this pins that the default
     request is unchanged by the newly exposed parameters.
     """
-    kwargs = _search_kwargs(size="Large", type_image="photo", layout="Square")
+    kwargs = await _search_kwargs(size="Large", type_image="photo", layout="Square")
 
     assert kwargs["size"] == "Large"
     assert kwargs["type_image"] == "photo"
@@ -77,6 +80,7 @@ def test_image_search_omits_unset_filters():
     assert kwargs.get("license_image") is None
 
 
+@pytest.mark.anyio
 @pytest.mark.parametrize(
     ("configured_limit", "env_value", "call_args", "expected_count", "warns"),
     [
@@ -95,7 +99,7 @@ def test_image_search_omits_unset_filters():
         pytest.param(None, "3", {"max_results": 0}, 5, True, id="zero-call-argument"),
     ],
 )
-def test_image_search_max_results_with_real_ddgs(monkeypatch, caplog, configured_limit, env_value, call_args, expected_count, warns):
+async def test_image_search_max_results_with_real_ddgs(monkeypatch, caplog, configured_limit, env_value, call_args, expected_count, warns):
     """Exercise DDGS image count arithmetic and slicing without contacting search engines.
 
     Environment substitution keeps `max_results: $VAR` as a string; DDGS then raised
@@ -131,7 +135,7 @@ def test_image_search_max_results_with_real_ddgs(monkeypatch, caplog, configured
     monkeypatch.setattr(DDGS, "_get_network_client", lambda self: None)
     monkeypatch.setattr(DDGS, "_get_engines", lambda self, category, backend: [engine])
 
-    parsed = json.loads(image_search_tool.invoke({"query": "Image", **call_args}))
+    parsed = json.loads(await image_search_tool.ainvoke({"query": "Image", **call_args}))
 
     assert "error" not in parsed
     assert parsed["total_results"] == len(parsed["results"]) == expected_count