让 Codex 和 Cursor 读写你的知识库：用 FastMCP 搓一个本地 MCP Server

kb/
├── docs/                  # KB_ROOT：唯一事实源，全是 .md
│   ├── conf.py            # Sphinx 配置
│   ├── index.md
│   ├── decision/          # 决策与取舍（为什么这么定）
│   ├── runbook/           # 操作手册（怎么干）
│   ├── domain/            # 业务与领域知识（只有我们知道的事）
│   └── reading/           # 读书与论文笔记
├── _build/                # make html 的产物，可重建，不进 git
├── Makefile               # make html / make clean
└── AGENTS.md              # 告诉 agent：查库请走 kb_* 工具

extensions = ["myst_parser", "sphinxcontrib.mermaid", "sphinxcontrib.plantuml"]

---
title: 为什么我们的重试上限定在 3 次
type: decision
status: current
updated: 2026-10-13
---
# 为什么我们的重试上限定在 3 次

下游 P99 已经 800ms，3 次重试叠加退避，最坏情况把调用方拖到超时线之外……

```mermaid
flowchart LR
    A[调用失败] --> B{第几次?}
    B -- "≤3" --> C[退避后重试]
    B -- ">3" --> D[熔断]

两个要点，都是老规矩：**Markdown 源文件是唯一事实源，`_build` 删了能一条命令长回来**——这套分层我在[知识库工作流](./2026-08-31-knowledge-base-workflow.html)那篇里详细写过；**`make html` 就是知识库的编译器**，Mermaid 代码块写坏了、内链断了，build 的时候全会暴露。这个性质后面有用。

## 三、柜台设计：五个工具，少而正交

| 工具 | 输入 | 输出 | 护栏 |
|---|---|---|---|
| `kb_catalog` | 无 | 全部文档的路径 + frontmatter 摘要 | 只读 |
| `kb_search` | 正则、条数上限 | `文件:行号: 摘要` | 只读，上限 50 条 |
| `kb_read` | 相对路径 | 整篇原文 | 路径白名单，只 `.md` |
| `kb_write` | 路径、内容、事由 | 成功或明确的拒绝理由 | 校验 frontmatter，原子写，留审计，不删 |
| `code_graph_query` | 自然语言问题 | 代码知识图谱的查询结果 | 只调本地 CLI，30 秒超时 |

设计上就两条原则：

**工具要少而正交。** 每加一个工具，agent 的选择空间就大一分，选错的概率也大一分。五个工具对应五个动作：看目录、找、读、写、查代码图，没有第七个动作的位置。

**返回必须带路径。** 搜索结果的每一条都落在具体文件的具体行上，读出来的内容自带出处。**不给路径的答案不算答案**——这一条能把大部分幻觉挡在门口，因为 agent 接下来要引用时，必须先去读原文。

![知识库 MCP 柜台架构](../images/tech_20261013_mcp_kb_architecture.png)

## 四、开张：一百多行代码

先装 FastMCP（Python 3.10+）：

```console
$ pip install fastmcp

# kb_mcp_server.py —— 知识库柜台：catalog / search / read / write / 代码图查询
import json
import os
import re
import subprocess
from datetime import datetime, timezone
from pathlib import Path

from fastmcp import FastMCP

KB_ROOT = Path(os.environ.get("KB_ROOT", "~/kb/docs")).expanduser().resolve()
CODE_ROOT = Path(os.environ.get("CODE_ROOT", KB_ROOT)).expanduser().resolve()
AUDIT_LOG = KB_ROOT.parent / ".kb_audit.jsonl"
ALLOWED_SUFFIX = ".md"
ALLOWED_GLOB = "*.md"
MAX_WRITE_BYTES = 200_000
VALID_STATUS = ("current", "draft", "superseded")
REQUIRED_META = ("title", "type", "status", "updated")

FRONTMATTER_RE = re.compile(r"\A---\s*\n(.*?)\n---\s*\n", re.DOTALL)

mcp = FastMCP("kb")


