Vibe coded web search/fetch MCP backed by SearXNG

sear-mcp.py

#!/usr/bin/env python3
"""SearXNG MCP server — exposes search and fetch tools to Claude Code."""

import re
from concurrent.futures import ThreadPoolExecutor

import keyring
import html2text
import niquests
import trafilatura
from mcp.server.fastmcp import FastMCP

SEARXNG_URL = "https://searxng.k8s.somemissing.info"
FETCH_MAX_CHARS = 50_000
FETCH_DEFAULT_CHARS = 15_000
FETCH_MAX_URLS = 8
CATEGORIES = {"general", "news", "it", "images", "videos", "science", "files", "q&a"}
TIME_RANGES = {"day", "week", "month", "year"}
HIGHLIGHT_MAX_CHARS = 2_000
SEARCH_FETCH_TIMEOUT = 8

mcp = FastMCP("searxng")


def _token() -> str:
    t = keyring.get_password("searxng", "bearer-token")
    if not t:
        raise RuntimeError(
            "Bearer token not found in keyring. "
            "Run: keyring set searxng bearer-token <token>"
        )
    return t


def _html_to_text(html: str) -> str:
    h = html2text.HTML2Text()
    h.ignore_links = False
    h.body_width = 0
    return h.handle(html)


def _extract_text(html: str) -> str:
    """Article body via trafilatura; html2text fallback for unparseable pages."""
    text = trafilatura.extract(html, include_links=False, favor_recall=True)
    return text if text else _html_to_text(html)


def _highlights(text: str, query: str, max_chars: int = HIGHLIGHT_MAX_CHARS) -> str:
    """Return the passages of `text` most relevant to `query` (keyword overlap,
    rare terms weighted higher), in document order."""
    terms = {t for t in re.findall(r"[a-z0-9_.-]+", query.lower()) if len(t) > 2}
    paras = [p.strip() for p in re.split(r"\n\s*\n", text) if len(p.strip()) > 40]
    if len(paras) < 3:
        # trafilatura separates blocks with single newlines; regroup lines
        # into ~paragraph-sized chunks
        paras, cur = [], ""
        for line in text.splitlines():
            cur = f"{cur}\n{line}" if cur else line
            if len(cur) > 300:
                paras.append(cur)
                cur = ""
        if cur.strip():
            paras.append(cur)
        paras = [p for p in paras if len(p.strip()) > 40]
    if not terms or not paras:
        return text[:max_chars]
    df = {t: sum(1 for p in paras if t in p.lower()) for t in terms}
    scored = []
    for i, p in enumerate(paras):
        low = p.lower()
        score = sum(len(paras) / df[t] for t in terms if df[t] and t in low)
        if score > 0:
            scored.append((score, i, p))
    if not scored:
        return text[:max_chars]
    picked, used = [], 0
    for _, i, p in sorted(scored, reverse=True):
        take = p[: max_chars - used]
        if len(take) < 80:
            continue
        picked.append((i, take))
        used += len(take)
        if used > max_chars * 0.8:
            break
    return "\n…\n".join(p for _, p in sorted(picked))


def _fetch_highlights(url: str, query: str) -> str:
    try:
        with niquests.Session() as s:
            r = s.get(url, timeout=SEARCH_FETCH_TIMEOUT)
        r.raise_for_status()
        body = r.text or ""
        if "text/html" not in r.headers.get("content-type", "text/html"):
            return body[:HIGHLIGHT_MAX_CHARS]
        return _highlights(_extract_text(body), query)
    except Exception as e:
        return f"(fetch failed: {e})"


