事件钩子（Event Hooks）

Hermes 有三个钩子系统，可在关键生命周期节点运行自定义代码：

系统	注册方式	运行于	用途
网关钩子	`~/.hermes/hooks/` 中的 `HOOK.yaml` + `handler.py`	仅网关	日志记录、警报、Webhook
插件钩子	插件中的 `ctx.register_hook()`	CLI + 网关	工具拦截、指标、防护栏
Shell 钩子	`~/.hermes/config.yaml` 中的 `hooks:` 块，指向 shell 脚本	CLI + 网关	用于阻断、自动格式化、上下文注入的即插脚本

三个系统都是非阻塞的——任何钩子中的错误都会被捕获并记录，永远不会使 Agent 崩溃。

网关事件钩子

网关钩子在网关运行期间（Telegram、Discord、Slack、WhatsApp、Teams）自动触发，不会阻塞主 Agent 流水线。

创建钩子

每个钩子是 ~/.hermes/hooks/ 下包含两个文件的目录：

~/.hermes/hooks/
└── my-hook/
    ├── HOOK.yaml      # 声明要监听的事件
    └── handler.py     # Python 处理函数

HOOK.yaml

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

events 列表决定哪些事件触发你的处理函数。你可以订阅任意事件组合，包括通配符，如 command:*。

handler.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——两者都有效
错误会被捕获并记录，永远不会使 Agent 崩溃

可用事件

事件	触发时机	上下文键
`gateway:startup`	网关进程启动	`platforms`（活跃平台名称列表）
`session:start`	新消息会话创建	`platform`、`user_id`、`session_id`、`session_key`
`session:end`	会话结束（重置前）	`platform`、`user_id`、`session_key`
`session:reset`	用户运行 `/new` 或 `/reset`	`platform`、`user_id`、`session_key`
`agent:start`	Agent 开始处理消息	`platform`、`user_id`、`session_id`、`message`
`agent:step`	工具调用循环的每次迭代	`platform`、`user_id`、`session_id`、`iteration`、`tool_names`
`agent:end`	Agent 完成处理	`platform`、`user_id`、`session_id`、`message`、`response`
`command:*`	执行任何斜杠命令	`platform`、`user_id`、`command`、`args`

通配符匹配

为 command:* 注册的处理函数会触发任何 command: 事件（command:model、command:reset 等）。通过单个订阅监控所有斜杠命令。

示例

Telegram 长任务警报

当 Agent 运行超过 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")

会话开始 Webhook

在新会话时向外部服务发送 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——每次网关启动时运行启动检查清单

社区中一个流行的模式：在 ~/.hermes/BOOT.md 放置一个 Markdown 检查清单，让 Agent 在每次网关启动时运行一次。适用于"每次启动时检查夜间 cron 失败情况，如有问题则在 Discord 通知我"，或"汇总最近 24 小时的 deploy.log 并发布到 Slack #ops"。

本教程展示如何以用户自定义钩子的方式自己构建它。Hermes 不附带内置的 BOOT.md 钩子——你可以精确地连接你想要的行为。

我们要构建的

一个位于 ~/.hermes/BOOT.md 的文件，包含自然语言启动指令。
一个在 gateway:startup 触发的网关钩子，使用你网关已解析的模型/凭证生成一个一次性 Agent，并运行 BOOT.md 指令。
一个 [SILENT] 约定，让 Agent 在没有需要报告的内容时选择不发送消息。

第一步：编写检查清单

创建 ~/.hermes/BOOT.md。就像给人类助理下指令一样编写它：

# Startup Checklist

Run `hermes cron list` and check if any scheduled jobs failed overnight.
If any failed, send a summary to Discord #ops using the `send_message` tool.
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.
If nothing went wrong, reply with only `[SILENT]` so no message is sent.

Agent 将此视为其提示词的一部分，因此任何可以用自然语言描述的内容都有效——工具调用、Shell 命令、发送消息、汇总文件。

第二步：创建钩子

~/.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 token 和凭证池。

没有这些，裸 AIAgent() 会回退到内置默认值，对任何非默认端点都会返回 401 错误。

第三步：测试

重启网关：

hermes gateway restart

查看日志：

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

你应该看到 Running BOOT.md (N chars)，然后是 boot-md completed: ...（Agent 操作摘要）或 boot-md completed (nothing to report)（Agent 回复了 [SILENT] 时）。

删除 ~/.hermes/BOOT.md 即可禁用检查清单——钩子保持加载状态，但在文件不存在时会静默跳过。

扩展模式

感知日程的检查清单： 在 BOOT.md 的指令中基于 datetime.now().weekday() 进行条件判断（"如果是星期一，还要检查每周部署日志"）。指令是自由格式文本，Agent 能推理的任何内容都可以。
多个检查清单： 将钩子指向不同的文件（STARTUP.md、MORNING.md 等），并为每个文件注册单独的钩子目录。
非 Agent 变体： 如果不需要完整的 Agent 循环，完全跳过 AIAgent，直接通过 httpx 从处理函数发布固定通知。成本更低、速度更快，且没有提供商依赖。

为什么不作为内置功能

Hermes 早期版本将此作为内置钩子，在每次网关启动时静默生成使用裸默认值的 Agent。这让使用自定义端点的用户感到意外，也让不知道它在运行的用户对该功能毫不知情。将其作为有文档记录的模式保留——由你构建，在你的钩子目录中——意味着你能清楚地看到它做什么，并通过编写文件来选择加入。

工作原理

网关启动时，HookRegistry.discover_and_load() 扫描 ~/.hermes/hooks/
每个包含 HOOK.yaml + handler.py 的子目录被动态加载
处理函数被注册到其声明的事件
在每个生命周期节点，hooks.emit() 触发所有匹配的处理函数
任何处理函数中的错误都会被捕获并记录——损坏的钩子永远不会使 Agent 崩溃

