Skip to main content

事件钩子

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/resetplatform, user_id, session_key
agent:startAgent 开始处理消息时platform, user_id, session_id, message
agent:step工具调用循环的每次迭代platform, user_id, session_id, iteration, tool_names
agent:endAgent 完成处理时platform, user_id, session_id, message, response
command:*执行任何斜杠命令时platform, user_id, command, args

通配符匹配

command:* 注册的处理程序会响应任何 command: 事件(如 command:modelcommand: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——你需要精确地接入你想要的行为。

我们要构建的内容

  1. 一个放在 ~/.hermes/BOOT.md 的文件,里面是自然语言描述的启动指令。
  2. 一个网关 Hook,监听 gateway:startup 事件,启动一个一次性 Agent,该 Agent 使用你的网关已配置的模型/凭证,并执行 BOOT.md 中的指令。
  3. 一个 [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.mdMORNING.md 等),并为每个文件注册单独的钩子目录。
  • 非 Agent 变体: 如果你不需要完整的 Agent 循环,可以完全跳过 AIAgent,让 handler 通过 httpx 直接发送固定通知。更便宜、更快,且没有提供者依赖。

为什么这不是内置功能

Hermes 的早期版本将这个功能作为内置 hook 发布,并在每次网关启动时静默生成一个使用默认配置的 Agent。这会给那些使用自定义端点的用户带来意外,也让不知道此功能正在运行的用户无法感知其存在。将其保留为一种文档化的模式——由你在 hooks 目录中自行构建——意味着你能清楚看到它的作用,并通过编写文件来选择启用。

工作原理

  1. 网关启动时,HookRegistry.discover_and_load() 扫描 ~/.hermes/hooks/ 目录
  2. 每个包含 HOOK.yamlhandler.py 的子目录都会被动态加载
  3. Handler 会被注册到它们声明的事件上
  4. 在每个生命周期节点,hooks.emit() 会触发所有匹配的 handler
  5. 任何 handler 中的错误都会被捕获并记录——一个出错的 hook 永远不会导致 Agent 崩溃
info

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_finalizeCLI/网关拆除活跃会话时(刷新、保存、统计)忽略
on_session_reset网关切换为新会话密钥时(例如 /new/reset忽略
subagent_stopdelegate_task 子任务退出时忽略
pre_gateway_dispatch网关收到用户消息后、认证和分发之前{"action": "skip" | "rewrite" | "allow", ...} 用于影响流程
pre_approval_request危险命令需要用户批准时,在发送提示/通知之前忽略
post_approval_response用户对批准提示做出响应(或超时)时忽略
transform_tool_result任何工具返回之后、结果交回模型之前str 用于替换结果,None 表示不修改
transform_terminal_outputterminal 工具内部,截断/ANSI 剥离/脱敏之前str 用于替换原始输出,None 表示不修改
transform_llm_output工具调用循环完成后、最终响应交付之前str 用于替换响应文本,None/空值表示不修改

pre_tool_call

在每次工具执行(包括内置工具和插件工具)之前立即触发。

回调签名:

def my_callback(tool_name: str, args: dict, task_id: str, **kwargs):
参数类型描述
tool_namestr即将执行的工具名称(例如 "terminal""web_search""read_file"
argsdict模型传递给工具的参数字典
task_idstr会话/任务标识符。如果未设置则为空字符串。

触发时机:model_tools.pyhandle_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_namestr刚刚执行完毕的工具名称
argsdict模型传递给工具的参数
resultstr工具的返回值(始终为 JSON 字符串)
task_idstr会话/任务标识符。如果未设置则为空字符串。
duration_msint工具调度所花费的时间,单位为毫秒(通过 registry.dispatch() 周围的 time.monotonic() 测量)。

触发时机:model_tools.pyhandle_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_idstr当前会话的唯一标识符
user_messagestr用户在本轮次中的原始消息(在技能注入之前)
conversation_historylist完整消息列表的副本(OpenAI 格式:[{"role": "user", "content": "..."}]
is_first_turnbool如果是新会话的第一轮则为 True,后续轮次为 False
modelstr模型标识符(例如 "anthropic/claude-sonnet-4.6"
platformstr会话运行所在的平台:"cli""telegram""discord"
触发时机:run_agent.pyrun_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_idstr当前会话的唯一标识符
user_messagestr该轮用户发送的原始消息
assistant_responsestr该轮 Agent 生成的最终文本响应
conversation_historylist该轮完成后完整消息列表的副本
modelstr模型标识符
platformstr会话运行所在的平台

触发时机:run_agent.pyrun_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_idstr新 session 的唯一标识符
modelstr模型标识符
platformstrsession 运行的位置

触发时机: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_idstr会话的唯一标识符
completedbool如果 Agent 生成了最终响应则为 True,否则为 False
interruptedbool如果该轮对话被中断(用户发送新消息、/stop 或退出)则为 True
modelstr模型标识符
platformstr会话运行所在的平台

触发位置: 两处:

  1. run_agent.py — 每次 run_conversation() 调用的末尾,在所有清理工作之后。即使该轮对话出错也会触发。
  2. cli.py — 在 CLI 的退出处理器中,但仅当退出发生时 Agent 正处于一轮对话中间(_agent_running=True)。这捕获了处理过程中的 Ctrl+C 和 /exit 操作。在这种情况下,completed=Falseinterrupted=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_idstrNone外发会话 ID。如果不存在活跃会话,则为 None
platformstr"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_idstr新会话的 ID(已轮换为新的值)。
platformstr消息平台名称。

触发时机: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_idstr发起委托的父 Agent 的会话 ID
child_rolestr | None设置在子 Agent 上的编排器角色标签(如果未启用该功能则为 None
child_summarystr | None子 Agent 返回给父 Agent 的最终响应
child_statusstr"completed""failed""interrupted""error"
duration_msint运行子 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)
info

在深度委派场景下(例如编排角色 × 5 个叶节点 × 嵌套深度),subagent_stop 在每个回合中会触发多次。请确保回调函数处理迅速;将耗时操作推入后台队列。


pre_gateway_dispatch

在网关中的每个传入的 MessageEvent 触发一次,触发时机在内部事件守卫之后、但认证/配对和 Agent 分发之前。这是实现网关级消息流策略(只读窗口、人工交接、按对话路由等)的拦截点,这些策略很难完全纳入某个单一平台适配器。

回调签名:

def my_callback(event, gateway, session_store, **kwargs):
参数类型描述
eventMessageEvent标准化后的入站消息(包含 .text.source.message_id.internal 等属性)。
gatewayGatewayRunner当前活动的网关运行器,插件可通过 gateway.adapters[platform].send(...) 发送旁路回复(如所有者通知等)。
session_storeSessionStore用于通过 session_store.append_to_transcript(...) 静默注入对话记录。
触发时机:gateway/run.pyGatewayRunner._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,
):
参数类型描述
commandstr待批准的 shell 命令
descriptionstr命令被标记的人类可读原因(多个模式匹配时合并)
pattern_keystr触发批准的主模式键(例如 "rm_rf""sudo"
pattern_keyslist[str]所有匹配到的模式键
session_keystr会话标识符,用于限定每个聊天的通知范围
surfacestr"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,外加:

参数类型描述
choicestr"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_namestr产生结果的工具(read_fileweb_extractdelegate_task 等)。
argumentsdict模型调用工具时使用的参数。
resultstr工具的原始结果字符串,已经过截断和 ANSI 转义符清理。
task_idstr | 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:
参数类型描述
commandstr产生输出的 shell 命令。
outputstr原始的合并 stdout/stderr(可能非常大——截断发生在钩子之后)。
exit_codeint进程退出码。
cwdstr命令运行的工作目录。

返回值: str 表示替换输出,None 表示保持不变。

使用场景: 为产生大量输出的命令(du -ahfindtree)注入摘要;用项目特定标记标记输出,以便下游钩子知道如何处理;去除运行之间波动且破坏提示缓存的时序噪声。

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_textstr本轮次助手的最终响应文本。
session_idstr当前会话的会话 ID(单次运行可能为空)。
modelstr生成该响应的模型名称(例如 anthropic/claude-sonnet-4.6)。
platformstr交付平台(clitelegramdiscord 等;未设置时为空)。

返回值: 非空 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_stopVALID_HOOKS网关生命周期(gateway:startupagent:*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_callsubagent_stop、会话生命周期),tool_nametool_inputnullextra 字典包含所有事件特定的关键字参数(user_messageconversation_historychild_roleduration_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)将跳过该提示。

有三种“逃生口”可以绕过交互式提示——任意一种即可:

  1. CLI 中的 --accept-hooks 标志(例如 hermes --accept-hooks chat
  2. HERMES_ACCEPT_HOOKS=1 环境变量
  3. 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 &lt;event&gt; [--for-tool X] [--payload-file F]对每个匹配的 hook 使用模拟载荷触发,并打印解析后的响应
hermes hooks revoke &lt;command&gt;移除所有匹配 &lt;command&gt; 的允许列表条目(下次重启生效)
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},聚合器就会立即返回。