From b5a749224de4e4c2137c85a4b7d1dbb3cee7b301 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 22 Mar 2026 18:17:51 +0000 Subject: [PATCH] Unified search: Kiwix offline + DDG live with freshness detection Replaced three separate tools (search_docs, web_search, search) with a smart unified search() that: - Always hits Kiwix first (instant, offline, no rate limit) - Checks query for freshness keywords (latest, 2026, release, CVE, etc) - If time-sensitive: also hits DDG, flags "prefer live results" - If timeless (algorithms, docs, concepts): Kiwix only, skips web hit - If Kiwix returns nothing: falls back to DDG automatically Model sees both result sets with clear guidance on which to trust. Keeps read_doc() for reading full Kiwix articles and web_search() for explicit live-only queries when verifying offline currency. 130GB of ZIMs earn their keep on timeless topics (no rate limit, instant, complete articles). DDG covers everything else. https://claude.ai/code/session_01PtYTPherSJaxDEVPgF6Nxu --- mcp_server.py | 168 ++++++++++++++++++++++++++++++++++---------------- 1 file changed, 114 insertions(+), 54 deletions(-) diff --git a/mcp_server.py b/mcp_server.py index a35ff2b..ce5aca0 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -129,66 +129,132 @@ def git_checkout(branch: str, repo: str = "", create: bool = False) -> str: args = ["checkout", "-b", branch] if create else ["checkout", branch] return _git(args, repo) -# ── Search: Kiwix (offline docs) + DuckDuckGo (web) ───────────────────────── -@mcp.tool() -def search_docs(query: str, limit: int = 5) -> str: - """Search offline docs (Wikipedia, Stack Overflow, DevDocs, Arch Wiki) via Kiwix. - Returns article titles, snippets, and URLs. Always try this before web_search.""" - import re +# ── Search: unified (Kiwix offline + DuckDuckGo live) ─────────────────────── +# Kiwix ZIMs have complete, high-quality articles but may be months old. +# DDG has live results but lower signal-to-noise. The unified search tool +# checks both and lets the model see freshness info to judge which to trust. +# +# Heuristic: topics that change fast (releases, CVEs, "latest X") get flagged +# as potentially stale in offline results. Timeless topics (algorithms, language +# docs, math) are fine from Kiwix and skip the web hit entirely. + +import re as _re +from datetime import datetime as _dt + +# Words that suggest the query needs fresh data +_FRESH_KEYWORDS = _re.compile( + r'\b(latest|newest|recent|2025|2026|update|release|version|changelog|CVE|vulnerability|' + r'breaking change|deprecat|current|today|this year|this month|announce|just released)\b', + _re.IGNORECASE +) + +def _kiwix_search(query: str, limit: int = 5) -> list[dict]: + """Search Kiwix, return list of {title, snippet, path, source}.""" try: - # Kiwix full-text search returns HTML — parse the results - r = httpx.get(f"{KIWIX_URL}/search", params={"pattern": query, "pageLength": limit}, + r = httpx.get(f"{KIWIX_URL}/search", + params={"pattern": query, "pageLength": limit}, timeout=15, follow_redirects=True) if r.status_code != 200: - return f"Kiwix returned {r.status_code}. Is kiwix running with ZIM files loaded?" + return [] html = r.text results = [] - # Parse search result entries from Kiwix HTML - # Kiwix wraps results in
tags or links with snippets - articles = re.findall( + # Try structured parse first + articles = _re.findall( r']+href="(/[^"]+)"[^>]*>\s*]*>([^<]*).*?' r'(?:]*>([^<]*))?.*?' r'(?:]*>(.*?)

)?', - html, re.DOTALL + html, _re.DOTALL ) - if not articles: - # Fallback: grab any links with text from the results - articles = re.findall(r']+href="(/[^"]+)"[^>]*>([^<]+)', html) - for path, title in articles[:limit]: - results.append(f"**{title.strip()}**\n URL: {KIWIX_URL}{path}\n") - else: + if articles: for path, title, cite, snippet in articles[:limit]: - snippet_clean = re.sub(r'<[^>]+>', '', snippet or '').strip() - entry = f"**{title.strip()}**" - if cite: - entry += f" ({cite.strip()})" - if snippet_clean: - entry += f"\n {snippet_clean[:300]}" - entry += f"\n URL: {KIWIX_URL}{path}" - results.append(entry) - if not results: - return f"No results for '{query}' in offline docs. Try web_search instead." - return "\n\n".join(results) - except httpx.ConnectError: - return "Kiwix not reachable. Is the kiwix container running with ZIM files?" - except Exception as e: - return f"Kiwix search error: {e}" + snippet_clean = _re.sub(r'<[^>]+>', '', snippet or '').strip()[:300] + results.append({"title": title.strip(), "snippet": snippet_clean, + "path": path, "source": cite.strip() if cite else "kiwix"}) + else: + # Fallback: grab any links + for path, title in _re.findall(r']+href="(/[^"]+)"[^>]*>([^<]+)', html)[:limit]: + results.append({"title": title.strip(), "snippet": "", + "path": path, "source": "kiwix"}) + return results + except Exception: + return [] + +def _ddg_search(query: str, limit: int = 5) -> list[dict]: + """Search DuckDuckGo, return list of {title, snippet, url}.""" + try: + from duckduckgo_search import DDGS + results = [] + with DDGS() as ddgs: + for r in ddgs.text(query, max_results=limit): + results.append({"title": r["title"], "snippet": r["body"], "url": r["href"]}) + return results + except Exception: + return [] + +@mcp.tool() +def search(query: str, limit: int = 5) -> str: + """Unified search: checks offline docs (Kiwix) AND live web (DuckDuckGo). + Returns results from both with freshness guidance. + For timeless topics (algorithms, docs): offline results are sufficient. + For time-sensitive topics (releases, CVEs): live results are flagged as preferred.""" + needs_fresh = bool(_FRESH_KEYWORDS.search(query)) + output_parts = [] + + # Always search Kiwix (fast, local) + kiwix_results = _kiwix_search(query, limit) + if kiwix_results: + header = "## Offline Docs (Kiwix)" + if needs_fresh: + header += " ⚠️ POSSIBLY STALE — query looks time-sensitive, prefer live results below" + output_parts.append(header) + for i, r in enumerate(kiwix_results, 1): + entry = f"{i}. **{r['title']}**" + if r["source"] and r["source"] != "kiwix": + entry += f" ({r['source']})" + if r["snippet"]: + entry += f"\n {r['snippet']}" + entry += f"\n → read_doc('{r['path']}')" + output_parts.append(entry) + + # Search DDG if: query needs fresh data, OR Kiwix returned nothing, OR always (to compare) + do_web = needs_fresh or not kiwix_results + ddg_results = [] + if do_web: + ddg_results = _ddg_search(query, limit) + + if ddg_results: + header = "## Live Web (DuckDuckGo)" + if needs_fresh: + header += " ✓ PREFER THESE for this query" + output_parts.append(header) + for i, r in enumerate(ddg_results, 1): + output_parts.append(f"{i}. **{r['title']}**\n {r['snippet']}\n {r['url']}") + elif do_web: + output_parts.append("## Live Web (DuckDuckGo)\n(no results or DDG unreachable)") + + if not kiwix_results and not ddg_results: + return f"No results for '{query}' from either offline docs or web search." + + # Freshness note + if kiwix_results and not needs_fresh and not ddg_results: + output_parts.append("\n_Offline results look sufficient for this topic. " + "Use web_search() if you need to verify currency._") + + return "\n\n".join(output_parts) @mcp.tool() def read_doc(path: str) -> str: - """Read a full article from Kiwix by its path (from search_docs results). + """Read a full article from Kiwix by its path (from search results). Example: read_doc('/wikipedia_en_all/A/Python_(programming_language)')""" try: r = httpx.get(f"{KIWIX_URL}{path}", timeout=15, follow_redirects=True) if r.status_code != 200: return f"Not found: {path} (HTTP {r.status_code})" - import re # Strip HTML tags, keep text content - text = re.sub(r']*>.*?', '', r.text, flags=re.DOTALL) - text = re.sub(r']*>.*?', '', text, flags=re.DOTALL) - text = re.sub(r'<[^>]+>', ' ', text) - text = re.sub(r'\s+', ' ', text).strip() - # Truncate to ~8K chars to fit in model context + text = _re.sub(r']*>.*?', '', r.text, flags=_re.DOTALL) + text = _re.sub(r']*>.*?', '', text, flags=_re.DOTALL) + text = _re.sub(r'<[^>]+>', ' ', text) + text = _re.sub(r'\s+', ' ', text).strip() if len(text) > 8000: text = text[:8000] + "\n\n[... truncated — article continues ...]" return text @@ -197,19 +263,13 @@ def read_doc(path: str) -> str: @mcp.tool() def web_search(query: str, num_results: int = 5) -> str: - """Search the live web via DuckDuckGo. No API key needed. - Use search_docs first for programming/wiki topics — it's faster and offline.""" - try: - from duckduckgo_search import DDGS - results = [] - with DDGS() as ddgs: - for r in ddgs.text(query, max_results=num_results): - results.append(f"**{r['title']}**\n {r['body']}\n {r['href']}") - return "\n\n".join(results) if results else f"No web results for '{query}'" - except ImportError: - return "duckduckgo-search not installed. Add it to mcp_requirements.txt." - except Exception as e: - return f"Web search error: {e}" + """Search ONLY the live web via DuckDuckGo. Use search() instead for most queries — + it checks both offline and live. Use this directly only when you specifically need + live-only results (e.g., verifying if offline info is current).""" + results = _ddg_search(query, num_results) + if not results: + return f"No web results for '{query}'" + return "\n\n".join(f"**{r['title']}**\n {r['snippet']}\n {r['url']}" for r in results) # ── Gitea API ───────────────────────────────────────────────────────────────── def _gitea(method: str, path: str, body: dict = {}) -> dict: