SearXNG MCP Server

Instructions

Implementation Reference

• searxng_mcp/searxng_mcp.py:105-239 (handler)

  Implementation of the 'web_search' tool handler. This async function is decorated with @mcp.tool(), defining the input schema via Pydantic Fields and executing the search logic by querying a SearXNG instance, handling authentication, progress reporting, error handling, and returning formatted results.
  @mcp.tool( annotations={ "title": "SearXNG Search", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False, }, tags={"search"}, ) async def web_search( query: str = Field(description="Search query", default=None), language: str = Field( description="Language code for search results (e.g., 'en', 'de', 'fr'). Default: 'en'", default="en", ), time_range: Optional[str] = Field( description="Time range for search results. Options: 'day', 'week', 'month', 'year'. Default: null (no time restriction).", default=None, ), categories: Optional[List[str]] = Field( description="Categories to search in (e.g., 'general', 'images', 'news'). Default: null (all categories).", default=None, ), engines: Optional[List[str]] = Field( description="Specific search engines to use. Default: null (all available engines).", default=None, ), safesearch: int = Field( description="Safe search level: 0 (off), 1 (moderate), 2 (strict). Default: 1 (moderate).", default=1, ), pageno: int = Field( description="Page number for results. Must be minimum 1. Default: 1.", default=1, ge=1, ), max_results: int = Field( description="Maximum number of search results to return. Range: 1-50. Default: 10.", default=10, ge=1, le=50, ), ctx: Context = Field( description="MCP context for progress reporting.", default=None ), ) -> Dict[str, Any]: """ Perform web searches using SearXNG, a privacy-respecting metasearch engine. Returns relevant web content with customizable parameters. Returns a Dictionary response with status, message, data (search results), and error if any. """ logger = logging.getLogger("SearXNG") logger.debug(f"[SearXNG] Searching for: {query}") try: if not query: return { "status": 400, "message": "Invalid input: query must not be empty", "data": None, "error": "query must not be empty", } # Prepare search parameters search_params = { "q": query, "format": "json", "language": language, "safesearch": safesearch, "pageno": pageno, } if time_range: search_params["time_range"] = time_range if categories: search_params["categories"] = ",".join(categories) if engines: search_params["engines"] = ",".join(engines) # Report initial progress if ctx is available if ctx: await ctx.report_progress(progress=0, total=100) logger.debug("Reported initial progress: 0/100") # Make request to SearXNG auth = (SEARXNG_USERNAME, SEARXNG_PASSWORD) if HAS_BASIC_AUTH else None response = requests.get( f"{SEARXNG_INSTANCE_URL}/search", params=search_params, auth=auth ) response.raise_for_status() search_response: Dict[str, Any] = response.json() # Limit results limited_results = search_response.get("results", [])[:max_results] # Construct final response final_response = { **search_response, "results": limited_results, "number_of_results": len(limited_results), } # Report completion if ctx: await ctx.report_progress(progress=100, total=100) logger.debug("Reported final progress: 100/100") logger.debug(f"[SearXNG] Search completed for query: {query}") return { "status": 200, "message": "Search completed successfully", "data": final_response, "error": None, } except requests.exceptions.HTTPError as e: status_code = e.response.status_code if e.response else None if status_code == 401: error_msg = "Authentication failed. Please check your SearXNG username and password." else: error_msg = f"SearXNG API error: {e.response.json().get('message', str(e)) if e.response else str(e)}" logger.error(f"[SearXNG Error] {error_msg}") return { "status": status_code or 500, "message": "Failed to perform search", "data": None, "error": error_msg, } except Exception as e: logger.error(f"[SearXNG Error] {str(e)}") return { "status": 500, "message": "Failed to perform search", "data": None, "error": str(e), }