信息

网关钩子仅在网关（Telegram、Discord、Slack、WhatsApp、Teams）中触发。CLI 不加载网关钩子。对于在任何地方都有效的钩子，请使用插件钩子。

插件钩子

插件可以注册在 CLI 和网关会话中都触发的钩子。这些钩子通过插件的 register() 函数中的 ctx.register_hook() 以编程方式注册。

def register(ctx):
    ctx.register_hook("pre_tool_call", my_tool_observer)
    ctx.register_hook("post_tool_call", my_tool_logger)
    ctx.register_hook("pre_llm_call", my_memory_callback)
    ctx.register_hook("post_llm_call", my_sync_callback)
    ctx.register_hook("on_session_start", my_init_callback)
    ctx.register_hook("on_session_end", my_cleanup_callback)

所有钩子的通用规则：

回调函数接收关键字参数。始终接受 **kwargs 以保持向前兼容性——未来版本可能会添加新参数而不破坏你的插件。
如果回调函数崩溃，它会被记录并跳过。其他钩子和 Agent 继续正常运行。行为不当的插件永远不会使 Agent 崩溃。
两个钩子的返回值会影响行为：pre_tool_call 可以阻断工具，pre_llm_call 可以注入上下文到 LLM 调用中。所有其他钩子都是即发即忘的观察者。

快速参考

钩子	触发时机	返回值
`pre_tool_call`	任何工具执行前	`{"action": "block", "message": str}` 以否决调用
`post_tool_call`	任何工具返回后	忽略
`pre_llm_call`	每轮次一次，工具调用循环前	`{"context": str}` 以在用户消息前添加上下文
`post_llm_call`	每轮次一次，工具调用循环后	忽略
`on_session_start`	新会话创建（仅第一轮次）	忽略
`on_session_end`	会话结束	忽略
`on_session_finalize`	CLI/网关终止活跃会话（刷新、保存、统计）	忽略
`on_session_reset`	网关换入新会话键（如 `/new`、`/reset`）	忽略
`subagent_stop`	`delegate_task` 子 Agent 已退出	忽略
`pre_gateway_dispatch`	网关收到用户消息，在认证+分发前	`{"action": "skip" \| "rewrite" \| "allow", ...}` 影响流程
`pre_approval_request`	危险命令需要用户批准，在发送提示/通知前	忽略
`post_approval_response`	用户响应批准提示（或超时）	忽略
`transform_tool_result`	任何工具返回后，在结果传回模型前	`str` 以替换结果，`None` 保持不变
`transform_terminal_output`	在 `terminal` 工具内部，在截断/ANSI 剥离/脱敏前	`str` 以替换原始输出，`None` 保持不变
`transform_llm_output`	工具调用循环完成后，最终响应投递前	`str` 以替换响应文本，`None`/空字符串保持不变

`pre_tool_call`

在每个工具执行之前立即触发——内置工具和插件工具均包括。

回调函数签名：

def my_callback(tool_name: str, args: dict, task_id: str, **kwargs):

参数	类型	说明
`tool_name`	`str`	即将执行的工具名称（如 `"terminal"`、`"web_search"`、`"read_file"`）
`args`	`dict`	模型传给工具的参数
`task_id`	`str`	会话/任务标识符。未设置时为空字符串。

触发位置： 在 model_tools.py 的 handle_function_call() 中，工具处理函数运行前。每次工具调用触发一次——如果模型并行调用 3 个工具，则触发 3 次。

返回值——否决调用：

return {"action": "block", "message": "Reason the tool call was blocked"}

Agent 以 message 作为返回给模型的错误来短路工具调用。第一个匹配的阻断指令获胜（Python 插件先注册，然后是 Shell 钩子）。任何其他返回值都被忽略，因此仅用于观察的现有回调保持正常工作。

使用场景： 日志记录、审计追踪、工具调用计数器、阻断危险操作、速率限制、每用户策略执行。

示例——工具调用审计日志：

import json, logging
from datetime import datetime

logger = logging.getLogger(__name__)

def audit_tool_call(tool_name, args, task_id, **kwargs):
    logger.info("TOOL_CALL session=%s tool=%s args=%s",
                task_id, tool_name, json.dumps(args)[:200])

def register(ctx):
    ctx.register_hook("pre_tool_call", audit_tool_call)

示例——对危险工具发出警告：

DANGEROUS = {"terminal", "write_file", "patch"}

def warn_dangerous(tool_name, **kwargs):
    if tool_name in DANGEROUS:
        print(f"⚠ Executing potentially dangerous tool: {tool_name}")

def register(ctx):
    ctx.register_hook("pre_tool_call", warn_dangerous)

`post_tool_call`

在每个工具执行返回之后立即触发。

回调函数签名：

def my_callback(tool_name: str, args: dict, result: str, task_id: str,
                duration_ms: int, **kwargs):

参数	类型	说明
`tool_name`	`str`	刚刚执行的工具名称
`args`	`dict`	模型传给工具的参数
`result`	`str`	工具的返回值（始终是 JSON 字符串）
`task_id`	`str`	会话/任务标识符。未设置时为空字符串。
`duration_ms`	`int`	工具分发所花费的时间，以毫秒为单位（用 `time.monotonic()` 在 `registry.dispatch()` 周围测量）。

触发位置： 在 model_tools.py 的 handle_function_call() 中，工具处理函数返回后。每次工具调用触发一次。如果工具抛出未处理的异常不会触发（错误被捕获并作为错误 JSON 字符串返回，而 post_tool_call 以该错误字符串作为 result 触发）。

返回值： 忽略。

