sidebar_position: 6 title: "Event Hooks" description: "Run custom code at key lifecycle points — log activity, send alerts, post to webhooks" lang: ru

Перехватчики событий

У Hermes есть три системы перехватчиков, которые запускают собственный код в ключевых точках жизненного цикла:

Все три системы являются неблокирующими — ошибки в любом хуке фиксируются и протоколируются, что никогда не приводит к сбою агента.

Перехватчики событий шлюза

Перехватчики шлюза срабатывают автоматически во время работы шлюза (Telegram, Discord, Slack, WhatsApp, Teams), не блокируя основной конвейер агента.

Создание крючка

Каждый хук представляет собой каталог ~/.hermes/hooks/, содержащий два файла:

~/.hermes/hooks/
└── my-hook/
    ├── HOOK.yaml      # Declares which events to listen for
    └── handler.py     # Python handler function

КРЮК.yaml

name: my-hook
description: Log all agent activity to a file
events:
  - agent:start
  - agent:end
  - agent:step

Список events определяет, какие события запускают ваш обработчик. Вы можете подписаться на любую комбинацию событий, включая подстановочные знаки, такие как command:*.

обработчик.py

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):
    """Called for each subscribed event. Must be named 'handle'."""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "event": event_type,
        **context,
    }
    with open(LOG_FILE, "a") as f:
        f.write(json.dumps(entry) + "\n")

Правила обработчика: - Должно быть имя handle - Получает event_type (строку) и context (дикт). - Может быть async def или обычный def — оба работают - Ошибки фиксируются и протоколируются, что не приводит к сбою агента.

Доступные события

Соответствие подстановочным знакам

Обработчики, зарегистрированные для command:*, срабатывают для любого события command: (command:model, command:reset и т. д.). Отслеживайте все слэш-команды с помощью одной подписки.

Примеры

Оповещение Telegram о длинных задачах

Отправьте себе сообщение, когда агент сделает более 10 шагов:

# ~/.hermes/hooks/long-task-alert/HOOK.yaml
name: long-task-alert
description: Alert when agent is taking many steps
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"⚠️ Agent has been running for {iteration} steps. Last tools: {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: Log slash command usage
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")

Вебхук запуска сеанса

POST во внешнюю службу в новых сеансах:

# ~/.hermes/hooks/session-webhook/HOOK.yaml
name: session-webhook
description: Notify external service on new sessions
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)

Учебное пособие: BOOT.md — запуск контрольного списка при каждой загрузке шлюза

Популярный шаблон сообщества: отправьте контрольный список Markdown по адресу ~/.hermes/BOOT.md и попросите агента запускать его один раз при каждом запуске шлюза. Полезно для «при каждой загрузке проверять сбои cron в ночное время и пинговать меня в Discord, если что-то не удалось», или «подводить итоги развертывания.log за последние 24 часа и публиковать его в Slack #ops».

В этом руководстве показано, как создать его самостоятельно в виде определяемого пользователем хука. Hermes не поставляет встроенный крючок BOOT.md — вы подключаете именно то поведение, которое хотите.

Что мы строим

1. Файл ~/.hermes/BOOT.md с инструкциями по запуску на естественном языке.
2. Перехватчик шлюза, который срабатывает на gateway:startup, порождает одноразовый агент с разрешенной моделью/учетными данными вашего шлюза и запускает инструкции BOOT.md.
3. Соглашение [SILENT], позволяющее агенту отказаться от отправки сообщения, когда сообщать не о чем.

Шаг 1. Напишите свой контрольный список

Создайте ~/.hermes/BOOT.md. Напишите это так, как если бы вы давали инструкции помощнику-человеку:

# Startup Checklist

1. Run `hermes cron list` and check if any scheduled jobs failed overnight.
2. If any failed, send a summary to Discord #ops using the `send_message` tool.
3. Check if `/opt/app/deploy.log` has any ERROR lines from the last 24 hours. If yes, summarize them and include in the same Discord message.
4. If nothing went wrong, reply with only `[SILENT]` so no message is sent.

Агент видит это как часть своего приглашения, поэтому все, что вы можете описать простым языком, работает — вызовы инструментов, команды оболочки, отправка сообщений, суммирование файлов.

Шаг 2: Создайте крючок

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

