外观
05.MCP模块化开发最佳实践
来源：语雀原文 | 整理日期：2026-09-22
示例代码
MCP核心架构与设计理念剖析
理论知识
Model Context Protocol (MCP) 是一个开源标准协议，由 Anthropic 于2024年发布，旨在为AI助手与外部数据源和工具之间建立统一的连接标准。MCP的核心设计理念包括：
统一标准化接口：MCP提供了一个通用协议来替代碎片化的集成方式，使AI应用能够通过标准化接口访问各种外部资源、工具和提示模板。
客户端-主机-服务器架构：
- 客户端（Client）：AI应用或工具，如Claude Desktop、Cursor等
- 主机（Host）：运行客户端的环境，可以管理多个客户端实例
- 服务器（Server）：提供具体功能的后端服务，如数据库访问、文件系统操作等
基于能力的协商机制：客户端和服务器在初始化时明确声明支持的功能特性，通过能力协商确定会话期间可用的协议功能。
安全性优先：2025年6月的规范更新引入了资源指示器和增强的授权处理机制，防止恶意服务器获取访问令牌。
最佳实践
- 安全性设计：始终实施强大的同意和授权流程，提供清晰的安全影响文档
- 模块化设计：采用松耦合的架构，便于维护和扩展
- 错误处理：实现完善的错误处理和回退机制
- 性能优化：合理使用缓存和连接池来提升性能
Python示例代码
plain
"""
MCP基础概念演示
"""
from mcp import ClientSession, StdioServerTransport
import asyncio
class MCPBasicExample:
def __init__(self):
self.session = None
async def initialize_connection(self, command: list[str]):
"""初始化MCP连接"""
# 创建STDIO传输
transport = StdioServerTransport(command)
# 建立客户端会话
self.session = ClientSession(transport)
# 初始化连接
await self.session.initialize()
print("MCP连接初始化完成")
async def get_server_capabilities(self):
"""获取服务器能力"""
if not self.session:
raise Exception("会话未初始化")
capabilities = self.session.server_capabilities
print(f"服务器能力: {capabilities}")
return capabilities
async def list_resources(self):
"""列出可用资源"""
if not self.session:
raise Exception("会话未初始化")
resources = await self.session.list_resources()
print(f"可用资源: {resources}")
return resources
# 使用示例
async def main():
example = MCPBasicExample()
await example.initialize_connection(["python", "simple_server.py"])
await example.get_server_capabilities()
await example.list_resources()
if __name__ == "__main__":
asyncio.run(main())MCP上下文架构深度剖析
理论知识
MCP的多上下文架构是其核心优势之一，通过以下机制支撑智能对话：
上下文管理机制：
- 会话上下文：维护单次对话的状态信息
- 资源上下文：管理外部数据源的访问状态
- 工具上下文：跟踪工具调用的执行状态
- 全局上下文：保持跨会话的持久化信息
上下文同步策略：
- 实时同步：关键状态变更立即同步
- 批量同步：非关键信息批量更新
- 延迟同步：后台异步处理大量数据
内存管理：通过LRU缓存、压缩存储和垃圾回收机制优化内存使用。
最佳实践
- 上下文隔离：不同会话间的上下文应完全隔离，避免数据泄露
- 状态持久化：关键状态信息应持久化存储，支持故障恢复
- 内存优化：合理设置上下文大小限制，防止内存溢出
- 并发处理：使用异步编程模型处理多个上下文
Python示例代码
plain
"""
多上下文管理系统
"""
import asyncio
import json
import time
from typing import Dict, Any, Optional
from dataclasses import dataclass
from collections import defaultdict
@dataclass
class ContextInfo:
context_id: str
created_at: float
last_accessed: float
data: Dict[str, Any]
session_id: str
class MultiContextManager:
def __init__(self, max_contexts: int = 1000):
self.contexts: Dict[str, ContextInfo] = {}
self.session_contexts: Dict[str, list[str]] = defaultdict(list)
self.max_contexts = max_contexts
async def create_context(self, session_id: str, context_data: Dict[str, Any]) -> str:
"""创建新的上下文"""
context_id = f"{session_id}_{int(time.time() * 1000)}"
# 检查上下文数量限制
if len(self.contexts) >= self.max_contexts:
await self._cleanup_old_contexts()
context_info = ContextInfo(
context_id=context_id,
created_at=time.time(),
last_accessed=time.time(),
data=context_data.copy(),
session_id=session_id
)
self.contexts[context_id] = context_info
self.session_contexts[session_id].append(context_id)
print(f"创建上下文: {context_id}")
return context_id
async def get_context(self, context_id: str) -> Optional[Dict[str, Any]]:
"""获取上下文数据"""
if context_id not in self.contexts:
return None
context_info = self.contexts[context_id]
context_info.last_accessed = time.time()
return context_info.data.copy()
async def update_context(self, context_id: str, updates: Dict[str, Any]):
"""更新上下文数据"""
if context_id not in self.contexts:
raise ValueError(f"上下文不存在: {context_id}")
context_info = self.contexts[context_id]
context_info.data.update(updates)
context_info.last_accessed = time.time()
print(f"更新上下文: {context_id}")
async def merge_contexts(self, target_context_id: str, source_context_id: str):
"""合并两个上下文"""
target_context = self.contexts.get(target_context_id)
source_context = self.contexts.get(source_context_id)
if not target_context or not source_context:
raise ValueError("目标或源上下文不存在")
# 合并数据
merged_data = {**target_context.data, **source_context.data}
target_context.data = merged_data
target_context.last_accessed = time.time()
# 删除源上下文
await self.delete_context(source_context_id)
print(f"合并上下文: {source_context_id} -> {target_context_id}")
async def get_session_contexts(self, session_id: str) -> list[str]:
"""获取会话的所有上下文"""
return self.session_contexts.get(session_id, [])
async def delete_context(self, context_id: str):
"""删除上下文"""
if context_id not in self.contexts:
return
context_info = self.contexts[context_id]
session_id = context_info.session_id
# 从会话上下文列表中移除
if session_id in self.session_contexts:
self.session_contexts[session_id].remove(context_id)
# 删除上下文
del self.contexts[context_id]
print(f"删除上下文: {context_id}")
async def _cleanup_old_contexts(self):
"""清理旧的上下文"""
current_time = time.time()
timeout = 3600 # 1小时超时
contexts_to_delete = [
context_id for context_id, context_info in self.contexts.items()
if current_time - context_info.last_accessed > timeout
]
for context_id in contexts_to_delete:
await self.delete_context(context_id)
print(f"清理了 {len(contexts_to_delete)} 个过期上下文")
# 使用示例
async def demo_multi_context():
manager = MultiContextManager()
# 创建会话上下文
session_id = "session_123"
context_id1 = await manager.create_context(session_id, {
"user_id": "user_456",
"conversation_history": ["Hello", "How can I help you?"]
})
context_id2 = await manager.create_context(session_id, {
"current_task": "file_analysis",
"file_path": "/data/report.txt"
})
# 获取上下文
context_data = await manager.get_context(context_id1)
print(f"上下文数据: {context_data}")
# 更新上下文
await manager.update_context(context_id1, {
"conversation_history": ["Hello", "How can I help you?", "I need help with data analysis"]
})
# 合并上下文
await manager.merge_contexts(context_id1, context_id2)
# 获取会话的所有上下文
session_contexts = await manager.get_session_contexts(session_id)
print(f"会话上下文: {session_contexts}")
if __name__ == "__main__":
asyncio.run(demo_multi_context())MCP 单上下文 vs 多上下文差异与优势
理论知识
单上下文模式：
- 特点：每次对话独立，无状态保持
- 优势：简单、资源消耗低、无并发冲突
- 劣势：无法保持连续对话、功能受限
多上下文模式：
- 特点：维护多个相关上下文，支持复杂交互
- 优势：连续对话、状态保持、功能丰富
- 劣势：复杂度高、资源消耗大
关键差异对比：
| 维度 | 单上下文 | 多上下文 |
| 内存使用 | 低 | 高 |
| 实现复杂度 | 简单 | 复杂 |
| 功能丰富度 | 基础 | 高级 |
| 并发处理 | 容易 | 困难 |
| 用户体验 | 基础 | 优秀 |
最佳实践
- 场景选择：简单任务使用单上下文，复杂应用使用多上下文
- 混合模式：根据功能需求动态选择上下文模式
- 资源监控：实时监控内存和CPU使用情况
- 优雅降级：在资源不足时自动降级到单上下文模式
Python示例代码
plain
"""
单上下文vs多上下文对比实现
"""
import asyncio
import time
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
@dataclass
class Message:
content: str
timestamp: float
user_id: str
message_type: str = "user"
class ContextHandler(ABC):
@abstractmethod
async def process_message(self, message: Message) -> str:
pass
@abstractmethod
async def get_context_info(self) -> Dict[str, Any]:
pass
class SingleContextHandler(ContextHandler):
"""单上下文处理器"""
def __init__(self):
self.current_message: Optional[Message] = None
async def process_message(self, message: Message) -> str:
"""处理单条消息，不保留历史"""
self.current_message = message
# 简单的响应逻辑
if "hello" in message.content.lower():
response = "Hello! How can I help you today?"
elif "help" in message.content.lower():
response = "I'm here to assist you with your questions."
else:
response = f"I understand you said: {message.content}"
return response
async def get_context_info(self) -> Dict[str, Any]:
"""获取当前上下文信息"""
return {
"type": "single_context",
"current_message": self.current_message.content if self.current_message else None,
"memory_usage": "low"
}
class MultiContextHandler(ContextHandler):
"""多上下文处理器"""
def __init__(self):
self.conversation_history: List[Message] = []
self.user_profiles: Dict[str, Dict[str, Any]] = {}
self.session_data: Dict[str, Any] = {}
async def process_message(self, message: Message) -> str:
"""处理消息，保持完整上下文"""
# 添加到历史记录
self.conversation_history.append(message)
# 更新用户档案
await self._update_user_profile(message)
# 基于历史上下文生成响应
response = await self._generate_contextual_response(message)
# 添加响应到历史记录
response_message = Message(
content=response,
timestamp=time.time(),
user_id="assistant",
message_type="assistant"
)
self.conversation_history.append(response_message)
return response
async def _update_user_profile(self, message: Message):
"""更新用户档案"""
user_id = message.user_id
if user_id not in self.user_profiles:
self.user_profiles[user_id] = {
"first_seen": message.timestamp,
"message_count": 0,
"topics": set()
}
profile = self.user_profiles[user_id]
profile["message_count"] += 1
profile["last_seen"] = message.timestamp
# 简单的主题提取
if "data" in message.content.lower():
profile["topics"].add("data_analysis")
if "help" in message.content.lower():
profile["topics"].add("help_seeking")
async def _generate_contextual_response(self, message: Message) -> str:
"""基于上下文生成响应"""
user_profile = self.user_profiles.get(message.user_id, {})
# 检查是否是回头客
if user_profile.get("message_count", 0) > 1:
greeting = f"Welcome back! We've chatted {user_profile['message_count']} times."
else:
greeting = "Nice to meet you!"
# 基于历史对话生成响应
recent_messages = self.conversation_history[-5:] # 最近5条消息
context_info = f"Based on our conversation history of {len(recent_messages)} recent messages..."
if "hello" in message.content.lower():
return f"{greeting} How can I help you today?"
elif "what did we discuss" in message.content.lower():
topics = list(user_profile.get("topics", set()))
return f"We've discussed topics like: {', '.join(topics)}"
else:
return f"{context_info} I understand: {message.content}"
async def get_context_info(self) -> Dict[str, Any]:
"""获取多上下文信息"""
return {
"type": "multi_context",
"conversation_length": len(self.conversation_history),
"active_users": len(self.user_profiles),
"memory_usage": "high",
"context_details": {
"total_messages": len(self.conversation_history),
"unique_users": list(self.user_profiles.keys())
}
}
class ContextComparator:
"""上下文模式对比工具"""
def __init__(self):
self.single_handler = SingleContextHandler()
self.multi_handler = MultiContextHandler()
async def compare_performance(self, messages: List[Message]):
"""对比两种模式的性能"""
print("=== 性能对比测试 ===\n")
# 测试单上下文模式
print("1. 单上下文模式测试:")
start_time = time.time()
for message in messages:
response = await self.single_handler.process_message(message)
print(f"用户: {message.content}")
print(f"助手: {response}\n")
single_time = time.time() - start_time
single_info = await self.single_handler.get_context_info()
print(f"单上下文模式耗时: {single_time:.4f}秒")
print(f"上下文信息: {single_info}\n")
# 测试多上下文模式
print("2. 多上下文模式测试:")
start_time = time.time()
for message in messages:
response = await self.multi_handler.process_message(message)
print(f"用户: {message.content}")
print(f"助手: {response}\n")
multi_time = time.time() - start_time
multi_info = await self.multi_handler.get_context_info()
print(f"多上下文模式耗时: {multi_time:.4f}秒")
print(f"上下文信息: {multi_info}\n")
# 性能对比结果
print("=== 对比结果 ===")
print(f"性能差异: {((multi_time - single_time) / single_time * 100):.2f}%")
print(f"功能丰富度: 多上下文模式显著优于单上下文模式")
# 使用示例
async def demo_context_comparison():
comparator = ContextComparator()
# 构造测试消息
test_messages = [
Message("Hello", time.time(), "user_1"),
Message("I need help with data analysis", time.time() + 1, "user_1"),
Message("What did we discuss before?", time.time() + 2, "user_1"),
Message("Hello", time.time() + 3, "user_2"),
]
await comparator.compare_performance(test_messages)
if __name__ == "__main__":
asyncio.run(demo_context_comparison())MCP多上下文能够解决那些场景问题
理论知识
多上下文技术能够解决以下关键问题：
1. 对话连续性问题
- 问题描述：传统AI助手无法记住之前的对话内容
- 解决方案：通过对话历史上下文维护完整的交互记录
- 应用场景：客服系统、教育辅导、个人助理
2. 个性化服务问题
- 问题描述：无法根据用户偏好提供定制化服务
- 解决方案：建立用户档案上下文，学习用户习惯和偏好
- 应用场景：推荐系统、内容定制、智能客服
3. 复杂任务处理问题
- 问题描述：单一上下文无法处理需要多步骤的复杂任务
- 解决方案：任务状态上下文跟踪整个执行过程
- 应用场景：项目管理、工作流自动化、数据分析
4. 多用户协作问题
- 问题描述：多人同时使用时出现状态混乱
- 解决方案：会话隔离和状态同步机制
- 应用场景：团队协作工具、在线教育平台
最佳实践
- 场景识别：准确识别需要多上下文的应用场景
- 上下文设计：合理设计上下文结构和生命周期
- 性能优化：使用缓存和索引提升检索性能
- 数据安全：确保不同用户的上下文完全隔离
Python示例代码
plain
"""
多上下文典型应用场景实现
"""
import asyncio
import json
import time
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import uuid
class TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class Task:
task_id: str
title: str
description: str
status: TaskStatus
created_at: float
updated_at: float
user_id: str
progress: int = 0
metadata: Dict[str, Any] = None
@dataclass
class UserProfile:
user_id: str
name: str
preferences: Dict[str, Any]
interaction_history: List[Dict[str, Any]]
created_at: float
last_active: float
class IntelligentAssistant:
"""智能助手 - 多上下文应用场景实现"""
def __init__(self):
# 对话上下文
self.conversations: Dict[str, List[Dict[str, Any]]] = {}
# 用户档案上下文
self.user_profiles: Dict[str, UserProfile] = {}
# 任务管理上下文
self.tasks: Dict[str, Task] = {}
self.user_tasks: Dict[str, List[str]] = {}
# 会话上下文
self.active_sessions: Dict[str, Dict[str, Any]] = {}
# =================== 对话连续性场景 ===================
async def handle_conversation(self, user_id: str, message: str) -> str:
"""处理连续对话"""
# 初始化用户对话历史
if user_id not in self.conversations:
self.conversations[user_id] = []
# 添加用户消息到历史
self.conversations[user_id].append({
"role": "user",
"content": message,
"timestamp": time.time()
})
# 基于对话历史生成响应
response = await self._generate_contextual_response(user_id, message)
# 添加助手响应到历史
self.conversations[user_id].append({
"role": "assistant",
"content": response,
"timestamp": time.time()
})
return response
async def _generate_contextual_response(self, user_id: str, message: str) -> str:
"""基于上下文生成响应"""
conversation = self.conversations.get(user_id, [])
# 分析对话历史
if len(conversation) == 1: # 首次对话
return f"Hello! Nice to meet you. You said: {message}"
# 查找相关历史对话
recent_topics = self._extract_topics_from_history(conversation[-5:])
if "continue" in message.lower() or "before" in message.lower():
return f"Based on our previous discussion about {', '.join(recent_topics)}, let me continue..."
return f"Continuing our conversation (we've discussed {', '.join(recent_topics)}): {message}"
def _extract_topics_from_history(self, messages: List[Dict[str, Any]]) -> List[str]:
"""从历史消息中提取主题"""
topics = set()
for msg in messages:
content = msg.get("content", "").lower()
if "data" in content:
topics.add("data analysis")
if "task" in content:
topics.add("task management")
if "file" in content:
topics.add("file processing")
return list(topics)
# =================== 个性化服务场景 ===================
async def update_user_profile(self, user_id: str, name: str, preferences: Dict[str, Any]):
"""更新用户档案"""
current_time = time.time()
if user_id in self.user_profiles:
profile = self.user_profiles[user_id]
profile.name = name
profile.preferences.update(preferences)
profile.last_active = current_time
else:
profile = UserProfile(
user_id=user_id,
name=name,
preferences=preferences,
interaction_history=[],
created_at=current_time,
last_active=current_time
)
self.user_profiles[user_id] = profile
print(f"用户档案已更新: {name}")
async def get_personalized_recommendation(self, user_id: str) -> str:
"""获取个性化推荐"""
profile = self.user_profiles.get(user_id)
if not profile:
return "请先设置您的个人档案以获得个性化推荐"
preferences = profile.preferences
name = profile.name
# 基于偏好生成推荐
recommendations = []
if preferences.get("interests", {}).get("technology", False):
recommendations.append("最新技术文章和工具推荐")
if preferences.get("work_style") == "productivity_focused":
recommendations.append("效率提升工具和方法")
if preferences.get("learning_preference") == "hands_on":
recommendations.append("实践性强的教程和项目")
return f"Hi {name}! 基于您的偏好，为您推荐: {'; '.join(recommendations)}"
# =================== 复杂任务处理场景 ===================
async def create_task(self, user_id: str, title: str, description: str) -> str:
"""创建任务"""
task_id = str(uuid.uuid4())
current_time = time.time()
task = Task(
task_id=task_id,
title=title,
description=description,
status=TaskStatus.PENDING,
created_at=current_time,
updated_at=current_time,
user_id=user_id,
metadata={}
)
self.tasks[task_id] = task
if user_id not in self.user_tasks:
self.user_tasks[user_id] = []
self.user_tasks[user_id].append(task_id)
return f"任务已创建: {title} (ID: {task_id})"
async def update_task_progress(self, task_id: str, progress: int, status: TaskStatus = None):
"""更新任务进度"""
if task_id not in self.tasks:
raise ValueError(f"任务不存在: {task_id}")
task = self.tasks[task_id]
task.progress = progress
task.updated_at = time.time()
if status:
task.status = status
print(f"任务进度更新: {task.title} - {progress}%")
async def get_task_status(self, user_id: str) -> str:
"""获取用户任务状态"""
user_task_ids = self.user_tasks.get(user_id, [])
if not user_task_ids:
return "您目前没有任何任务"
status_summary = {}
for task_id in user_task_ids:
task = self.tasks[task_id]
status = task.status.value
if status not in status_summary:
status_summary[status] = 0
status_summary[status] += 1
summary_text = []
for status, count in status_summary.items():
summary_text.append(f"{status}: {count}个")
return f"任务状态汇总: {', '.join(summary_text)}"
# =================== 多用户协作场景 ===================
async def create_collaboration_session(self, session_name: str, participants: List[str]) -> str:
"""创建协作会话"""
session_id = str(uuid.uuid4())
session_data = {
"session_id": session_id,
"name": session_name,
"participants": participants,
"created_at": time.time(),
"shared_context": {},
"messages": [],
"active": True
}
self.active_sessions[session_id] = session_data
return f"协作会话已创建: {session_name} (ID: {session_id}), 参与者: {', '.join(participants)}"
async def add_to_shared_context(self, session_id: str, user_id: str, key: str, value: Any):
"""添加到共享上下文"""
if session_id not in self.active_sessions:
raise ValueError(f"会话不存在: {session_id}")
session = self.active_sessions[session_id]
if user_id not in session["participants"]:
raise ValueError(f"用户 {user_id} 不是会话参与者")
session["shared_context"][key] = {
"value": value,
"added_by": user_id,
"timestamp": time.time()
}
print(f"共享上下文已更新: {key} = {value}")
async def get_session_summary(self, session_id: str) -> str:
"""获取会话摘要"""
if session_id not in self.active_sessions:
return "会话不存在"
session = self.active_sessions[session_id]
summary = f"""
协作会话摘要:
- 会话名称: {session['name']}
- 参与者: {', '.join(session['participants'])}
- 创建时间: {time.ctime(session['created_at'])}
- 共享上下文项目: {len(session['shared_context'])}
- 消息数量: {len(session['messages'])}
""".strip()
return summary
# =================== 综合使用示例 ===================
async def demo_multi_context_scenarios():
"""演示多上下文应用场景"""
assistant = IntelligentAssistant()
print("=== 多上下文应用场景演示 ===\n")
# 1. 对话连续性场景
print("1. 对话连续性场景:")
response1 = await assistant.handle_conversation("user_1", "Hello, I need help with data analysis")
print(f"用户: Hello, I need help with data analysis")
print(f"助手: {response1}")
response2 = await assistant.handle_conversation("user_1", "Can you continue from where we left off?")
print(f"用户: Can you continue from where we left off?")
print(f"助手: {response2}\n")
# 2. 个性化服务场景
print("2. 个性化服务场景:")
await assistant.update_user_profile("user_1", "Alice", {
"interests": {"technology": True, "data_science": True},
"work_style": "productivity_focused",
"learning_preference": "hands_on"
})
recommendation = await assistant.get_personalized_recommendation("user_1")
print(f"个性化推荐: {recommendation}\n")
# 3. 复杂任务处理场景
print("3. 复杂任务处理场景:")
task_msg = await assistant.create_task("user_1", "数据分析项目", "分析销售数据并生成报告")
print(task_msg)
# 模拟任务进度更新
task_id = list(assistant.tasks.keys())[0]
await assistant.update_task_progress(task_id, 25, TaskStatus.IN_PROGRESS)
await assistant.update_task_progress(task_id, 75, TaskStatus.IN_PROGRESS)
status = await assistant.get_task_status("user_1")
print(f"任务状态: {status}\n")
# 4. 多用户协作场景
print("4. 多用户协作场景:")
session_msg = await assistant.create_collaboration_session("数据分析团队", ["user_1", "user_2", "user_3"])
print(session_msg)
session_id = list(assistant.active_sessions.keys())[0]
await assistant.add_to_shared_context(session_id, "user_1", "dataset_path", "/data/sales_2024.csv")
await assistant.add_to_shared_context(session_id, "user_2", "analysis_method", "regression_analysis")
session_summary = await assistant.get_session_summary(session_id)
print(f"会话摘要:\n{session_summary}")
if __name__ == "__main__":
asyncio.run(demo_multi_context_scenarios())MCP模块化开发最佳实践
理论知识
模块化MCP开发是构建可扩展、可维护MCP应用的关键方法。主要包括以下核心概念：
模块化架构设计：
- 传输层模块：负责底层通信，支持STDIO、HTTP+SSE等传输方式
- 协议层模块：实现JSON-RPC 2.0协议处理和消息路由
- 业务逻辑模块：处理具体的资源、工具和提示功能
- 上下文管理模块：管理会话状态和多上下文处理
模块间通信机制：
- 事件驱动：使用事件系统实现松耦合的模块通信
- 依赖注入：通过依赖注入容器管理模块依赖关系
- 接口抽象：定义清晰的接口契约，支持模块替换和扩展
插件化设计：
- 动态加载：支持运行时加载和卸载功能模块
- 配置驱动：通过配置文件定义模块启用状态和参数
- 热更新：支持不停机更新功能模块
最佳实践
- 单一职责：每个模块只负责一个特定的功能域
- 接口隔离：定义最小化的接口，避免不必要的依赖
- 配置外置：将配置信息从代码中分离出来
- 错误边界：在模块边界实现错误隔离和恢复
- 测试友好：设计易于单元测试的模块结构
Python示例代码
plain
"""
模块化MCP架构实现
"""
import asyncio
import json
import abc
from typing import Dict, Any, List, Optional, Callable, Type
from dataclasses import dataclass
import importlib
import logging
from pathlib import Path
# =================== 核心接口定义 ===================
class MCPModule(abc.ABC):
"""MCP模块基类"""
@property
@abc.abstractmethod
def module_name(self) -> str:
"""模块名称"""
pass
@property
@abc.abstractmethod
def version(self) -> str:
"""模块版本"""
pass
@abc.abstractmethod
async def initialize(self, config: Dict[str, Any]) -> bool:
"""初始化模块"""
pass
@abc.abstractmethod
async def cleanup(self):
"""清理模块资源"""
pass
@abc.abstractmethod
async def health_check(self) -> bool:
"""健康检查"""
pass
class TransportModule(MCPModule):
"""传输模块接口"""
@abc.abstractmethod
async def send_message(self, message: Dict[str, Any]) -> bool:
"""发送消息"""
pass
@abc.abstractmethod
async def receive_message(self) -> Optional[Dict[str, Any]]:
"""接收消息"""
pass
@abc.abstractmethod
async def connect(self) -> bool:
"""建立连接"""
pass
@abc.abstractmethod
async def disconnect(self):
"""断开连接"""
pass
class ProtocolModule(MCPModule):
"""协议模块接口"""
@abc.abstractmethod
async def process_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""处理请求"""
pass
@abc.abstractmethod
async def validate_message(self, message: Dict[str, Any]) -> bool:
"""验证消息格式"""
pass
class BusinessModule(MCPModule):
"""业务模块接口"""
@abc.abstractmethod
async def handle_resource_request(self, resource_uri: str) -> Any:
"""处理资源请求"""
pass
@abc.abstractmethod
async def handle_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""处理工具调用"""
pass
# =================== 具体模块实现 ===================
class StdioTransportModule(TransportModule):
"""STDIO传输模块"""
def __init__(self):
self.connected = False
self.process = None
@property
def module_name(self) -> str:
return "stdio_transport"
@property
def version(self) -> str:
return "1.0.0"
async def initialize(self, config: Dict[str, Any]) -> bool:
"""初始化STDIO传输"""
self.command = config.get("command", [])
logging.info(f"初始化STDIO传输: {self.command}")
return True
async def connect(self) -> bool:
"""建立STDIO连接"""
try:
# 这里应该启动子进程
self.connected = True
logging.info("STDIO连接已建立")
return True
except Exception as e:
logging.error(f"STDIO连接失败: {e}")
return False
async def send_message(self, message: Dict[str, Any]) -> bool:
"""发送消息到STDIO"""
if not self.connected:
return False
try:
message_json = json.dumps(message)
logging.debug(f"发送STDIO消息: {message_json}")
return True
except Exception as e:
logging.error(f"发送STDIO消息失败: {e}")
return False
async def receive_message(self) -> Optional[Dict[str, Any]]:
"""从STDIO接收消息"""
if not self.connected:
return None
# 模拟接收消息
return {"jsonrpc": "2.0", "method": "ping", "id": 1}
async def disconnect(self):
"""断开STDIO连接"""
self.connected = False
logging.info("STDIO连接已断开")
async def cleanup(self):
"""清理资源"""
await self.disconnect()
async def health_check(self) -> bool:
"""健康检查"""
return self.connected
class JsonRpcProtocolModule(ProtocolModule):
"""JSON-RPC协议模块"""
def __init__(self):
self.request_id = 0
@property
def module_name(self) -> str:
return "jsonrpc_protocol"
@property
def version(self) -> str:
return "2.0.0"
async def initialize(self, config: Dict[str, Any]) -> bool:
"""初始化协议模块"""
self.timeout = config.get("timeout", 30)
logging.info("JSON-RPC协议模块已初始化")
return True
async def validate_message(self, message: Dict[str, Any]) -> bool:
"""验证JSON-RPC消息格式"""
required_fields = ["jsonrpc"]
for field in required_fields:
if field not in message:
return False
if message["jsonrpc"] != "2.0":
return False
# 检查是否是请求或响应
is_request = "method" in message
is_response = "result" in message or "error" in message
return is_request or is_response
async def process_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""处理JSON-RPC请求"""
if not await self.validate_message(request):
return self._create_error_response(
request.get("id"), -32600, "Invalid Request"
)
method = request.get("method")
params = request.get("params", {})
request_id = request.get("id")
logging.info(f"处理JSON-RPC请求: {method}")
# 这里应该路由到具体的处理器
result = {"status": "success", "method": method}
return {
"jsonrpc": "2.0",
"result": result,
"id": request_id
}
def _create_error_response(self, request_id: Any, code: int, message: str) -> Dict[str, Any]:
"""创建错误响应"""
return {
"jsonrpc": "2.0",
"error": {"code": code, "message": message},
"id": request_id
}
async def cleanup(self):
"""清理资源"""
pass
async def health_check(self) -> bool:
"""健康检查"""
return True
class FileSystemBusinessModule(BusinessModule):
"""文件系统业务模块"""
def __init__(self):
self.base_path = None
@property
def module_name(self) -> str:
return "filesystem_business"
@property
def version(self) -> str:
return "1.0.0"
async def initialize(self, config: Dict[str, Any]) -> bool:
"""初始化文件系统模块"""
self.base_path = Path(config.get("base_path", "."))
self.allowed_extensions = config.get("allowed_extensions", [".txt", ".json", ".py"])
logging.info(f"文件系统模块初始化: {self.base_path}")
return True
async def handle_resource_request(self, resource_uri: str) -> Any:
"""处理文件资源请求"""
try:
if resource_uri.startswith("file://"):
file_path = Path(resource_uri[7:]) # 移除 file:// 前缀
# 安全检查
if not self._is_safe_path(file_path):
raise ValueError("不安全的文件路径")
full_path = self.base_path / file_path
if full_path.exists() and full_path.is_file():
content = full_path.read_text(encoding="utf-8")
return {
"uri": resource_uri,
"content": content,
"mimeType": "text/plain"
}
else:
raise FileNotFoundError(f"文件不存在: {full_path}")
except Exception as e:
logging.error(f"处理资源请求失败: {e}")
raise
async def handle_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""处理工具调用"""
if tool_name == "list_files":
return await self._list_files(arguments.get("directory", "."))
elif tool_name == "read_file":
return await self._read_file(arguments.get("path"))
elif tool_name == "write_file":
return await self._write_file(arguments.get("path"), arguments.get("content"))
else:
raise ValueError(f"未知工具: {tool_name}")
async def _list_files(self, directory: str) -> List[Dict[str, Any]]:
"""列出文件"""
dir_path = self.base_path / directory
if not self._is_safe_path(Path(directory)):
raise ValueError("不安全的目录路径")
files = []
for item in dir_path.iterdir():
if item.is_file() and item.suffix in self.allowed_extensions:
files.append({
"name": item.name,
"path": str(item.relative_to(self.base_path)),
"size": item.stat().st_size,
"type": "file"
})
elif item.is_dir():
files.append({
"name": item.name,
"path": str(item.relative_to(self.base_path)),
"type": "directory"
})
return files
async def _read_file(self, path: str) -> str:
"""读取文件"""
file_path = Path(path)
if not self._is_safe_path(file_path):
raise ValueError("不安全的文件路径")
full_path = self.base_path / file_path
return full_path.read_text(encoding="utf-8")
async def _write_file(self, path: str, content: str) -> bool:
"""写入文件"""
file_path = Path(path)
if not self._is_safe_path(file_path):
raise ValueError("不安全的文件路径")
full_path = self.base_path / file_path
full_path.parent.mkdir(parents=True, exist_ok=True)
full_path.write_text(content, encoding="utf-8")
return True
def _is_safe_path(self, path: Path) -> bool:
"""检查路径安全性"""
# 防止目录遍历攻击
try:
resolved_path = (self.base_path / path).resolve()
return str(resolved_path).startswith(str(self.base_path.resolve()))
except:
return False
async def cleanup(self):
"""清理资源"""
pass
async def health_check(self) -> bool:
"""健康检查"""
return self.base_path.exists() if self.base_path else False
# =================== 模块管理器 ===================
@dataclass
class ModuleConfig:
module_class: Type[MCPModule]
config: Dict[str, Any]
enabled: bool = True
dependencies: List[str] = None
class ModularMCPServer:
"""模块化MCP服务器"""
def __init__(self):
self.modules: Dict[str, MCPModule] = {}
self.module_configs: Dict[str, ModuleConfig] = {}
self.initialized = False
def register_module(self, name: str, module_config: ModuleConfig):
"""注册模块"""
self.module_configs[name] = module_config
logging.info(f"注册模块: {name}")
async def initialize_all(self):
"""初始化所有模块"""
# 按依赖顺序初始化
initialized_modules = set()
while len(initialized_modules) < len(self.module_configs):
progress_made = False
for name, config in self.module_configs.items():
if name in initialized_modules or not config.enabled:
continue
# 检查依赖是否已初始化
dependencies = config.dependencies or []
if all(dep in initialized_modules for dep in dependencies):
# 初始化模块
module_instance = config.module_class()
success = await module_instance.initialize(config.config)
if success:
self.modules[name] = module_instance
initialized_modules.add(name)
progress_made = True
logging.info(f"模块初始化成功: {name}")
else:
logging.error(f"模块初始化失败: {name}")
if not progress_made:
raise Exception("模块依赖循环或初始化失败")
self.initialized = True
logging.info("所有模块初始化完成")
async def get_module(self, name: str, module_type: Type = None) -> Optional[MCPModule]:
"""获取模块实例"""
module = self.modules.get(name)
if module and module_type:
if isinstance(module, module_type):
return module
else:
return None
return module
async def health_check_all(self) -> Dict[str, bool]:
"""检查所有模块健康状态"""
results = {}
for name, module in self.modules.items():
try:
results[name] = await module.health_check()
except Exception as e:
logging.error(f"模块健康检查失败 {name}: {e}")
results[name] = False
return results
async def cleanup_all(self):
"""清理所有模块"""
for name, module in self.modules.items():
try:
await module.cleanup()
logging.info(f"模块清理完成: {name}")
except Exception as e:
logging.error(f"模块清理失败 {name}: {e}")
self.modules.clear()
self.initialized = False
# =================== 配置和启动 ===================
async def setup_modular_mcp_server():
"""设置模块化MCP服务器"""
server = ModularMCPServer()
# 注册传输模块
server.register_module("stdio_transport", ModuleConfig(
module_class=StdioTransportModule,
config={"command": ["python", "mcp_server.py"]},
dependencies=[]
))
# 注册协议模块
server.register_module("jsonrpc_protocol", ModuleConfig(
module_class=JsonRpcProtocolModule,
config={"timeout": 30},
dependencies=[]
))
# 注册业务模块
server.register_module("filesystem_business", ModuleConfig(
module_class=FileSystemBusinessModule,
config={
"base_path": ".",
"allowed_extensions": [".txt", ".json", ".py", ".md"]
},
dependencies=[]
))
return server
async def demo_modular_mcp():
"""演示模块化MCP"""
print("=== 模块化MCP服务器演示 ===\n")
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 创建服务器
server = await setup_modular_mcp_server()
try:
# 初始化所有模块
print("初始化模块...")
await server.initialize_all()
# 健康检查
print("\n健康检查结果:")
health_status = await server.health_check_all()
for module_name, status in health_status.items():
print(f" {module_name}: {'✓' if status else '✗'}")
# 测试传输模块
print("\n测试传输模块:")
transport = await server.get_module("stdio_transport", StdioTransportModule)
if transport:
await transport.connect()
await transport.send_message({"test": "message"})
# 测试协议模块
print("\n测试协议模块:")
protocol = await server.get_module("jsonrpc_protocol", JsonRpcProtocolModule)
if protocol:
request = {"jsonrpc": "2.0", "method": "test", "id": 1}
response = await protocol.process_request(request)
print(f"协议响应: {response}")
# 测试业务模块
print("\n测试业务模块:")
business = await server.get_module("filesystem_business", FileSystemBusinessModule)
if business:
files = await business.handle_tool_call("list_files", {"directory": "."})
print(f"文件列表: {len(files)} 个文件")
except Exception as e:
print(f"演示过程中出现错误: {e}")
finally:
# 清理资源
print("\n清理资源...")
await server.cleanup_all()
print("演示完成")
if __name__ == "__main__":
asyncio.run(demo_modular_mcp())总结与最佳实践汇总
技术架构总结
通过本教程的学习，我们深入了解了MCP（Model Context Protocol）的核心技术架构和实现方法：
1. 核心设计理念
- MCP是一个开源标准协议，为AI助手与外部系统建立统一连接标准
- 采用客户端-主机-服务器架构，支持多种传输方式（STDIO、HTTP+SSE）
- 基于JSON-RPC 2.0协议，提供标准化的消息格式和处理机制
2. 多上下文管理机制
- 实现会话上下文、资源上下文、工具上下文的统一管理
- 通过上下文同步策略和内存优化技术支撑大规模应用
- 相比单上下文模式，多上下文在功能丰富度和用户体验方面具有显著优势
3. 实际应用价值
- 解决对话连续性、个性化服务、复杂任务处理、多用户协作等关键问题
- 广泛应用于智能客服、教育辅导、项目管理、团队协作等场景
- 通过模块化设计实现高度可扩展和可维护的系统架构
开发最佳实践
1. 安全性优先
- 实施强大的授权和认证机制，防止未授权访问
- 使用资源指示器防止恶意服务器获取访问令牌
- 确保不同用户和会话的上下文完全隔离
2. 性能优化策略
- 合理使用缓存和连接池提升系统性能
- 实现LRU缓存、压缩存储和垃圾回收机制优化内存使用
- 采用异步编程模型处理并发请求
3. 模块化设计原则
- 遵循单一职责原则，确保每个模块只负责特定功能域
- 实现接口隔离，定义最小化的接口契约
- 支持动态加载和热更新，提高系统灵活性
4. 错误处理和监控
- 实现完善的错误处理和回退机制
- 在模块边界实现错误隔离和恢复
- 建立健康检查和监控机制，确保系统稳定运行
技术发展趋势
1. 生态系统扩展
- 主要技术公司（Google、Microsoft、Anthropic）积极参与MCP生态建设
- 提供丰富的预构建MCP服务器，支持企业级系统集成
- 开发者社区活跃，不断贡献新的工具和最佳实践
2. 规范持续演进
- 2025年6月规范更新引入增强安全特性和授权处理机制
- 支持多种传输协议，适应不同部署环境需求
- 持续优化性能和稳定性，支持大规模生产环境
3. 应用场景拓展
- 从简单的数据访问扩展到复杂的业务流程自动化
- 支持实时协作和多模态交互
- 与新兴AI技术深度融合，如多模型协同、边缘计算等
通过掌握这些核心概念和最佳实践，开发者可以构建出功能强大、安全可靠的MCP应用，为AI助手与外部系统的集成提供标准化解决方案。