""" Web search tool using SearXNG. """ from dataclasses import dataclass from datetime import datetime import httpx from src.domains.tools.base import BaseTool, ToolResult from src.shared.config import get_settings from src.shared.logging import logged, get_logger logger = get_logger(__name__) @dataclass class SearchResult: """A single search result.""" title: str url: str content: str engine: str published_date: str | None = None class WebSearchTool(BaseTool): """ Search the web using SearXNG metasearch engine. Returns relevant web results for queries about current events, documentation, or anything beyond the LLM's knowledge cutoff. """ name = "web_search" description = """Search the web for current information. Args: query: Search query string num_results: Maximum results to return (default: 5, max: 10) engines: Comma-separated engine list (optional, e.g., "google,brave,duckduckgo") categories: Search category (optional: "general", "images", "news", "science", "it") Returns: List of search results with title, URL, and snippet. Include a "Sources:" section with URLs in your response. Examples: - query="Python 3.12 new features" - Find latest Python docs - query="FastAPI best practices 2024" - Find recent tutorials - query="CVE-2024" categories="it" - Search IT/security news IMPORTANT: - Use this for current events, recent documentation, or facts you're unsure about - Always cite sources with URLs in your response - Today's date is {date} - use current year in queries for recent info """.format(date=datetime.now().strftime("%Y-%m-%d")) def __init__( self, searxng_url: str | None = None, timeout: int | None = None, max_results: int = 10, ): """ Initialize WebSearchTool. Args: searxng_url: SearXNG instance URL (default: from config) timeout: Request timeout in seconds (default: from config) max_results: Maximum results to return """ settings = get_settings() self.searxng_url = (searxng_url or settings.searxng_url).rstrip("/") self.timeout = timeout or settings.searxng_timeout self.max_results = max_results @logged() async def execute( self, query: str, num_results: int = 5, engines: str | None = None, categories: str | None = None, ) -> ToolResult: """ Execute web search. Args: query: Search query num_results: Number of results (1-10) engines: Specific engines to use categories: Search category Returns: ToolResult with search results """ if not query or not query.strip(): return self._error("Query cannot be empty") num_results = min(max(1, num_results), self.max_results) # Build SearXNG API request params = { "q": query.strip(), "format": "json", } if engines: params["engines"] = engines if categories: params["categories"] = categories try: async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.get( f"{self.searxng_url}/search", params=params, ) response.raise_for_status() data = response.json() except httpx.TimeoutException: return self._error(f"Search timed out after {self.timeout}s") except httpx.HTTPStatusError as e: return self._error(f"Search failed: HTTP {e.response.status_code}") except httpx.RequestError as e: return self._error(f"Search request failed: {e}") except Exception as e: logger.exception(f"Unexpected search error: {e}") return self._error(f"Search error: {e}") # Parse results raw_results = data.get("results", [])[:num_results] if not raw_results: return self._success( f"No results found for: {query}", result_count=0, query=query, ) # Format results for LLM consumption results = [] for r in raw_results: result = SearchResult( title=r.get("title", "Untitled"), url=r.get("url", ""), content=r.get("content", "No description"), engine=r.get("engine", "unknown"), published_date=r.get("publishedDate"), ) results.append(result) # Format as readable text output_lines = [f"Search results for: {query}", ""] for i, r in enumerate(results, 1): output_lines.append(f"{i}. **{r.title}**") output_lines.append(f" URL: {r.url}") output_lines.append(f" {r.content}") if r.published_date: output_lines.append(f" Published: {r.published_date}") output_lines.append("") return self._success( "\n".join(output_lines), result_count=len(results), query=query, engines_used=list({r.engine for r in results}), )