search

Instructions

Implementation Reference

• mcp_server/tools.py:144-291 (handler)

  The @mcp.tool decorated async function 'search' that executes the SearXNG web search. It accepts query, categories, engines, language, time_range, safesearch, pageno, pages, max_results, and format parameters, queries the SearXNG backend, caches results, and returns JSON output.

  @mcp.tool(
      annotations=ToolAnnotations(
          readOnlyHint=True,
          destructiveHint=False,
          idempotentHint=True,
          openWorldHint=True,
      )
  )
  async def search(
      query: Annotated[str, Field(
          description="The search query to use",
      )],
      categories: Annotated[str, Field(
          description=(
              "Comma-separated category names to focus on (e.g., 'general,news,science'). "
              "Prefer this over 'engines' to narrow results — categories leverage multiple engines automatically. "
              "Call engine_info to discover available categories."
          ),
      )] = "",
      engines: Annotated[str, Field(
          description=(
              "Comma-separated engine names to use (e.g., 'google,arxiv,wikipedia'). "
              "Only use when you need a specific source; otherwise prefer 'categories'. "
              "Overrides category-based engine selection when set."
          ),
      )] = "",
      language: Annotated[str, Field(
          description=(
              "Search language code (e.g., 'en', 'zh', 'ja', 'de'). "
              "Filters results to the specified language. Omit to search all languages."
          ),
      )] = "",
      time_range: Annotated[
          Literal["day", "week", "month", "year"] | None,
          Field(description="Restrict results to those published within this time window. Omit for no time restriction."),
      ] = None,
      safesearch: Annotated[
          Literal[0, 1, 2],
          Field(description="Safe search level: 0=off, 1=moderate, 2=strict"),
      ] = 0,
      pageno: Annotated[int, Field(
          ge=1,
          description="Starting page number. Use with 'pages' for pagination.",
      )] = 1,
      pages: Annotated[int, Field(
          ge=1, le=5,
          description=(
              "Number of pages to fetch in parallel (multi-page fanout). "
              "Higher values return more results but increase latency. Use 2-3 for comprehensive research."
          ),
      )] = 1,
      max_results: Annotated[int, Field(
          ge=1, le=100,
          description="Maximum number of results to return. Applied after aggregation across pages.",
      )] = 10,
      format: Annotated[
          Literal["compact", "full"],
          Field(description="Result detail level: 'compact' returns title/url/content only, 'full' includes engines/score/category/date/thumbnails"),
      ] = "compact",
  ) -> str:
      """Search the web using SearXNG metasearch engine.
  
      Aggregates results from 200+ search engines (Google, Bing, DuckDuckGo, Brave, etc.)
      with privacy. Returns results, answers, suggestions, corrections, and infoboxes.
      Use 'categories' to focus on specific content types. Use 'pages' for more results.
  
      Not suitable for autocomplete suggestions (use autocomplete tool) or discovering
      available engines/categories (use engine_info tool). Results are cached for 60 seconds.
      """
      fields = COMPACT_FIELDS if format == "compact" else FULL_FIELDS
  
      params: dict = {"q": query, "format": "json"}
      if categories:
          params["categories"] = categories
      if language:
          params["language"] = language
      if time_range is not None:
          params["time_range"] = time_range
      if safesearch:
          params["safesearch"] = str(safesearch)
      if engines:
          params["engines"] = engines
  
      cache_params = {**params, "pageno": pageno, "pages": pages, "format": format}
      cache_k = _cache_key(cache_params)
      cached = _get_cached(cache_k)
      if cached is not None:
          results = cached["results"][:max_results]
          output = {**cached, "results": results, "number_of_results": len(results), "cached": True}
          return json.dumps(output, ensure_ascii=False)
  
      all_results = []
      all_answers: set[str] = set()
      all_suggestions: set[str] = set()
      all_corrections: set[str] = set()
      all_infoboxes = []
      errors: list[str] = []
  
      client = await _get_client()
      tasks = []
      for page in range(pageno, pageno + pages):
          page_params = {**params, "pageno": str(page)}
          tasks.append(
              client.get(
                  f"{SEARXNG_BASE_URL}/search",
                  params=page_params,
                  timeout=30.0,
              )
          )
      responses = await asyncio.gather(*tasks, return_exceptions=True)
  
      for resp in responses:
          if isinstance(resp, Exception):
              errors.append(str(resp))
              continue
          if resp.status_code != 200:
              errors.append(f"HTTP {resp.status_code}")
              continue
          data = resp.json()
          all_results.extend(_trim_result(r, fields) for r in data.get("results", []))
          all_answers.update(data.get("answers", []))
          all_suggestions.update(data.get("suggestions", []))
          all_corrections.update(data.get("corrections", []))
          all_infoboxes.extend(data.get("infoboxes", []))
          for engine_name, error_msg in data.get("unresponsive_engines", []):
              errors.append(f"{engine_name}: {error_msg}")
  
      if not all_results and not all_answers:
          return json.dumps(_build_diagnostics(query, params, errors), ensure_ascii=False)
  
      output: dict = {
          "results": all_results,
          "number_of_results": len(all_results),
      }
      if all_answers:
          output["answers"] = list(all_answers)
      if all_suggestions:
          output["suggestions"] = list(all_suggestions)
      if all_corrections:
          output["corrections"] = list(all_corrections)
      if all_infoboxes:
          output["infoboxes"] = all_infoboxes
  
      _set_cache(cache_k, output)
  
      results = output["results"][:max_results]
      return_data = {**output, "results": results, "number_of_results": len(results)}
      return json.dumps(return_data, ensure_ascii=False)