使用场景： 记录工具结果、指标收集、追踪工具成功/失败率、延迟仪表板、每工具预算警报、特定工具完成时发送通知。

示例——追踪工具使用指标：

from collections import Counter, defaultdict
import json

_tool_counts = Counter()
_error_counts = Counter()
_latency_ms = defaultdict(list)

def track_metrics(tool_name, result, duration_ms=0, **kwargs):
    _tool_counts[tool_name] += 1
    _latency_ms[tool_name].append(duration_ms)
    try:
        parsed = json.loads(result)
        if "error" in parsed:
            _error_counts[tool_name] += 1
    except (json.JSONDecodeError, TypeError):
        pass

def register(ctx):
    ctx.register_hook("post_tool_call", track_metrics)

`pre_llm_call`

每轮次触发一次，在工具调用循环开始前。这是唯一使用返回值的钩子——它可以将上下文注入到当前轮次的用户消息中。

回调函数签名：

def my_callback(session_id: str, user_message: str, conversation_history: list,
                is_first_turn: bool, model: str, platform: str, **kwargs):

参数	类型	说明
`session_id`	`str`	当前会话的唯一标识符
`user_message`	`str`	本轮次用户的原始消息（技能注入前）
`conversation_history`	`list`	完整消息列表的副本（OpenAI 格式：`[{"role": "user", "content": "..."}]`）
`is_first_turn`	`bool`	如果这是新会话的第一轮次则为 `True`，后续轮次为 `False`
`model`	`str`	模型标识符（如 `"anthropic/claude-sonnet-4.6"`）
`platform`	`str`	会话运行的位置：`"cli"`、`"telegram"`、`"discord"` 等

触发位置： 在 run_agent.py 的 run_conversation() 中，上下文压缩后、主 while 循环前。每次 run_conversation() 调用触发一次（即每轮次用户消息触发一次），而非工具循环中的每次 API 调用触发一次。

返回值： 如果回调返回带有 "context" 键的字典，或非空的普通字符串，文本将被追加到当前轮次的用户消息中。返回 None 则不注入。

# 注入上下文
return {"context": "Recalled memories:\n- User likes Python\n- Working on hermes-agent"}

# 普通字符串（等效）
return "Recalled memories:\n- User likes Python"

# 不注入
return None

上下文注入位置： 始终是用户消息，而不是系统提示词。这保留了提示词缓存——系统提示词在轮次之间保持相同，因此缓存的 token 被重用。系统提示词是 Hermes 的领地（模型指导、工具执行、个性化、技能）。插件与用户的输入一起贡献上下文。

所有注入的上下文都是临时的——仅在 API 调用时添加。原始用户消息在对话历史中永远不会被修改，也不会持久化到会话数据库中。

当多个插件返回上下文时，它们的输出按插件发现顺序（按目录名字母顺序）用双换行连接。

使用场景： 记忆回溯、RAG 上下文注入、防护栏、每轮次分析。

示例——记忆回溯：

import httpx

MEMORY_API = "https://your-memory-api.example.com"

def recall(session_id, user_message, is_first_turn, **kwargs):
    try:
        resp = httpx.post(f"{MEMORY_API}/recall", json={
            "session_id": session_id,
            "query": user_message,
        }, timeout=3)
        memories = resp.json().get("results", [])
        if not memories:
            return None
        text = "Recalled context:\n" + "\n".join(f"- {m['text']}" for m in memories)
        return {"context": text}
    except Exception:
        return None

def register(ctx):
    ctx.register_hook("pre_llm_call", recall)

示例——防护栏：

POLICY = "Never execute commands that delete files without explicit user confirmation."

def guardrails(**kwargs):
    return {"context": POLICY}

def register(ctx):
    ctx.register_hook("pre_llm_call", guardrails)

`post_llm_call`

每轮次触发一次，在工具调用循环完成且 Agent 产生最终响应后。仅在成功的轮次触发——如果轮次被中断则不触发。

回调函数签名：

def my_callback(session_id: str, user_message: str, assistant_response: str,
                conversation_history: list, model: str, platform: str, **kwargs):

参数	类型	说明
`session_id`	`str`	当前会话的唯一标识符
`user_message`	`str`	本轮次用户的原始消息
`assistant_response`	`str`	Agent 本轮次的最终文本响应
`conversation_history`	`list`	轮次完成后的完整消息列表副本
`model`	`str`	模型标识符
`platform`	`str`	会话运行的位置

触发位置： 在 run_agent.py 的 run_conversation() 中，工具循环以最终响应退出后。由 if final_response and not interrupted 保护——因此当用户在轮次中途中断或 Agent 达到迭代限制而没有产生响应时不会触发。

返回值： 忽略。

使用场景： 将对话数据同步到外部记忆系统、计算响应质量指标、记录轮次摘要、触发后续动作。

示例——同步到外部记忆：

import httpx

MEMORY_API = "https://your-memory-api.example.com"

def sync_memory(session_id, user_message, assistant_response, **kwargs):
    try:
        httpx.post(f"{MEMORY_API}/store", json={
            "session_id": session_id,
            "user": user_message,
            "assistant": assistant_response,
        }, timeout=5)
    except Exception:
        pass  # best-effort

def register(ctx):
    ctx.register_hook("post_llm_call", sync_memory)

示例——追踪响应长度：

import logging
logger = logging.getLogger(__name__)

def log_response_length(session_id, assistant_response, model, **kwargs):
    logger.info("RESPONSE session=%s model=%s chars=%d",
                session_id, model, len(assistant_response or ""))

def register(ctx):
    ctx.register_hook("post_llm_call", log_response_length)

`on_session_start`

当全新会话创建时触发一次。会话继续时（用户在现有会话中发送第二条消息）不会触发。

回调函数签名：

