yunzerwebsite/yunzer_go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加任务管理模块

Browse Source
This commit is contained in:
hero920103
2025-11-12 17:32:03 +08:00
parent 12a0ff8afc
commit db16ee70de
54 changed files with 3638 additions and 672 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/mcp-server/client.py
+166
View File
@@ -0,0 +1,166 @@
#!/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()
server/mcp-server/config.json
+20
View File
@@ -0,0 +1,20 @@
{
"mysql": {
"user": "gotest",
"password": "2nZhRdMPCNZrdzsd",
"host": "212.64.112.158",
"port": 3388,
"database": "gotest",
"charset": "utf8mb4",
"timeout": "10s",
"readTimeout": "30s",
"writeTimeout": "30s",
"maxIdleConns": 10,
"maxOpenConns": 100,
"connMaxLifetime": "30m"
},
"server": {
"logLevel": "info",
"enableQueryLogging": false
}
}
server/mcp-server/go.mod
+5
View File
@@ -0,0 +1,5 @@
module mcp-server
go 1.17
require github.com/go-sql-driver/mysql v1.7.0
server/mcp-server/go.sum
+2
View File
@@ -0,0 +1,2 @@
github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc=
github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=
server/mcp-server/main.go
+387
View File
@@ -0,0 +1,387 @@
package main
import (
"bufio"
"database/sql"
"encoding/json"
"fmt"
"os"
"strings"
_ "github.com/go-sql-driver/mysql"
)
// MCPRequest 表示 MCP 请求
type MCPRequest struct {
JSONRPC string `json:"jsonrpc"`
ID interface{} `json:"id"`
Method string `json:"method"`
Params json.RawMessage `json:"params"`
}
// MCPResponse 表示 MCP 响应
type MCPResponse struct {
JSONRPC string `json:"jsonrpc"`
ID interface{} `json:"id"`
Result interface{} `json:"result,omitempty"`
Error *MCPError `json:"error,omitempty"`
}
// MCPError 表示 MCP 错误
type MCPError struct {
Code int `json:"code"`
Message string `json:"message"`
Data string `json:"data,omitempty"`
}
// QueryParams 表示查询参数
type QueryParams struct {
SQL string `json:"sql"`
Args []interface{} `json:"args"`
}
// ExecuteParams 表示执行参数
type ExecuteParams struct {
SQL string `json:"sql"`
Args []interface{} `json:"args"`
}
var db *sql.DB
func main() {
// 初始化数据库
if err := initDatabase(); err != nil {
fmt.Fprintf(os.Stderr, "Failed to initialize database: %v\n", err)
os.Exit(1)
}
defer db.Close()
// 启动 MCP 服务器
reader := bufio.NewReader(os.Stdin)
for {
line, err := reader.ReadString('\n')
if err != nil {
break
}
line = strings.TrimSpace(line)
if line == "" {
continue
}
// 解析请求
var req MCPRequest
if err := json.Unmarshal([]byte(line), &req); err != nil {
sendError(nil, -32700, "Parse error", err.Error())
continue
}
// 处理请求
handleRequest(&req)
}
}
func initDatabase() error {
// 从环境变量或默认值读取配置
user := getEnv("MYSQL_USER", "gotest")
pass := getEnv("MYSQL_PASS", "2nZhRdMPCNZrdzsd")
urls := getEnv("MYSQL_URLS", "212.64.112.158:3388")
dbName := getEnv("MYSQL_DB", "gotest")
dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4&parseTime=True&loc=Local&timeout=10s&readTimeout=30s&writeTimeout=30s",
user, pass, urls, dbName)
var err error
db, err = sql.Open("mysql", dsn)
if err != nil {
return fmt.Errorf("failed to open database: %w", err)
}
// 测试连接
if err := db.Ping(); err != nil {
return fmt.Errorf("failed to ping database: %w", err)
}
// 配置连接池
db.SetMaxIdleConns(10)
db.SetMaxOpenConns(100)
fmt.Fprintf(os.Stderr, "Database connected successfully\n")
return nil
}
func handleRequest(req *MCPRequest) {
switch req.Method {
case "initialize":
handleInitialize(req)
case "query":
handleQuery(req)
case "execute":
handleExecute(req)
case "get_tables":
handleGetTables(req)
case "get_table_schema":
handleGetTableSchema(req)
default:
sendError(req.ID, -32601, "Method not found", fmt.Sprintf("Unknown method: %s", req.Method))
}
}
func handleInitialize(req *MCPRequest) {
result := map[string]interface{}{
"protocolVersion": "1.0",
"capabilities": map[string]interface{}{
"tools": []map[string]interface{}{
{
"name": "query",
"description": "Execute a SELECT query and return results",
"inputSchema": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"sql": map[string]interface{}{
"type": "string",
"description": "SQL SELECT query",
},
"args": map[string]interface{}{
"type": "array",
"description": "Query parameters (optional)",
},
},
"required": []string{"sql"},
},
},
{
"name": "execute",
"description": "Execute an INSERT, UPDATE, or DELETE query",
"inputSchema": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"sql": map[string]interface{}{
"type": "string",
"description": "SQL INSERT/UPDATE/DELETE query",
},
"args": map[string]interface{}{
"type": "array",
"description": "Query parameters (optional)",
},
},
"required": []string{"sql"},
},
},
{
"name": "get_tables",
"description": "Get all table names in the database",
"inputSchema": map[string]interface{}{
"type": "object",
},
},
{
"name": "get_table_schema",
"description": "Get the schema of a specific table",
"inputSchema": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"table": map[string]interface{}{
"type": "string",
"description": "Table name",
},
},
"required": []string{"table"},
},
},
},
},
"serverInfo": map[string]interface{}{
"name": "MySQL MCP Server",
"version": "1.0.0",
},
}
sendResponse(req.ID, result)
}
func handleQuery(req *MCPRequest) {
var params QueryParams
if err := json.Unmarshal(req.Params, &params); err != nil {
sendError(req.ID, -32602, "Invalid params", err.Error())
return
}
if params.SQL == "" {
sendError(req.ID, -32602, "Invalid params", "SQL query is required")
return
}
// 确保是 SELECT 查询
if !strings.HasPrefix(strings.ToUpper(strings.TrimSpace(params.SQL)), "SELECT") {
sendError(req.ID, -32602, "Invalid query", "Only SELECT queries are allowed")
return
}
rows, err := db.Query(params.SQL, params.Args...)
if err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
defer rows.Close()
// 获取列名
columns, err := rows.Columns()
if err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
// 读取数据
var results []map[string]interface{}
for rows.Next() {
values := make([]interface{}, len(columns))
valuePtrs := make([]interface{}, len(columns))
for i := range columns {
valuePtrs[i] = &values[i]
}
if err := rows.Scan(valuePtrs...); err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
entry := make(map[string]interface{})
for i, col := range columns {
val := values[i]
b, ok := val.([]byte)
if ok {
entry[col] = string(b)
} else {
entry[col] = val
}
}
results = append(results, entry)
}
if err := rows.Err(); err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
sendResponse(req.ID, map[string]interface{}{
"rows": results,
"count": len(results),
})
}
func handleExecute(req *MCPRequest) {
var params ExecuteParams
if err := json.Unmarshal(req.Params, &params); err != nil {
sendError(req.ID, -32602, "Invalid params", err.Error())
return
}
if params.SQL == "" {
sendError(req.ID, -32602, "Invalid params", "SQL query is required")
return
}
result, err := db.Exec(params.SQL, params.Args...)
if err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
lastID, _ := result.LastInsertId()
rowsAffected, _ := result.RowsAffected()
sendResponse(req.ID, map[string]interface{}{
"lastInsertId": lastID,
"rowsAffected": rowsAffected,
})
}
func handleGetTables(req *MCPRequest) {
rows, err := db.Query("SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = DATABASE()")
if err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
defer rows.Close()
var tables []string
for rows.Next() {
var tableName string
if err := rows.Scan(&tableName); err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
tables = append(tables, tableName)
}
sendResponse(req.ID, map[string]interface{}{
"tables": tables,
})
}
func handleGetTableSchema(req *MCPRequest) {
var params struct {
Table string `json:"table"`
}
if err := json.Unmarshal(req.Params, &params); err != nil {
sendError(req.ID, -32602, "Invalid params", err.Error())
return
}
rows, err := db.Query(fmt.Sprintf("DESCRIBE %s", params.Table))
if err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
defer rows.Close()
var schema []map[string]interface{}
for rows.Next() {
var field, typeStr, null, key, defaultVal, extra string
if err := rows.Scan(&field, &typeStr, &null, &key, &defaultVal, &extra); err != nil {
sendError(req.ID, -32603, "Database error", err.Error())
return
}
schema = append(schema, map[string]interface{}{
"field": field,
"type": typeStr,
"null": null,
"key": key,
"default": defaultVal,
"extra": extra,
})
}
sendResponse(req.ID, schema)
}
func sendResponse(id interface{}, result interface{}) {
response := MCPResponse{
JSONRPC: "2.0",
ID: id,
Result: result,
}
data, _ := json.Marshal(response)
fmt.Println(string(data))
}
func sendError(id interface{}, code int, message string, data string) {
response := MCPResponse{
JSONRPC: "2.0",
ID: id,
Error: &MCPError{
Code: code,
Message: message,
Data: data,
},
}
jsonData, _ := json.Marshal(response)
fmt.Println(string(jsonData))
}
func getEnv(key, defaultValue string) string {
if value := os.Getenv(key); value != "" {
return value
}
return defaultValue
}
server/mcp-server/mcp-server.exe
BIN
View File
Binary file not shown.
server/mcp-server/start.bat
+23
View File
@@ -0,0 +1,23 @@
@echo off
REM MySQL MCP Server 启动脚本
REM 设置环境变量（可选，从 app.conf 读取）
set MYSQL_USER=gotest
set MYSQL_PASS=2nZhRdMPCNZrdzsd
set MYSQL_URLS=212.64.112.158:3388
set MYSQL_DB=gotest
REM 编译
echo Building MCP Server...
go build -o mcp-server.exe main.go
if errorlevel 1 (
echo Build failed!
exit /b 1
)
echo Build successful! Starting MCP Server...
echo.
REM 启动服务器
mcp-server.exe
server/mcp-server/start.sh
+24
View File
@@ -0,0 +1,24 @@
#!/bin/bash
# MySQL MCP Server 启动脚本
# 设置环境变量（可选）
export MYSQL_USER=gotest
export MYSQL_PASS=2nZhRdMPCNZrdzsd
export MYSQL_URLS=212.64.112.158:3388
export MYSQL_DB=gotest
# 编译
echo "Building MCP Server..."
go build -o mcp-server main.go
if [ $? -ne 0 ]; then
echo "Build failed!"
exit 1
fi
echo "Build successful! Starting MCP Server..."
echo ""
# 启动服务器
./mcp-server
server/mcp-server/test.py
+174
View File
@@ -0,0 +1,174 @@
#!/usr/bin/env python3
"""
MySQL MCP Server 测试脚本
"""
import subprocess
import json
import time
import sys
import os
from pathlib import Path
class MCPClient:
"""MCP 客户端"""
def __init__(self, binary_path):
"""初始化客户端"""
self.binary_path = binary_path
self.process = None
self.request_id = 0
def start(self):
"""启动 MCP 服务器"""
print("Starting MCP Server...")
self.process = subprocess.Popen(
[self.binary_path],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
time.sleep(1) # 等待服务器启动
print("✓ MCP Server started")
def stop(self):
"""停止服务器"""
if self.process:
self.process.terminate()
self.process.wait()
print("✓ MCP Server stopped")
def send_request(self, method, params=None):
"""发送请求"""
self.request_id += 1
request = {
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
"params": params or {}
}
json_str = json.dumps(request)
print(f"\n→ Sending: {method}")
print(f" Request: {json_str}")
self.process.stdin.write(json_str + "\n")
self.process.stdin.flush()
# 读取响应
response_str = self.process.stdout.readline()
print(f" Response: {response_str.strip()}")
try:
response = json.loads(response_str)
return response
except json.JSONDecodeError as e:
print(f" Error parsing response: {e}")
return None
def test_mcp_server():
"""测试 MCP 服务器"""
# 获取二进制文件路径
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: go build -o mcp-server.exe main.go")
return False
client = MCPClient(str(binary_path))
try:
# 启动服务器
client.start()
# 测试 1: 初始化
print("\n" + "="*50)
print("Test 1: Initialize")
print("="*50)
response = client.send_request("initialize")
if response and "result" in response:
print("✓ Initialize successful")
else:
print("✗ Initialize failed")
return False
# 测试 2: 获取表列表
print("\n" + "="*50)
print("Test 2: Get Tables")
print("="*50)
response = client.send_request("get_tables")
if response and "result" in response:
tables = response["result"].get("tables", [])
print(f"✓ Get tables successful, found {len(tables)} tables")
if tables:
print(f" Tables: {', '.join(tables[:5])}")
else:
print("✗ Get tables failed")
return False
# 测试 3: 查询数据
print("\n" + "="*50)
print("Test 3: Query Data")
print("="*50)
response = client.send_request("query", {
"sql": "SELECT * FROM users LIMIT 5",
"args": []
})
if response and "result" in response:
count = response["result"].get("count", 0)
print(f"✓ Query successful, returned {count} rows")
if count > 0:
print(f" Sample row: {response['result']['rows'][0]}")
else:
print("✗ Query failed")
if "error" in response:
print(f" Error: {response['error']['message']}")
# 测试 4: 获取表结构
print("\n" + "="*50)
print("Test 4: Get Table Schema")
print("="*50)
response = client.send_request("get_table_schema", {
"table": "users"
})
if response and "result" in response:
schema = response["result"]
print(f"✓ Get schema successful, found {len(schema)} columns")
for col in schema[:3]:
print(f" - {col['field']}: {col['type']}")
else:
print("✗ Get schema failed")
# 测试 5: 参数化查询
print("\n" + "="*50)
print("Test 5: Parameterized Query")
print("="*50)
response = client.send_request("query", {
"sql": "SELECT * FROM users WHERE id = ?",
"args": [1]
})
if response and "result" in response:
count = response["result"].get("count", 0)
print(f"✓ Parameterized query successful, returned {count} rows")
else:
print("✗ Parameterized query failed")
print("\n" + "="*50)
print("All tests completed!")
print("="*50)
return True
except Exception as e:
print(f"Error: {e}")
return False
finally:
client.stop()
if __name__ == "__main__":
success = test_mcp_server()
sys.exit(0 if success else 1)
server/mcp-server/vscode-config-example.json
+16
View File
@@ -0,0 +1,16 @@
{
"comment": "VS Code MCP 配置示例 - 将此配置添加到你的 VS Code settings.json",
"modelContextProtocol": {
"servers": {
"mysql": {
"command": "e:\\Demos\\DemoOwns\\Go\\yunzer_go\\server\\mcp-server\\mcp-server.exe",
"env": {
"MYSQL_USER": "gotest",
"MYSQL_PASS": "2nZhRdMPCNZrdzsd",
"MYSQL_URLS": "212.64.112.158:3388",
"MYSQL_DB": "gotest"
}
}
}
}
}