# MCP 构建指南

本文档指导如何将 md-vector-db 封装为 MCP（Model Context Protocol）服务。

参考实现：[PaddleOCR MCP Server](D:/Code/language/Python/PaddleOCR/ocr_mcp_server.py) — 使用 `FastMCP` + `Pydantic` 模式。

---

## PaddleOCR 的 MCP 模式（参考）

PaddleOCR 的 MCP 实现非常简洁，核心模式如下：

```python
from mcp.server.fastmcp import FastMCP, Context
from pydantic import BaseModel, Field

# 1. 一行创建 server
mcp = FastMCP("server_name")

# 2. Pydantic 定义参数模型（自动生成 inputSchema）
class MyInput(BaseModel):
    file_path: str = Field(..., description="参数说明", min_length=1)
    option: bool = Field(default=False, description="可选参数")

# 3. @mcp.tool 装饰器注册工具，参数模型自动转换为 JSON Schema
@mcp.tool(name="tool_name", annotations={...})
async def tool_name(params: MyInput, ctx: Context) -> str:
    await ctx.info("处理中...")             # 日志
    await ctx.report_progress(0.5, ...)     # 进度
    return "结果字符串"

# 4. 一行启动
if __name__ == "__main__":
    mcp.run()
```

---

## md-vector-db 的 MCP Server 实现

### 1. 安装依赖

```bash
uv add "mcp>=1.0.0"
```

### 2. 创建 `src/mcp_server.py`