def my_callback(session_id: str, model: str, platform: str, **kwargs):

参数	类型	说明
`session_id`	`str`	新会话的唯一标识符
`model`	`str`	模型标识符
`platform`	`str`	会话运行的位置

触发位置： 在 run_agent.py 的 run_conversation() 中，新会话的第一轮次——具体在系统提示词构建后、工具循环开始前。检查条件为 if not conversation_history（没有先前消息 = 新会话）。

返回值： 忽略。

使用场景： 初始化会话范围的状态、预热缓存、向外部服务注册会话、记录会话开始。

示例——初始化会话缓存：

_session_caches = {}

def init_session(session_id, model, platform, **kwargs):
    _session_caches[session_id] = {
        "model": model,
        "platform": platform,
        "tool_calls": 0,
        "started": __import__("datetime").datetime.now().isoformat(),
    }

def register(ctx):
    ctx.register_hook("on_session_start", init_session)

`on_session_end`

在每次 run_conversation() 调用结束时触发，无论结果如何。如果用户在 Agent 处于轮次中途时退出，也会从 CLI 的退出处理器触发。

回调函数签名：

def my_callback(session_id: str, completed: bool, interrupted: bool,
                model: str, platform: str, **kwargs):

参数	类型	说明
`session_id`	`str`	会话的唯一标识符
`completed`	`bool`	如果 Agent 产生了最终响应则为 `True`，否则为 `False`
`interrupted`	`bool`	如果轮次被中断（用户发送新消息、`/stop` 或退出）则为 `True`
`model`	`str`	模型标识符
`platform`	`str`	会话运行的位置

触发位置：

run_agent.py — 每次 run_conversation() 调用结束时，所有清理之后。始终触发，即使轮次出错。
cli.py — 在 CLI 的 atexit 处理器中，但仅当 Agent 在退出时处于轮次中途（_agent_running=True）时。这捕获了处理过程中的 Ctrl+C 和 /exit。在这种情况下，completed=False 且 interrupted=True。

返回值： 忽略。

使用场景： 刷新缓冲区、关闭连接、持久化会话状态、记录会话时长、清理在 on_session_start 中初始化的资源。

示例——刷新并清理：

_session_caches = {}

def cleanup_session(session_id, completed, interrupted, **kwargs):
    cache = _session_caches.pop(session_id, None)
    if cache:
        # Flush accumulated data to disk or external service
        status = "completed" if completed else ("interrupted" if interrupted else "failed")
        print(f"Session {session_id} ended: {status}, {cache['tool_calls']} tool calls")

def register(ctx):
    ctx.register_hook("on_session_end", cleanup_session)

示例——会话时长追踪：

import time, logging
logger = logging.getLogger(__name__)

_start_times = {}

def on_start(session_id, **kwargs):
    _start_times[session_id] = time.time()

def on_end(session_id, completed, interrupted, **kwargs):
    start = _start_times.pop(session_id, None)
    if start:
        duration = time.time() - start
        logger.info("SESSION_DURATION session=%s seconds=%.1f completed=%s interrupted=%s",
                     session_id, duration, completed, interrupted)

def register(ctx):
    ctx.register_hook("on_session_start", on_start)
    ctx.register_hook("on_session_end", on_end)

`on_session_finalize`

当 CLI 或网关终止活跃会话时触发——例如用户运行 /new、网关 GC 了空闲会话，或 CLI 在有活跃 Agent 时退出。这是在会话身份消失前刷新与出站会话绑定的状态的最后机会。

回调函数签名：

def my_callback(session_id: str | None, platform: str, **kwargs):

参数	类型	说明
`session_id`	`str` 或 `None`	出站会话 ID。如果不存在活跃会话则可能为 `None`。
`platform`	`str`	`"cli"` 或消息平台名称（`"telegram"`、`"discord"` 等）。

触发位置： 在 cli.py（/new / CLI 退出时）和 gateway/run.py（会话重置或 GC 时）。在网关侧始终与 on_session_reset 配对。

返回值： 忽略。

使用场景： 在会话 ID 被丢弃前持久化最终会话指标，关闭每会话资源，发出最终遥测事件，排空排队的写入。

`on_session_reset`

当网关为活跃聊天换入新会话键时触发——用户调用了 /new、/reset、/clear，或适配器在空闲窗口后选择了新会话。这让插件可以对对话状态已被清除这一事实做出响应，而无需等待下一个 on_session_start。

回调函数签名：

def my_callback(session_id: str, platform: str, **kwargs):

参数	类型	说明
`session_id`	`str`	新会话的 ID（已轮换为新值）。
`platform`	`str`	消息平台名称。

触发位置： 在 gateway/run.py 中，新会话键分配后、处理下一条入站消息前立即触发。在网关上，顺序为：on_session_finalize(old_id) → 切换 → on_session_reset(new_id) → 第一条入站轮次的 on_session_start(new_id)。

返回值： 忽略。

使用场景： 重置以 session_id 为键的每会话缓存，发出"会话已轮换"分析事件，为新状态桶做准备。

参见**构建插件指南**获取完整演练，包括工具 schema、处理器和高级钩子模式。

`subagent_stop`

delegate_task 完成后每个子 Agent 触发一次。无论你委派了单个任务还是三个批次，该钩子对每个子 Agent 各触发一次，在父线程上序列化。

回调函数签名：

def my_callback(parent_session_id: str, child_role: str | None,
                child_summary: str | None, child_status: str,
                duration_ms: int, **kwargs):

参数	类型	说明
`parent_session_id`	`str`	委派父 Agent 的会话 ID
`child_role`	`str \| None`	子 Agent 上设置的编排角色标签（功能未启用时为 `None`）
`child_summary`	`str \| None`	子 Agent 返回给父 Agent 的最终响应
`child_status`	`str`	`"completed"`、`"failed"`、`"interrupted"` 或 `"error"`
`duration_ms`	`int`	运行子 Agent 所花费的挂钟时间，以毫秒为单位

触发位置： 在 tools/delegate_tool.py 中，ThreadPoolExecutor.as_completed() 耗尽所有子 future 后。触发被编组到父线程，因此钩子作者不必考虑并发回调执行。

返回值： 忽略。

使用场景： 记录编排活动，为计费积累子 Agent 时长，写入委派后审计记录。

示例——记录编排活动：

import logging
logger = logging.getLogger(__name__)

def log_subagent(parent_session_id, child_role, child_status, duration_ms, **kwargs):
    logger.info(
        "SUBAGENT parent=%s role=%s status=%s duration_ms=%d",
        parent_session_id, child_role, child_status, duration_ms,
    )

def register(ctx):
    ctx.register_hook("subagent_stop", log_subagent)

信息

进行大量委派时（如编排角色 × 5 个叶子节点 × 嵌套深度），subagent_stop 每轮次会触发多次。保持回调快速；将昂贵工作推送到后台队列。

`pre_gateway_dispatch`

在网关中每条传入 MessageEvent 触发一次，在内部事件保护后但在认证/配对和 Agent 分发前。这是网关级别消息流策略（只听模式窗口、人工接管、每聊天路由等）的拦截点，这些策略不适合整洁地放入任何单个平台适配器。

回调函数签名：

def my_callback(event, gateway, session_store, **kwargs):

参数	类型	说明
`event`	`MessageEvent`	标准化的入站消息（有 `.text`、`.source`、`.message_id`、`.internal` 等）。
`gateway`	`GatewayRunner`	活跃的网关运行器，供插件调用 `gateway.adapters[platform].send(...)` 进行侧通道回复（所有者通知等）。
`session_store`	`SessionStore`	通过 `session_store.append_to_transcript(...)` 进行静默转录摄取。

触发位置： 在 gateway/run.py 的 GatewayRunner._handle_message() 中，is_internal 计算完成后立即触发。内部事件完全跳过该钩子（它们是系统生成的——后台进程完成等——绝对不能被用户面向策略拦截）。

返回值： None 或字典。第一个被识别的动作字典获胜；剩余的插件结果被忽略。插件回调中的异常被捕获并记录；网关在出错时始终回落到正常分发。

返回值	效果
`{"action": "skip", "reason": "..."}`	丢弃消息——无 Agent 回复、无配对流程、无认证。插件假定已处理（如静默摄取到转录）。
`{"action": "rewrite", "text": "new text"}`	替换 `event.text`，然后继续以修改后的事件进行正常分发。用于将缓冲的环境消息合并成单个提示词。
`{"action": "allow"}` / `None`	正常分发——运行完整的认证/配对/Agent 循环链。

使用场景： 只听群聊（仅在被 @ 时回复；将环境消息缓冲到上下文中）；人工接管（所有者手动处理聊天时静默摄取客户消息）；每配置文件速率限制；策略驱动的路由。

示例——静默丢弃未授权私信而不触发配对代码：

def deny_unauthorized_dms(event, **kwargs):
    src = event.source
    if src.chat_type == "dm" and not _is_approved_user(src.user_id):
        return {"action": "skip", "reason": "unauthorized-dm"}
    return None

def register(ctx):
    ctx.register_hook("pre_gateway_dispatch", deny_unauthorized_dms)

示例——在提及时将环境消息缓冲区改写为单个提示词：

_buffers = {}

def buffer_or_rewrite(event, **kwargs):
    key = (event.source.platform, event.source.chat_id)
    buf = _buffers.setdefault(key, [])
    if _bot_mentioned(event.text):
        combined = "\n".join(buf + [event.text])
        buf.clear()
        return {"action": "rewrite", "text": combined}
    buf.append(event.text)
    return {"action": "skip", "reason": "ambient-buffered"}

def register(ctx):
    ctx.register_hook("pre_gateway_dispatch", buffer_or_rewrite)

`pre_approval_request`

在向用户显示批准请求之前立即触发——涵盖每个界面：交互式 CLI、Ink TUI、网关平台（Telegram、Discord、Slack、WhatsApp、Matrix 等）和 ACP 客户端（VS Code、Zed、JetBrains）。

这是连接自定义通知器的合适位置——例如弹出允许/拒绝通知的 macOS 菜单栏应用，或记录每个带上下文的批准请求的审计日志。

回调函数签名：

def my_callback(
    command: str,
    description: str,
    pattern_key: str,
    pattern_keys: list[str],
    session_key: str,
    surface: str,
    **kwargs,
):

参数	类型	说明
`command`	`str`	等待批准的 Shell 命令
`description`	`str`	命令被标记的人类可读原因（多个模式匹配时组合）
`pattern_key`	`str`	触发批准的主模式键（如 `"rm_rf"`、`"sudo"`）
`pattern_keys`	`list[str]`	所有匹配的模式键
`session_key`	`str`	会话标识符，用于每聊天通知范围
`surface`	`str`	交互式 CLI/TUI 提示为 `"cli"`，异步平台批准为 `"gateway"`

返回值： 忽略。这里的钩子仅用于观察；它们不能否决或预先回答批准。使用 pre_tool_call 在工具到达批准系统前阻断它。

使用场景： 桌面通知、推送警报、审计日志、Slack Webhook、升级路由、指标。

示例——macOS 桌面通知：

import subprocess

def notify_approval(command, description, session_key, **kwargs):
    title = "Hermes needs approval"
    body = f"{description}: {command[:80]}"
    subprocess.Popen([
        "osascript", "-e",
        f'display notification "{body}" with title "{title}"',
    ])