• mcp_server/tools.py:152-203 (schema)

  Pydantic Field annotations defining the input schema for the 'search' tool, including query, categories, engines, language, time_range, safesearch, pageno, pages, max_results, and format.

  async def search(
      query: Annotated[str, Field(
          description="The search query to use",
      )],
      categories: Annotated[str, Field(
          description=(
              "Comma-separated category names to focus on (e.g., 'general,news,science'). "
              "Prefer this over 'engines' to narrow results — categories leverage multiple engines automatically. "
              "Call engine_info to discover available categories."
          ),
      )] = "",
      engines: Annotated[str, Field(
          description=(
              "Comma-separated engine names to use (e.g., 'google,arxiv,wikipedia'). "
              "Only use when you need a specific source; otherwise prefer 'categories'. "
              "Overrides category-based engine selection when set."
          ),
      )] = "",
      language: Annotated[str, Field(
          description=(
              "Search language code (e.g., 'en', 'zh', 'ja', 'de'). "
              "Filters results to the specified language. Omit to search all languages."
          ),
      )] = "",
      time_range: Annotated[
          Literal["day", "week", "month", "year"] | None,
          Field(description="Restrict results to those published within this time window. Omit for no time restriction."),
      ] = None,
      safesearch: Annotated[
          Literal[0, 1, 2],
          Field(description="Safe search level: 0=off, 1=moderate, 2=strict"),
      ] = 0,
      pageno: Annotated[int, Field(
          ge=1,
          description="Starting page number. Use with 'pages' for pagination.",
      )] = 1,
      pages: Annotated[int, Field(
          ge=1, le=5,
          description=(
              "Number of pages to fetch in parallel (multi-page fanout). "
              "Higher values return more results but increase latency. Use 2-3 for comprehensive research."
          ),
      )] = 1,
      max_results: Annotated[int, Field(
          ge=1, le=100,
          description="Maximum number of results to return. Applied after aggregation across pages.",
      )] = 10,
      format: Annotated[
          Literal["compact", "full"],
          Field(description="Result detail level: 'compact' returns title/url/content only, 'full' includes engines/score/category/date/thumbnails"),
      ] = "compact",
  ) -> str:

• mcp_server/tools.py:144-151 (registration)

  The @mcp.tool() decorator registers the 'search' function as an MCP tool with annotations marking it as read-only, non-destructive, idempotent, and open-world.

  @mcp.tool(
      annotations=ToolAnnotations(
          readOnlyHint=True,
          destructiveHint=False,
          idempotentHint=True,
          openWorldHint=True,
      )
  )

• mcp_server/tools.py:115-117 (helper)

  Helper function '_trim_result' that keeps only specified fields from a search result dict. Used by the search handler to filter results based on format (compact vs full).

  def _trim_result(result: dict, fields: list[str]) -> dict:
      """Keep only specified fields from a search result."""
      return {k: v for k, v in result.items() if k in fields and v}

• mcp_server/tools.py:120-141 (helper)

  Helper function '_build_diagnostics' that builds diagnostic info with suggestions when search returns zero results.

  def _build_diagnostics(query: str, params: dict, errors: list[str]) -> dict:
      """Build diagnostic info when search returns zero results."""
      tips = [
          "Try broader or different keywords",
          "Remove time_range filter if set",
          "Try different categories or engines",
      ]
      if params.get("language"):
          tips.append(f"Try removing language filter (currently: {params['language']})")
      if params.get("engines"):
          tips.append("Some specified engines may be unresponsive — try without engines filter")
      if params.get("time_range"):
          tips.append(f"Time range '{params['time_range']}' may be too restrictive")
  
      diag: dict = {
          "query": query,
          "message": "No results found",
          "suggestions": tips,
      }
      if errors:
          diag["errors"] = errors
      return diag