"""MCP-сервер панели: любые ИИ-агенты читают состояние серверов по /mcp. Транспорт — streamable HTTP (stateless + JSON-ответы), приложение монтируется в FastAPI в app.main на /mcp. Клиент Claude Code: claude mcp add --transport http ghard https://panel.example.com/mcp Инструкции агенту (AGENT_INSTRUCTIONS) отдаются и как instructions инициализации, и как prompt agent_guide — паттерн reference-реализации gnexus-creds (data_api). Все тулы read-only: агент — наблюдатель. """ import json from datetime import datetime, timedelta, timezone from fastapi import Request from starlette.responses import Response from mcp.server.fastmcp import FastMCP from app.api.events import list_events as _list_events from app.api.servers import get_server, list_servers from app.api.services import list_services as _list_services from app.api.services import service_incidents as _service_incidents from app.api.shares import list_shares as _list_shares from app.db import get_db AGENT_INSTRUCTIONS = """\ # GHard Monitor — инструкция для ИИ-агента ## Назначение Этот сервер — MCP-интерфейс панели мониторинга GHard Monitor (Gnexus Hardware Monitor). Ты — наблюдатель-диагност: отвечаешь пользователю о состоянии его серверов и сервисов. Все инструменты — **только чтение**: менять настройки, удалять и перезапускать ты не можешь (и это хорошо — панельные изменения делает человек через веб-интерфейс). ## Данные - **Серверы** — машины с установленным агентом hard-monitor: CPU/RAM/swap, load, диски по маунтам, сеть (счёт с агента), процессы, docker-контейнеры, апптайм, заметка пользователя. - **Сервисы** — health-чеки внешних HTTP-эндпоинтов (gnexus-сервисы: /health и похожие). Панель пробует их по расписанию и хранит историю откликов + журнал моментов недоступности. - **Сетевые хранилища** — примонтированные NAS-шары и их заполнение. - **Журнал событий** — всё, что панель зафиксировала: offline/пороги/ контейнеры/сервисы (хранение 90 дней). ## Язык Отвечай пользователю на его языке. Вывод тулов технический (статусы и события — английские слаги), переводи их в человеческую речь сам. ## Карта инструментов — что вызывать | Вопрос пользователя | Инструмент | |---|---| | «как мои серверы?» / «что вообще происходит?» | `panel_overview` — одним вызовом: все серверы + сервисы + хранилища | | «что с web-01?» / «что жрёт память?» / «почему тормозит?» | `server_details` (id — из `panel_overview`) | | «растёт ли диск на db-01 за сутки?» / тренд | `server_history` | | «сервис падал?», «с какими сервисами проблемы?» | `services_overview`, потом `service_incidents` (id — из overview) | | «что происходило ночью / за последние часы?» | `events_recent` | Порядок для составных вопросов: обзор → выбранная детализация. `panel_overview` дешёвый — начинай с него, если не уверен. ## Словарь статусов - Сервер: `online` (пакеты идут) / `offline` (панель не получает метрики — агент умер, сеть или машина выключена) / данных нет. - Сервис: `up` (здоров) / `degraded` (работает, но сообщает о деградации) / `down` (не отвечает) / `pending` (данных ещё нет). Поле `mode` (maintenance / readonly / draining) — это **режим работы**, а не падение; не пугай пользователя режимом. - Журнал: severity `info` / `warning` / `critical`. ## Словарь событий (поле type в журнале) - `server_offline` / `server_online` — сервер перестал/возобновил слать метрики. - `cpu_high` / `ram_high` / `swap_high` / `disk_high` / `load_high` — ресурс превысил порог (порог и значение — в payload); `*_recovered` — вернулся в норму (гистерезис: открытие и закрытие — разные пороги). - `container_added` / `container_removed` / `container_started` / `container_exited` — docker-контейнер появился/исчез/запустился/завершился (в payload: name, image, exit_code если есть). - `service_down` / `service_degraded` / `service_recovered` — сервис перестал отвечать / деградировал / восстановился. - Пороги сжимаются в человеческую речь, например: `cpu 95% (≥90%)`, `disk / 98% (≥90%)`, `load1 2.1×cores (≥2.0×)`. ## Правила - ID серверов и сервисов бери из обзорных тулов — угадывание запрещено. - `server_history(hours)`: разумно 1 (детали), 6, 24 (сутки) или 168 (неделя; глубже недели данных нет — история метрик хранится 7 дней). - Числа передавай как есть из тулов (проценты, МБ/с, ms) — не пересчитывай единицы самостоятельно. - Если сервер offline — не паникуй и не повторяй вопросы тулами; скажи последнее известное состояние (в `server_details`) и время последнего пакета. - Пользователь просит «что-нибудь сделать» (перезапустить, удалить) — объясни, что доступ только на чтение, изменения — через веб-интерфейс. ## Подключение (для администратора панели) ``` claude mcp add --transport http ghard https://panel.example.com/mcp ``` Transport — streamable HTTP. Авторизация: если у панели задан `GHARD_ADMIN_TOKEN`, добавь заголовок `Authorization: Bearer ` в конфигурацию MCP-клиента. """ mcp = FastMCP( "GHard Monitor", instructions=AGENT_INSTRUCTIONS, stateless_http=True, # без сессий — работает за reverse proxy json_response=True, streamable_http_path="/", ) @mcp.prompt() def agent_guide() -> str: """📘 Гайд: как отвечать на вопросы о серверах и сервисах через GHard Monitor.""" return AGENT_INSTRUCTIONS def _dumps(data) -> str: return json.dumps(data, ensure_ascii=False, indent=1) def _worst_disk(disks: list[dict] | None) -> dict | None: if not disks: return None return max(disks, key=lambda d: d.get("percent", 0)) @mcp.tool() async def panel_overview() -> str: """Обзор всего: серверы (статус, CPU/RAM/диски, сеть, последний пакет), сервисы (health), сетевые хранилища. Самый быстрый способ ответить на «как мои серверы?» / «что вообще происходит?».""" servers = await list_servers() online = sum(1 for s in servers if s["status"] == "online") lines = [f"Серверов: {len(servers)}, онлайн: {online}, не в сети: {len(servers) - online}"] for s in servers: m = s.get("metrics") if m: worst = _worst_disk(m.get("disks")) disk = f", диск {worst['mount']} {worst['percent']:.0f}%" if worst else "" lines.append( f"#{s['id']} {s['name']} [{s['status']}] CPU {m['cpu']:.0f}%, " f"RAM {m['ram']['used'] / 2**30:.1f}/{m['ram']['total'] / 2**30:.1f} GiB{disk}, " f"сеть {m['net_in_mbs']:.2f}/{m['net_out_mbs']:.2f} MB/s, " f"пакет {s['last_seen']}" ) else: status = s['status'] lines.append(f"#{s['id']} {s['name']} [{status}] — данных ещё нет") services = await _list_services() if services: bad = sum(1 for x in services if x["status"] in ("down", "degraded")) lines.append(f"Сервисов: {len(services)}, проблемных (down/degraded): {bad}") for svc in services: report = svc.get("report") or {} identity = "" if report.get("version"): identity = f", {report.get('service') or '?'} {report['version']}" mode = f", режим {report['mode']}" if report.get("mode") else "" msg = f" ({svc['message']})" if svc.get("message") else "" lines.append( f"#{svc['id']} {svc['name']} [{svc['status']}] {svc['latency_ms'] or '—'} ms{identity}{mode}{msg}" ) shares = await _list_shares() if shares: lines.append(f"Сетевые хранилища: {len(shares)}") for sh in shares: if sh["total"]: percent = sh["used"] / sh["total"] * 100 lines.append( f"#{sh['id']} {sh['name']} [{sh['status']}] " f"{sh['used'] / 2**30:.1f}/{sh['total'] / 2**30:.1f} GiB ({percent:.0f}%), {sh['path']}" ) else: lines.append(f"#{sh['id']} {sh['name']} [{sh['status']}] — нет данных, {sh['path']}") return "\n".join(lines) @mcp.tool() async def server_details(server_id: int) -> str: """Полное текущее состояние одного сервера: метрики, диски, топ процессов, docker-контейнеры, заметка. id бери из panel_overview.""" try: server = await get_server(server_id) except Exception: return f"Сервер #{server_id} не найден" return _dumps(server) @mcp.tool() async def server_history(server_id: int, hours: float = 1.0, max_points: int = 200) -> str: """История метрик сервера за `hours` часов, прореженная до `max_points`. Возвращает точки: ts, cpu, ram%, swap, load, сеть MB/s, диски. История метрик хранится 7 дней — глубже точки удаляются. """ since = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat() db = get_db() cursor = await db.execute( """SELECT ts, cpu, load1, load5, load15, ram_used, ram_total, swap_used, swap_total, uptime, net_in_mbs, net_out_mbs, disks_json FROM metrics WHERE server_id = ? AND ts >= ? ORDER BY ts DESC LIMIT 5000""", (server_id, since), ) rows = await cursor.fetchall() points = [ { "ts": row["ts"], "cpu": row["cpu"], "load": [row["load1"], row["load5"], row["load15"]], "ram_percent": round(row["ram_used"] / row["ram_total"] * 100, 1) if row["ram_total"] else None, "swap_percent": round(row["swap_used"] / row["swap_total"] * 100, 1) if row["swap_total"] else None, "uptime": row["uptime"], "net_in_mbs": row["net_in_mbs"], "net_out_mbs": row["net_out_mbs"], "disks": json.loads(row["disks_json"]), } for row in reversed(rows) ] if len(points) > max_points: step = len(points) / max_points points = [points[int(i * step)] for i in range(max_points)] return _dumps({"server_id": server_id, "hours": hours, "points": points}) # --- Сервисы (health-чеки) ----------------------------------------------------- @mcp.tool() async def services_overview() -> str: """Все наблюдаемые сервисы (health-чеки): state, latency, HTTP-код, self-report (версия/окружение/режим). id нужен для service_incidents.""" services = await _list_services() lines = [f"Сервисов: {len(services)}"] for svc in services: report = svc.get("report") or {} identity = [] for key in ("service", "version", "environment"): if report.get(key): identity.append(str(report[key])) id_str = f" — {' '.join(identity)}" if identity else "" mode = f", режим {report['mode']}" if report.get("mode") else "" code = f" код {svc['code']}" if svc.get("code") is not None else "" msg = f" ({svc['message']})" if svc.get("message") else "" lines.append( f"#{svc['id']} {svc['name']} [{svc['status']}] " f"{svc['latency_ms'] or '—'} ms{code}{mode}{msg}{id_str}" ) return "\n".join(lines) @mcp.tool() async def service_incidents(service_id: int, limit: int = 20) -> str: """Журнал моментов недоступности сервиса: периоды down+degraded. Возвращает: start, end (null = идёт сейчас), duration_s, worst (худшее состояние в периоде), code, message (причина из первой точки). """ try: incidents = await _service_incidents(service_id, limit=limit) except Exception: return f"Сервис #{service_id} не найден" return _dumps(incidents) # --- Журнал событий ------------------------------------------------------------ @mcp.tool() async def events_recent(limit: int = 30, severity: str = "") -> str: """Хвост единого журнала событий (90 дней хранения). severity — необязательный фильтр: info | warning | critical. Возвращает: id, ts, type, severity, message, data (payload события). """ if limit < 1 or limit > 1000: return "limit: 1..1000" events = await _list_events(limit=min(limit, 1000)) if severity: events = [e for e in events if e.get("severity") == severity] return _dumps(events) mcp_app = mcp.streamable_http_app() # ASGI-эндпоинт MCP: регистрируется прямо в FastAPI на /mcp (см. app.main), # потому что Mount() не совпадает с путём без хвостового слэша, а lifespan # смонтированного sub-app не запускается сам — менеджер сессий стартует # в lifespan app.main (mcp.session_manager.run()). mcp_asgi = mcp_app.router.routes[0].app async def mcp_endpoint(request: Request) -> "Response": """Мост FastAPI → ASGI streamable-http-app. FastAPI валидирует сигнатуру endpoint'а: голый ASGI-объект (scope, receive, send) был бы прочитан как query-параметры → 422. Поэтому мост: забираем тело, вручную докручиваем ASGI-диалог, ответ собираем из сообщений send (stateless + json_response=True — ответ всегда один JSON, SSE-стримов тут нет). """ body = await request.body() received = False start: dict = {} chunks: list[bytes] = [] async def receive(): nonlocal received if received: return {"type": "http.disconnect"} received = True return {"type": "http.request", "body": body, "more_body": False} async def send(message) -> None: if message["type"] == "http.response.start": start.update(message) else: chunks.append(message.get("body", b"")) await mcp_asgi(request.scope, receive, send) response = Response( content=b"".join(chunks) if chunks else b"", status_code=start["status"] ) response.raw_headers.extend(start.get("headers") or []) return response