~/.hermes/hooks/boot-md/HOOK.yaml

name: boot-md
description: Run ~/.hermes/BOOT.md on gateway startup
events:
  - gateway:startup

~/.hermes/hooks/boot-md/handler.py

"""Run ~/.hermes/BOOT.md on every gateway startup."""

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 (
        "You are running a startup boot checklist. Follow the instructions "
        "below exactly.\n\n"
        "---\n"
        f"{content}\n"
        "---\n\n"
        "Execute each instruction. Use the send_message tool to deliver any "
        "messages to platforms like Discord or Slack.\n"
        "If nothing needs attention and there is nothing to report, reply "
        "with ONLY: [SILENT]"
    )


def _run_boot_agent(content: str) -> None:
    """Spawn a one-shot agent and execute the checklist.

    Uses the gateway's resolved model and runtime credentials so this works
    against custom endpoints, aggregators, and OAuth-based providers alike.
    """
    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 completed: %s", response[:200])
        else:
            logger.info("boot-md completed (nothing to report)")
    except Exception as e:
        logger.error("boot-md agent failed: %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("Running BOOT.md (%d chars)", len(content))

    # Background thread so gateway startup isn't blocked on a full agent turn.
    thread = threading.Thread(
        target=_run_boot_agent,
        args=(content,),
        name="boot-md",
        daemon=True,
    )
    thread.start()

Две ключевые линии:

• _resolve_gateway_model() считывает текущую конфигурируемую модель шлюза.
• _resolve_runtime_agent_kwargs() разрешает учетные данные поставщика так же, как это делает обычный шлюз, включая ключи API, базовые URL-адреса, токены OAuth и пулы учетных данных.

Без них простой AIAgent() возвращается к встроенным настройкам по умолчанию и выдает ошибку 401 для любой конечной точки, отличной от умолчанию.

Шаг 3: Проверьте это

Перезапустите шлюз:

hermes gateway restart

Смотрите логи:

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

Вы должны увидеть Running BOOT.md (N chars), за которым следует либо boot-md completed: ... (сводка того, что сделал агент), либо boot-md completed (nothing to report), когда агент ответил [SILENT].

Удалите ~/.hermes/BOOT.md, чтобы отключить контрольный список — перехватчик остается загруженным, но автоматически пропускает его, когда файла нет.

Расширение шаблона

• Контрольные списки с учетом расписания: выключите datetime.now().weekday() в инструкциях BOOT.md («если сегодня понедельник, проверьте также еженедельный журнал развертывания»). Инструкции представляют собой текст в свободной форме, поэтому все, о чем агент может рассуждать, является честной игрой.
• Несколько контрольных списков: укажите перехватчик на другой файл (STARTUP.md, MORNING.md и т. д.) и зарегистрируйте для каждого отдельные каталоги перехватчиков.
• Вариант без агента: если вам не нужен полный цикл агента, полностью пропустите AIAgent и попросите обработчик отправить фиксированное уведомление непосредственно через httpx. Дешевле, быстрее и не зависит от поставщика.

Почему это не встроенный

Более ранняя версия Hermes поставляла это как встроенный перехватчик и автоматически создавала агент с пустыми настройками по умолчанию при каждой загрузке шлюза. Это удивило пользователей с настраиваемыми конечными точками и сделало эту функцию невидимой для пользователей, которые не знали, что она работает. Сохранение его в виде документированного шаблона, созданного вами в каталоге хуков, означает, что вы точно видите, что он делает, и соглашаетесь на него, записывая файлы.

Как это работает

1. При запуске шлюза HookRegistry.discover_and_load() сканирует ~/.hermes/hooks/.
2. Каждый подкаталог с HOOK.yaml + handler.py загружается динамически.
3. Обработчики регистрируются на заявленные ими события.
4. В каждой точке жизненного цикла hooks.emit() запускает все соответствующие обработчики.
5. Ошибки в любом обработчике фиксируются и протоколируются — сломанный хук никогда не приводит к сбою агента.

:::информация Перехватчики шлюза срабатывают только на шлюзе (Telegram, Discord, Slack, WhatsApp, Teams). CLI не загружает перехватчики шлюза. Для хуков, которые работают везде, используйте хуки плагинов.