```python
"""md-vector-db MCP Server — 将向量知识库暴露为 MCP 工具.

基于 FastMCP + Pydantic 模式（参考 PaddleOCR/ocr_mcp_server.py）。
"""
import sys
from pathlib import Path

# 确保 src/ 在路径中
sys.path.insert(0, str(Path(__file__).parent))

from mcp.server.fastmcp import FastMCP, Context
from pydantic import BaseModel, Field, ConfigDict

from src.core.config import load_config
from src.core.db import VectorDB
from src.core.embedder import create_embedder
from src.core.ingest import DocumentIngestor
from src.core.search import Searcher

# ── 初始化（进程启动时执行一次） ──
cfg = load_config()
db = VectorDB(persist_dir=cfg.chroma.persist_dir)
embedder = create_embedder(cfg.embed)

_searchers: dict[str, Searcher] = {}
_ingestors: dict[str, DocumentIngestor] = {}

def _get_searcher(collection: str) -> Searcher:
    if collection not in _searchers:
        _searchers[collection] = Searcher(db, embedder, collection)
    return _searchers[collection]

def _get_ingestor(collection: str) -> DocumentIngestor:
    if collection not in _ingestors:
        _ingestors[collection] = DocumentIngestor(db, embedder, collection)
    return _ingestors[collection]

# ── MCP Server ──
mcp = FastMCP("md-vector-db")


# ========== Pydantic 输入模型 ==========
class SearchInput(BaseModel):
    """语义检索参数."""
    model_config = ConfigDict(str_strip_whitespace=True)

    query: str = Field(
        ...,
        description="自然语言查询，例如 'Git 分支管理策略' 或 'Docker Compose 多容器编排'",
        min_length=1,
    )
    collection: str = Field(
        default="obsidian_blog",
        description="知识库集合名。可选值: obsidian_blog (博客笔记), default (测试数据)",
    )
    top_k: int = Field(
        default=5,
        ge=1,
        le=50,
        description="返回结果数量 (1-50)",
    )


class IngestInput(BaseModel):
    """文档入库参数."""
    model_config = ConfigDict(str_strip_whitespace=True)

    content: str = Field(
        ...,
        description="Markdown 格式的文档内容",
        min_length=1,
    )
    file_name: str = Field(
        ...,
        description="文档名（用于标识来源，如 'git-guide.md'）",
        min_length=1,
    )
    collection: str = Field(
        default="default",
        description="目标集合名",
    )


class ListCollectionsInput(BaseModel):
    """列出集合（无参数）."""
    pass


# ========== MCP 工具 ==========
@mcp.tool(
    name="search_knowledge_base",
    annotations={
        "title": "语义搜索知识库",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True,
    },
)
async def search_knowledge_base(params: SearchInput, ctx: Context) -> str:
    """语义检索知识库，返回最相关的 Markdown 文档片段。

    基于 bge-small-zh-v1.5 模型 + ChromaDB，支持跨文档模糊语义匹配。
    当前知识库包含 51 篇博客文章（3,611 chunks），涵盖 Git、Docker、深度学习、
    LaTeX、AI 编程等主题。

    Args:
        params: SearchInput — query/collection/top_k
    Returns:
        格式化的 Markdown 搜索结果，包含相似度、来源、章节和内容
    """
    searcher = _get_searcher(params.collection)

    await ctx.info(f"搜索: {params.query} (集合: {params.collection})")

    results = searcher.search(query=params.query, top_k=params.top_k)

    if not results:
        return f"未找到与 '{params.query}' 相关的结果（集合: {params.collection}）"

    lines = [f"## 搜索结果: {params.query}\n"]
    for i, r in enumerate(results, 1):
        lines.append(
            f"### 结果 {i} (相似度: {r['score']:.2%})\n"
            f"- **来源**: {r.get('source_file', '?')}\n"
            f"- **章节**: {r.get('section_title', '—')}\n"
            f"\n{r['content']}\n"
        )

    await ctx.info(f"返回 {len(results)} 条结果")
    return "\n".join(lines)


@mcp.tool(
    name="list_collections",
    annotations={
        "title": "列出知识库集合",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True,
    },
)
async def list_collections(params: ListCollectionsInput, ctx: Context) -> str:
    """列出所有知识库集合及文档统计."""
    await ctx.info("列出集合...")

    lines = ["## 知识库集合\n"]
    try:
        collections = db.client.list_collections()
        if not collections:
            lines.append("（无集合）")
        for coll in collections:
            count = coll.count()
            # 获取源文件数
            try:
                sources = coll.get()["metadatas"]
                unique_sources = len(set(
                    m.get("source_file", "") for m in sources if m
                )) if sources else 0
            except Exception:
                unique_sources = "?"
            lines.append(f"- **{coll.name}**: {count} chunks / {unique_sources} 文件")
    except Exception as e:
        lines.append(f"错误: {e}")

    return "\n".join(lines)


@mcp.tool(
    name="ingest_document",
    annotations={
        "title": "入库 Markdown 文档",
        "readOnlyHint": False,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True,
    },
)
async def ingest_document(params: IngestInput, ctx: Context) -> str:
    """入库 Markdown 文档到知识库。同名文件会覆盖旧版本。

    Args:
        params: IngestInput — content/file_name/collection
    Returns:
        入库结果摘要
    """
    ingestor = _get_ingestor(params.collection)

    await ctx.info(f"入库: {params.file_name} → 集合 {params.collection}")

    n = ingestor.ingest_content(params.content, params.file_name)

    return (
        f"✅ 已入库: **{params.file_name}**\n"
        f"- Chunks: {n}\n"
        f"- 集合: `{params.collection}`"
    )


# ── 入口 ──
if __name__ == "__main__":
    mcp.run()
```

### 3. 注册入口

```toml
# pyproject.toml
[project.scripts]
md-vector-db = "src.cli.main:app"
md-vector-db-mcp = "src.mcp_server:mcp"    # FastMCP 的 run() 通过入口调用
```

> 注意：FastMCP 用 `mcp.run()` 启动，入口指向模块的 `mcp` 对象或直接用 `python -m src.mcp_server`。

### 4. MCP 客户端配置

#### Claude Code（`.claude/mcp.json`）


```json
{
  "mcpServers": {
    "md-vector-db": {
      "type": "stdio",
      "command": "D:/Code/doing_exercises/programs/md-vector-db/.venv/Scripts/python.exe",
      "args": [
        "D:/Code/doing_exercises/programs/md-vector-db/src/mcp_server.py"
      ]
    }
  }
}
```


> PaddleOCR 也是用绝对路径 python + 脚本路径的方式，**不通过 `uv run`**，避免 uv 的环境解析覆盖 CUDA torch。

#### Claude Desktop（`claude_desktop_config.json`）

```json
{
  "mcpServers": {
    "md-vector-db": {
      "command": "D:/Code/doing_exercises/programs/md-vector-db/.venv/Scripts/python.exe",
      "args": [
        "D:/Code/doing_exercises/programs/md-vector-db/src/mcp_server.py"
      ]
    }
  }
}
```

