test_mcp_protocol.py•3.54 kB
#!/usr/bin/env python3
"""
测试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_protocol():
"""测试完整的MCP协议流程"""
print("🧪 测试MCP协议流程")
print("=" * 50)
server = MemOSMCPServer()
# 1. 测试初始化
print("1. 测试初始化...")
init_request = {
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
}
init_response = await server.handle_request(init_request)
print(f"初始化响应: {json.dumps(init_response, indent=2, ensure_ascii=False)}")
# 2. 测试工具列表
print("\n2. 测试工具列表...")
list_request = {
"id": 2,
"method": "tools/list",
"params": {}
}
list_response = await server.handle_request(list_request)
print(f"工具列表响应: {json.dumps(list_response, indent=2, ensure_ascii=False)}")
# 3. 测试查询工具
print("\n3. 测试查询工具...")
query_request = {
"id": 3,
"method": "tools/call",
"params": {
"name": "query_memos_context",
"arguments": {
"query": "项目技术架构",
"max_memories": 3
}
}
}
query_response = await server.handle_request(query_request)
print(f"查询响应: {json.dumps(query_response, indent=2, ensure_ascii=False)}")
# 4. 测试添加记忆工具
print("\n4. 测试添加记忆工具...")
add_request = {
"id": 4,
"method": "tools/call",
"params": {
"name": "add_memos_memory",
"arguments": {
"content": "MCP协议测试成功",
"tags": ["测试", "MCP", "成功"]
}
}
}
add_response = await server.handle_request(add_request)
print(f"添加记忆响应: {json.dumps(add_response, indent=2, ensure_ascii=False)}")
def test_json_rpc_format():
"""测试JSON-RPC格式"""
print("\n\n📋 标准JSON-RPC格式测试")
print("=" * 50)
# 标准的MCP初始化流程
requests = [
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "claude-desktop",
"version": "1.0.0"
}
}
},
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
},
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "query_memos_context",
"arguments": {
"query": "测试查询",
"max_memories": 2
}
}
}
]
for i, request in enumerate(requests, 1):
print(f"\n请求 {i}:")
print(json.dumps(request, indent=2, ensure_ascii=False))
if __name__ == "__main__":
asyncio.run(test_mcp_protocol())
test_json_rpc_format()