import json import os import sys import requests BASE_URL = 'https://status.rrrosmontis.icu' # ============================================================ # 内部辅助 # ============================================================ _session = None def get_session(): """获取或创建带 cookie 的 requests Session""" global _session if _session is None: _session = requests.Session() return _session def ensure_login(): """确保已登录，如果未登录则自动登录""" s = get_session() resp = s.get(f'{BASE_URL}/api/me', timeout=10) data = resp.json() if data.get('logged_in'): return account = os.environ.get('ROS_SERVERS_STATUS_ACCOUNT', '') password = os.environ.get('ROS_SERVERS_STATUS_PASSWORD', '') if not account or not password: raise Exception('未登录，且环境变量 ROS_SERVERS_STATUS_ACCOUNT / ROS_SERVERS_STATUS_PASSWORD 未设置') payload = {'username': account, 'password': password} resp = s.post(f'{BASE_URL}/api/login', json=payload, timeout=10) result = resp.json() if result.get('status') != 'success': raise Exception(f"登录失败: {result.get('message', '未知错误')}") def api_get(path, params=None, need_auth=True): if need_auth: ensure_login() s = get_session() resp = s.get(f'{BASE_URL}{path}', params=params, timeout=15) return resp.json() # ============================================================ # 工具函数 # ============================================================ def get_nodes(): """获取所有节点基本信息列表""" return api_get('/api/nodes') def get_node_recent(uuid): """获取指定节点最近1分钟实时状态（嵌套对象结构）""" return api_get(f'/api/recent/{uuid}') def get_load_records(uuid, hours=4, load_type='all'): """获取指定节点负载历史记录（扁平化结构）""" params = {'uuid': uuid, 'hours': hours} if load_type and load_type != 'all': params['load_type'] = load_type return api_get('/api/records/load', params=params) def get_ping_records(uuid=None, task_id=None, hours=4): """获取 Ping 历史记录，uuid 和 task_id 至少提供一个""" params = {'hours': hours} if uuid: params['uuid'] = uuid if task_id is not None: params['task_id'] = task_id return api_get('/api/records/ping', params=params) def get_ping_tasks(): """获取公开的 Ping 任务配置列表""" return api_get('/api/task/ping') def get_site_info(): """获取站点公开设置属性""" return api_get('/api/public', need_auth=False) def get_version(): """获取 Komari 服务端版本信息""" return api_get('/api/version', need_auth=False) def check_login(): """检查当前登录状态""" s = get_session() resp = s.get(f'{BASE_URL}/api/me', timeout=10) return resp.json() # ============================================================ # MCP JSON-RPC 处理 # ============================================================ TOOLS = [ { "name": "get_nodes", "description": "获取所有节点的基本信息列表（名称、CPU、内存、磁盘、价格、到期时间、流量限制等）。未登录时 hidden=true 的节点不可见。", "inputSchema": { "type": "object", "properties": {}, "required": [] } }, { "name": "get_node_recent", "description": "获取指定节点最近1分钟的实时状态数据（CPU/内存/磁盘/网络/负载/连接数/GPU/运行时间），数据结构为嵌套对象格式。", "inputSchema": { "type": "object", "properties": { "uuid": {"type": "string", "description": "节点的唯一标识符 UUID"} }, "required": ["uuid"] } }, { "name": "get_load_records", "description": "获取指定节点的负载历史记录（扁平化结构），可按指标类型筛选，用于生成图表和趋势分析。", "inputSchema": { "type": "object", "properties": { "uuid": {"type": "string", "description": "节点的唯一标识符 UUID"}, "hours": {"type": "number", "description": "查询时间范围（小时），默认 4"}, "load_type": {"type": "string", "description": "指标筛选类型: cpu / ram / swap / load / temp / disk / network / process / connections / all（默认）"} }, "required": ["uuid"] } }, { "name": "get_ping_records", "description": "获取 Ping 历史记录。uuid 和 task_id 至少提供一个：仅 uuid 返回该节点参与的任务记录，仅 task_id 返回该任务下所有节点记录，同时提供则交集查询。", "inputSchema": { "type": "object", "properties": { "uuid": {"type": "string", "description": "节点 UUID（可选，与 task_id 至少提供一个）"}, "task_id": {"type": "number", "description": "Ping 任务 ID（可选，与 uuid 至少提供一个）"}, "hours": {"type": "number", "description": "查询时间范围（小时），默认 4"} }, "required": [] } }, { "name": "get_ping_tasks", "description": "获取公开的 Ping 任务配置列表，包含任务名称、类型、执行间隔、关联节点等。", "inputSchema": { "type": "object", "properties": {}, "required": [] } }, { "name": "get_site_info", "description": "获取站点公开设置（站点名、描述、主题、OAuth 配置、CORS、记录保留时间等），无需登录。", "inputSchema": { "type": "object", "properties": {}, "required": [] } }, { "name": "get_version", "description": "获取 Komari 服务端版本号和 Git Hash，无需登录。", "inputSchema": { "type": "object", "properties": {}, "required": [] } }, { "name": "check_login", "description": "检查当前登录状态，返回用户名、UUID、是否启用2FA、SSO 绑定等信息。", "inputSchema": { "type": "object", "properties": {}, "required": [] } } ] TOOL_MAP = { "get_nodes": get_nodes, "get_node_recent": get_node_recent, "get_load_records": get_load_records, "get_ping_records": get_ping_records, "get_ping_tasks": get_ping_tasks, "get_site_info": get_site_info, "get_version": get_version, "check_login": check_login } def handle_request(req): method = req.get("method", "") req_id = req.get("id") if method == "initialize": return { "jsonrpc": "2.0", "id": req_id, "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {} }, "serverInfo": { "name": "ros-servers-status", "version": "1.0.0" } } } if method == "ping": return {"jsonrpc": "2.0", "id": req_id, "result": {}} if method == "notifications/initialized": return None if method == "tools/list": return {"jsonrpc": "2.0", "id": req_id, "result": {"tools": TOOLS}} if method == "tools/call": tool_name = req["params"]["name"] arguments = req["params"].get("arguments", {}) tool_func = TOOL_MAP.get(tool_name) if not tool_func: return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"未知工具: {tool_name}"}} try: result = tool_func(**arguments) text = json.dumps(result, ensure_ascii=False, indent=2) return {"jsonrpc": "2.0", "id": req_id, "result": {"content": [{"type": "text", "text": text}]}} except Exception as e: return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32000, "message": str(e)}} return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"未知方法: {method}"}} def main(): for line in sys.stdin: line = line.strip() if not line: continue try: req = json.loads(line) resp = handle_request(req) if resp is None: continue sys.stdout.write(json.dumps(resp, ensure_ascii=False) + '\n') sys.stdout.flush() except json.JSONDecodeError: pass if __name__ == '__main__': main()