---

## FastMCP vs 低层 API 对比

|             | FastMCP（推荐）                                          | 低层 API（不推荐）                                 |
| ----------- | -------------------------------------------------------- | -------------------------------------------------- |
| 创建 Server | `mcp = FastMCP("name")`                                | `server = Server("name")`                        |
| 注册工具    | `@mcp.tool()` 装饰器                                   | `@server.list_tools()` + `@server.call_tool()` |
| 参数定义    | **Pydantic `BaseModel`** → 自动生成 JSON Schema | 手写`inputSchema` dict                           |
| 进度/日志   | `ctx.info()` / `ctx.report_progress()`               | 不支持（需手动实现）                               |
| 启动        | `mcp.run()`                                            | `asyncio.run(stdio_server(...))`                 |
| 工具元数据  | `annotations` 字典（标准 MCP 注解）                    | 不支持                                             |

---

## 工具注解说明

MCP 规范定义 4 个注解，帮助客户端理解工具行为：

| 注解                | 含义           | 示例                          |
| ------------------- | -------------- | ----------------------------- |
| `readOnlyHint`    | 是否只读       | `True` — 搜索不修改数据    |
| `destructiveHint` | 是否破坏性     | `False` — 入库不会删除数据 |
| `idempotentHint`  | 是否幂等       | `True` — 重复调用结果一致  |
| `openWorldHint`   | 是否与外部交互 | `True` — 连接外部 ChromaDB |

---

## 设计要点总结

### 1. 用 Pydantic，不用手写 JSON Schema

```python
# ✅ FastMCP 方式 — Pydantic 自动生成 inputSchema
class SearchInput(BaseModel):
    query: str = Field(..., description="...", min_length=1)
    top_k: int = Field(default=5, ge=1, le=50)

@mcp.tool()
async def search(params: SearchInput, ctx: Context) -> str:
    ...

# ❌ 低层 API — 手写容易出错
@server.list_tools()
async def list_tools():
    return [Tool(name="search", inputSchema={...手写 JSON Schema...})]
```

### 2. 用 `ctx` 报告进度

```python
await ctx.info("搜索: Docker 部署")                          # 日志
await ctx.report_progress(0.5, message="嵌入查询中...")       # 进度条
```

### 3. 返回 Markdown 格式字符串

Claude 能正确渲染 Markdown，所以直接返回格式化的 Markdown：

```python
return f"## 搜索结果\n\n### 1. {title}\n{content}"
```

### 4. 异常处理 — 返回错误字符串而非抛异常

```python
try:
    results = searcher.search(query, top_k)
except Exception as e:
    return f"搜索失败: {e}"
```

### 5. 启动用绝对路径 Python，不用 `uv run`

```
command: .venv/Scripts/python.exe    # 直接用 venv 的 python
args: [src/mcp_server.py]            # 脚本路径
```

---

## 测试 MCP Server

### MCP Inspector

```bash
npx @modelcontextprotocol/inspector \
  D:/Code/doing_exercises/programs/md-vector-db/.venv/Scripts/python.exe \
  D:/Code/doing_exercises/programs/md-vector-db/src/mcp_server.py
```

### 手动 JSON-RPC 测试

```bash
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | \
  .venv/Scripts/python.exe src/mcp_server.py
```

---

## 部署到 NAS

### 1. 同步项目文件

```bash
rsync -avz --exclude '.venv' --exclude 'data' \
  D:/Code/doing_exercises/programs/md-vector-db/ \
  LHY@192.168.5.8:/volume2/办公/Code/md-vector-db/
```

### 2. NAS 上创建 venv 并安装

```bash
ssh LHY@192.168.5.8
cd /volume2/办公/Code/md-vector-db
uv sync  # NAS 上不需要 GPU，CPU torch 即可
```

### 3. 客户端配置（通过 SSH 隧道连接）

如果 Claude Code 运行在 Windows 上，MCP Server 在 NAS 上，需要 SSH 隧道：

```json
{
  "mcpServers": {
    "md-vector-db": {
      "command": "ssh",
      "args": [
        "LHY@192.168.5.8",
        "cd /volume2/办公/Code/md-vector-db && .venv/bin/python src/mcp_server.py"
      ]
    }
  }
}
```