def safe_path(rel: str) -> Path:
    """把 agent 给的路径关进 KB_ROOT，出界或非 .md 一律拒绝。"""
    p = (KB_ROOT / rel).resolve()
    if not p.is_relative_to(KB_ROOT):
        raise ValueError(f"path escapes KB_ROOT: {rel}")
    if p.suffix != ALLOWED_SUFFIX:
        raise ValueError(f"only .md is served, got: {rel}")
    return p


def parse_frontmatter(text: str) -> dict:
    meta = {}
    m = FRONTMATTER_RE.match(text)
    if m:
        for line in m.group(1).splitlines():
            k, sep, v = line.partition(":")
            if sep:
                meta[k.strip()] = v.strip()
    return meta


def audit(action: str, path: str, extra: dict | None = None) -> None:
    entry = {"ts": datetime.now(timezone.utc).isoformat(), "action": action,
             "path": path, **(extra or {})}
    with open(AUDIT_LOG, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")


@mcp.tool
def kb_catalog() -> str:
    """列出知识库全部文档的路径和 frontmatter 摘要。先看这个，再决定读哪篇。"""
    items = []
    for p in sorted(KB_ROOT.rglob(ALLOWED_GLOB)):
        rel = p.relative_to(KB_ROOT)
        if any(part.startswith(".") for part in rel.parts):
            continue
        meta = parse_frontmatter(p.read_text(encoding="utf-8"))
        items.append({
            "path": str(rel),
            "title": meta.get("title", ""),
            "type": meta.get("type", ""),
            "status": meta.get("status", ""),
            "updated": meta.get("updated", ""),
        })
    return json.dumps(items, ensure_ascii=False, indent=1)


@mcp.tool
def kb_search(query: str, limit: int = 10) -> str:
    """按正则搜索全部 .md 文档，返回 "文件:行号: 摘要"。"""
    limit = max(1, min(limit, 50))
    rx = re.compile(query, re.IGNORECASE)
    hits = []
    for p in sorted(KB_ROOT.rglob(ALLOWED_GLOB)):
        for n, line in enumerate(p.read_text(encoding="utf-8").splitlines(), 1):
            if rx.search(line):
                hits.append(f"{p.relative_to(KB_ROOT)}:{n}: {line.strip()[:120]}")
                if len(hits) >= limit:
                    return "\n".join(hits)
    return "\n".join(hits) if hits else "no hits"


@mcp.tool
def kb_read(path: str) -> str:
    """读一篇文档的完整原文。"""
    return safe_path(path).read_text(encoding="utf-8")


@mcp.tool
def kb_write(path: str, content: str, reason: str) -> str:
    """新增或更新一篇文档。content 必须带 frontmatter
    （title/type/status/updated），status 只能是 current/draft/superseded。
    不提供删除——删东西走 git，让人来决定。"""
    p = safe_path(path)
    if len(content.encode("utf-8")) > MAX_WRITE_BYTES:
        return f"rejected: content over {MAX_WRITE_BYTES} bytes"
    meta = parse_frontmatter(content)
    missing = [k for k in REQUIRED_META if not meta.get(k)]
    if missing:
        return f"rejected: frontmatter missing {missing}"
    if meta["status"] not in VALID_STATUS:
        return f"rejected: status must be one of {VALID_STATUS}"
    p.parent.mkdir(parents=True, exist_ok=True)
    tmp = p.with_name(p.name + ".tmp")
    tmp.write_text(content, encoding="utf-8")
    os.replace(tmp, p)
    audit("write", path, {"reason": reason, "bytes": len(content)})
    return f"ok: {path} ({len(content)} chars, reason recorded)"


@mcp.tool
def code_graph_query(question: str) -> str:
    """查代码知识图谱（graphify-out/graph.json），回答"谁调用谁、改哪会影响哪"。
    前提：先在 CODE_ROOT 跑过 graphify。"""
    graph = CODE_ROOT / "graphify-out" / "graph.json"
    if not graph.exists():
        return "no graph found: run `graphify update` in CODE_ROOT first"
    try:
        out = subprocess.run(
            ["graphify", "query", question],
            cwd=CODE_ROOT, capture_output=True, text=True, timeout=30,
        )
    except FileNotFoundError:
        return "graphify CLI not installed: uv tool install graphifyy"
    return (out.stdout or out.stderr)[:8000]


if __name__ == "__main__":
    mcp.run()   # stdio：本地 agent 够用，零部署

SEARCH: decision/retry.md:2: title: 为什么重试上限定 3 次
WRITE no-frontmatter: rejected: frontmatter missing ['title', 'type', 'status', 'updated']
TRAVERSAL blocked: Error calling tool 'kb_read': path escapes KB_ROOT: ../../etc/passwd.md
WRITE ok: ok: runbook/deploy.md (85 chars, reason recorded)

{"ts": "2026-10-13T06:30:51.964687+00:00", "action": "write",
 "path": "runbook/deploy.md", "reason": "test", "bytes": 85}

[mcp_servers.kb]
command = "python3"
args = ["/Users/walter/kb/kb_mcp_server.py"]
env = { KB_ROOT = "/Users/walter/kb/docs", CODE_ROOT = "/Users/walter/src/myproject" }

{
  "mcpServers": {
    "kb": {
      "command": "python3",
      "args": ["/Users/walter/kb/kb_mcp_server.py"],
      "env": {
        "KB_ROOT": "/Users/walter/kb/docs",
        "CODE_ROOT": "/Users/walter/src/myproject"
      }
    }
  }
}

$ claude mcp add kb --env KB_ROOT=/Users/walter/kb/docs \
    -- python3 /Users/walter/kb/kb_mcp_server.py

## 知识库怎么用
- 查知识库一律走 kb_* 工具：先 kb_catalog 看目录，再 kb_search / kb_read。
- 引用结论必须给出 kb 里的文件路径。
- 写入用 kb_write，必须带 reason；删除不归你管，提给人类。

@mcp.tool
def ua_graph_summary() -> str:
    """读 Understand-Anything 图谱的摘要：节点数、边数、核心节点。"""
    g = CODE_ROOT / ".ua" / "knowledge-graph.json"
    if not g.exists():
        return "no graph: run /understand first"
    data = json.loads(g.read_text(encoding="utf-8"))
    return json.dumps({
        "nodes": len(data.get("nodes", [])),
        "edges": len(data.get("edges", [])),
        "top": [n.get("label", n.get("id")) for n in data.get("nodes", [])[:20]],
    }, ensure_ascii=False)

@startmindmap
<style>
mindmapDiagram {
  node {
    BackgroundColor #F8F9FA
    LineColor #64748B
    FontColor #172033
    RoundCorner 10
    Padding 10
    FontSize 13
  }
  :depth(0) {
    BackgroundColor #1E3A5F
    LineColor #1E3A5F
    FontColor white
    FontSize 18
    FontStyle bold
  }
  :depth(1) {
    FontSize 15
    FontStyle bold
  }
  :depth(2) {
    FontSize 13
  }
}
</style>

* 知识库 MCP 柜台
** 痛点
*** 看不见：库不在工作目录
*** 全看见：grep 烧 token、乱写无护栏
*** 解法：开柜台，不给钥匙
** 为什么是 MCP
*** 三家客户端共用一份
*** 工具 schema 自描述
*** 护栏收在 server，不散在 prompt
** 知识库一侧
*** Sphinx + MyST + Mermaid
*** Markdown 是唯一事实源
*** make html 就是编译器
** 柜台五个工具
*** kb_catalog 目录
*** kb_search 正则搜路径:行号
*** kb_read 整篇原文
*** kb_write 校验+原子写+审计
*** code_graph_query 查代码图
** 写护栏
*** 路径白名单 is_relative_to
*** frontmatter 四字段必填
*** 200KB 上限、原子替换
*** 审计 JSONL、不提供删除
*** 防呆不防贼，硬约束靠沙箱
** 代码图上柜台
*** graphify：免费地图，零 token 常挂
*** Understand-Anything：贵导游，按需
*** 图是指路不是作证，回源码核对
** 升级路径
*** stdio → http 加鉴权
*** grep → FTS5 → pgvector
*** 单库 → 领域联邦
*** 验收用 golden question
@endmindmap