mcp: multisync_server + bridge + firefox
This commit is contained in:
Your Name
@@ -0,0 +1,117 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Firefox Browser MCP — 控制 Firefox 进行 Web 测试"""
|
||||
import json, time, base64, asyncio
|
||||
from selenium import webdriver
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.firefox.options import Options
|
||||
from selenium.webdriver.firefox.service import Service
|
||||
from mcp.server import Server
|
||||
from mcp.server.stdio import stdio_server
|
||||
|
||||
server = Server("firefox-mcp")
|
||||
driver = None
|
||||
|
||||
def _get_driver():
|
||||
global driver
|
||||
if driver is None:
|
||||
opts = Options()
|
||||
opts.add_argument("--headless")
|
||||
opts.add_argument("--no-sandbox")
|
||||
opts.add_argument("--disable-dev-shm-usage")
|
||||
driver = webdriver.Firefox(options=opts)
|
||||
driver.set_page_load_timeout(30)
|
||||
return driver
|
||||
|
||||
# ================================================================
|
||||
# T1: 导航
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def firefox_navigate(url: str = "https://192.168.124.10/login.php"):
|
||||
"""Firefox 打开指定 URL，返回页面标题和 URL"""
|
||||
try:
|
||||
d = _get_driver()
|
||||
d.get(url)
|
||||
return f"标题: {d.title}\nURL: {d.current_url}"
|
||||
except Exception as e:
|
||||
return f"导航失败: {e}"
|
||||
|
||||
# ================================================================
|
||||
# T2: 截图
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def firefox_screenshot():
|
||||
"""截取当前页面截图，返回 base64 PNG"""
|
||||
try:
|
||||
d = _get_driver()
|
||||
png = d.get_screenshot_as_base64()
|
||||
return f"截图大小: {len(png)} bytes (base64). 可用工具查看。\n前100字符: {png[:100]}..."
|
||||
except Exception as e:
|
||||
return f"截图失败: {e}"
|
||||
|
||||
# ================================================================
|
||||
# T3: 获取页面文本
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def firefox_get_text():
|
||||
"""获取当前页面可见文本内容（用于检查错误信息）"""
|
||||
try:
|
||||
d = _get_driver()
|
||||
body = d.find_element(By.TAG_NAME, "body")
|
||||
text = body.text[:3000]
|
||||
return text if text else "(空页面)"
|
||||
except Exception as e:
|
||||
return str(e)
|
||||
|
||||
# ================================================================
|
||||
# T4: 填充输入框并提交
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def firefox_fill(selector: str = "input[name='username']", value: str = "admin"):
|
||||
"""在指定 CSS selector 的输入框中填入值"""
|
||||
try:
|
||||
d = _get_driver()
|
||||
el = d.find_element(By.CSS_SELECTOR, selector)
|
||||
el.clear()
|
||||
el.send_keys(value)
|
||||
return f"已填入 {selector} = {value}"
|
||||
except Exception as e:
|
||||
return str(e)
|
||||
|
||||
# ================================================================
|
||||
# T5: 点击
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def firefox_click(selector: str = "button[type='submit']"):
|
||||
"""点击指定 CSS selector 的元素"""
|
||||
try:
|
||||
d = _get_driver()
|
||||
el = d.find_element(By.CSS_SELECTOR, selector)
|
||||
el.click()
|
||||
time.sleep(2)
|
||||
return f"已点击 {selector}。当前页面: {d.title} ({d.current_url})"
|
||||
except Exception as e:
|
||||
return str(e)
|
||||
|
||||
# ================================================================
|
||||
# T6: 控制台日志
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def firefox_console():
|
||||
"""获取浏览器控制台日志"""
|
||||
try:
|
||||
d = _get_driver()
|
||||
logs = d.get_log("browser")
|
||||
entries = [f"[{e['level']}] {e['message'][:200]}" for e in logs[-20:]]
|
||||
return "\n".join(entries) if entries else "(无日志)"
|
||||
except Exception as e:
|
||||
return str(e)
|
||||
|
||||
# ================================================================
|
||||
# Main
|
||||
# ================================================================
|
||||
async def main():
|
||||
async with stdio_server() as (read, write):
|
||||
await server.run(read, write, server.create_initialization_options())
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,13 @@
|
||||
"""MCP Bridge — launches remote MCP server via SSH and bridges stdio"""
|
||||
import sys
|
||||
import subprocess
|
||||
|
||||
ssh = subprocess.Popen([
|
||||
"ssh",
|
||||
"-o", "StrictHostKeyChecking=no",
|
||||
"-i", "/c/Users/uzuma/.ssh/id_rsa_mcp",
|
||||
"root@47.76.187.108",
|
||||
"python3", "/www/wwwroot/api.synaglobal.vip/mcp/multisync_server.py"
|
||||
], stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
|
||||
|
||||
sys.exit(ssh.wait())
|
||||
@@ -0,0 +1,173 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Nexus MCP Server — 测试/运维工具集 (mcp 1.x API)"""
|
||||
import json, os, re, subprocess
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
API_BASE = "http://127.0.0.1:8600"
|
||||
API_KEY = ""
|
||||
|
||||
def _load_db_creds():
|
||||
"""Load DB credentials from .env file (never hardcode)"""
|
||||
env_path = "/www/wwwroot/api.synaglobal.vip/.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-mcp")
|
||||
|
||||
# ================================================================
|
||||
# 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 multisync 数据库。只读查询"""
|
||||
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(['systemctl', action, 'multisync'], capture_output=True, text=True, timeout=15)
|
||||
return p.stdout or p.stderr or f"multisync {action}: OK"
|
||||
except Exception as e:
|
||||
return str(e)
|
||||
|
||||
# ================================================================
|
||||
# T4: 日志查看
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def logs(lines: int = 50):
|
||||
"""查看 Python 后端日志。lines: 行数"""
|
||||
try:
|
||||
p = subprocess.run(['journalctl', '-u', 'multisync', '-n', str(lines), '--no-pager'],
|
||||
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', '/www/wwwroot/api.synaglobal.vip/tests/test_api.py'],
|
||||
capture_output=True, text=True, timeout=60,
|
||||
cwd='/www/wwwroot/api.synaglobal.vip')
|
||||
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='/www/wwwroot/api.synaglobal.vip')
|
||||
return p.stdout
|
||||
except Exception as e:
|
||||
return str(e)
|
||||
|
||||
# ================================================================
|
||||
# T7: 部署同步
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def deploy():
|
||||
"""同步最新代码到运行目录并重启服务"""
|
||||
try:
|
||||
cmds = [
|
||||
"cd /www/wwwroot/api.synaglobal.vip && git fetch --all 2>&1 && git reset --hard origin/master 2>&1",
|
||||
"systemctl restart multisync 2>&1",
|
||||
]
|
||||
result = []
|
||||
for cmd in cmds:
|
||||
p = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
|
||||
result.append(p.stdout or p.stderr or "OK")
|
||||
return "\n---\n".join(result)
|
||||
except Exception as e:
|
||||
return str(e)
|
||||
|
||||
# ================================================================
|
||||
# T8: 配置查看
|
||||
# ================================================================
|
||||
@server.tool()
|
||||
async def config_view():
|
||||
"""查看当前非敏感配置（隐藏密码/密钥）"""
|
||||
try:
|
||||
with open('/www/wwwroot/api.synaglobal.vip/.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)
|
||||
|
||||
# ================================================================
|
||||
# Main
|
||||
# ================================================================
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
asyncio.run(server.run_stdio_async())
|
||||
Reference in New Issue
Block a user