Add reproducible Docker and WireGuard host bootstrap

platform/web-search/web_search_mcp.py

@@ -31,6 +31,7 @@ SERVER_VERSION = "2.1.0"
 TINYSEARCH_CONTAINER = os.environ.get(
     "TINYSEARCH_CONTAINER", "mike-ai-web-search-tinysearch-1"
 )
+SEARXNG_URL = os.environ.get("SEARXNG_URL", "").rstrip("/")
 CHILD_TIMEOUT_SECONDS = float(os.environ.get("TINYSEARCH_CHILD_TIMEOUT", "110"))
 HTTP_TIMEOUT_SECONDS = float(os.environ.get("WEB_API_TIMEOUT", "18"))
 GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "").strip()
@@ -349,6 +350,31 @@ def api_json(url: str, service: str) -> Any:
     return decoded
 
 
+def searxng_json(query: str) -> Any:
+    """Query only the administrator-configured internal SearXNG endpoint.
+
+    Public API fetches intentionally reject private addresses.  SearXNG is the
+    one explicit internal exception; callers cannot influence its scheme,
+    authority or path, only the encoded search term.
+    """
+    base = urlparse(SEARXNG_URL)
+    if base.scheme not in {"http", "https"} or not base.hostname:
+        raise RuntimeError("invalid configured SearXNG URL")
+    url = f"{SEARXNG_URL}/search?" + urlencode({
+        "q": query, "format": "json", "language": "auto"})
+    try:
+        with urlopen(Request(url, headers={
+                "Accept": "application/json",
+                "User-Agent": f"mike-ai-web/{SERVER_VERSION}",
+        }), timeout=HTTP_TIMEOUT_SECONDS) as response:
+            payload = response.read(2_000_000)
+    except HTTPError as exc:
+        raise RuntimeError(f"searxng returned HTTP {exc.code}") from exc
+    except (URLError, TimeoutError) as exc:
+        raise RuntimeError("searxng unavailable") from exc
+    return json.loads(payload.decode("utf-8", errors="replace"))
+
+
 def infer_backend(query: str, requested: str = "auto") -> str:
     if requested not in {"auto", "web", "github", "huggingface"}:
         raise ValueError("backend must be auto, web, github or huggingface")
@@ -746,7 +772,19 @@ def general_discovery(query: str, limit: int) -> tuple[list[dict[str, Any]], lis
     results: list[dict[str, Any]] = []
     warnings: list[str] = []
     try:
-        results.extend(parse_search_xml(client().call("search", {"query": query}), limit))
+        if SEARXNG_URL:
+            data = searxng_json(query)
+            for row in (data.get("results") or [])[:limit]:
+                results.append({
+                    "title": clean_text(str(row.get("title", "")), 240),
+                    "url": str(row.get("url", "")),
+                    "preview": clean_text(str(row.get("content", "")), 500),
+                    "source": "searxng",
+                    "published_at": row.get("publishedDate") or row.get("published_date"),
+                })
+        else:
+            results.extend(parse_search_xml(
+                client().call("search", {"query": query}), limit))
     except Exception as exc:
         warnings.append(clean_text(str(exc), 240))
     if len(results) < limit and BRAVE_SEARCH_API_KEY: