MCP资源发现系统：LLM应用的高效数据访问协议

1. 项目概述

MCP（Model Context Protocol）资源发现系统是一种专为大语言模型（LLM）应用设计的标准化数据访问协议。它解决了LLM应用中一个关键问题：如何让语言模型安全、高效地访问服务器端的各类数据资源。

1.1 核心价值

传统LLM应用通常通过API调用来获取服务器数据，这种方式存在几个痛点：

1. 每个API都需要单独设计和实现，开发成本高
2. 数据格式不统一，需要大量适配工作
3. 缺乏标准化的资源发现机制

MCP资源系统通过以下创新点解决了这些问题：

• 统一资源模型：将服务器数据抽象为标准的Resource对象
• 自动发现机制：客户端可以动态发现服务器提供的资源
• 语义检索支持：内置向量化索引能力，支持基于内容的智能检索

1.2 系统架构

MCP资源系统的核心架构包含三个关键组件：

1. 资源服务器：负责暴露和管理数据资源
2. 向量索引引擎：实现文本的向量化表示和相似度检索
3. 客户端代理：协调LLM与资源系统的交互

这种架构特别适合构建RAG（Retrieval-Augmented Generation）系统，能够显著提升LLM回答的准确性和专业性。

2. 技术实现详解

2.1 环境准备

2.1.1 基础环境配置

推荐使用Python 3.10+环境，这是目前最稳定的MCP支持版本。环境检查命令：

bash复制# 检查Python版本
python --version
# 应输出：Python 3.10.x或更高

# 确认虚拟环境
which python
# 应输出虚拟环境路径

2.1.2 依赖安装

MCP系统依赖多个专业库，建议使用uv工具进行安装（比pip快3-5倍）：

bash复制# 服务器端依赖
cd server
uv sync

# 客户端依赖
cd ../client
uv sync

关键依赖说明：

• mcp[cli]>=1.6.0：MCP核心库及命令行工具
• faiss-cpu>=1.10.0：Facebook开源的向量相似度搜索库
• openai>=1.75.0：用于文本向量化嵌入

2.2 资源服务器实现

2.2.1 基础资源服务器

最简单的资源服务器只需要实现资源列表功能：

python复制import asyncio
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server

app = Server("example-server")

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="file:///logs/app.log",
            name="Application Log"
        )
    ]

async def main():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], 
                     app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

这个基础版本虽然简单，但包含了MCP资源系统的核心要素：

• @app.list_resources()装饰器声明资源列表接口
• types.Resource定义标准资源对象
• 使用标准输入输出进行进程间通信

2.2.2 完整资源服务器

生产环境需要更完整的实现，包括资源读取功能：

python复制import os
from pathlib import Path

DOC_DIR = str(Path(__file__).parent / "medical_docs")

@app.read_resource()
async def read_resource(uri: str) -> str:
    path = str(uri).replace("file://", "")
    with open(path, encoding="utf-8") as f:
        return f.read()

关键改进点：

1. 使用Path处理跨平台路径问题
2. 添加资源内容读取接口
3. 支持动态发现目录下的文档资源

2.3 向量化检索实现

2.3.1 文本向量化

使用OpenAI的embedding接口将文本转换为向量：

python复制async def embed_text(texts: List[str]) -> np.ndarray:
    resp = openai.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
        encoding_format="float"
    )
    return np.array([d.embedding for d in resp.data], dtype="float32")

性能参数：

• 模型：text-embedding-3-small
• 维度：1536
• 成本：约$0.02/百万token
• 延迟：1-2秒/千token

2.3.2 FAISS索引构建

python复制_index = faiss.IndexFlatL2(1536)  # L2距离度量
_docs: List[str] = []  # 原始文档存储

async def index_docs(docs: List[str]) -> str:
    global _index, _docs
    emb = await embed_text(docs)
    _index.add(emb)
    _docs.extend(docs)
    return f"已索引 {len(docs)} 篇文档"

索引特点：

• 使用FlatL2索引，精度最高
• 适合文档量<1万的场景
• 内存占用：1536×4×N字节（N为文档数）

2.3.3 语义检索

python复制async def retrieve_docs(query: str, top_k: int = 3) -> str:
    q_emb = await embed_text([query])
    D, I = _index.search(q_emb, top_k)
    return "\n".join(f"【文档{i}】{_docs[i][:200]}..." for i in I[0])

检索流程：

1. 将查询语句向量化
2. 在FAISS索引中搜索相似文档
3. 返回最相关的文档片段

3. 系统集成与优化

