test_mcp_server.py•4.35 kB
#!/usr/bin/env python3
"""
测试MemOS MCP服务器
"""
import json
import asyncio
import sys
from pathlib import Path
# 添加当前目录到路径
sys.path.insert(0, str(Path(__file__).parent))
from memos_mcp_server import MemOSMCPServer
async def test_mcp_server():
"""测试MCP服务器功能"""
print("🧪 测试MemOS MCP服务器")
print("=" * 50)
server = MemOSMCPServer()
# 测试工具列表
print("1. 测试工具列表...")
list_request = {
"id": 1,
"method": "tools/list",
"params": {}
}
list_response = await server.handle_request(list_request)
print(f"✅ 工具列表: {len(list_response.get('tools', []))} 个工具")
for tool in list_response.get('tools', []):
print(f" - {tool['name']}: {tool['description']}")
# 测试查询功能
print("\n2. 测试查询功能...")
query_request = {
"id": 2,
"method": "tools/call",
"params": {
"name": "query_memos_context",
"arguments": {
"query": "项目的技术架构",
"max_memories": 3
}
}
}
query_response = await server.handle_request(query_request)
if "content" in query_response:
print("✅ 查询成功")
print("返回内容:")
for content in query_response["content"]:
if content["type"] == "text":
print(content["text"])
else:
print("❌ 查询失败")
print(query_response)
# 测试添加记忆功能
print("\n3. 测试添加记忆功能...")
add_request = {
"id": 3,
"method": "tools/call",
"params": {
"name": "add_memos_memory",
"arguments": {
"content": "MCP服务器测试成功，可以正常调用MemOS功能",
"tags": ["测试", "MCP", "成功"]
}
}
}
add_response = await server.handle_request(add_request)
if "content" in add_response:
print("✅ 添加记忆成功")
for content in add_response["content"]:
if content["type"] == "text":
print(content["text"])
else:
print("❌ 添加记忆失败")
print(add_response)
# 再次查询验证
print("\n4. 验证新添加的记忆...")
verify_request = {
"id": 4,
"method": "tools/call",
"params": {
"name": "query_memos_context",
"arguments": {
"query": "MCP测试",
"max_memories": 2
}
}
}
verify_response = await server.handle_request(verify_request)
if "content" in verify_response:
print("✅ 验证成功，可以检索到新添加的记忆")
for content in verify_response["content"]:
if content["type"] == "text":
print(content["text"])
else:
print("❌ 验证失败")
print(verify_response)
def test_json_rpc_format():
"""测试JSON-RPC格式"""
print("\n\n📋 JSON-RPC格式示例")
print("=" * 50)
# 工具列表请求
list_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
print("工具列表请求:")
print(json.dumps(list_request, indent=2, ensure_ascii=False))
# 查询请求
query_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "query_memos_context",
"arguments": {
"query": "项目技术架构",
"max_memories": 3
}
}
}
print("\n查询请求:")
print(json.dumps(query_request, indent=2, ensure_ascii=False))
# 添加记忆请求
add_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "add_memos_memory",
"arguments": {
"content": "这是一个测试记忆",
"tags": ["测试", "示例"]
}
}
}
print("\n添加记忆请求:")
print(json.dumps(add_request, indent=2, ensure_ascii=False))
if __name__ == "__main__":
asyncio.run(test_mcp_server())
test_json_rpc_format()