"""DuckDuckGo web search tool for the research agents (free, no API key). Uses the maintained ``ddgs`` package. Degrades gracefully on rate limits / empty results so the writing step can still proceed from whatever was gathered. """ from __future__ import annotations import time from dataclasses import dataclass from typing import List @dataclass class SearchResult: title: str url: str snippet: str def as_markdown(self) -> str: return f"- [{self.title}]({self.url})\n {self.snippet}" def web_search(query: str, max_results: int = 5, *, retries: int = 2) -> List[SearchResult]: """Return up to ``max_results`` results for ``query``. Never raises.""" try: from ddgs import DDGS except ImportError: # pragma: no cover - older package name fallback try: from duckduckgo_search import DDGS # type: ignore except ImportError: return [] last_err: Exception | None = None for attempt in range(retries + 1): try: with DDGS() as ddgs: hits = ddgs.text(query, max_results=max_results) results = [ SearchResult( title=h.get("title", "").strip(), url=h.get("href") or h.get("url", ""), snippet=(h.get("body") or h.get("snippet") or "").strip(), ) for h in (hits or []) ] return [r for r in results if r.url] except Exception as exc: # rate limit / network hiccup last_err = exc if attempt < retries: time.sleep(1.5 * (attempt + 1)) # Give up quietly; caller handles empty findings. print(f"[search] '{query}' failed after {retries + 1} attempts: {last_err}") return []