#!/usr/bin/env python3 """Nexus MCP Server — 测试/运维工具集 (mcp 1.x API) All paths read from NEXUS_DEPLOY_DIR env var (default: /opt/nexus). """ import json, os, re, subprocess from mcp.server.fastmcp import FastMCP API_BASE = "http://127.0.0.1:8600" API_KEY = "" DEPLOY_DIR = os.environ.get("NEXUS_DEPLOY_DIR", "/opt/nexus") def _load_db_creds(): """Load DB credentials from .env file (never hardcode)""" env_path = os.path.join(DEPLOY_DIR, ".env") creds = {"user": "nexus", "pass": "", "db": "nexus"} try: if os.path.exists(env_path): with open(env_path) as f: for line in f: line = line.strip() if line.startswith("NEXUS_DATABASE_URL="): url = line.split("=", 1)[1].strip() m = re.match(r"mysql\+pymysql://([^:]+):([^@]+)@([^/]+)/(.+)", url) if m: creds["user"] = m.group(1) creds["pass"] = m.group(2) creds["db"] = m.group(4) except Exception: pass return creds server = FastMCP("nexus") # ================================================================ # T1: API 调用 # ================================================================ @server.tool() async def api_call(method: str = "GET", path: str = "/health", body: str = "{}"): """调用 Nexus Python API。method: GET/POST/PUT/DELETE, path: /api/servers/等, body: JSON字符串""" import urllib.request url = f"{API_BASE}{path}" data = body.encode() if method in ("POST","PUT") else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Content-Type", "application/json") if API_KEY: req.add_header("X-API-Key", API_KEY) try: resp = urllib.request.urlopen(req, timeout=10) return json.dumps(json.loads(resp.read()), indent=2, ensure_ascii=False) except Exception as e: return str(e) # ================================================================ # T2: 数据库查询 (read-only, credentials from .env) # ================================================================ @server.tool() async def db_query(sql: str = "SELECT COUNT(*) FROM servers"): """直查 MySQL nexus 数据库。只读查询""" sql_upper = sql.strip().upper() if not sql_upper.startswith("SELECT") and not sql_upper.startswith("SHOW") and not sql_upper.startswith("DESCRIBE"): return "Error: Only SELECT/SHOW/DESCRIBE queries allowed" if ";" in sql.rstrip(";").strip() and len(sql.rstrip(";").strip().split(";")) > 1: return "Error: Multiple statements not allowed" try: creds = _load_db_creds() p = subprocess.run( ['mysql', f'-u{creds["user"]}', f'-p{creds["pass"]}', creds["db"], '-e', sql], capture_output=True, text=True, timeout=10 ) return p.stdout if p.returncode == 0 else p.stderr except Exception as e: return str(e) # ================================================================ # T3: 服务管理 # ================================================================ @server.tool() async def service(action: str = "status"): """管理 Python 后端服务。action: status/start/stop/restart""" try: p = subprocess.run(['supervisorctl', action, 'nexus'], capture_output=True, text=True, timeout=15) return p.stdout or p.stderr or f"nexus {action}: OK" except Exception as e: return str(e) # ================================================================ # T4: 日志查看 # ================================================================ @server.tool() async def logs(lines: int = 50): """查看 Python 后端日志。lines: 行数""" try: log_path = os.path.join(DEPLOY_DIR, "logs", "nexus.log") p = subprocess.run(['tail', '-n', str(lines), log_path], capture_output=True, text=True, timeout=10) return p.stdout[-3000:] except Exception as e: return str(e) # ================================================================ # T5: 运行测试 # ================================================================ @server.tool() async def run_test(): """运行 test_api.py 端到端测试""" try: p = subprocess.run(['python3', os.path.join(DEPLOY_DIR, 'tests', 'test_api.py')], capture_output=True, text=True, timeout=60, cwd=DEPLOY_DIR) return p.stdout except Exception as e: return str(e) # ================================================================ # T6: Git 日志 # ================================================================ @server.tool() async def git_log(count: int = 10): """查看 project Git 提交历史""" try: p = subprocess.run(['git', 'log', f'-{count}', '--oneline'], capture_output=True, text=True, timeout=5, cwd=DEPLOY_DIR) return p.stdout except Exception as e: return str(e) # ================================================================ # T7: 部署同步 # ================================================================ @server.tool() async def deploy(): """同步最新代码到运行目录并重启服务（需通过 pre-deploy 门控检查）""" # ── Pull latest code first ── try: pull = subprocess.run( f"cd {DEPLOY_DIR} && git fetch --all 2>&1 && git reset --hard origin/main 2>&1", shell=True, capture_output=True, text=True, timeout=30, ) if pull.returncode != 0: return f"🚫 Git pull failed:\n\n{pull.stdout}\n{pull.stderr}" except Exception as e: return f"🚫 Git pull error: {e}" # ── Gate Check (after pull so files are up to date) ── gate_script = os.path.join(DEPLOY_DIR, "deploy", "pre_deploy_check.sh") if os.path.exists(gate_script): try: g = subprocess.run( ['bash', gate_script], capture_output=True, text=True, timeout=30, env={**os.environ, "NEXUS_DEPLOY_DIR": DEPLOY_DIR}, ) if g.returncode != 0: return f"🚫 DEPLOY BLOCKED — Pre-deploy gate check failed:\n\n{g.stdout}\n\nFix the blocked gates before deploying." except Exception as e: return f"🚫 Gate check error: {e}" else: # Gate script not on server yet — warn but allow (bootstrap scenario) pass # ── Restart service ── try: p = subprocess.run("supervisorctl restart nexus 2>&1", shell=True, capture_output=True, text=True, timeout=30) return f"Pull:\n{pull.stdout}\n---\nRestart:\n{p.stdout or p.stderr or 'OK'}" except Exception as e: return str(e) # ================================================================ # T8: 配置查看 # ================================================================ @server.tool() async def config_view(): """查看当前非敏感配置（隐藏密码/密钥）""" try: with open(os.path.join(DEPLOY_DIR, '.env')) as f: lines = [] for line in f: line = line.strip() if not line or line.startswith('#'): lines.append(line) continue for sensitive in ('DATABASE_URL', 'SECRET_KEY', 'API_KEY', 'ADMIN_PASSWORD', 'TELEGRAM_BOT_TOKEN', 'TELEGRAM_CHAT_ID', 'ENCRYPTION_KEY'): if line.startswith(f'NEXUS_{sensitive}='): key = line.split('=', 1)[0] lines.append(f'{key}=***REDACTED***') break else: lines.append(line) return '\n'.join(lines) except Exception as e: return str(e) # ================================================================ # T9: 门控日志 # ================================================================ @server.tool() async def gate_log(lines: int = 20): """查看门控执行日志。lines: 显示最近N条记录""" try: log_path = os.path.join(DEPLOY_DIR, "deploy", "gate_log.jsonl") if not os.path.exists(log_path): return "No gate log found — no gate checks have been recorded yet" result = [] with open(log_path, encoding="utf-8") as f: all_lines = f.readlines() for line in all_lines[-lines:]: d = json.loads(line) icon = {"PASS": "✅", "BLOCK": "🚫", "SKIP": "⏭️", "WARN": "⚠️"}.get(d["result"], "?") ts = d.get("ts", "")[:19] gate = d.get("gate", "?") detail = d.get("detail", "") result.append(f"{ts} {icon} {gate}: {detail}") return "\n".join(result) if result else "No entries" except Exception as e: return str(e) # ================================================================ # Main # ================================================================ if __name__ == "__main__": import asyncio asyncio.run(server.run_stdio_async())