feat: add web search tool using SearXNG

Add WebSearchTool that queries the self-hosted SearXNG metasearch engine
for current information, documentation, and facts beyond training data.

- Add SEARXNG_URL and SEARXNG_TIMEOUT config settings
- Create WebSearchTool with query, num_results, categories params
- Register web_search tool with explore agent
- Add 10 tests for search functionality

Usage: Agents can now use web_search(query="...") to find current info.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-11 14:09:21 +01:00
co-authored by Claude Opus 4.5
parent f6256363a2
commit 82a816a5b5
6 changed files with 434 additions and 2 deletions
+169
View File
@@ -0,0 +1,169 @@
"""
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}),
)