"""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 <GHARD_ADMIN_TOKEN>` в конфигурацию 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
