MCP模型上下文协议实战:个人应用项目如何集成MCP?
MCP模型上下文协议实战:个人应用项目如何集成MCP?
一、什么是MCP?
MCP(Model Context Protocol,模型上下文协议) 是一种标准化的协议,用于在应用程序与大语言模型(LLM)之间高效、结构化地传递上下文信息。MCP的核心目标是:
- 统一上下文格式:确保不同模型和应用之间上下文传递的一致性
- 提升交互效率:减少冗余信息传输,优化token使用
- 增强可维护性:通过标准化协议降低集成复杂度
MCP通常包含以下核心组件:
- 上下文元数据(Context Metadata)
- 对话历史(Conversation History)
- 工具调用信息(Tool Calls)
- 用户意图识别(User Intent)
- 状态管理(State Management)
二、为什么个人项目需要MCP?
对于个人开发者而言,集成MCP带来以下优势:
- 降低学习成本:统一的协议标准减少对不同API的适配工作
- 提高开发效率:可复用的上下文管理组件
- 更好的调试体验:结构化的上下文便于问题定位
- 未来兼容性:为后续升级到更复杂的AI系统奠定基础
三、实战:个人笔记助手项目集成MCP
3.1 项目背景
我们开发一个智能笔记助手,具备以下功能:
- 自动整理用户输入的零散笔记
- 根据上下文生成摘要
- 支持多轮对话式的笔记编辑
- 能够调用外部工具(如日历、待办事项)
3.2 MCP集成架构设计
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ 用户界面 │ │ MCP上下文 │ │ LLM服务 │
│ (Frontend) │───▶│ 管理器 │───▶│ (Backend) │
└─────────────────┘ └─────────────────┘ └─────────────────┘│▼┌─────────────────┐│ 工具调用 ││ 处理器 │└─────────────────┘
3.3 MCP上下文结构定义
首先定义MCP上下文的数据结构:
from dataclasses import dataclass
from typing import List, Dict, Optional, Any
from enum import Enum
import json
from datetime import datetimeclass ContextType(Enum):CONVERSATION = "conversation"DOCUMENT = "document"TOOL_CALL = "tool_call"@dataclass
class MCPContext:"""MCP上下文协议数据结构"""session_id: str # 会话IDuser_id: str # 用户IDcontext_type: ContextTypemessages: List[Dict[str, Any]] # 消息历史tools: List[Dict[str, Any]] # 可用工具metadata: Dict[str, Any] # 元数据timestamp: datetimestate: Dict[str, Any] # 当前状态def to_dict(self) -> Dict[str, Any]:"""转换为字典格式"""return {"session_id": self.session_id,"user_id": self.user_id,"context_type": self.context_type.value,"messages": self.messages,"tools": self.tools,"metadata": self.metadata,"timestamp": self.timestamp.isoformat(),"state": self.state}@classmethoddef from_dict(cls, data: Dict[str, Any]) -> 'MCPContext':"""从字典创建实例"""return cls(session_id=data["session_id"],user_id=data["user_id"],context_type=ContextType(data["context_type"]),messages=data["messages"],tools=data["tools"],metadata=data["metadata"],timestamp=datetime.fromisoformat(data["timestamp"]),state=data["state"])
3.4 上下文管理器实现
创建MCP上下文管理器来处理上下文的生命周期:
import uuid
from typing import Optionalclass MCPContextManager:"""MCP上下文管理器"""def __init__(self):self.contexts: Dict[str, MCPContext] = {}def create_context(self, user_id: str, context_type: ContextType = ContextType.CONVERSATION) -> MCPContext:"""创建新的上下文"""session_id = str(uuid.uuid4())context = MCPContext(session_id=session_id,user_id=user_id,context_type=context_type,messages=[],tools=self._get_available_tools(),metadata={"created_at": datetime.now().isoformat()},timestamp=datetime.now(),state={})self.contexts[session_id] = contextreturn contextdef get_context(self, session_id: str) -> Optional[MCPContext]:"""获取上下文"""return self.contexts.get(session_id)def update_context(self, session_id: str, message: Dict[str, Any], state_updates: Optional[Dict] = None):"""更新上下文"""if session_id not in self.contexts:raise ValueError(f"Context {session_id} not found")context = self.contexts[session_id]context.messages.append(message)context.timestamp = datetime.now()if state_updates:context.state.update(state_updates)def _get_available_tools(self) -> List[Dict[str, Any]]:"""获取可用工具列表"""return [{"name": "create_todo","description": "创建待办事项","parameters": {"type": "object", "properties": {"task": {"type": "string"}}}},{"name": "format_note","description": "格式化笔记内容","parameters": {"type": "object", "properties": {"content": {"type": "string"}, "format": {"type": "string", "enum": ["markdown", "plain"]}}}}]def serialize_context(self, session_id: str) -> str:"""序列化上下文为JSON字符串"""context = self.get_context(session_id)if context:return json.dumps(context.to_dict(), ensure_ascii=False)return ""def deserialize_context(self, context_json: str) -> MCPContext:"""从JSON字符串反序列化上下文"""data = json.loads(context_json)context = MCPContext.from_dict(data)self.contexts[context.session_id] = contextreturn context
3.5 与LLM服务集成
实现与LLM服务的集成,使用MCP上下文:
import openai
from typing import Dict, Any, Listclass LLMService:"""LLM服务封装"""def __init__(self, api_key: str):openai.api_key = api_keyself.context_manager = MCPContextManager()def process_message(self, user_id: str, user_message: str, session_id: Optional[str] = None) -> Dict[str, Any]:"""处理用户消息"""# 获取或创建上下文if session_id and self.context_manager.get_context(session_id):context = self.context_manager.get_context(session_id)else:context = self.context_manager.create_context(user_id)session_id = context.session_id# 添加用户消息到上下文user_msg = {"role": "user", "content": user_message}self.context_manager.update_context(session_id, user_msg)# 准备LLM请求messages = self._prepare_messages(context)tools = context.tools# 调用LLMtry:response = openai.ChatCompletion.create(model="gpt-3.5-turbo",messages=messages,tools=tools,tool_choice="auto")# 处理响应assistant_message = response.choices[0].message# 更新上下文self.context_manager.update_context(session_id, assistant_message.to_dict())return {"session_id": session_id,"response": assistant_message.get("content", ""),"tool_calls": assistant_message.get("tool_calls", [])}except Exception as e:return {"error": str(e), "session_id": session_id}def _prepare_messages(self, context: MCPContext) -> List[Dict[str, Any]]:"""准备发送给LLM的消息"""# 系统消息system_message = {"role": "system","content": "你是一个智能笔记助手,帮助用户整理笔记、创建待办事项。"}return [system_message] + context.messages
3.6 工具调用处理器
实现工具调用的处理逻辑:
class ToolHandler:"""工具调用处理器"""def __init__(self):self.todo_list = []def handle_tool_calls(self, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:"""处理工具调用"""results = []for tool_call in tool_calls:tool_name = tool_call["function"]["name"]arguments = json.loads(tool_call["function"]["arguments"])if tool_name == "create_todo":result = self._create_todo(arguments["task"])elif tool_name == "format_note":result = self._format_note(arguments["content"], arguments["format"])else:result = f"Unknown tool: {tool_name}"results.append({"tool_call_id": tool_call["id"],"role": "tool","name": tool_name,"content": json.dumps(result)})return resultsdef _create_todo(self, task: str) -> Dict[str, Any]:"""创建待办事项"""todo_item = {"id": len(self.todo_list) + 1,"task": task,"created_at": datetime.now().isoformat(),"completed": False}self.todo_list.append(todo_item)return {"status": "success", "todo": todo_item}def _format_note(self, content: str, format_type: str) -> Dict[str, Any]:"""格式化笔记"""if format_type == "markdown":# 简单的Markdown格式化formatted = f"# Note\n\n{content}"else:formatted = contentreturn {"status": "success", "formatted_content": formatted}
3.7 完整的API端点
创建Flask API端点来暴露服务:
from flask import Flask, request, jsonifyapp = Flask(__name__)
llm_service = LLMService("your-api-key")
tool_handler = ToolHandler()@app.route('/api/chat', methods=['POST'])
def chat():"""聊天API端点"""data = request.jsonuser_id = data.get('user_id', 'anonymous')message = data.get('message', '')session_id = data.get('session_id')# 处理消息response = llm_service.process_message(user_id, message, session_id)if "error" in response:return jsonify(response), 500# 处理工具调用if response.get("tool_calls"):tool_results = tool_handler.handle_tool_calls(response["tool_calls"])# 将工具结果添加到上下文并再次调用LLMsession_id = response["session_id"]context = llm_service.context_manager.get_context(session_id)for tool_result in tool_results:llm_service.context_manager.update_context(session_id, tool_result)# 再次调用LLM获取最终响应final_messages = llm_service._prepare_messages(context)final_response = openai.ChatCompletion.create(model="gpt-3.5-turbo",messages=final_messages)final_content = final_response.choices[0].message.get("content", "")response["response"] = final_contentreturn jsonify({"session_id": response["session_id"],"message": response["response"]})if __name__ == '__main__':app.run(debug=True)
四、最佳实践与注意事项
4.1 上下文长度管理
- 设置最大上下文长度:避免token超限
- 实现上下文压缩:对历史消息进行摘要
- 智能上下文清理:移除不相关的旧消息
def trim_context(self, context: MCPContext, max_tokens: int = 3000):"""修剪上下文以控制token数量"""# 实现基于重要性的消息保留策略pass
4.2 错误处理与恢复
- 上下文持久化:定期保存上下文状态
- 异常恢复机制:在错误发生时能够恢复上下文
- 用户友好的错误提示:提供清晰的错误信息
4.3 安全性考虑
- 输入验证:验证用户输入的安全性
- 敏感信息过滤:避免在上下文中存储敏感数据
- 访问控制:确保用户只能访问自己的上下文
五、总结
通过在个人笔记助手项目中集成MCP模型上下文协议,我们实现了:
- 结构化的上下文管理:清晰的数据结构和生命周期管理
- 高效的LLM交互:减少冗余信息,优化token使用
- 可扩展的架构:易于添加新功能和工具
- 良好的维护性:标准化的协议降低了代码复杂度
MCP不仅适用于大型企业项目,在个人应用开发中同样能带来显著的价值。通过标准化上下文处理,开发者可以更专注于业务逻辑的实现,而不是底层的协议适配。
下一步建议:
- 实现上下文持久化到数据库
- 添加上下文版本控制
- 集成更多的外部工具和服务
- 优化上下文压缩算法
通过持续迭代和完善MCP集成,你的个人项目将具备更强的AI能力和更好的用户体验。