Files
elmer/....vscode/mcp_server.py
T
2026-06-10 15:46:20 +04:00

42 lines
1.6 KiB
Python

#!/usr/bin/env python3
"""MCP сервер для Elmer — БД, логи, ssh."""
import json, subprocess, sys
def handle(req):
method = req.get("method", "")
params = req.get("params", {})
if method == "list_tools":
return {
"tools": [
{"name": "query_db", "description": "SQL-запрос к elmer.db", "inputSchema": {"type": "object", "properties": {"sql": {"type": "string"}}}},
{"name": "server_logs", "description": "Логи сервера (последние N строк)", "inputSchema": {"type": "object", "properties": {"lines": {"type": "number", "default": 30}}}},
{"name": "ssh", "description": "Выполнить bash-команду на ВМ", "inputSchema": {"type": "object", "properties": {"cmd": {"type": "string"}}}},
]
}
if method == "call_tool":
name = params.get("name", "")
args = params.get("arguments", {})
if name == "query_db":
ssh(f"sqlite3 /opt/elmer/elmer.db \"{args['sql']}\"")
elif name == "server_logs":
ssh(f"sudo journalctl -u elmer --no-pager -n {args.get('lines', 30)}")
elif name == "ssh":
ssh(args["cmd"])
else: return {"error": f"unknown tool: {name}"}
return {"result": "ok"}
def ssh(cmd):
r = subprocess.run(["ssh", "-i", "/home/naeel/.ssh/naeel_vm_id_ed25519", "naeel@5.172.178.213", cmd], capture_output=True, text=True)
return {"stdout": r.stdout, "stderr": r.stderr}
for line in sys.stdin:
line = line.strip()
if line:
resp = handle(json.loads(line))
print(json.dumps(resp), flush=True)