167 lines
5.6 KiB
Python
167 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
MySQL MCP Server 交互式客户端示例
|
|
可用于测试和与 MCP 服务器交互
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import subprocess
|
|
import os
|
|
from pathlib import Path
|
|
|
|
def pretty_print_json(obj):
|
|
"""美化打印 JSON 对象"""
|
|
print(json.dumps(obj, indent=2, ensure_ascii=False))
|
|
|
|
def run_interactive_client():
|
|
"""运行交互式客户端"""
|
|
|
|
script_dir = Path(__file__).parent
|
|
binary_path = script_dir / "mcp-server.exe"
|
|
|
|
if not binary_path.exists():
|
|
print(f"❌ Error: Binary not found at {binary_path}")
|
|
print("Please build the project first:")
|
|
print(" cd e:\\Demos\\DemoOwns\\Go\\yunzer_go\\server\\mcp-server")
|
|
print(" go build -o mcp-server.exe main.go")
|
|
return
|
|
|
|
print("🚀 Starting MySQL MCP Server...")
|
|
|
|
# 启动进程
|
|
process = subprocess.Popen(
|
|
[str(binary_path)],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
bufsize=1
|
|
)
|
|
|
|
print("✅ Server started. Type 'help' for commands.\n")
|
|
|
|
request_id = 0
|
|
|
|
try:
|
|
while True:
|
|
try:
|
|
user_input = input(">>> ").strip()
|
|
|
|
if not user_input:
|
|
continue
|
|
|
|
if user_input.lower() == "help":
|
|
print("""
|
|
Available commands:
|
|
init - Initialize server
|
|
tables - List all tables
|
|
schema <table> - Get table schema
|
|
query <sql> - Execute SELECT query
|
|
exec <sql> - Execute INSERT/UPDATE/DELETE
|
|
json <json_string> - Send raw JSON-RPC request
|
|
help - Show this help
|
|
exit / quit - Exit the client
|
|
|
|
Examples:
|
|
> tables
|
|
> schema users
|
|
> query SELECT * FROM users LIMIT 5
|
|
> exec INSERT INTO users (name, email) VALUES ('John', 'john@example.com')
|
|
> json {"jsonrpc":"2.0","id":1,"method":"query","params":{"sql":"SELECT COUNT(*) as count FROM users"}}
|
|
""")
|
|
continue
|
|
|
|
if user_input.lower() in ["exit", "quit"]:
|
|
break
|
|
|
|
request_id += 1
|
|
|
|
# 解析命令
|
|
if user_input.lower() == "init":
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"method": "initialize",
|
|
"params": {}
|
|
}
|
|
|
|
elif user_input.lower() == "tables":
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"method": "get_tables",
|
|
"params": {}
|
|
}
|
|
|
|
elif user_input.lower().startswith("schema "):
|
|
table = user_input[7:].strip()
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"method": "get_table_schema",
|
|
"params": {"table": table}
|
|
}
|
|
|
|
elif user_input.lower().startswith("query "):
|
|
sql = user_input[6:].strip()
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"method": "query",
|
|
"params": {"sql": sql, "args": []}
|
|
}
|
|
|
|
elif user_input.lower().startswith("exec "):
|
|
sql = user_input[5:].strip()
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"method": "execute",
|
|
"params": {"sql": sql, "args": []}
|
|
}
|
|
|
|
elif user_input.lower().startswith("json "):
|
|
json_str = user_input[5:].strip()
|
|
try:
|
|
request = json.loads(json_str)
|
|
except json.JSONDecodeError as e:
|
|
print(f"❌ Invalid JSON: {e}")
|
|
continue
|
|
|
|
else:
|
|
print("❌ Unknown command. Type 'help' for available commands.")
|
|
continue
|
|
|
|
# 发送请求
|
|
request_json = json.dumps(request)
|
|
process.stdin.write(request_json + "\n")
|
|
process.stdin.flush()
|
|
|
|
# 读取响应
|
|
response_str = process.stdout.readline()
|
|
if response_str:
|
|
try:
|
|
response = json.loads(response_str)
|
|
print("\n✅ Response:")
|
|
pretty_print_json(response)
|
|
print()
|
|
except json.JSONDecodeError as e:
|
|
print(f"❌ Failed to parse response: {e}")
|
|
print(f"Raw response: {response_str}")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n^C Exiting...")
|
|
break
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
|
|
finally:
|
|
print("\n🛑 Stopping server...")
|
|
process.terminate()
|
|
process.wait()
|
|
print("✓ Server stopped")
|
|
|
|
if __name__ == "__main__":
|
|
run_interactive_client()
|