MCP协议的深度分析与应用示例
Anthropic于2024年发布的Model Context Protocol(MCP)是目前LLM和Agent工具的基础。
这里基于网络资料，构建一个MCP知识库助手，实现跨知识库语义检索、问答记录、定时分析。
示例MCP集成SQLite存储与FAISS向量检索，支持资源暴露、工具调用与提示模板三大核心能力。
1 MCP分析
1.1 MCP引入
在LLM工程落地阶段，每接入一个新工具或数据源，就需要编写一套定制化的适配代码。
以典型的AI助手为例，它需要同时对接数据库查询、文件读写、API调用等十几个能力点。
在MCP 出现之前，这些集成通常依赖各厂商私有的Function Calling 机制实现。
OpenAI、Anthropic等各家 LLM 平台函数调用规范各不相同，JSON Schema格式各异。
这种碎片化造成了LLM和工具的集成问题，M 个模型 × N 个工具，需要维护 M×N 套定制集成。
任何一方发生变化，集成便可能失效。
MCP是一个开放、厂商中立的标准化协议，通过统一客户端-服务器架构和JSON-RPC 2.0消息格式，使任意LLM应用都能发现并调用外部工具、读取数据资源和获取提示模板。
MCP旨在统一 AI 与外部工具的连接标准，核心功能归纳为三点：
1）标准化
定义统一的接口规范，使任意 AI 模型均可通过相同方式调用符合规范的工具与数据源。
2）解耦化
将模型与工具的实现解耦，工具以独立的 MCP Server 形式部署，模型通过协议调用，互不依赖。
3）安全化
内置访问控制、权限声明和沙箱机制，避免模型越权操作外部资源。
2025 年MCP被正式移交 Linux 基金会管理，OpenAI、Google、微软等均宣布支持，社区已拥有大量Server，集成 MCP的应用广泛投入实际使用。
1.2 MCP架构
MCP采用 Host-Client-Server 三层架构。
1）MCP Host
运行LLM方(Claude Desktop、IDE 插件、自定义 AI 应用)负责发起用户意图、管理上下文。
2）MCP Client
协议层客户端，内嵌于Host中，负责与MCP Server建立连接、编解码消息、维护会话状态。
3）MCP Server
MCP Server是轻量级独立程序，实现MCP协议，暴露特定能力供LLM模型调用。
MCP交互流程如下
用户通过Host发出指令
→Host内部的Client将指令转换为JSON-RPC请求
→ MCP Server 接收并执行操作
→ 返回结构化结果
→ Host将结果反馈给LLM继续推理。
MCP定义了三种核心能力类型，分别对应不同的交互模式：
1）工具
代表MCP Server暴露的具体可执行动作，使 LLM 能够基于用户意图执行具体操作，范围涵盖数据库查询、文件系统交互、API 调用等。工具是类似 POST 端点的主动执行能力。
2）资源
提供只读数据内容，作为 LLM 交互的上下文。
资源类似GET端点，用于将信息加载到 LLM 的上下文中。
例如数据库 Schema、文件内容、API 文档等。
3）提示
预定义交互模板，由MCP Server暴露给客户端。
允许用户显式地向LLM交互添加上下文并引导特定工作流。
例如应用的启动提示、GitHub PR评论的动态摘要等。
1.3 MCP vs Function Calling
以下是对MCP和Function Calling的对比。
| 维度 | Function Calling | MCP |
|---|---|---|
| 架构 | 模型直接调用预定义函数 | Client-Server 三层架构，模型与工具解耦 |
| 集成方式 | 每个模型平台需单独适配 | 一次开发，多模型复用 |
| 工具发现 | 静态预定义，需在请求中携带 | 动态发现，Server 实时告知可用工具 |
| 跨平台 | 各厂商格式不统一 | 厂商中立的开放标准 |
| 安全性 | 缺乏统一权限管理 | 内置多层安全机制 |
| 适用场景 | 简单、单一的工具调用 | 企业级多工具协作、多 Agent 系统 |
Function Calling是LLM 内置的交互机制，让模型直接调用预定义工具。
MCP 是协议标准，标准化了 LLM 如何发现、使用和管理跨不同模型和供应商的工具。
二者可协同工作，形成“意图解析-协议传输-工具执行”的分层架构。
2 MCP示例
这里提供一个完整、可运行的 MCP 应用示例，MCP 智能知识库助手。
集成SQLite 和 FAISS，支持语义检索、问答记录、统计分析等功能，展示MCP原语的协同运用。
2.1 应用架构
该MCP智能知识库助手示例，具备以下能力。
| 能力类型 | 名称 | 实现方式 | MCP 原语 |
|---|---|---|---|
| 语义检索 | search_knowledge_base |
FAISS 向量检索 | Tool |
| 问答记录 | save_qa_record |
SQLite 持久化 | Tool |
| 知识库统计 | get_kb_stats |
SQLite 聚合查询 | Tool |
| 知识库索引信息 | kb://stats |
实时计算统计信息 | Resource |
| 分析报告模板 | analyze_knowledge_base |
结构化 Prompt 生成 | Prompt |
| 最近问答记录 | qa://recent |
查询最近 5 条问答 | Resource |
2.2 环境配置
运行以下示例，需要如下依赖安装。
pip install mcp litellm sqlite3 faiss-cpu numpy
由于涉及到向量模型，需要从hf下载，这里配置hf-mirror国内站点。
同时配置openai格式的api key和base url，示例代码如下所示。
import os
os.environ['HF_ENDPOINT'] = "https://hf-mirror.com"
import config
model_name = gpt_model_name # LLM名称,比如deepseek-r1, qwen3.5-8b
os.environ['OPENAI_API_KEY'] = gpt_api_key # LLM供应商提供的api key
os.environ['OPENAI_BASE_URL'] = gpt_api_url # LLM供应商提供llm访问api的url
2.3 工具包导入
代码实现需要import如下工具包。
#!/usr/bin/env python3
import os
import json
import sqlite3
import hashlib
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict
from sentence_transformers import SentenceTransformer
import numpy as np
import faiss
from mcp.server.fastmcp import FastMCP
import litellm2.3 数据模型和基础设施
1）基础设施
这里定义数据模型和基础设施。
# ============================================================================
# 第一部分：数据模型与基础设施
# ============================================================================
@dataclass
class Document:
"""知识库文档模型"""
id: str
title: str
content: str
source: str
created_at: str
embedding: Optional[List[float]] = None
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
class SQLiteStore:
"""SQLite 数据存储层 - 管理文档元数据和问答记录"""
def __init__(self, db_path: str = "knowledge_base.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""初始化数据库表结构"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
# 文档表
cursor.execute("""
CREATE TABLE IF NOT EXISTS documents (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
content TEXT NOT NULL,
source TEXT,
created_at TEXT NOT NULL,
embedding_hash TEXT
)
""")
# 问答记录表 - 用于追踪 LLM 调用历史
cursor.execute("""
CREATE TABLE IF NOT EXISTS qa_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
question TEXT NOT NULL,
answer TEXT NOT NULL,
retrieved_docs TEXT,
created_at TEXT NOT NULL,
latency_ms INTEGER
)
""")
conn.commit()
def insert_document(self, doc: Document) -> bool:
try:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO documents
(id, title, content, source, created_at, embedding_hash)
VALUES (?, ?, ?, ?, ?, ?)
""", (doc.id, doc.title, doc.content, doc.source,
doc.created_at, self._hash_embedding(doc.embedding)))
conn.commit()
return True
except Exception as e:
print(f"Failed to insert document: {e}")
return False
def get_document(self, doc_id: str) -> Optional[Dict]:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT id, title, content, source, created_at FROM documents WHERE id = ?",
(doc_id,))
row = cursor.fetchone()
if row:
return {
"id": row[0], "title": row[1], "content": row[2],
"source": row[3], "created_at": row[4]
}
return None
def insert_qa_record(self, question: str, answer: str,
retrieved_docs: List[str], latency_ms: int) -> int:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO qa_records (question, answer, retrieved_docs, created_at, latency_ms)
VALUES (?, ?, ?, ?, ?)
""", (question, answer, json.dumps(retrieved_docs),
datetime.now().isoformat(), latency_ms))
conn.commit()
return cursor.lastrowid
def get_recent_qa(self, limit: int = 5) -> List[Dict]:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT question, answer, retrieved_docs, created_at, latency_ms
FROM qa_records ORDER BY created_at DESC LIMIT ?
""", (limit,))
rows = cursor.fetchall()
return [{
"question": r[0], "answer": r[1], "retrieved_docs": json.loads(r[2]),
"created_at": r[3], "latency_ms": r[4]
} for r in rows]
def get_stats(self) -> Dict[str, Any]:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM documents")
doc_count = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM qa_records")
qa_count = cursor.fetchone()[0]
cursor.execute("SELECT AVG(latency_ms) FROM qa_records")
avg_latency = cursor.fetchone()[0] or 0
return {
"total_documents": doc_count,
"total_qa_records": qa_count,
"avg_latency_ms": round(avg_latency, 2)
}
@staticmethod
def _hash_embedding(emb: Optional[List[float]]) -> str:
if emb is None:
return ""
return hashlib.md5(np.array(emb).tobytes()).hexdigest()[:16]
2）Faiss向量索引
Faiss向量索引，这时语义检索的核心。
class FAISSIndex:
"""FAISS 向量索引 - 语义检索核心"""
def __init__(self, dimension: int = 768, index_path: str = "kb_index.faiss"):
self.dimension = dimension
self.index_path = index_path
self.index: Optional[faiss.IndexFlatIP] = None
self.doc_ids: List[str] = []
self._load_or_create()
def _load_or_create(self):
if os.path.exists(self.index_path):
self.index = faiss.read_index(self.index_path)
# 尝试加载对应的 ID 映射文件
id_map_path = self.index_path.replace(".faiss", "_ids.json")
if os.path.exists(id_map_path):
with open(id_map_path, "r") as f:
self.doc_ids = json.load(f)
else:
self.index = faiss.IndexFlatIP(self.dimension)
def add(self, doc_id: str, embedding: List[float]):
if self.index is None:
self.index = faiss.IndexFlatIP(self.dimension)
vec = np.array(embedding, dtype=np.float32).reshape(1, -1)
faiss.normalize_L2(vec)
self.index.add(vec)
self.doc_ids.append(doc_id)
def search(self, query_embedding: List[float], k: int = 5) -> List[tuple]:
if self.index is None or self.index.ntotal == 0:
return []
vec = np.array(query_embedding, dtype=np.float32).reshape(1, -1)
faiss.normalize_L2(vec)
distances, indices = self.index.search(vec, min(k, self.index.ntotal))
results = []
for dist, idx in zip(distances[0], indices[0]):
if idx >= 0 and idx < len(self.doc_ids):
results.append((self.doc_ids[idx], float(dist)))
return results
def save(self):
if self.index:
faiss.write_index(self.index, self.index_path)
id_map_path = self.index_path.replace(".faiss", "_ids.json")
with open(id_map_path, "w") as f:
json.dump(self.doc_ids, f)
3）嵌入服务
这里使用Sentence-Transformer调用真实 LLM 生成向量
class EmbeddingService:
"""嵌入服务 - 使用Sentence-Transformer调用真实 LLM 生成向量"""
def __init__(self, model: str = "maidalun1020/bce-embedding-base_v1"):
self.model = model
self.embedder = SentenceTransformer(model)
def generate(self, text: str) -> List[float]:
"""
生成文本嵌入向量。
实际部署时可替换为本地模型（如 sentence-transformers）
"""
try:
embedding = self.embedder.encode(text[:8000])
return embedding
except Exception as e:
print(f"Embedding generation failed: {e}")
# 降级：返回随机向量（仅用于演示，生产环境应使用本地模型）
return np.random.randn(768)
2.4 MCP核心实现
这里定义MCP Server的核心实现，具体为调用各种基础设施，实现特定功能。
1）初始化Server
# ============================================================================
# 第二部分：MCP Server 核心实现
# ============================================================================
# 初始化 MCP Server
mcp = FastMCP(
"KnowledgeBaseAssistant",
json_response=True
)
# 初始化底层服务
db = SQLiteStore()
faiss_index = FAISSIndex(dimension=768)
emb_service = EmbeddingService(model="maidalun1020/bce-embedding-base_v1")
embedding = emb_service.generate("hello!").tolist()
print(len(embedding))
2）工具定义
即定义AI 可以主动调用的功能。
# ----------------------------------------------------------------------------
# 工具定义 - AI 可以主动调用的功能
# ----------------------------------------------------------------------------
@mcp.tool()
def search_knowledge_base(query: str, top_k: int = 3) -> str:
"""
在知识库中进行语义检索，返回最相关的文档内容。
Args:
query: 用户查询的自然语言问题
top_k: 返回的文档数量，默认 3 条，最大 10 条
Returns:
检索到的文档内容，以结构化格式呈现
"""
# 生成查询向量
query_emb = emb_service.generate(query)
# FAISS 向量检索
results = faiss_index.search(query_emb, k=min(top_k, 10))
if not results:
return "知识库中未找到相关文档。"
# 获取文档详情
output_parts = [f"检索到 {len(results)} 条相关文档：\n"]
for i, (doc_id, score) in enumerate(results, 1):
doc = db.get_document(doc_id)
if doc:
output_parts.append(
f"{i}. 【{doc['title']}】(相关度: {score:.3f})\n"
f" 来源: {doc['source']}\n"
f" 内容: {doc['content'][:500]}...\n"
)
return "\n".join(output_parts)
@mcp.tool()
def save_qa_record(question: str, answer: str,
retrieved_doc_ids: List[str] = None) -> str:
"""
保存问答记录到数据库，用于追踪和统计。
Args:
question: 用户的原始问题
answer: LLM 生成的回答
retrieved_doc_ids: 检索到的文档 ID 列表
Returns:
保存结果的状态信息
"""
record_id = db.insert_qa_record(
question=question,
answer=answer,
retrieved_docs=retrieved_doc_ids or [],
latency_ms=0 # 实际应计算调用耗时
)
return f"问答记录已保存，ID: {record_id}"
@mcp.tool()
def get_kb_stats() -> str:
"""
获取知识库的统计信息。
Returns:
包含文档数量、问答记录数、平均延迟等统计数据的文本
"""
stats = db.get_stats()
return (
f"📊 知识库统计信息：\n"
f" - 文档总数: {stats['total_documents']}\n"
f" - 问答记录数: {stats['total_qa_records']}\n"
f" - 平均响应延迟: {stats['avg_latency_ms']} ms\n"
f" - 向量索引大小: {faiss_index.index.ntotal if faiss_index.index else 0} 条"
)
@mcp.tool()
def add_document(title: str, content: str, source: str = "manual") -> str:
"""
向知识库添加新文档。
Args:
title: 文档标题
content: 文档内容
source: 文档来源（如 URL、文件名等）
Returns:
添加结果
"""
doc_id = hashlib.md5(f"{title}:{source}".encode()).hexdigest()[:16]
# 生成嵌入向量
embedding = emb_service.generate(content)
doc = Document(
id=doc_id,
title=title,
content=content,
source=source,
created_at=datetime.now().isoformat(),
embedding=embedding
)
# 保存到 SQLite
if not db.insert_document(doc):
return f"❌ 文档保存失败"
# 添加到 FAISS 索引
faiss_index.add(doc_id, embedding)
faiss_index.save()
return f"✅ 文档已成功添加到知识库，ID: {doc_id}"
3）资源定义
即定义AI 可以读取的只读数据。
# ----------------------------------------------------------------------------
# 资源定义 - AI 可以读取的只读数据
# ----------------------------------------------------------------------------
@mcp.resource("kb://stats")
def get_kb_resource_stats() -> str:
"""
知识库统计信息资源。
AI 可通过读取 kb://stats 获取当前知识库的统计摘要。
"""
stats = db.get_stats()
return json.dumps(stats, indent=2, ensure_ascii=False)
@mcp.resource("qa://recent")
def get_recent_qa_resource() -> str:
"""
最近问答记录资源。
AI 可通过读取 qa://recent 获取最近的问答历史。
"""
records = db.get_recent_qa(limit=5)
if not records:
return "暂无问答记录。"
output = ["📝 最近 5 条问答记录：\n"]
for i, r in enumerate(records, 1):
output.append(f"{i}. Q: {r['question'][:100]}...")
output.append(f" A: {r['answer'][:100]}...")
output.append(f" 时间: {r['created_at']}\n")
return "\n".join(output)
@mcp.resource("doc://{doc_id}")
def get_document_resource(doc_id: str) -> str:
"""
获取指定 ID 的文档内容。
"""
doc = db.get_document(doc_id)
if not doc:
return f"未找到文档 ID: {doc_id}"
return json.dumps(doc, indent=2, ensure_ascii=False)
4）提示定义
这里定义预设的交互模板。
# ----------------------------------------------------------------------------
# 提示定义 - 预设的交互模板
# ----------------------------------------------------------------------------
@mcp.prompt()
def analyze_knowledge_base(topic: str = "整体情况") -> str:
"""
生成分析知识库的提示模板。
引导 LLM 对知识库进行结构化分析。
"""
stats = db.get_stats()
return f"""你是一个知识库分析专家。请基于以下信息分析知识库：
**知识库概况：**
- 文档总数：{stats['total_documents']}
- 问答记录数：{stats['total_qa_records']}
- 平均响应延迟：{stats['avg_latency_ms']} ms
**分析主题：** {topic}
**要求：**
1. 评估知识库的覆盖范围和质量
2. 识别可能的知识盲区
3. 提出改进建议
请开始你的分析："""
@mcp.prompt()
def answer_with_context(question: str) -> str:
"""
生成带检索增强的回答提示模板。
用于引导 LLM 基于检索到的文档回答问题。
"""
# 先检索相关文档
query_emb = emb_service.generate(question)
results = faiss_index.search(query_emb, k=3)
context_parts = []
for doc_id, _ in results:
doc = db.get_document(doc_id)
if doc:
context_parts.append(f"【文档：{doc['title']}】\n{doc['content']}")
context = "\n\n".join(context_parts) if context_parts else "（未找到相关文档）"
return f"""请基于以下参考文档回答问题。
**参考文档：**
{context}
**用户问题：** {question}
**要求：**
1. 回答必须基于参考文档的内容
2. 如果文档中没有相关信息，请明确说明
3. 引用具体的文档内容作为依据
请回答："""
2.5 MCP Client 测试
这里进一步定义MCP Client测试，即真实LLM交互。
示例代码如下
# ============================================================================
# 第三部分：MCP Client 测试 - 真实 LLM 交互
# ============================================================================
class MCPKnowledgeBaseClient:
"""
MCP 客户端测试类。
通过 LiteLLM 调用真实 LLM，展示完整的 MCP 工作流。
"""
def __init__(self, model: str = "deepseek/deepseek-chat"):
self.model = f"openai/{model}"
self.api_key = os.getenv("OPENAI_API_KEY")
self.base_url = os.getenv("OPENAI_BASE_URL")
self.conversation_history = []
def _get_tools_for_llm(self) -> List[Dict]:
"""将 MCP 工具转换为 LLM 可理解的 Function Calling 格式"""
return [
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "在知识库中进行语义检索，返回最相关的文档内容。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "用户查询的自然语言问题"
},
"top_k": {
"type": "integer",
"description": "返回的文档数量，默认 3 条，最大 10 条",
"default": 3
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "get_kb_stats",
"description": "获取知识库的统计信息。",
"parameters": {"type": "object", "properties": {}}
}
},
{
"type": "function",
"function": {
"name": "save_qa_record",
"description": "保存问答记录到数据库。",
"parameters": {
"type": "object",
"properties": {
"question": {"type": "string", "description": "用户问题"},
"answer": {"type": "string", "description": "回答内容"},
"retrieved_doc_ids": {
"type": "array",
"items": {"type": "string"},
"description": "检索到的文档ID"
}
},
"required": ["question", "answer"]
}
}
},
{
"type": "function",
"function": {
"name": "add_document",
"description": "向知识库添加新文档。",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "文档标题"},
"content": {"type": "string", "description": "文档内容"},
"source": {"type": "string", "description": "文档来源"}
},
"required": ["title", "content"]
}
}
}
]
def chat(self, user_message: str) -> str:
"""
单轮对话，支持工具调用。
展示 MCP 工具如何在真实 LLM 交互中使用。
"""
import time
messages = self.conversation_history + [
{"role": "user", "content": user_message}
]
# 调用 LLM（可能触发工具调用）
response = litellm.completion(
model=self.model,
messages=messages,
api_base = self.base_url,
api_key=self.api_key,
tools=self._get_tools_for_llm(),
tool_choice="auto"
)
assistant_message = response.choices[0].message
# 处理工具调用
if assistant_message.tool_calls:
tool_results = []
for tool_call in assistant_message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
start_time = time.time()
# 执行对应的 MCP 工具
if func_name == "search_knowledge_base":
result = search_knowledge_base(**func_args)
elif func_name == "get_kb_stats":
result = get_kb_stats()
elif func_name == "save_qa_record":
result = save_qa_record(**func_args)
elif func_name == "add_document":
result = add_document(**func_args)
else:
result = f"未知工具: {func_name}"
latency = int((time.time() - start_time) * 1000)
tool_results.append({
"tool_call_id": tool_call.id,
"role": "tool",
"content": result
})
# 如果是问答，自动保存记录
if func_name == "search_knowledge_base":
save_qa_record(user_message, result, [])
# 将工具结果发回 LLM 生成最终回答
final_messages = messages + [
{"role": "assistant", "content": None, "tool_calls": [
{"id": tc.id, "type": "function", "function": {
"name": tc.function.name, "arguments": tc.function.arguments
}} for tc in assistant_message.tool_calls
]}
] + tool_results
final_response = litellm.completion(
model=self.model,
api_base = self.base_url,
api_key=self.api_key,
messages=final_messages
)
return final_response.choices[0].message.content
return assistant_message.content or ""
def run_test_scenarios(self):
"""
运行预设的测试场景，验证 MCP 所有功能。
"""
print("=" * 60)
print("MCP 智能知识库助手 - 工业级测试")
print("=" * 60)
# 场景 1：初始化测试数据
print("\n【场景 1】初始化知识库...")
test_docs = [
("MCP 协议简介", "Model Context Protocol (MCP) 是 Anthropic 于 2024 年 11 月发布的开放标准，用于标准化 LLM 与外部工具的连接。MCP 采用客户端-服务器架构，基于 JSON-RPC 2.0 通信。", "test"),
("MCP 核心原语", "MCP 定义了三种核心能力：Tools（工具）用于执行操作，Resources（资源）用于提供只读数据，Prompts（提示）用于预设交互模板。", "test"),
("MCP 传输层", "MCP 支持 stdio 和 Streamable HTTP 两种传输方式。stdio 适用于本地工具调用，Streamable HTTP 适用于远程服务。", "test"),
]
for title, content, source in test_docs:
result = add_document(title, content, source)
print(f" 添加文档: {title} -> {result}")
# 场景 2：测试统计信息工具
print("\n【场景 2】测试工具 - 获取知识库统计...")
print(f" 结果: {get_kb_stats()}")
# 场景 3：测试语义检索工具
print("\n【场景 3】测试工具 - 语义检索...")
result = search_knowledge_base("MCP 是什么？")
print(f" 结果: {result[:300]}...")
# 场景 4：测试资源读取
print("\n【场景 4】测试资源 - 读取统计信息...")
print(f" kb://stats 资源内容: {get_kb_resource_stats()}")
print("\n【场景 5】测试资源 - 读取问答记录...")
print(f" qa://recent 资源内容: {get_recent_qa_resource()}")
# 场景 6：测试提示模板
print("\n【场景 6】测试提示 - 生成分析模板...")
prompt = analyze_knowledge_base("MCP 文档覆盖情况")
print(f" 生成的提示: {prompt[:200]}...")
# 场景 7：真实 LLM 交互测试
print("\n【场景 7】真实 LLM 交互测试...")
print(" 向 LLM 提问: 请帮我查询 MCP 协议的核心原语有哪些？")
response = self.chat("请帮我查询 MCP 协议的核心原语有哪些？")
print(f" LLM 回答: {response[:300]}...")
# 场景 8：验证问答记录
print("\n【场景 8】验证问答记录已保存...")
records = db.get_recent_qa(limit=2)
for r in records:
print(f" Q: {r['question'][:50]}...")
print(f" A: {r['answer'][:50]}...")
print(f" 时间: {r['created_at']}")
print("\n" + "=" * 60)
print("所有测试场景执行完毕！")
print("=" * 60)
3.5 运行入口测试
这里为MCP智能知识库助手的入口。
包括运行测试，和启动MCP Server，提供MCP服务。
# ============================================================================
# 第四部分：入口
# ============================================================================
# if __name__ == "__main__":
# import sys
# if len(sys.argv) > 1 and sys.argv[1] == "--test":
# # 运行测试模式
# client = MCPKnowledgeBaseClient()
# client.run_test_scenarios()
# else:
# # 启动 MCP Server（stdio 传输，供 Claude Desktop 等客户端连接）
# print("启动 MCP Knowledge Base Server (stdio)...", file=sys.stderr)
# mcp.run(transport="stdio")这里主要示例运行MCP测试，示例代码如下
client = MCPKnowledgeBaseClient(model=model_name)
client.run_test_scenarios()输出如下所示
============================================================
MCP 智能知识库助手 - 工业级测试
============================================================【场景 1】初始化知识库...
Batches: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 10.91it/s]
添加文档: MCP 协议简介 -> ✅ 文档已成功添加到知识库，ID: 2a11f5fb0f6cb545
Batches: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 22.71it/s]
添加文档: MCP 核心原语 -> ✅ 文档已成功添加到知识库，ID: 59887be0e482c244
Batches: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 22.24it/s]
添加文档: MCP 传输层 -> ✅ 文档已成功添加到知识库，ID: f70e0c46ebd501ba【场景 2】测试工具 - 获取知识库统计...
结果: 📊 知识库统计信息：
- 文档总数: 3
- 问答记录数: 1
- 平均响应延迟: 0 ms
- 向量索引大小: 21 条【场景 3】测试工具 - 语义检索...
Batches: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 25.33it/s]
10:49:26 - LiteLLM:INFO: utils.py:3874 -
LiteLLM completion() model= xxxx; provider = openai
结果: 检索到 3 条相关文档：1. 【MCP 协议简介】(相关度: 0.682)
来源: test
内容: Model Context Protocol (MCP) 是 Anthropic 于 2024 年 11 月发布的开放标准，用于标准化 LLM 与外部工具的连接。MCP 采用客户端-服务器架构，基于 JSON-RPC 2.0 通信。...2. 【MCP 协议简介】(相关度: 0.682)
来源: test
内容: Model Context Protocol (MCP) 是 Anthropic 于 2024 年 11 月发布的开放标准，用于标准化 LLM 与外部...【场景 4】测试资源 - 读取统计信息...
kb://stats 资源内容: {
"total_documents": 3,
"total_qa_records": 1,
"avg_latency_ms": 0
}【场景 5】测试资源 - 读取问答记录...
qa://recent 资源内容: 📝 最近 5 条问答记录：1. Q: 请帮我查询 MCP 协议的核心原语有哪些？...
A: 检索到 3 条相关文档：1. 【MCP 核心原语】(相关度: 0.620)
来源: test
内容: MCP 定义了三种核心能力：Tools（工具）用于执行操作，Resources（资...
时间: 2026-04-11T10:43:34.222709
【场景 6】测试提示 - 生成分析模板...
生成的提示: 你是一个知识库分析专家。请基于以下信息分析知识库：**知识库概况：**
- 文档总数：3
- 问答记录数：1
- 平均响应延迟：0 ms**分析主题：** MCP 文档覆盖情况
**要求：**
1. 评估知识库的覆盖范围和质量
2. 识别可能的知识盲区
3. 提出改进建议请开始你的分析：...
【场景 7】真实 LLM 交互测试...
向 LLM 提问: 请帮我查询 MCP 协议的核心原语有哪些？
[04/11/26 10:49:26] INFO utils.py:3874
LiteLLM completion() model= xxxx; provider = openai
10:49:27 - LiteLLM:INFO: utils.py:1623 - Wrapper: Completed Call, calling success_handler
[04/11/26 10:49:27] INFO Wrapper: Completed Call, calling success_handler utils.py:1623
Batches: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 3.47it/s]
10:49:28 - LiteLLM:INFO: utils.py:3874 -
LiteLLM completion() model= xxxxx; provider = openai
[04/11/26 10:49:28] INFO utils.py:3874
LiteLLM completion() model= xxxxx; provider = openai
10:49:31 - LiteLLM:INFO: utils.py:1623 - Wrapper: Completed Call, calling success_handler
[04/11/26 10:49:31] INFO Wrapper: Completed Call, calling success_handler utils.py:1623
LLM 回答:MCP 协议的核心原语包含以下三种基础能力：
1. **Tools（工具）**
用于执行具体操作或调用外部功能，例如触发 API、运行脚本或控制设备。工具通常具有明确的输入参数和输出结果，支持动态交互。2. **Resources（资源）**
提供只读数据访问能力，例如读取文件内容、查询数据库或获取系统状态。资源强调数据的被动获取，不涉及修改或操作。3. **Prompts（提示）**
预设的交互模板或引导指令，用于规范用户输入或系统响应格式。例如定义对话流程、生成结构化请求或约束输出范围。这三种原语共同构成了 MCP 协议的基础交互框架，分别对...
【场景 8】验证问答记录已保存...
Q: 请帮我查询 MCP 协议的核心原语有哪些？...
A: 检索到 3 条相关文档：1. 【MCP 核心原语】(相关度: 0.620)
来源: tes...
时间: 2026-04-11T10:49:28.132973
Q: 请帮我查询 MCP 协议的核心原语有哪些？...
A: 检索到 3 条相关文档：1. 【MCP 核心原语】(相关度: 0.620)
来源: tes...
时间: 2026-04-11T10:43:34.222709============================================================
所有测试场景执行完毕！
============================================================
3.6.测试场景说明
该示例包含 8 个测试场景，覆盖 MCP 全部核心能力。
| 场景 | 测试内容 | MCP 原语 | 技术验证点 |
|---|---|---|---|
| 1 | 初始化测试数据 | Tool: add_document |
SQLite 写入 + FAISS 向量化 |
| 2 | 获取知识库统计 | Tool: get_kb_stats |
SQLite 聚合查询 |
| 3 | 语义检索 | Tool: search_knowledge_base |
FAISS 相似度检索 |
| 4 | 读取统计信息 | Resource: kb://stats |
资源 URI 解析 |
| 5 | 读取问答记录 | Resource: qa://recent |
动态资源生成 |
| 6 | 生成分析模板 | Prompt: analyze_knowledge_base |
提示模板动态注入上下文 |
| 7 | 真实 LLM 交互 | 全部 Tool | LiteLLM 调用真实模型 + 工具编排 |
| 8 | 验证问答记录 | Tool: save_qa_record |
端到端数据持久化 |
各功能模型的特性说明如下：
| 特性 | 实现方式 | 价值 |
|---|---|---|
| 向量检索 | FAISS IndexFlatIP + 余弦相似度 | 语义级知识发现，非简单关键词匹配 |
| 数据持久化 | SQLite（符合“轻量级工具”要求） | 文档管理、问答追踪、统计聚合 |
| 真实 LLM 集成 | LiteLLM 统一适配 OpenAI/Anthropic/DeepSeek | 生产环境可直接对接主流模型 |
| 结构化输出 | JSON 格式响应 + json_response=True |
符合 MCP 2025-06-18 规范 |
| 模块化设计 | 数据层/索引层/服务层分离 | 便于扩展和单元测试 |
| 错误处理 | 嵌入生成降级、空索引处理 | 生产环境鲁棒性保障 |
这里提供的知识库助手示例完整展示了 MCP 在生产环境中的应用模式。
通过 FAISS 实现语义检索、通过 SQLite 实现轻量级持久化、通过 LiteLLM 对接真实 LLM。
该示例可直接运行，也可作为构建更复杂 MCP 应用的基础模板。
reference
---
Key Changes
https://modelcontextprotocol.io/specification/2025-11-25/changelog
Inaugural MCP Dev Summit Charts AI Integration's Future
MCP (Model Context Protocol): The USB-C for AI Applications
https://www.sitepoint.com/model-context-protocol-mcp/?utm_source=rss
Model Context Protocol: Game changer or just hype?
https://tray.ai/resources/blog/model-context-protocol-mcp.md
SEP-1577: Sampling With Tools
https://modelcontextprotocol.io/seps/1577--sampling-with-tools
MCP是什么？从Function Calling到AI工具生态中枢的范式转变
https://cloud.tencent.com.cn/developer/article/2611533
如何基于sentence_transformers构建向量计算工具
https://blog.csdn.net/liliang199/article/details/160043994
大模型融合访问开源工具 - LiteLLM
https://blog.csdn.net/liliang199/article/details/155907814
MCP-与本地大模型集成实现工具调用
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
1
0- 0
已为社区贡献15条内容
所有评论(0)