Serendipity
832201186d
H1: AppConfig 自定义 __init__ 改用 from_dict() 类方法 H2: list_collections_with_stats 改为从 ChromaDB 直接查询 H3: RateLimiter 过期 key 自动清理, 防止内存泄漏 M1: delete_by_source 区分 ValueError 与真实异常, 记日志 M2: VectorDB 新增 list_collections() 封装方法 M3: _remove_by_source 异常记日志, 不再静默吞掉 M4: CLI 集合回退支持 MD_VECTOR_DB_COLLECTION 环境变量 L1: DEFAULT_CONFIG_PATH 自动从项目根目录解析 L2: ingest 命令内重复 import 移至模块顶部 L3: 新增死循环回归测试 + 密集分隔符分块测试 L4: 统一 logger 名称为 md-vector-db 删除旧版审计文档 测试: 48 passed
438 lines
12 KiB
Markdown
438 lines
12 KiB
Markdown
# 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"
|
||
]
|
||
}
|
||
}
|
||
}
|
||
```
|