esh-pfi-infrastructure/services/searxng-mcp/src/searxng_mcp/server.py

"""MCP server exposing the fleet's SearXNG instance as a Claude Code tool.

Registered at USER scope so every Claude Code session on the box has it without
per-project setup. The instance lives on nh3-docker (10.100.50.40:9996) —
deliberately at NH3 rather than the colo, because search engines gate
datacenter IP ranges and the NH3 site egresses residentially.

⚠ WRITTEN AGAINST mcp 2.x. In 2.x `FastMCP` was renamed `MCPServer` and the
v1 decorator API (`@app.list_tools()` on `mcp.server.Server`) no longer exists —
it fails at import with `'Server' object has no attribute 'list_tools'`. If you
are copying an older MCP example, that is why it will not start. Pin `mcp<2`
only if you intend to keep v1 code.

⚠ WHY A ZERO-RESULT SEARCH RAISES RATHER THAN RETURNING AN EMPTY LIST.
On 2026-09-03 the previous instance was found returning zero results for every
query — for an unknown number of weeks — while its container reported `healthy`
and its dashboard card was green. SearXNG's /healthz proves the web app answers
and says nothing about whether any engine works. An empty result list is
indistinguishable from a broken aggregator, so zero results surface as a
failure with the engine errors attached, never as a quiet "no matches".
"""

from __future__ import annotations

import os

import httpx
from mcp.server.mcpserver import MCPServer

SEARXNG_URL = os.environ.get("SEARXNG_URL", "http://10.100.50.40:9996")
TIMEOUT = float(os.environ.get("SEARXNG_TIMEOUT", "45"))

mcp = MCPServer(
    name="searxng",
    instructions=(
        "Private web search via the PFI fleet's SearXNG metasearch instance. "
        "Use for current information, documentation lookups, and anything past "
        "the training cutoff."
    ),
)


@mcp.tool(
    name="web_search",
    description=(
        "Search the web through the PFI fleet's private SearXNG metasearch "
        "instance. Returns titles, URLs and snippets, with the engine that "
        "produced each result. Supports SearXNG bang syntax to target one "
        "engine, e.g. '!github asyncio' or '!stackoverflow mmap'."
    ),
)
async def web_search(
    query: str,
    categories: str | None = None,
    max_results: int = 10,
) -> str:
    """Search the web.

    Args:
        query: The search query. Supports bang syntax such as '!github ...'.
        categories: Optional comma-separated SearXNG categories, e.g.
            'general', 'it', 'science', 'news'.
        max_results: Maximum number of results to return.
    """
    query = (query or "").strip()
    if not query:
        raise ValueError("query must not be empty")

    params: dict[str, str] = {"q": query, "format": "json"}
    if categories:
        params["categories"] = categories

    try:
        async with httpx.AsyncClient(timeout=TIMEOUT) as client:
            resp = await client.get(f"{SEARXNG_URL}/search", params=params)
            resp.raise_for_status()
            data = resp.json()
    except httpx.HTTPError as exc:
        # An unreachable instance is an OUTAGE, not an absence of results, and
        # the two must never render the same way to the caller.
        raise RuntimeError(
            f"SearXNG at {SEARXNG_URL} could not be reached: {exc}. "
            "This is an outage, not an empty result set."
        ) from exc

    results = data.get("results") or []
    errors = data.get("unresponsive_engines") or []

    if not results:
        raise RuntimeError(
            f"SearXNG returned no results for {query!r}. Unresponsive engines: "
            f"{errors or 'none reported'}. Zero results across every engine "
            "usually means the instance is stale — SearXNG engine scrapers rot "
            "as upstream sites change their markup — rather than that the query "
            "has no matches. Check scripts/searxng-health.sh in eshpfi."
        )

    lines = [f"{len(results)} results for {query!r} via {SEARXNG_URL}"]
    if errors:
        lines.append(f"(engines that failed: {', '.join(e[0] for e in errors)})")
    lines.append("")
    for i, r in enumerate(results[:max_results], 1):
        lines.append(f"{i}. {r.get('title', '(untitled)')}")
        lines.append(f"   {r.get('url', '')}")
        if r.get("content"):
            lines.append(f"   {r['content'].strip()[:300]}")
        lines.append(f"   [engine: {r.get('engine', '?')}]")
        lines.append("")
    return "\n".join(lines)


def main() -> None:
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()