def register(ctx):
    ctx.register_hook("pre_approval_request", notify_approval)

`post_approval_response`

用户响应批准提示（或提示超时）之后触发。

回调函数签名：

def my_callback(
    command: str,
    description: str,
    pattern_key: str,
    pattern_keys: list[str],
    session_key: str,
    surface: str,
    choice: str,
    **kwargs,
):

与 pre_approval_request 相同的参数，另加：

参数	类型	说明
`choice`	`str`	`"once"`、`"session"`、`"always"`、`"deny"` 或 `"timeout"` 之一

返回值： 忽略。

使用场景： 关闭匹配的桌面通知、在审计日志中记录最终决定、更新指标、推进速率限制器。

def log_decision(command, choice, session_key, **kwargs):
    logger.info("approval %s: %s for session %s", choice, command[:60], session_key)

def register(ctx):
    ctx.register_hook("post_approval_response", log_decision)

`transform_tool_result`

工具返回之后、结果被追加到对话之前触发。让插件在模型看到结果之前改写任何工具的结果字符串——不仅仅是终端输出。

回调函数签名：

def my_callback(
    tool_name: str,
    arguments: dict,
    result: str,
    task_id: str | None,
    **kwargs,
) -> str | None:

参数	类型	说明
`tool_name`	`str`	产生结果的工具（`read_file`、`web_extract`、`delegate_task`，……）。
`arguments`	`dict`	模型调用工具时传入的参数。
`result`	`str`	工具的原始结果字符串，截断后且 ANSI 剥离后。
`task_id`	`str \| None`	在 RL/基准环境中运行时的任务/会话 ID。

返回值： str 以替换结果（返回的字符串是模型看到的），None 保持不变。

使用场景： 从 web_extract 输出中脱敏特定组织的 PII，将长 JSON 工具响应包装在摘要头中，将检索增强提示注入 read_file 结果，将 delegate_task 子 Agent 报告改写为项目特定的 schema。

import re
SECRET = re.compile(r"sk-[A-Za-z0-9]{32,}")

def redact_secrets(tool_name, result, **kwargs):
    if SECRET.search(result):
        return SECRET.sub("[REDACTED]", result)
    return None

def register(ctx):
    ctx.register_hook("transform_tool_result", redact_secrets)

适用于每个工具。对于仅终端的改写，见下方的 transform_terminal_output——它更窄且在流水线中运行得更早（截断前、脱敏前）。

`transform_terminal_output`

在 terminal 工具的前台输出流水线内部触发，在默认的 50 KB 截断、ANSI 剥离和密钥脱敏之前。让插件在任何下游处理之前改写 Shell 命令的原始 stdout/stderr。

回调函数签名：

def my_callback(
    command: str,
    output: str,
    exit_code: int,
    cwd: str,
    task_id: str | None,
    **kwargs,
) -> str | None:

参数	类型	说明
`command`	`str`	产生输出的 Shell 命令。
`output`	`str`	原始合并的 stdout/stderr（可能非常大——截断在钩子之后发生）。
`exit_code`	`int`	进程退出码。
`cwd`	`str`	命令运行的工作目录。

返回值： str 以替换输出，None 保持不变。

使用场景： 为产生大量输出的命令注入摘要（du -ah、find、tree），用项目特定标记标记输出以便下游钩子知道如何处理，剥除在运行间抖动的计时噪声以避免提示词缓存失效。

def summarize_find(command, output, **kwargs):
    if command.startswith("find ") and len(output) > 50_000:
        lines = output.count("\n")
        head = "\n".join(output.splitlines()[:40])
        return f"{head}\n\n[summary: {lines} paths total, showing first 40]"
    return None

def register(ctx):
    ctx.register_hook("transform_terminal_output", summarize_find)

与 transform_tool_result 配合使用效果更好（它涵盖所有其他工具）。

`transform_llm_output`

在每轮工具调用循环完成、模型产生最终响应之后、该响应投递给用户（CLI、网关或程序调用方）之前触发一次。允许插件使用经典编程方法改写助手的最终文本——无需为 SOUL 风格文本或技能驱动的转换消耗额外推理 token。

回调签名：

def my_callback(
    response_text: str,
    session_id: str,
    model: str,
    platform: str,
    **kwargs,
) -> str | None:

参数	类型	描述
`response_text`	`str`	本轮助手的最终响应文本。
`session_id`	`str`	本次对话的会话 ID（单次运行时可能为空）。
`model`	`str`	生成响应的模型名称（如 `anthropic/claude-sonnet-4.6`）。
`platform`	`str`	投递平台（`cli`、`telegram`、`discord`、…；未设置时为空）。

返回值： 非空 str 以替换响应文本，None 或空字符串保持不变。当多个插件注册时，第一个非空字符串胜出——与 transform_tool_result 行为一致。

使用场景： 应用个性/词汇转换（海盗腔、Spongebob），从最终文本中脱敏用户特定标识符，追加项目特定签名页脚，在不消耗 token 于 SOUL 指令的情况下执行文档风格指南。

import os, re

def spongebob(response_text, **kwargs):
    if os.environ.get("SPONGEBOB_MODE") != "on":
        return None  # pass through unchanged
    return re.sub(r"!", "!! Tartar sauce!", response_text)

def register(ctx):
    ctx.register_hook("transform_llm_output", spongebob)

该钩子仅在响应非空且未被中断时触发——不会在停止按钮中断或空轮次上触发。异常会被记录为警告，不会中断 Agent 执行。

Shell 钩子

在你的 cli-config.yaml 中声明 Shell 脚本钩子，Hermes 在对应的插件钩子事件触发时将它们作为子进程运行——在 CLI 和网关会话中均有效。无需编写 Python 插件。

