событий Хуки

~/.hermes/hooks/
└── my-hook/
    ├── HOOK.yaml      # Объявляет, какие события прослушивать
    └── handler.py     # Функция-обработчик на Python

name: my-hook
description: Логировать всю активность агента в файл
events:
  - agent:start
  - agent:end
  - agent:step

import json
from datetime import datetime
from pathlib import Path

LOG_FILE = Path.home() / ".hermes" / "hooks" / "my-hook" / "activity.log"

async def handle(event_type: str, context: dict):
    """Вызывается для каждого подписанного события. Имя функции должно быть 'handle'."""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "event": event_type,
        **context,
    }
    with open(LOG_FILE, "a") as f:
        f.write(json.dumps(entry) + "\n")

# ~/.hermes/hooks/long-task-alert/HOOK.yaml
name: long-task-alert
description: Оповещение, когда агент выполняет много шагов
events:
  - agent:step

# ~/.hermes/hooks/long-task-alert/handler.py
import os
import httpx

THRESHOLD = 10
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
CHAT_ID = os.getenv("TELEGRAM_HOME_CHANNEL")

async def handle(event_type: str, context: dict):
    iteration = context.get("iteration", 0)
    if iteration == THRESHOLD and BOT_TOKEN and CHAT_ID:
        tools = ", ".join(context.get("tool_names", []))
        text = f"⚠️ Агент работает уже {iteration} шагов. Последние инструменты: {tools}"
        async with httpx.AsyncClient() as client:
            await client.post(
                f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
                json={"chat_id": CHAT_ID, "text": text},
            )

# ~/.hermes/hooks/command-logger/HOOK.yaml
name: command-logger
description: Логирование использования слэш-команд
events:
  - command:*

# ~/.hermes/hooks/command-logger/handler.py
import json
from datetime import datetime
from pathlib import Path

LOG = Path.home() / ".hermes" / "logs" / "command_usage.jsonl"

def handle(event_type: str, context: dict):
    LOG.parent.mkdir(parents=True, exist_ok=True)
    entry = {
        "ts": datetime.now().isoformat(),
        "command": context.get("command"),
        "args": context.get("args"),
        "platform": context.get("platform"),
        "user": context.get("user_id"),
    }
    with open(LOG, "a") as f:
        f.write(json.dumps(entry) + "\n")

# ~/.hermes/hooks/session-webhook/HOOK.yaml
name: session-webhook
description: Уведомлять внешний сервис о новых сессиях
events:
  - session:start
  - session:reset

# ~/.hermes/hooks/session-webhook/handler.py
import httpx

WEBHOOK_URL = "https://your-service.example.com/hermes-events"

async def handle(event_type: str, context: dict):
    async with httpx.AsyncClient() as client:
        await client.post(WEBHOOK_URL, json={
            "event": event_type,
            **context,
        }, timeout=5)

# Стартовый чек-лист

1. Выполни `hermes cron list` и проверь, не упали ли какие-либо запланированные задачи за ночь.
2. Если какие-то упали, отправь сводку в Discord #ops с помощью инструмента `send_message`.
3. Проверь, есть ли в `/opt/app/deploy.log` строки ERROR за последние 24 часа. Если да, суммируй их и включи в то же сообщение в Discord.
4. Если ничего не пошло не так, ответь только `[SILENT]`, чтобы сообщение не отправлялось.

~/.hermes/hooks/boot-md/
├── HOOK.yaml
└── handler.py

name: boot-md
description: Выполнять ~/.hermes/BOOT.md при запуске шлюза
events:
  - gateway:startup

"""Выполняет ~/.hermes/BOOT.md при каждом запуске шлюза."""

import logging
import threading
from pathlib import Path

logger = logging.getLogger("hooks.boot-md")

BOOT_FILE = Path.home() / ".hermes" / "BOOT.md"


def _build_prompt(content: str) -> str:
    return (
        "Вы выполняете стартовый чек-лист. Следуйте приведённым ниже инструкциям "
        "точно.\n\n"
        "---\n"
        f"{content}\n"
        "---\n\n"
        "Выполните каждую инструкцию. Используйте инструмент send_message для доставки "
        "любых сообщений на платформы, такие как Discord или Slack.\n"
        "Если ничего не требует внимания и нечего сообщать, ответьте "
        "ТОЛЬКО: [SILENT]"
    )


def _run_boot_agent(content: str) -> None:
    """Порождает одноразового агента и выполняет чек-лист.

    Использует разрешённую модель шлюза и учётные данные времени выполнения,
    так что это работает с пользовательскими конечными точками, агрегаторами
    и провайдерами на основе OAuth.
    """
    try:
        from gateway.run import _resolve_gateway_model, _resolve_runtime_agent_kwargs
        from run_agent import AIAgent

        agent = AIAgent(
            model=_resolve_gateway_model(),
            **_resolve_runtime_agent_kwargs(),
            platform="gateway",
            quiet_mode=True,
            skip_context_files=True,
            skip_memory=True,
            max_iterations=20,
        )
        result = agent.run_conversation(_build_prompt(content))
        response = result.get("final_response", "")
        if response and "[SILENT]" not in response:
            logger.info("boot-md завершён: %s", response[:200])
        else:
            logger.info("boot-md завершён (нечего сообщать)")
    except Exception as e:
        logger.error("Ошибка агента boot-md: %s", e)


async def handle(event_type: str, context: dict) -> None:
    if not BOOT_FILE.exists():
        return
    content = BOOT_FILE.read_text(encoding="utf-8").strip()
    if not content:
        return

    logger.info("Выполнение BOOT.md (%d символов)", len(content))

    # Фоновый поток, чтобы запуск шлюза не блокировался полным оборотом агента.
    thread = threading.Thread(
        target=_run_boot_agent,
        args=(content,),
        name="boot-md",
        daemon=True,
    )
    thread.start()

hermes gateway restart

hermes logs --follow --level INFO | grep boot-md