用Python和FastMCP为AI助手打造专属文档搜索工具:从本地Stdio到云端SSE部署全流程
用Python和FastMCP构建AI文档搜索引擎:从本地到云端的智能解决方案
在AI编程助手日益普及的今天,开发者们面临一个共同挑战:如何让这些助手准确理解并回答特定技术框架的问题?传统方法依赖静态知识库,但技术文档更新频繁,导致回答经常过时。本文将介绍如何利用Python和FastMCP技术,构建一个实时搜索主流AI框架官方文档的智能工具,彻底解决这一痛点。
1. 技术选型与架构设计
1.1 为什么选择MCP协议
MCP(Model Context Protocol)作为新兴的标准化接口协议,解决了AI工具集成中的三个核心问题:
- 接口统一化:不同厂商的Function Call格式差异导致集成困难,MCP提供了类似USB-C的统一接口
- 工具管理标准化:通过声明式描述自动处理输入输出转换,减少胶水代码
- 传输协议多样化:同时支持本地Stdio和云端SSE两种通信模式
对比传统Function Call方式,MCP在文档搜索场景的优势尤为明显:
| 特性 | 传统Function Call | MCP协议 |
|---|---|---|
| 开发效率 | 低(需手动适配) | 高(自动转换) |
| 维护成本 | 高 | 低 |
| 跨平台支持 | 有限 | 完善 |
| 实时性 | 依赖实现 | 内置长连接支持 |
1.2 系统架构设计
整个文档搜索工具采用微服务架构,核心组件包括:
class DocumentSearchSystem: def __init__(self): self.search_engine = SerperAPIWrapper() self.content_extractor = BeautifulSoupParser() self.mcp_server = FastMCPServer() self.cache_layer = LRUCache() async def process_query(self, query: str, library: str): """端到端查询处理流程""" cached = self.cache_layer.get((query, library)) if cached: return cached results = await self.search_engine.search(query, library) extracted = await self.content_extractor.parse(results) self.cache_layer.set((query, library), extracted) return extracted关键设计考虑:
- 缓存层:减少重复搜索的开销
- 异步处理:最大化I/O密集型操作的效率
- 模块化设计:便于更换搜索引擎或解析器
2. 本地Stdio模式实现
2.1 环境配置与依赖安装
推荐使用uv作为Python环境管理工具,跨平台安装命令如下:
# MacOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # Windows powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"项目初始化步骤:
- 创建项目目录并进入
- 初始化虚拟环境
- 安装核心依赖
uv init doc-search && cd doc-search uv venv source .venv/bin/activate # Windows: .venv\Scripts\activate uv add "fastmcp[cli]" httpx beautifulsoup4 python-dotenv2.2 核心搜索功能实现
文档搜索的核心在于精准定位和内容提取,我们采用站点限定搜索策略:
docs_urls = { "langchain": "python.langchain.com/docs", "llama-index": "docs.llamaindex.ai/en/stable", "autogen": "microsoft.github.io/autogen/stable", # 其他框架配置... } async def search_docs(query: str, library: str) -> str: """执行限定站点的文档搜索""" if library not in docs_urls: raise ValueError(f"不支持的框架: {library}") site = docs_urls[library] query = f"site:{site} {query}" results = await serper_search(query) return await extract_content(results)提示:Serper API的免费套餐足够个人开发使用,生产环境建议升级到付费计划
2.3 MCP工具封装
将搜索功能封装为MCP工具,使其可以被AI助手调用:
from mcp import tool @tool() async def get_ai_docs(query: str, framework: str) -> str: """ 获取AI框架技术文档 参数: query: 搜索关键词 framework: 框架名称(langchain|llama-index|...) 返回: 相关文档内容文本 """ return await search_docs(query, framework)3. 云端SSE部署方案
3.1 SSE协议优势分析
相比本地Stdio模式,SSE(Server-Sent Events)特别适合云端部署:
- 跨网络访问:无需SSH隧道或端口转发
- 自动重连:内置心跳机制保持连接
- 多客户端支持:单个服务端可服务多个AI助手
- 资源隔离:搜索服务与客户端环境解耦
3.2 服务端实现
基于Starlette构建SSE服务端:
from fastmcp import FastMCP from mcp.server.sse import SseServerTransport import uvicorn app = FastMCP("ai-docs-searcher") @app.tool() async def search_ai_docs(query: str, framework: str): # 工具实现同上 pass def create_sse_app(): sse_transport = SseServerTransport("/mcp-events/") return sse_transport.create_app(app) if __name__ == "__main__": uvicorn.run(create_sse_app(), host="0.0.0.0", port=8020)启动命令:
uv run server.py --host 0.0.0.0 --port 80203.3 客户端配置
云端部署后,客户端通过HTTP连接服务端:
async with mcp_client.SSEClient("http://your-server:8020/mcp-events") as client: result = await client.call_tool("search_ai_docs", {"query": "agent memory", "framework": "langchain"}) print(result)4. 集成到开发环境
4.1 Cursor集成配置
在Cursor中启用MCP服务只需两步:
- 项目根目录创建
.cursor/mcp.json - 添加服务配置:
{ "mcpServers": { "ai-docs-searcher": { "command": "uv", "args": ["run", "server.py"] } } }集成后效果:
- 输入"/docs"触发文档搜索
- 自动补全框架名称
- 结果直接嵌入对话流
4.2 性能优化技巧
实际使用中发现三个关键优化点:
- 缓存策略:对相同查询返回缓存结果,将API调用减少70%
- 内容精简:提取文档核心内容,去除导航和广告等噪音
- 连接池:复用HTTP连接,降低SSE模式下的延迟
# 优化后的搜索实现 async def optimized_search(query: str, framework: str) -> str: cache_key = f"{query}:{framework}" if cached := cache.get(cache_key): return cached async with httpx.AsyncClient(timeout=30.0) as client: results = await client.post(SERPER_URL, json={"q": query}) cleaned = clean_content(results.text) cache.set(cache_key, cleaned, ttl=3600) return cleaned5. 扩展与定制
5.1 支持新框架的步骤
添加对新AI框架的支持非常简单:
- 在docs_urls字典中添加条目
- 测试站点搜索语法是否有效
- 验证内容提取准确性
例如新增支持HuggingFace Agents:
docs_urls["huggingface"] = "huggingface.co/docs/agents"5.2 高级功能扩展
基于现有架构可以轻松实现更多实用功能:
- 多框架联合搜索:同时查询多个框架的文档
- 版本切换:支持查看不同版本的文档
- 代码示例提取:专门提取文档中的代码片段
- 本地文档索引:对高频访问文档建立本地索引
@tool() async def search_multiple_frameworks(query: str, frameworks: List[str]) -> Dict[str, str]: """同时在多个框架中搜索文档""" results = {} for framework in frameworks: results[framework] = await search_docs(query, framework) return results在实际项目中,这套系统显著提升了开发效率。一个典型场景是当AI助手遇到不熟悉的概念时,会自动触发文档搜索并将最相关的内容整合到回答中,而不是给出可能过时的通用建议。
