事件钩子
Hermes 提供三种钩子系统,用于在关键生命周期节点运行自定义代码:
| 系统 | 注册方式 | 运行环境 | 使用场景 |
|---|---|---|---|
| 网关钩子 | ~/.hermes/hooks/ 目录下的 HOOK.yaml + handler.py | 仅网关 | 日志记录、告警、Webhooks |
| 插件钩子 | 插件 中的 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: 将所有 Agent 活动记录到文件
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: 当 Agent 执行步骤过多时发出提醒
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 已运行 {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")
Session Start Webhook
在新会话上通过 POST 请求通知外部服务:
# ~/.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)
教程:BOOT.md — 每次网关启动时运行启动检查清单
社区中流行的一种模式:把一份 Markdown 格式的检查清单放到 ~/.hermes/BOOT.md,然后让 Agent 在每次网关启动时运行一次。适用于“每次启动时,检查隔夜的 Cron 任务失败情况,如果有失败就在 Discord 上通知我”,或者“总结过去 24 小时的 deploy.log 并发布到 Slack #ops 频道”。
本教程展示了如何自己构建一个用户定义的 Hook。Hermes 并未内置 BOOT.md Hook——你需要精确地接入你想要的行为。
我们要构建的内容
- 一个放在
~/.hermes/BOOT.md的文件,里面是自然语言描述的启动指令。 - 一个网关 Hook,监听
gateway:startup事件,启动一个一次性 Agent,该 Agent 使用你的网关已配置的模型/凭证,并执行 BOOT.md 中的指令。 - 一个
[SILENT]约定,使得 Agent 在无内容需要报告时可以选择不发送消息。
第一步:编写检查清单
创建 ~/.hermes/BOOT.md。内容就像你给人类助手下达指令那样写:
# 启动检查清单
1. 运行 `hermes cron list`,检查是否有定时任务在夜间执行失败。
2. 如果有任何失败的任务,使用 `send_message` 工具向 Discord #ops 频道发送摘要。
3. 检查 `/opt/app/deploy.log` 在过去 24 小时内是否有 ERROR 行。如果有,汇总这些错误并包含在同一条 Discord 消息中。
4. 如果一切正常,只回复 `[SILENT]`,这样就不会发送任何消息。
Agent 会将此视为其提示的一部分,因此你只需用自然语言描述任何操作即可——调用工具、运行 shell 命令、发送消息、汇总文件等。
第二步:创建 Hook
~/.hermes/hooks/boot-md/
├── HOOK.yaml
└── handler.py
~/.hermes/hooks/boot-md/HOOK.yaml
name: boot-md
description: 在网关启动时运行 ~/.hermes/BOOT.md
events:
- gateway:startup
~/.hermes/hooks/boot-md/handler.py
"""每次网关启动时运行 ~/.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:
"""启动一个一次性 Agent 并执行检查清单。
使用网关解析的模型和运行时凭证,因此可兼容自定义端点、聚合器以及基于 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 agent 执行失败:%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))
# 在后台线程中执行,避免网关启动被完整的 Agent 回合阻塞。
thread = threading.Thread(
target=_run_boot_agent,
args=(content,),
name="boot-md",
daemon=True,
)
thread.start()
两个关键行:
_resolve_gateway_model()读取网关当前配置的模型。_resolve_runtime_agent_kwargs()以正常网关 turn 相同的方式解析提供者凭据——包括 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: ...(Agent 执行内容的摘要)或者当 Agent 回复了 [SILENT] 时的 boot-md completed (nothing to report)。
删除 ~/.hermes/BOOT.md 即可禁用该检查清单——钩子仍会加载,但文件不存在时会静默跳过。
扩展模式
- 定时检查清单: 在 BOOT.md 的指令中通过
datetime.now().weekday()判断星期几(例如“如果是周一,还要检查本周部署日志”)。指令是自由文本,因此 Agent 能推理的任何内容都可以使用。 - 多份检查清单: 将钩子指向不同的文件(
STARTUP.md、MORNING.md等),并为每个文件注册单独的钩子目录。 - 非 Agent 变体: 如果你不需要完整的 Agent 循环,可以完全跳过
AIAgent,让 handler 通过httpx直接发送固定通知。更便宜、更快,且没有提供者依赖。
为什么这不是内置功能
Hermes 的早期版本将这个功能作为内置 hook 发布,并在每次网关启动时静默生成一个使用默认配置的 Agent。这会给那些使用自定义端点的用户带来意外,也让不知道此功能正在运行的用户无法感知其存在。将其保留为一种文档化的模式——由你在 hooks 目录中自行构建——意味着你能清楚看到它的作用,并通过编写文件来选择启用。
工作原理
- 网关启动时,
HookRegistry.discover_and_load()扫描~/.hermes/hooks/目录 - 每个包含
HOOK.yaml和handler.py的子目录都会被动态加载 - Handler 会被注册到它们声明的事件上
- 在每个生命周期节点,
hooks.emit()会触发所有匹配的 handler - 任何 handler 中的错误都会被捕获并记录——一个出错的 hook 永远不会导致 Agent 崩溃
Gateway hooks 仅在网关(Telegram、Discord、Slack、WhatsApp、Teams)中触发。CLI 不会加载 gateway hooks。如果你需要能在任何地方工作的 hooks,请使用 plugin hooks。
Plugin Hooks
插件 可以注册在 CLI 和网关 会话中都会触发的 hooks。这些 hooks 是在插件的 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 子任务退出时 | 忽略 |
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"⚠ 正在执行潜在危险工具:{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 | 工具调度所花费的时间,单位为毫秒(通过 registry.dispatch() 周围的 time.monotonic() 测量)。 |
触发时机: 在 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)
示例 — 安全护栏(guardrails):
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 # 尽力而为
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
当一个全新的 session 被创建时,仅触发一次。在 session 延续(用户在已有 session 中发送第二条消息)时不会触发。
回调签名:
def my_callback(session_id: str, model: str, platform: str, **kwargs):
| 参数 | 类型 | 描述 |
|---|---|---|
session_id | str | 新 session 的唯一标识符 |
model | str | 模型标识符 |
platform | str | session 运行的位置 |
触发时机: 在 run_agent.py 中的 run_conversation() 里,新 session 的第一个回合期间——具体来说,是在系统提示构建之后、tool loop 开始之前。检查条件是 if not conversation_history(没有之前消息 = 新 session)。
返回值: 忽略。
使用场景: 初始化会话作用域的状态、预热缓存、向外部服务注册会话、记录会话启动日志。
示例 — 初始化一个会话缓存:
_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 的退出处理器中,但仅当退出发生时 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:
# 将累积数据刷新到磁盘或外部服务
status = "completed" if completed else ("interrupted" if interrupted else "failed")
print(f"会话 {session_id} 结束: {status}, {cache['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,网关垃圾回收了空闲会话,或者 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(当会话被重置或垃圾回收时)触发。在 gateway 端始终与 on_session_reset 成对出现。
返回值: 忽略。
使用场景: 在会话 ID 被丢弃前持久化最终会话指标、关闭每个会话的资源、发送最终遥测事件、排空已入队的写入。
on_session_reset
当 gateway 为活跃聊天切换到新的会话键时触发——用户调用了 /new、/reset、/clear,或者适配器在空闲窗口后选取了新的会话。这让插件能在对话状态被清空时做出反应,而无需等待下一次 on_session_start。
回调签名:
def my_callback(session_id: str, platform: str, **kwargs):
| 参数 | 类型 | 描述 |
|---|---|---|
session_id | str | 新会话的 ID(已轮换为新的值)。 |
platform | str | 消息平台名称。 |
触发时机: 在 gateway/run.py 中,分配新会话键后、处理下一条入站消息之前立即触发。在 gateway 端,顺序为:on_session_finalize(old_id) → 切换 → on_session_reset(new_id) → 第一个入站轮次时触发 on_session_start(new_id)。
返回值: 忽略。
使用场景: 重置以 session_id 为键的每会话缓存,发出“会话轮转”分析事件,为新会话状态桶初始化。
完整的工具架构、处理器和高级钩子模式示例,请参阅 构建插件指南。
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() 清空所有子 futures 后触发。触发过程被编排到父线程,因此钩子作者无需担心并发回调的执行顺序。
返回值: 忽略。 用例: 记录编排活动、累加子任务耗时用于计费、编写委派后审计记录。
示例——记录编排器活动:
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 之后立即触发。内部事件完全跳过该 hook(因为它们是由系统生成的——如后台进程完成等——不应受用户层面策略的阻拦)。 |
返回值: None 或一个字典。第一个被识别到的 action 字典生效;其余插件结果被忽略。插件回调中的异常会被捕获并记录日志;出错时网关始终回退到正常分发流程。
| 返回值 | 效果 |
|---|---|
{"action": "skip", "reason": "..."} | 丢弃该消息——无 Agent 回复、无配对流程、无认证。假定插件已处理该消息(例如静默写入对话记录)。 |
{"action": "rewrite", "text": "new text"} | 替换 event.text,然后使用修改后的事件继续正常分发。适用于将缓冲的 ambient 消息合并为一条提示词。 |
{"action": "allow"} / None | 正常分发——执行完整的认证/配对/Agent 循环链。 |
使用场景: 仅监听的群聊(仅在被@时回复;将 ambient 消息缓冲到上下文中);人工交接(静默摄入客户消息,同时所有者手动处理聊天);按配置文件限流;策略驱动的路由。
示例——静默丢弃未授权的私信,不触发配对代码:
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)
示例 — 将 ambient 消息缓冲区重写为提及时的单条提示:
_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" 表示交互式 CLI/TUI 提示,"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 相同的 kwargs,外加:
| 参数 | 类型 | 描述 |
|---|---|---|
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 报告重写为项目特定的格式。
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 剥离和机密脱敏之前。让插件在任何下游处理接触原始 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、网关或程序化调用者)之前执行。此钩子允许插件通过经典的编程方法来重写助手的最终文本,而无需额外消耗推理 token 来生成 SOUL 风格的提示文本或技能驱动的转换。
回调签名:
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 的行为一致。
使用场景: 应用个性/词汇转换(如海盗用语、海绵宝宝风格)、从最终文本中屏蔽用户特定标识符、附加项目特定的签名页脚、在不消耗 SOUL 指令 token 的情况下强制执行内部风格指南。
import os, re
def spongebob(response_text, **kwargs):
if os.environ.get("SPONGEBOB_MODE") != "on":
return None # 原样通过,不做修改
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 插件。
当你需要一个即插即用的单文件脚本(Bash、Python 或任何带有 shebang 的脚本)时,可以使用 shell 钩子来:
- 阻止工具调用 —— 拒绝危险的
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) | 否 |
| 授权方式 | 首次使用时按 (事件, 命令) 对提示 | 隐式(信任 Python 插件) | 隐式(信任目录) |
| 进程间隔离 | 是(子进程) | 否(进程内) | 否(进程内) |
配置结构
hooks:
<事件名称>: # 必须在 VALID_HOOKS 中
- matcher: "<正则表达式>" # 可选;仅用于 pre/post_tool_call
command: "<shell 命令>" # 必填;通过 shlex.split 运行,shell=False
timeout: <秒数> # 可选;默认 60,上限 300
hooks_auto_accept: false # 参见下方“授权模型”
事件名称必须是插件钩子事件之一;拼写错误会产生"您是不是想用 X?"的警告并跳过。单个条目中的未知键会被忽略;缺少 command 会跳过并给出警告。timeout > 300 会被截断并给出警告。
JSON 线路协议
每次事件触发时,Hermes 都会为每个匹配的钩子(在匹配器允许的情况下)生成一个子进程,通过 stdin 管道传输 JSON 负载,并读取 stdout 返回的 JSON。
stdin — 脚本接收的负载:
{
"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 字典包含所有事件特定的关键字参数(user_message、conversation_history、child_role、duration_ms 等)。不可序列化的值会被字符串化,而不是被省略。
stdout — 可选响应:
// 阻止 pre_tool_call(两种格式均可接受;内部会进行标准化):
{"decision": "block", "reason": "禁止: rm -rf"} // Claude-Code 风格
{"action": "block", "message": "禁止: rm -rf"} // Hermes 规范风格
// 为 pre_llm_call 注入上下文:
{"context": "今天是星期五,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 # 丢弃标准输入的有效载荷
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. 记录每个 subagent 的完成
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 或 gateway)将跳过该提示。
有三种“逃生口”可以绕过交互式提示——任意一种即可:
- CLI 中的
--accept-hooks标志(例如hermes --accept-hooks chat) HERMES_ACCEPT_HOOKS=1环境变量cli-config.yaml中的hooks_auto_accept: true
非 TTY 运行(gateway、cron、CI)需要这三种之一——否则任何新添加的 hook 都会静默地保持未注册状态,并记录一条警告。
脚本编辑会静默信任。 允许列表以精确的命令字符串为键,而不是脚本的哈希值,因此编辑磁盘上的脚本不会使同意失效。hermes hooks doctor 会标记 mtime 变化,以便您发现编辑并决定是否重新批准。
hermes hooks CLI
| 命令 | 作用 |
|---|---|
hermes hooks list | 转储已配置的 hooks,包含匹配器、超时和同意状态 |
hermes hooks test <event> [--for-tool X] [--payload-file F] | 对每个匹配的 hook 使用模拟载荷触发,并打印解析后的响应 |
hermes hooks revoke <command> | 移除所有匹配 <command> 的允许列表条目(下次重启生效) |
hermes hooks doctor | 对每个配置的 hook:检查可执行位、允许列表状态、mtime 变化、JSON 输出有效性以及大致执行时间 |
安全性
Shell 钩子会以你的完整用户凭据运行——其信任边界与 cron 条目或 shell 别名相同。请将 config.yaml 中的 hooks: 块视为特权配置:
- 只引用你亲自编写或完全审查过的脚本。
- 将脚本保存在
~/.hermes/agent-hooks/目录下,以便路径易于审计。 - 拉取共享配置后,重新运行
hermes hooks doctor,以便在钩子注册前发现新增的钩子。 - 如果你的 config.yaml 在团队中进行版本控制,请像审查 CI 配置一样审查修改
hooks:部分的 PR。
顺序与优先级
Python 插件钩子和 Shell 钩子都通过同一个 invoke_hook() 分发器执行。Python 插件先注册(discover_and_load()),Shell 钩子后注册(register_from_config()),因此在平局情况下,Python 的 pre_tool_call 块决策具有更高优先级。第一个有效的块胜出——只要任何回调返回包含非空消息的 {"action": "block", "message": str},聚合器就会立即返回。