在以下情况下使用 Shell 钩子，你希望用一个即插的单文件脚本（Bash、Python 或任何带有 shebang 的脚本）来：

阻断工具调用 — 拒绝危险的 terminal 命令，执行每目录策略，要求对破坏性的 write_file / patch 操作进行批准。
工具调用后运行 — 自动格式化 Agent 刚刚写入的 Python 或 TypeScript 文件，记录 API 调用，触发 CI 工作流。
向下一次 LLM 轮次注入上下文 — 在用户消息前添加 git status 输出、当前工作日或检索到的文档（参见 pre_llm_call）。
观察生命周期事件 — 子 Agent 完成时写一行日志（subagent_stop）或会话开始时（on_session_start）。

Shell 钩子通过在 CLI 启动（hermes_cli/main.py）和网关启动（gateway/run.py）时调用 agent.shell_hooks.register_from_config(cfg) 来注册。它们与 Python 插件钩子自然组合——两者都流经相同的分发器。

一览对比

维度	Shell 钩子	插件钩子	网关钩子
声明于	`~/.hermes/config.yaml` 中的 `hooks:` 块	插件 `plugin.yaml` 中的 `register()`	`HOOK.yaml` + `handler.py` 目录
位于	`~/.hermes/agent-hooks/`（约定）	`~/.hermes/plugins/<name>/`	`~/.hermes/hooks/<name>/`
语言	任意（Bash、Python、Go 二进制，……）	仅 Python	仅 Python
运行于	CLI + 网关	CLI + 网关	仅网关
事件	`VALID_HOOKS`（包括 `subagent_stop`）	`VALID_HOOKS`	网关生命周期（`gateway:startup`、`agent:`、`command:`）
可阻断工具调用	是（`pre_tool_call`）	是（`pre_tool_call`）	否
可注入 LLM 上下文	是（`pre_llm_call`）	是（`pre_llm_call`）	否
许可	每 `(event, command)` 对首次使用时提示	隐式（Python 插件信任）	隐式（目录信任）
进程间隔离	是（子进程）	否（进程内）	否（进程内）

配置 schema

hooks:
  <event_name>:                  # 必须在 VALID_HOOKS 中
    - matcher: "<regex>"         # 可选；仅用于 pre/post_tool_call
      command: "<shell command>" # 必填；通过 shlex.split 运行，shell=False
      timeout: <seconds>         # 可选；默认 60，上限 300

hooks_auto_accept: false         # 见下方"许可模型"

事件名称必须是插件钩子事件之一；拼写错误会产生"你是否是指 X？"警告并被跳过。单个条目中的未知键会被忽略；缺少 command 则跳过并发出警告。timeout > 300 会被夹紧并发出警告。

JSON 通信协议

每次事件触发时，Hermes 为每个匹配的钩子（在匹配器允许的情况下）生成一个子进程，将 JSON 负载通过标准输入传递，并从标准输出读回 JSON。

标准输入——脚本接收的负载：

{
  "hook_event_name": "pre_tool_call",
  "tool_name":       "terminal",
  "tool_input":      {"command": "rm -rf /"},
  "session_id":      "sess_abc123",
  "cwd":             "/home/user/project",
  "extra":           {"task_id": "...", "tool_call_id": "..."}
}

对于非工具事件（pre_llm_call、subagent_stop、会话生命周期），tool_name 和 tool_input 为 null。extra 字典包含所有特定于事件的 kwargs（user_message、conversation_history、child_role、duration_ms，……）。不可序列化的值被字符串化而非省略。

标准输出——可选响应：

// 阻断 pre_tool_call（两种形式均被接受；内部标准化）：
{"decision": "block", "reason":  "Forbidden: rm -rf"}   // Claude-Code 风格
{"action":   "block", "message": "Forbidden: rm -rf"}   // Hermes 标准

// 为 pre_llm_call 注入上下文：
{"context": "Today is Friday, 2026-04-17"}

// 静默无操作——任何空/不匹配的输出都可以：

格式错误的 JSON、非零退出码和超时会记录警告，但永远不会中止 Agent 循环。

实际示例

1. 每次写入后自动格式化 Python 文件

# ~/.hermes/config.yaml
hooks:
  post_tool_call:
    - matcher: "write_file|patch"
      command: "~/.hermes/agent-hooks/auto-format.sh"

#!/usr/bin/env bash
# ~/.hermes/agent-hooks/auto-format.sh
payload="$(cat -)"
path=$(echo "$payload" | jq -r '.tool_input.path // empty')
[[ "$path" == *.py ]] && command -v black >/dev/null && black "$path" 2>/dev/null
printf '{}\n'

Agent 在上下文中的文件视图不会自动重新读取——重新格式化只影响磁盘上的文件。后续的 read_file 调用会获取格式化后的版本。

2. 阻断破坏性的 `terminal` 命令

hooks:
  pre_tool_call:
    - matcher: "terminal"
      command: "~/.hermes/agent-hooks/block-rm-rf.sh"
      timeout: 5

#!/usr/bin/env bash
# ~/.hermes/agent-hooks/block-rm-rf.sh
payload="$(cat -)"
cmd=$(echo "$payload" | jq -r '.tool_input.command // empty')
if echo "$cmd" | grep -qE 'rm[[:space:]]+-rf?[[:space:]]+/'; then
  printf '{"decision": "block", "reason": "blocked: rm -rf / is not permitted"}\n'
else
  printf '{}\n'
fi

3. 将 `git status` 注入每轮次（Claude-Code `UserPromptSubmit` 等效）

hooks:
  pre_llm_call:
    - command: "~/.hermes/agent-hooks/inject-cwd-context.sh"

