test_mcp_v2.py•2.76 kB
#!/usr/bin/env python3
"""
测试MemOS MCP服务器 v2.0的所有功能
"""
import asyncio
import json
from memos_mcp_server_v2 import MemOSMCPServerV2
async def test_all_functions():
"""测试所有MCP v2.0功能"""
print("🚀 开始测试MemOS MCP服务器 v2.0")
server = MemOSMCPServerV2()
# 1. 测试初始化
print("\n1️⃣ 测试初始化...")
init_request = {
"jsonrpc": "2.0",
"method": "initialize",
"params": {},
"id": 1
}
result = await server.handle_request(init_request)
print(f"✅ 初始化成功: {result['result']['serverInfo']['name']} v{result['result']['serverInfo']['version']}")
# 2. 测试工具列表
print("\n2️⃣ 测试工具列表...")
tools_request = {
"jsonrpc": "2.0",
"method": "tools/list",
"params": {},
"id": 2
}
result = await server.handle_request(tools_request)
tools = result['result']['tools']
print(f"✅ 发现 {len(tools)} 个工具:")
for tool in tools:
print(f" - {tool['name']}: {tool['description']}")
# 3. 测试添加记忆
print("\n3️⃣ 测试添加记忆...")
add_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "add_memos_memory",
"arguments": {
"content": "MCP v2.0测试记忆：新服务器工作正常",
"tags": ["测试", "MCP v2.0", "验证"]
}
},
"id": 3
}
result = await server.handle_request(add_request)
print("✅ 添加记忆结果:")
print(result['result']['content'][0]['text'])
# 4. 测试查询记忆
print("\n4️⃣ 测试查询记忆...")
query_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "query_memos_context",
"arguments": {
"query": "MCP v2.0测试",
"max_memories": 3
}
},
"id": 4
}
result = await server.handle_request(query_request)
print("✅ 查询记忆结果:")
print(result['result']['content'][0]['text'])
# 5. 测试反馈统计
print("\n5️⃣ 测试反馈统计...")
stats_request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "get_feedback_stats",
"arguments": {}
},
"id": 5
}
result = await server.handle_request(stats_request)
print("✅ 反馈统计结果:")
print(result['result']['content'][0]['text'])
print("\n🎉 所有测试完成！MCP v2.0服务器工作正常")
if __name__ == "__main__":
asyncio.run(test_all_functions())