md-vector-db/docs/MCP构建指南.md

MCP 构建指南

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

参考实现：PaddleOCR MCP Server — 使用 FastMCP + Pydantic 模式。

---

PaddleOCR 的 MCP 模式（参考）

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

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. 安装依赖

uv add "mcp>=1.0.0"

2. 创建 src/mcp_server.py

"""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. 注册入口

# 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）

{
  "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）

{
  "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

# ✅ 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 报告进度

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

3. 返回 Markdown 格式字符串

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

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

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

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

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 测试

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

---

部署到 NAS

1. 同步项目文件

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 并安装

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 隧道：

{
  "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"
      ]
    }
  }
}