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