test_expert_solution.py•8.06 kB
#!/usr/bin/env python3
"""
测试专家解决方案的完整MCP流程
"""
import json
import subprocess
import sys
import time
def test_expert_mcp_solution():
"""测试专家建议的完整MCP解决方案"""
print("🧪 测试专家解决方案 - 完整MemOS MCP集成")
print("=" * 60)
# 测试请求序列
requests = [
{
"name": "初始化",
"request": {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "claude-desktop",
"version": "1.0.0"
}
}
}
},
{
"name": "工具列表",
"request": {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
},
{
"name": "查询记忆",
"request": {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "query_memos_context",
"arguments": {
"query": "AI记忆增强对话",
"max_memories": 3
}
}
}
},
{
"name": "添加记忆",
"request": {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "add_memos_memory",
"arguments": {
"content": "专家解决方案测试成功：使用.venv虚拟环境+启动脚本完美解决了MCP集成问题",
"tags": ["专家解决方案", "MCP", "成功", "虚拟环境"]
}
}
}
}
]
success_count = 0
total_tests = len(requests)
for i, test_case in enumerate(requests, 1):
print(f"\n{i}. 测试 {test_case['name']}...")
try:
# 创建输入
input_data = json.dumps(test_case['request']) + "\n"
# 运行MCP服务器
start_time = time.time()
process = subprocess.Popen(
["./run_mcp.sh"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd="/home/qqinshu/视频/MemOS"
)
# 发送请求
stdout, stderr = process.communicate(input=input_data, timeout=30)
end_time = time.time()
if stderr and "Poetry could not find" not in stderr:
print(f"❌ 错误: {stderr}")
continue
if stdout:
try:
# 解析响应（可能有多行，取最后一行JSON）
lines = stdout.strip().split('\n')
json_line = None
for line in reversed(lines):
if line.strip().startswith('{"'):
json_line = line.strip()
break
if json_line:
response = json.loads(json_line)
response_time = end_time - start_time
print(f"✅ 成功 (响应时间: {response_time:.2f}s)")
success_count += 1
# 详细分析响应
if test_case['name'] == "初始化":
if "result" in response and "serverInfo" in response["result"]:
server_info = response["result"]["serverInfo"]
print(f" 服务器: {server_info.get('name', 'unknown')} v{server_info.get('version', 'unknown')}")
elif test_case['name'] == "工具列表":
if "result" in response and "tools" in response["result"]:
tools = response["result"]["tools"]
print(f" 发现 {len(tools)} 个工具:")
for tool in tools:
print(f" - {tool['name']}: {tool.get('description', '无描述')}")
elif test_case['name'] == "查询记忆":
if "result" in response and "content" in response["result"]:
content = response["result"]["content"]
if content and len(content) > 0:
text = content[0].get("text", "")
# 提取关键信息
if "LLM组织的上下文" in text:
print(" ✅ DeepSeek-V3上下文组织正常")
if "相关记忆详情" in text:
print(" ✅ 向量搜索功能正常")
if "相关度:" in text:
print(" ✅ 相似度计算正常")
elif test_case['name'] == "添加记忆":
if "result" in response and "content" in response["result"]:
content = response["result"]["content"]
if content and len(content) > 0:
text = content[0].get("text", "")
if "成功添加记忆" in text:
print(" ✅ 记忆添加功能正常")
else:
print(f"❌ 未找到有效的JSON响应")
print(f"输出内容: {stdout}")
except json.JSONDecodeError as e:
print(f"❌ JSON解析错误: {e}")
print(f"响应内容: {stdout}")
else:
print("❌ 无响应")
except subprocess.TimeoutExpired:
print("❌ 超时")
process.kill()
except Exception as e:
print(f"❌ 异常: {e}")
# 总结
print(f"\n{'='*60}")
print(f"🎯 测试总结")
print(f"{'='*60}")
print(f"成功: {success_count}/{total_tests}")
print(f"成功率: {success_count/total_tests*100:.1f}%")
if success_count == total_tests:
print("\n🎉 专家解决方案完全成功！")
print("✅ 虚拟环境隔离正常")
print("✅ 启动脚本工作正常")
print("✅ 完整MemOS功能可用")
print("✅ MCP协议完全兼容")
print("✅ DeepSeek-V3智能上下文组织正常")
print("✅ 向量数据库搜索正常")
print(f"\n📋 Claude Desktop配置:")
print(f"使用配置文件: /home/qqinshu/视频/~/mcp-template-main/claude-desktop-expert.json")
print(f"启动命令: /home/qqinshu/视频/MemOS/run_mcp.sh")
print(f"\n🚀 下一步:")
print(f"1. 将claude-desktop-expert.json的内容复制到Claude Desktop配置")
print(f"2. 重启Claude Desktop")
print(f"3. 开始享受完整的AI记忆增强对话功能！")
else:
print(f"\n⚠️ 部分测试失败，需要进一步调试")
return success_count == total_tests
if __name__ == "__main__":
success = test_expert_mcp_solution()
sys.exit(0 if success else 1)