#!/usr/bin/env python3
"""
简单的MCP客户端测试脚本
用于测试SQLx MCP服务器的基本功能
"""
import json
import subprocess
import sys
def test_mcp_server():
"""测试MCP服务器的基本功能"""
# 启动MCP服务器进程
process = subprocess.Popen(
[r".\target\release\sqlx-mcp.exe"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=r"d:\code\sqlx-mcp"
)
try:
# 发送initialize请求
initialize_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
}
request_str = json.dumps(initialize_request) + "\n"
process.stdin.write(request_str)
process.stdin.flush()
# 读取响应
response_line = process.stdout.readline()
print(f"Initialize响应: {response_line.strip()}")
# 发送tools/list请求
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
request_str = json.dumps(tools_request) + "\n"
process.stdin.write(request_str)
process.stdin.flush()
# 读取工具列表响应
response_line = process.stdout.readline()
print(f"Tools列表响应: {response_line.strip()}")
response_data = json.loads(response_line.strip())
if "result" in response_data and "tools" in response_data["result"]:
tools = response_data["result"]["tools"]
print(f"\n发现的工具数量: {len(tools)}")
for tool in tools:
print(f"- {tool.get('name', 'Unknown')}: {tool.get('description', 'No description')}")
except Exception as e:
print(f"测试过程中出错: {e}")
finally:
# 清理进程
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
if __name__ == "__main__":
print("开始测试SQLx MCP服务器...")
test_mcp_server()
print("测试完成！")
