rdnpradipta/knowledge-base-mcp

1"""MCP server exposing the knowledge base to Claude (claude.ai / Claude Desktop).2 3The server provides RETRIEVAL ONLY — Claude on the client side is the LLM and4must answer exclusively from these tool results (see `instructions` below).5 6Run modes:7    python mcp_server.py --transport stdio   # Claude Desktop (fully local)8    python mcp_server.py --transport http    # streamable-HTTP for claude.ai9                                              # (expose via: make tunnel)10 11Auth (http mode): the MCP endpoint is mounted under a secret path,12/<MCP_AUTH_TOKEN>/mcp — claude.ai custom connectors can't send custom headers13without full OAuth, so the URL itself is the credential. Keep it secret;14regenerate MCP_AUTH_TOKEN in .env to rotate.15"""16from __future__ import annotations17 18import argparse19import sys20 21from mcp.server.fastmcp import FastMCP22 23from rag.config import settings24 25INSTRUCTIONS = """This server is the ONLY authoritative source for answering user questions.26Rules for the assistant:27- Answer questions ONLY from search_knowledge_base / lookup_error_code results.28- Cite the source_file (and section/timestamp) for every claim.29- If the knowledge base returns no relevant results, tell the user the answer is30  not in the knowledge base. NEVER answer from general knowledge, the web, or31  any other source."""32 33 34def _build_server(host: str, port: int, path_token: str | None) -> FastMCP:35    kwargs: dict = {"host": host, "port": port}36    if path_token:37        kwargs["streamable_http_path"] = f"/{path_token}/mcp"38    mcp = FastMCP("knowledge-base", instructions=INSTRUCTIONS, **kwargs)39 40    @mcp.tool()41    def search_knowledge_base(query: str, top_k: int = 8) -> str:42        """Hybrid (semantic + keyword) search over the ingested knowledge base.43        ALWAYS call this before answering any user question. Answer ONLY from44        the returned chunks and cite their source_file/section."""45        from retrieval import hybrid_search46 47        chunks = hybrid_search(query, top_k=min(top_k, 20))48        if not chunks:49            return ("NO RESULTS. The knowledge base contains nothing relevant. "50                    "Tell the user this instead of answering from other knowledge.")51        blocks = []52        for i, c in enumerate(chunks, start=1):53            meta = c.source_file54            if c.section:55                meta += f" § {c.section}"56            if c.page_no:57                meta += f" p.{c.page_no}"58                if c.page_verified is False:   # VLM flagged: text not seen on this page59                    meta += " (page unverified)"60            if c.timestamp_ref:61                meta += f" @ {c.timestamp_ref}"62            if c.error_code:63                meta += f" [error_code={c.error_code}]"64            blocks.append(f"[{i}] ({meta}, score={c.score:.4f})\n{c.content}")65        return "\n\n".join(blocks)66 67    @mcp.tool()68    def lookup_error_code(code: str) -> str:69        """Exact lookup of a unit error code (e.g. "E628", "C901") in the70        knowledge base. Use for any error-code question. If it returns NOT FOUND,71        relay the suggestion to the user — do not invent a meaning for the code.72 73        This delegates to the deterministic Flow-A lookup against the relational74        `error_codes` table (the SAME path the local app uses), NOT the vector75        index — error codes are exact tokens, so the relational table is76        authoritative. Accepts free text; the code is extracted from it."""77        from error_lookup import format_row, lookup78 79        result = lookup(code)80        if result.message:           # ask-back / "did you mean?" / empty input81            return result.message82        return "\n\n---\n\n".join(83            f"{format_row(r)}\n\n— source: {r.source_file}" for r in result.rows84        )85 86    @mcp.tool()87    def list_sources() -> str:88        """List the documents in the knowledge base (what topics it can answer89        about), with chunk counts."""90        from rag.db import get_conn91 92        with get_conn(vector=False) as conn:93            rows = conn.execute(94                "SELECT source_file, source_type, count(*) FROM chunks GROUP BY 1, 2 ORDER BY 1"95            ).fetchall()96        if not rows:97            return "The knowledge base is empty — nothing has been ingested."98        return "\n".join(f"{r[0]} ({r[1]}): {r[2]} chunks" for r in rows)99 100    return mcp101 102 103if __name__ == "__main__":104    ap = argparse.ArgumentParser()105    ap.add_argument("--transport", choices=["stdio", "http"], default="stdio")106    ap.add_argument("--host", default="127.0.0.1")107    ap.add_argument("--port", type=int, default=settings.mcp_port)108    args = ap.parse_args()109 110    if args.transport == "http":111        if not settings.mcp_auth_token:112            sys.exit("MCP_AUTH_TOKEN is empty in .env — generate one first:\n"113                     "  python -c \"import secrets; print(secrets.token_hex(32))\"")114        server = _build_server(args.host, args.port, settings.mcp_auth_token)115        print(f"MCP endpoint: http://{args.host}:{args.port}/{settings.mcp_auth_token}/mcp")116        server.run(transport="streamable-http")117    else:118        _build_server(args.host, args.port, None).run(transport="stdio")119