#!/usr/bin/env bash
# ~/.hermes/agent-hooks/inject-cwd-context.sh
cat - >/dev/null   # discard stdin payload
if status=$(git status --porcelain 2>/dev/null) && [[ -n "$status" ]]; then
  jq --null-input --arg s "$status" \
     '{context: ("Uncommitted changes in cwd:\n" + $s)}'
else
  printf '{}\n'
fi

Claude Code 的 UserPromptSubmit 事件在 Hermes 中故意不作为单独事件——pre_llm_call 在同一位置触发并已支持上下文注入。在这里使用它。

4. 记录每个子 Agent 完成情况

hooks:
  subagent_stop:
    - command: "~/.hermes/agent-hooks/log-orchestration.sh"

#!/usr/bin/env bash
# ~/.hermes/agent-hooks/log-orchestration.sh
log=~/.hermes/logs/orchestration.log
jq -c '{ts: now, parent: .session_id, extra: .extra}' < /dev/stdin >> "$log"
printf '{}\n'

许可模型

每个唯一的 (event, command) 对在 Hermes 首次看到时会提示用户批准，然后将决定持久化到 ~/.hermes/shell-hooks-allowlist.json。后续运行（CLI 或网关）跳过提示。

三个规避方式可以绕过交互式提示——任意一个即可：

CLI 上的 --accept-hooks 标志（如 hermes --accept-hooks chat）
HERMES_ACCEPT_HOOKS=1 环境变量
cli-config.yaml 中的 hooks_auto_accept: true

非 TTY 运行（网关、cron、CI）需要这三者之一——否则任何新添加的钩子都会静默保持未注册状态并记录警告。

脚本编辑被静默信任。 允许列表以确切的命令字符串为键，而非脚本的哈希，因此编辑磁盘上的脚本不会使许可失效。hermes hooks doctor 会标记 mtime 漂移，以便你发现编辑并决定是否重新批准。

`hermes hooks` CLI

命令	功能
`hermes hooks list`	导出已配置的钩子，含匹配器、超时和许可状态
`hermes hooks test <event> [--for-tool X] [--payload-file F]`	对合成负载触发每个匹配的钩子并打印解析后的响应
`hermes hooks revoke <command>`	删除所有匹配 `<command>` 的允许列表条目（下次重启生效）
`hermes hooks doctor`	对每个已配置的钩子：检查执行位、允许列表状态、mtime 漂移、JSON 输出有效性和大致执行时间

安全性

Shell 钩子以你的完整用户凭证运行——与 cron 条目或 Shell 别名的信任边界相同。将 config.yaml 中的 hooks: 块视为特权配置：

仅引用你编写或完全审查过的脚本。
将脚本保存在 ~/.hermes/agent-hooks/ 中，使路径易于审计。
拉取共享配置后重新运行 hermes hooks doctor，在新添加的钩子注册前发现它们。
如果你的 config.yaml 在团队中进行版本控制，审查更改 hooks: 部分的 PR，就像审查 CI 配置一样。

排序和优先级

Python 插件钩子和 Shell 钩子都流经同一个 invoke_hook() 分发器。Python 插件先注册（discover_and_load()），Shell 钩子后注册（register_from_config()），因此在平局情况下 Python pre_tool_call 阻断决定优先。第一个有效的阻断获胜——聚合器在任何回调产生带有非空消息的 {"action": "block", "message": str} 时立即返回。

网关事件钩子​

创建钩子​

HOOK.yaml​

handler.py​

可用事件​

通配符匹配​

示例​

Telegram 长任务警报​

命令使用日志记录器​

会话开始 Webhook​

教程：BOOT.md——每次网关启动时运行启动检查清单​

我们要构建的​

第一步：编写检查清单​

第二步：创建钩子​

第三步：测试​

扩展模式​

为什么不作为内置功能​

工作原理​

插件钩子​

快速参考​

pre_tool_call​

post_tool_call​

pre_llm_call​

post_llm_call​

on_session_start​

on_session_end​

on_session_finalize​

on_session_reset​

subagent_stop​

pre_gateway_dispatch​

pre_approval_request​

post_approval_response​

transform_tool_result​

transform_terminal_output​

transform_llm_output​

Shell 钩子​

一览对比​

配置 schema​

JSON 通信协议​

实际示例​

1. 每次写入后自动格式化 Python 文件​

2. 阻断破坏性的 terminal 命令​

3. 将 git status 注入每轮次（Claude-Code UserPromptSubmit 等效）​

4. 记录每个子 Agent 完成情况​

许可模型​

hermes hooks CLI​

安全性​

排序和优先级​

网关事件钩子

创建钩子

HOOK.yaml

handler.py

可用事件

通配符匹配

示例

Telegram 长任务警报

命令使用日志记录器

会话开始 Webhook

教程：BOOT.md——每次网关启动时运行启动检查清单

我们要构建的

第一步：编写检查清单

第二步：创建钩子

第三步：测试

扩展模式

为什么不作为内置功能

工作原理

插件钩子

快速参考

`pre_tool_call`

`post_tool_call`

`pre_llm_call`

`post_llm_call`

`on_session_start`

`on_session_end`

`on_session_finalize`

`on_session_reset`

`subagent_stop`

`pre_gateway_dispatch`

`pre_approval_request`

`post_approval_response`

`transform_tool_result`

`transform_terminal_output`

`transform_llm_output`

Shell 钩子

一览对比

配置 schema

JSON 通信协议

实际示例

1. 每次写入后自动格式化 Python 文件

2. 阻断破坏性的 `terminal` 命令

3. 将 `git status` 注入每轮次（Claude-Code `UserPromptSubmit` 等效）

4. 记录每个子 Agent 完成情况

许可模型

`hermes hooks` CLI

安全性

排序和优先级