3.1 FastMCP优化

使用FastMCP框架可以大幅简化代码：

python复制from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    server_name="rag",
    version="1.0.0",
    capabilities={"resources": {}}
)

@mcp.resource(
    "file:///path/to/doc.txt",
    name="doc",
    description="医学文档",
    mime_type="text/plain"
)
async def resource_func():
    with open("/path/to/doc.txt", encoding="utf-8") as f:
        return f.read()

@mcp.tool()
async def retrieve_docs(query: str, top_k: int = 3) -> str:
    # 检索实现...
    
mcp.run(transport="stdio")

优化效果：

• 代码量减少40%
• 资源注册更直观
• 自动生成API文档

3.2 客户端实现

智能问答客户端的核心逻辑：

python复制class RagClient:
    async def query(self, q: str) -> str:
        messages = [{"role": "user", "content": q}]
        
        while True:
            resp = self.openai.chat.completions.create(
                model="deepseek-chat",
                messages=messages,
                tools=self.tools
            )
            
            msg = resp.choices[0].message
            messages.append(msg)
            
            if msg.tool_calls:
                for call in msg.tool_calls:
                    result = await self.session.call_tool(
                        call.function.name,
                        json.loads(call.function.arguments)
                    )
                    messages.append({
                        "role": "tool",
                        "content": result.content[0].text,
                        "tool_call_id": call.id
                    })
            else:
                return msg.content

工具调用循环的工作流程：

1. 将用户问题发送给LLM
2. 检查LLM是否要求调用工具
3. 执行工具调用并收集结果
4. 将结果反馈给LLM生成最终回答

4. 生产环境实践

4.1 性能优化技巧

4.1.1 批量处理

python复制# 批量embedding减少API调用
texts = [doc1, doc2, doc3]
embeddings = await embed_text(texts)  # 一次调用处理多文档

4.1.2 缓存机制

python复制_embedding_cache = {}

async def embed_text_cached(texts: List[str]) -> np.ndarray:
    uncached = [t for t in texts if t not in _embedding_cache]
    if uncached:
        emb = await embed_text(uncached)
        for t, e in zip(uncached, emb):
            _embedding_cache[t] = e
    return np.array([_embedding_cache[t] for t in texts])

4.1.3 索引持久化

python复制import pickle

# 保存索引
with open('faiss_index.pkl', 'wb') as f:
    pickle.dump(_index, f)
    
# 加载索引
with open('faiss_index.pkl', 'rb') as f:
    _index = pickle.load(f)

4.2 错误处理与监控

4.2.1 重试机制

python复制from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_embedding_api(texts):
    return await openai.embeddings.create(...)

4.2.2 健康检查

python复制async def health_check():
    try:
        await session.list_resources()
        return {"status": "healthy"}
    except Exception as e:
        return {"status": "error", "detail": str(e)}

5. 典型问题解决方案

5.1 路径处理问题

问题现象：

code复制Error: [Errno 2] No such file or directory

解决方案：

python复制# 错误做法
DOC_DIR = "/hard/coded/path"

# 正确做法
from pathlib import Path
DOC_DIR = str(Path(__file__).parent / "medical_docs")

5.2 API限流处理

问题现象：

code复制RateLimitError: 429 - Too many requests

解决方案：

1. 实现请求缓存
2. 使用本地embedding模型作为备选
3. 升级API配额

python复制from sentence_transformers import Sentence[Transformer](https://taotoken.net?utm_source=ai)

model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(texts)  # 离线向量化

5.3 中文路径问题

问题现象：

code复制URI包含%E8%B5%84%E6%BA%90等编码字符

解决方案：

python复制from urllib.parse import unquote

path = unquote(str(uri).replace("file://", ""))

6. 应用场景扩展

6.1 多模态支持

通过扩展MCP资源类型，可以支持图片、视频等非文本资源：

python复制types.Resource(
    uri="file:///data/xray.jpg",
    name="X光片",
    mimeType="image/jpeg",
    embedding=image_embedding  # 使用CLIP等模型生成
)

6.2 实时更新机制

使用文件系统监控实现索引自动更新：

python复制from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class DocHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith(".txt"):
            update_index(event.src_path)

6.3 分布式部署

对于大规模文档集，可以采用分片索引：

python复制# 分片策略
shards = [
    faiss.IndexFlatL2(1536) for _ in range(4)
]

# 查询时合并结果
all_results = []
for shard in shards:
    D, I = shard.search(q_emb, top_k)
    all_results.extend(zip(D[0], I[0]))