webber/webber-api/src/domains/tools/search/web.py

"""
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}),
        )