@mcp.tool()
def search(
    query: str,
    max_results: int = 10,
    categories: str = "",
    time_range: str = "",
    fetch_top: int = 3,
) -> list[dict]:
    """Search the web via SearXNG. Returns title, url, content snippet, and engine
    per result, plus query-relevant "highlights" extracted from the top pages.

    Phrase queries as KEYWORDS, not descriptions — the engines are keyword-based.
    Quote exact error strings. ("kured 1.23.0 client-go compatibility", not
    "a page explaining which kubernetes versions kured 1.23.0 supports")

    categories: optional, one of general/news/it/images/videos/science/files/q&a —
        use "news" for current events, "it" for programming topics.
    time_range: optional, one of day/week/month/year to restrict result age.
    fetch_top: fetch the top N result pages in parallel and inline their most
        query-relevant passages as "highlights" (default 3; 0 to disable for
        faster link-only results).
    """
    if categories and categories not in CATEGORIES:
        return [{"error": f"invalid categories {categories!r}, expected one of {sorted(CATEGORIES)}"}]
    if time_range and time_range not in TIME_RANGES:
        return [{"error": f"invalid time_range {time_range!r}, expected one of {sorted(TIME_RANGES)}"}]
    params = {"q": query, "format": "json"}
    if categories:
        params["categories"] = categories
    if time_range:
        params["time_range"] = time_range
    try:
        with niquests.Session() as s:
            s.headers["Authorization"] = f"Bearer {_token()}"
            r = s.get(f"{SEARXNG_URL}/search", params=params, timeout=15)
        if r.status_code == 401:
            return [{"error": "Authentication failed — check bearer token in keyring"}]
        r.raise_for_status()
        data = r.json()
        results = [
            {
                "title": item.get("title", ""),
                "url": item.get("url", ""),
                "content": item.get("content", ""),
                "engine": item.get("engine", ""),
            }
            for item in data.get("results", [])
            if item.get("content")
        ][:max_results]
        if not results:
            hint = {"results": "none — try fewer or different keywords"}
            for k in ("suggestions", "corrections", "answers"):
                if data.get(k):
                    hint[k] = data[k]
            return [hint]
        top = results[: max(0, fetch_top)]
        if top:
            with ThreadPoolExecutor(max_workers=len(top)) as pool:
                futures = [pool.submit(_fetch_highlights, item["url"], query) for item in top]
                for item, fut in zip(top, futures):
                    item["highlights"] = fut.result()
        return results
    except (niquests.ConnectionError, niquests.Timeout):
        return [{"error": f"Could not reach {SEARXNG_URL} — is the cluster up?"}]
    except Exception as e:
        return [{"error": str(e)}]


def _fetch_text(url: str, max_chars: int, start_char: int) -> str:
    try:
        with niquests.Session() as s:
            r = s.get(url, timeout=15)
        r.raise_for_status()
        content_type = r.headers.get("content-type", "")
        body = r.text or ""
        # JSON and plain text must be returned whole — splitting either at an
        # arbitrary char boundary is semantically broken (mid-object JSON,
        # mid-function source code, mid-config YAML). Plain text capped at
        # 500 000 chars to guard against genuinely massive files.
        if "application/json" in content_type:
            return body
        if "text/plain" in content_type:
            if len(body) > 500_000:
                return body[:500_000] + "\n\n[file too large to return in full — truncated at 500 000 chars]"
            return body
        if "text/html" in content_type or not content_type:
            text = _extract_text(body)
        else:
            text = body
        end = start_char + max_chars
        clipped = text[start_char:end]
        if end < len(text):
            clipped += (
                f"\n\n[truncated: chars {start_char}–{end} of {len(text)};"
                f" pass start_char={end} to continue]"
            )
        return clipped
    except (niquests.ConnectionError, niquests.Timeout):
        return f"Error: could not connect to {url}"
    except niquests.HTTPError:
        return f"Error: HTTP {r.status_code} from {url}"
    except Exception as e:
        return f"Error: {e}"


@mcp.tool()
def fetch(
    urls: list[str] | str,
    max_chars: int = FETCH_DEFAULT_CHARS,
    start_char: int = 0,
) -> dict[str, str]:
    """Fetch one or more URLs (in parallel) and return their main content as
    readable plain text, keyed by URL.

    max_chars: per-URL character limit (default 15 000, max 50 000) — size it to
        what you need; truncated output ends with a note giving the start_char
        to pass for the next chunk.
    start_char: character offset to continue a previously truncated fetch.
    """
    if isinstance(urls, str):
        urls = [urls]
    urls = urls[:FETCH_MAX_URLS]
    if not urls:
        return {"error": "no URLs given"}
    max_chars = max(1, min(max_chars, FETCH_MAX_CHARS))
    start_char = max(0, start_char)
    with ThreadPoolExecutor(max_workers=len(urls)) as pool:
        texts = pool.map(lambda u: _fetch_text(u, max_chars, start_char), urls)
        return dict(zip(urls, texts))


if __name__ == "__main__":
    mcp.run()