"""MCP server exposing the wiki knowledge base to Claude Code. Tools: wiki_search(query, k) hybrid BM25 + embedding search over the wiki wiki_get_page(title) full markdown of one mirrored page wiki_list_pages(prefix) enumerate mirrored page titles """ from __future__ import annotations import sys from typing import Any from . import config from .scraper import load_manifest, sanitize_filename from .search import search def wiki_search( query: str, k: int = config.DEFAULT_TOP_K, variant: str = config.DEFAULT_VARIANT ) -> list[dict[str, Any]]: """Search the wiki knowledge base (hybrid BM25 + embeddings). ``variant`` scopes results when the wiki covers two editions of one subject (see wikikb.variant); it excludes the other edition while always keeping pages that apply to both or carry no edition tag. Every hit reports its source and variant, so a result can never be silently attributed to the wrong wiki or edition. Returns ranked hits with title, source, variant, section, url and passage text. """ return [ { "title": r.title, "source": r.source, "variant": r.variant, "section": r.section, "url": r.url, "text": r.text, } for r in search(query, k=k, variant=variant) ] def wiki_get_page(title: str) -> str: """Return the full markdown of a mirrored wiki page by exact title.""" entry = load_manifest().get(title) filename = entry["filename"] if entry else sanitize_filename(title) path = config.CORPUS_DIR / filename if not path.exists(): return f"No mirrored page found for {title!r}. Try wiki_list_pages or wiki_search." return path.read_text(encoding="utf-8") def wiki_list_pages(prefix: str = "", limit: int = 100) -> list[str]: """List mirrored wiki page titles, filtered by case-insensitive prefix.""" titles = sorted(load_manifest()) if prefix: lowered = prefix.lower() titles = [t for t in titles if t.lower().startswith(lowered)] return titles[:limit] def build_server(server: Any = None) -> Any: """Construct (or reuse) a FastMCP server with the wiki tools registered.""" if server is None: # pragma: no cover - real server needs the mcp runtime from mcp.server.fastmcp import FastMCP server = FastMCP(config.MCP_NAMESPACE) server.tool()(wiki_search) server.tool()(wiki_get_page) server.tool()(wiki_list_pages) return server def main() -> int: # pragma: no cover - starts the stdio transport loop from . import cpu cpu.isolate() # queries are light, but still keep off the reserved game cores build_server().run() return 0 if __name__ == "__main__": sys.exit(main())