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
This commit is contained in:
Claude
+114
-54
@@ -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 <article> tags or <a> links with snippets
|
||||
articles = re.findall(
|
||||
# Try structured parse first
|
||||
articles = _re.findall(
|
||||
r'<a[^>]+href="(/[^"]+)"[^>]*>\s*<span[^>]*>([^<]*)</span>.*?'
|
||||
r'(?:<cite[^>]*>([^<]*)</cite>)?.*?'
|
||||
r'(?:<p[^>]*>(.*?)</p>)?',
|
||||
html, re.DOTALL
|
||||
html, _re.DOTALL
|
||||
)
|
||||
if not articles:
|
||||
# Fallback: grab any links with text from the results
|
||||
articles = re.findall(r'<a[^>]+href="(/[^"]+)"[^>]*>([^<]+)</a>', 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'<a[^>]+href="(/[^"]+)"[^>]*>([^<]+)</a>', 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'<script[^>]*>.*?</script>', '', r.text, flags=re.DOTALL)
|
||||
text = re.sub(r'<style[^>]*>.*?</style>', '', 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'<script[^>]*>.*?</script>', '', r.text, flags=_re.DOTALL)
|
||||
text = _re.sub(r'<style[^>]*>.*?</style>', '', 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:
|
||||
|
||||
Reference in New Issue
Block a user