client.pyā¢5.86 kB
"""SearXNG API client for making search requests."""
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin
import httpx
from pydantic import BaseModel, Field
class SearchResult(BaseModel):
"""A single search result from SearXNG."""
title: str
url: str
content: str = ""
engine: str = ""
category: str = ""
score: float = 0.0
thumbnail: Optional[str] = None
publishedDate: Optional[str] = None
class SearchResponse(BaseModel):
"""Response from SearXNG search API."""
query: str
number_of_results: int = Field(alias="number_of_results")
results: List[SearchResult]
suggestions: List[str] = []
answers: List[Any] = [] # Can be strings or dicts
infoboxes: List[Dict[str, Any]] = []
unresponsive_engines: List[str] = []
class SearXNGClient:
"""Client for interacting with SearXNG search API."""
def __init__(
self,
base_url: str,
timeout: float = 30.0,
verify_ssl: bool = True,
):
"""Initialize the SearXNG client.
Args:
base_url: Base URL of the SearXNG instance (e.g., "https://search.example.com")
timeout: Request timeout in seconds
verify_ssl: Whether to verify SSL certificates
"""
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.verify_ssl = verify_ssl
self._client: Optional[httpx.AsyncClient] = None
async def __aenter__(self) -> "SearXNGClient":
"""Async context manager entry."""
self._client = httpx.AsyncClient(
timeout=self.timeout,
verify=self.verify_ssl,
follow_redirects=True,
headers={
"User-Agent": "searxng-mcp-server/0.1.0 (https://github.com/martinchen448/searxng-mcp-server)",
"Accept": "application/json",
},
)
return self
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
"""Async context manager exit."""
if self._client:
await self._client.aclose()
@property
def client(self) -> httpx.AsyncClient:
"""Get the HTTP client instance."""
if self._client is None:
raise RuntimeError("Client not initialized. Use async context manager.")
return self._client
async def search(
self,
query: str,
categories: Optional[List[str]] = None,
engines: Optional[List[str]] = None,
language: str = "en",
page: int = 1,
time_range: Optional[str] = None,
safesearch: int = 0,
format: str = "json",
) -> SearchResponse:
"""Perform a search query on SearXNG.
Args:
query: The search query string
categories: List of categories to search (e.g., ["general", "images"])
engines: List of specific engines to use
language: Language code for results (default: "en")
page: Page number for pagination (default: 1)
time_range: Time range filter ("day", "month", "year")
safesearch: Safe search level (0=off, 1=moderate, 2=strict)
format: Response format (default: "json")
Returns:
SearchResponse containing search results
Raises:
httpx.HTTPError: If the request fails
"""
url = urljoin(self.base_url, "/search")
params: Dict[str, Any] = {
"q": query,
"format": format,
"language": language,
"pageno": page,
"safesearch": safesearch,
}
if categories:
params["categories"] = ",".join(categories)
if engines:
params["engines"] = ",".join(engines)
if time_range:
params["time_range"] = time_range
response = await self.client.get(url, params=params)
response.raise_for_status()
data = response.json()
return SearchResponse(**data)
async def get_suggestions(
self,
query: str,
language: str = "en",
) -> List[str]:
"""Get search suggestions for a query prefix.
Args:
query: The query prefix
language: Language code for suggestions
Returns:
List of suggestion strings
Raises:
httpx.HTTPError: If the request fails
"""
url = urljoin(self.base_url, "/autocomplete")
params = {
"q": query,
"language": language,
}
response = await self.client.get(url, params=params)
response.raise_for_status()
# Autocomplete returns a simple list
return response.json()
async def health_check(self) -> Dict[str, Any]:
"""Check the health status of the SearXNG instance.
Returns:
Dictionary containing health status information
Raises:
httpx.HTTPError: If the request fails
"""
url = urljoin(self.base_url, "/healthz")
try:
response = await self.client.get(url)
response.raise_for_status()
return {"status": "ok", "message": "SearXNG instance is healthy"}
except httpx.HTTPError as e:
return {
"status": "error",
"message": f"Health check failed: {str(e)}",
}
async def get_config(self) -> Dict[str, Any]:
"""Get the configuration of the SearXNG instance.
Returns:
Dictionary containing instance configuration
Raises:
httpx.HTTPError: If the request fails
"""
url = urljoin(self.base_url, "/config")
response = await self.client.get(url)
response.raise_for_status()
return response.json()