SearxNG MCP Server

"""Client for interacting with SearxNG search API."""

import logging
from typing import Any

import httpx
from pydantic import AnyHttpUrl, BaseModel, Field

# Configure logging
logger = logging.getLogger(__name__)


class SearxngResult(BaseModel):
    """Model for a single SearxNG search result."""

    url: str = Field(..., alias="url")
    title: str = Field(..., alias="title")
    content: str = Field(..., alias="content")
    thumbnail: Any = Field(None, alias="thumbnail")
    engine: str = Field(..., alias="engine")
    template: str = Field(..., alias="template")
    parsed_url: list[str] = Field(..., alias="parsed_url")
    img_src: str = Field(..., alias="img_src")
    priority: str = Field(..., alias="priority")
    engines: list[str] = Field(..., alias="engines")
    positions: list[int] = Field(..., alias="positions")
    score: float = Field(..., alias="score")
    category: str = Field(..., alias="category")
    published_date: Any = Field(None, alias="publishedDate")


class SearxngResponse(BaseModel):
    """Model for SearxNG API response."""

    query: str
    number_of_results: int
    results: list[SearxngResult]


class SearxNGClient:
    """Client for interacting with SearxNG search API."""

    def __init__(self, base_url: AnyHttpUrl, timeout: int) -> None:
        """
        Initialize the SearxNG client.

        Args:
            base_url: URL of the SearxNG instance to use
            timeout: HTTP request timeout in seconds

        """
        self.base_url = str(base_url).rstrip("/")
        self.timeout = timeout

    async def search(
        self,
        query: str,
        categories: list[str] | None = None,
        language: str | None = None,
        time_range: str | None = None,
    ) -> SearxngResponse:
        """
        Perform a web search using the SearxNG API.

        Returns search results in a dictionary format containing the results
        and metadata such as number of total results found.

        Args:
            query: The search query string to look for on the web
            categories: Categories to search (e.g., "general", "images", "news")
            language: Language code for results (e.g., "all", "en", "ru")
            time_range: Time range (options: "day", "week", "month", "year")

        Returns:
            Dictionary containing search results and metadata

        """
        # Build search parameters
        params = {
            "q": query,
            "format": "json",
        }

        # Add optional parameters if provided
        if categories:
            params["categories"] = ",".join(categories)

        if time_range:
            params["time_range"] = time_range

        # Handle language parameter - always pass language
        if language:
            params["language"] = language

        try:
            # Make the API request
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    f"{self.base_url}/search",
                    params=params,
                    timeout=self.timeout,
                )

                # Raise an exception for HTTP errors
                response.raise_for_status()

                logger.info(f"Response: {response.text[:100]}...")

                # Parse the JSON response
                json_response = response.json()
                return SearxngResponse(**json_response)

        except Exception as e:
            logger.exception("Unexpected error occurred")
            msg = f"Error during search: {e}"
            raise ValueError(msg) from e

    def format_results(self, data: dict[str, Any]) -> str:
        """
        Format search results as a readable string.

        Args:
            data: Search results from the search() method

        Returns:
            Formatted string of search results

        """
        results = data.get("results", [])

        if not results:
            return "No results found."

        # Extract information about the search
        search_info = []
        if "query" in data:
            search_info.append(f"Query: {data['query']}")

        # Add information about the number of results
        if "number_of_results" in data and data["number_of_results"] > 0:
            search_info.append(
                f"Found approximately {data['number_of_results']} results"
            )

        # Format individual results
        formatted_results = []
        for i, result in enumerate(results, 1):
            title = result.get("title", "No title")
            url = result.get("url", "No URL")
            content = result.get("content", "No description")

            result_text = f"{i}. {title}\n   URL: {url}\n   {content}"

            # Add optional metadata if available
            if "publishedDate" in result:
                result_text += f"\n   Date: {result['publishedDate']}"
            if "score" in result:
                result_text += f"\n   Score: {result['score']:.2f}"

            formatted_results.append(result_text)

        # Combine search info and results
        search_info_text = "\n".join(search_info)
        results_text = "\n\n".join(formatted_results)

        return f"{search_info_text}\n\n{results_text}"