feat(hooks): outbound webhooks — push signed lifecycle events to external HTTP endpoints

The inverse of the inbound webhook platform: hooks.outbound in
config.yaml lists HTTP targets + the plugin-hook events they subscribe
to (on_session_end, subagent_stop, post_tool_call, ...). Each firing
POSTs a JSON payload (same top-level shape as shell hooks' stdin wire)
signed GitHub-style with HMAC-SHA256 (X-Hermes-Signature-256).

Rides the existing hook bus — notify-only callbacks registered on the
plugin manager at the same CLI/gateway/main entry points as shell
hooks. Delivery is fire-and-forget via a bounded queue + single daemon
worker thread, so a dead endpoint can never stall a tool call. Bounded
retries (5xx/conn errors once; 4xx never). secret_env preferred over
inline secret. HERMES_SAFE_MODE skips registration. hermes hooks list
shows outbound targets with signed/UNSIGNED status.

Zero new model tools, zero new subsystems.
This commit is contained in:
Teknium 2026-07-22 07:26:29 -07:00
parent 17155e3ae0
commit a39bfbd803
No known key found for this signature in database
9 changed files with 1129 additions and 43 deletions

529
agent/outbound_webhooks.py Normal file
View file

@ -0,0 +1,529 @@
"""
Outbound webhook notifications.
Reads the ``hooks.outbound:`` list from ``config.yaml`` and registers
notify-only callbacks on the existing plugin hook manager, so every
``invoke_hook()`` site can push lifecycle events to external HTTP
endpoints CI systems, dashboards, other agents with zero changes to
call sites and zero polling on the receiving end.
This is the outbound mirror of the inbound webhook platform
(``gateway/platforms/webhook.py``): inbound wakes Hermes when the world
changes; outbound tells the world when Hermes does something.
Design notes
------------
* Delivery is fire-and-forget through a bounded in-process queue and a
single daemon worker thread. ``invoke_hook()`` runs inside the agent
loop, so callbacks must never block on network I/O they serialize,
enqueue, and return ``None`` immediately. Outbound targets can never
block a tool call, inject context, or otherwise influence agent flow.
* Payloads are signed with HMAC-SHA256 (GitHub-style
``X-Hermes-Signature-256: sha256=<hexdigest>`` over the raw body) when
a secret is configured. Receivers verify exactly like they verify
GitHub webhooks.
* No consent prompt: unlike shell hooks, an outbound target executes no
code on this machine it POSTs JSON to a URL the user themselves put
in config. ``HERMES_SAFE_MODE=1`` still skips registration, matching
plugins / MCP / shell hooks.
* Registration is idempotent safe to invoke from both the CLI entry
point and the gateway entry point.
Config schema (``~/.hermes/config.yaml``)::
hooks:
outbound:
- url: https://ci.example.com/hermes-events
events: [on_session_end, subagent_stop]
# secret literal (discouraged) or env var name (preferred):
secret_env: HERMES_OUTBOUND_WEBHOOK_SECRET
# optional regex, honored for pre/post_tool_call only:
matcher: "terminal|delegate_task"
timeout: 10 # per-attempt seconds, clamped to [1, 60]
name: ci-notify # optional label for logs / `hermes hooks list`
Wire format (POST body)::
{
"hook_event_name": "on_session_end",
"tool_name": null,
"tool_input": null,
"session_id": "sess_abc123",
"cwd": "/home/user/project",
"extra": {...}, # event-specific kwargs
"delivery_id": "3f2c...", # uuid4, unique per POST
"timestamp": "2026-07-22T14:00:00Z"
}
Headers::
Content-Type: application/json
User-Agent: Hermes-Agent-Outbound-Webhook
X-Hermes-Event: <hook event name>
X-Hermes-Delivery: <delivery_id>
X-Hermes-Signature-256: sha256=<hmac hexdigest> # only when secret set
"""
from __future__ import annotations
import hashlib
import hmac
import json
import logging
import os
import queue
import re
import threading
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple
from urllib import error as urlerror
from urllib import request as urlrequest
logger = logging.getLogger(__name__)
DEFAULT_TIMEOUT_SECONDS = 10
MAX_TIMEOUT_SECONDS = 60
MAX_DELIVERY_ATTEMPTS = 2
RETRY_BACKOFF_SECONDS = 1.0
QUEUE_MAX_SIZE = 256
# Events whose ``matcher`` field is honored (mirrors shell hooks).
_TOOL_SCOPED_EVENTS = {"pre_tool_call", "post_tool_call"}
# kwargs promoted to top-level payload keys (mirrors shell hooks wire).
_TOP_LEVEL_PAYLOAD_KEYS = {"tool_name", "args", "session_id", "parent_session_id"}
# (event, url) pairs already wired to the plugin manager in this process.
_registered: Set[Tuple[str, str]] = set()
_registered_lock = threading.Lock()
_delivery_queue: "queue.Queue[Optional[Dict[str, Any]]]" = queue.Queue(
maxsize=QUEUE_MAX_SIZE
)
_worker_lock = threading.Lock()
_worker: Optional[threading.Thread] = None
@dataclass
class WebhookTarget:
"""Parsed and validated representation of one ``hooks.outbound`` entry."""
url: str
events: List[str]
name: str = ""
secret: Optional[str] = None
matcher: Optional[str] = None
timeout: int = DEFAULT_TIMEOUT_SECONDS
compiled_matcher: Optional[re.Pattern] = field(default=None, repr=False)
def __post_init__(self) -> None:
if isinstance(self.matcher, str):
stripped = self.matcher.strip()
self.matcher = stripped if stripped else None
if self.matcher:
try:
self.compiled_matcher = re.compile(self.matcher)
except re.error as exc:
logger.warning(
"outbound webhook matcher %r is invalid (%s) — treating "
"as literal equality", self.matcher, exc,
)
self.compiled_matcher = None
@property
def label(self) -> str:
return self.name or self.url
def matches_tool(self, tool_name: Optional[str]) -> bool:
if not self.matcher:
return True
if tool_name is None:
return False
if self.compiled_matcher is not None:
return self.compiled_matcher.fullmatch(tool_name) is not None
return tool_name == self.matcher
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def register_from_config(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget]:
"""Register every configured outbound webhook on the plugin manager.
``cfg`` is the full parsed config dict. Missing, empty, or malformed
``hooks.outbound`` is treated as zero targets config parsing never
raises, because a broken webhook entry must not crash the agent.
Returns the targets that ended up wired (deduplicated across repeat
calls, so the CLI and gateway can both invoke this safely).
"""
if not isinstance(cfg, dict):
return []
from utils import env_var_enabled
if env_var_enabled("HERMES_SAFE_MODE"):
logger.info("HERMES_SAFE_MODE=1 — outbound webhook registration skipped")
return []
hooks_cfg = cfg.get("hooks")
targets = _parse_outbound_block(
hooks_cfg.get("outbound") if isinstance(hooks_cfg, dict) else None
)
if not targets:
return []
from hermes_cli.plugins import get_plugin_manager
manager = get_plugin_manager()
registered: List[WebhookTarget] = []
with _registered_lock:
for target in targets:
wired_any = False
for event in target.events:
key = (event, target.url)
if key in _registered:
continue
manager._hooks.setdefault(event, []).append(
_make_callback(event, target)
)
_registered.add(key)
wired_any = True
logger.info(
"outbound webhook registered: %s -> %s (matcher=%s, "
"timeout=%ds)",
event, target.label, target.matcher, target.timeout,
)
if wired_any:
registered.append(target)
return registered
def iter_configured_targets(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget]:
"""Parse ``hooks.outbound`` without registering anything.
Used by ``hermes hooks list``."""
if not isinstance(cfg, dict):
return []
hooks_cfg = cfg.get("hooks")
return _parse_outbound_block(
hooks_cfg.get("outbound") if isinstance(hooks_cfg, dict) else None
)
def flush(timeout: float = 5.0) -> bool:
"""Block until all queued deliveries are done (or *timeout* elapses).
Returns ``True`` when the queue fully drained. Test/shutdown helper."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
with _delivery_queue.all_tasks_done:
if _delivery_queue.unfinished_tasks == 0:
return True
time.sleep(0.02)
with _delivery_queue.all_tasks_done:
return _delivery_queue.unfinished_tasks == 0
def reset_for_tests() -> None:
"""Clear the idempotence set and drain the queue. Test-only helper."""
with _registered_lock:
_registered.clear()
try:
while True:
_delivery_queue.get_nowait()
_delivery_queue.task_done()
except queue.Empty:
pass
# ---------------------------------------------------------------------------
# Config parsing
# ---------------------------------------------------------------------------
def _parse_outbound_block(raw: Any) -> List[WebhookTarget]:
if raw is None:
return []
if not isinstance(raw, list):
logger.warning(
"hooks.outbound must be a list of webhook targets; got %s",
type(raw).__name__,
)
return []
targets: List[WebhookTarget] = []
for i, entry in enumerate(raw):
target = _parse_single_target(i, entry)
if target is not None:
targets.append(target)
return targets
def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]:
from hermes_cli.plugins import VALID_HOOKS
if not isinstance(raw, dict):
logger.warning(
"hooks.outbound[%d] must be a mapping with 'url' and 'events' "
"keys; got %s", index, type(raw).__name__,
)
return None
url = raw.get("url")
if not isinstance(url, str) or not url.strip():
logger.warning("hooks.outbound[%d] is missing a non-empty 'url'", index)
return None
url = url.strip()
if not url.lower().startswith(("http://", "https://")):
logger.warning(
"hooks.outbound[%d].url must be http(s); got %r — skipped",
index, url,
)
return None
if url.lower().startswith("http://"):
logger.warning(
"hooks.outbound[%d].url uses plain http:// — payloads (including "
"tool inputs) travel unencrypted. Prefer https.", index,
)
events_raw = raw.get("events")
if not isinstance(events_raw, list) or not events_raw:
logger.warning(
"hooks.outbound[%d] needs a non-empty 'events' list (valid: %s)",
index, ", ".join(sorted(VALID_HOOKS)),
)
return None
events: List[str] = []
for ev in events_raw:
if ev in VALID_HOOKS:
events.append(ev)
else:
logger.warning(
"hooks.outbound[%d]: unknown event %r ignored (valid: %s)",
index, ev, ", ".join(sorted(VALID_HOOKS)),
)
if not events:
logger.warning(
"hooks.outbound[%d] has no valid events — skipped", index,
)
return None
matcher = raw.get("matcher")
if matcher is not None and not isinstance(matcher, str):
logger.warning(
"hooks.outbound[%d].matcher must be a string regex; ignoring",
index,
)
matcher = None
if matcher is not None and not any(e in _TOOL_SCOPED_EVENTS for e in events):
logger.warning(
"hooks.outbound[%d].matcher=%r will be ignored — matcher is only "
"honored for pre_tool_call / post_tool_call.", index, matcher,
)
matcher = None
timeout_raw = raw.get("timeout", DEFAULT_TIMEOUT_SECONDS)
try:
timeout = int(timeout_raw)
except (TypeError, ValueError):
logger.warning(
"hooks.outbound[%d].timeout must be an int (got %r); using "
"default %ds", index, timeout_raw, DEFAULT_TIMEOUT_SECONDS,
)
timeout = DEFAULT_TIMEOUT_SECONDS
timeout = max(1, min(timeout, MAX_TIMEOUT_SECONDS))
secret = _resolve_secret(index, raw)
name = raw.get("name")
if not isinstance(name, str):
name = ""
return WebhookTarget(
url=url,
events=events,
name=name.strip(),
secret=secret,
matcher=matcher,
timeout=timeout,
)
def _resolve_secret(index: int, raw: Dict[str, Any]) -> Optional[str]:
"""``secret_env`` (env var name, preferred) wins over inline ``secret``."""
secret_env = raw.get("secret_env")
if isinstance(secret_env, str) and secret_env.strip():
value = os.environ.get(secret_env.strip(), "")
if value:
return value
logger.warning(
"hooks.outbound[%d].secret_env=%r is not set in the environment "
"— deliveries will be UNSIGNED", index, secret_env.strip(),
)
return None
secret = raw.get("secret")
if isinstance(secret, str) and secret:
return secret
return None
# ---------------------------------------------------------------------------
# Callback + delivery
# ---------------------------------------------------------------------------
def _make_callback(event: str, target: WebhookTarget):
"""Build the notify-only closure ``invoke_hook()`` calls per firing."""
def _callback(**kwargs: Any) -> None:
if event in _TOOL_SCOPED_EVENTS:
if not target.matches_tool(kwargs.get("tool_name")):
return None
try:
body = _serialize_payload(event, kwargs)
except Exception: # defensive — a bad payload must not hurt the loop
logger.warning(
"outbound webhook payload serialization failed (event=%s "
"target=%s)", event, target.label, exc_info=True,
)
return None
_enqueue(_build_delivery(event, target, body))
return None
_callback.__name__ = f"outbound_webhook[{event}:{target.label}]"
_callback.__qualname__ = _callback.__name__
return _callback
def _serialize_payload(event: str, kwargs: Dict[str, Any]) -> bytes:
"""Render the POST body. Same top-level shape as shell hooks' stdin
(documented in :mod:`agent.shell_hooks`), plus delivery metadata."""
extras = {k: v for k, v in kwargs.items() if k not in _TOP_LEVEL_PAYLOAD_KEYS}
try:
cwd = str(Path.cwd())
except OSError:
cwd = ""
payload = {
"hook_event_name": event,
"tool_name": kwargs.get("tool_name"),
"tool_input": kwargs.get("args") if isinstance(kwargs.get("args"), dict) else None,
"session_id": kwargs.get("session_id") or kwargs.get("parent_session_id") or "",
"cwd": cwd,
"extra": extras,
"delivery_id": uuid.uuid4().hex,
"timestamp": datetime.now(tz=timezone.utc)
.isoformat()
.replace("+00:00", "Z"),
}
return json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8")
def _build_delivery(
event: str, target: WebhookTarget, body: bytes,
) -> Dict[str, Any]:
headers = {
"Content-Type": "application/json",
"User-Agent": "Hermes-Agent-Outbound-Webhook",
"X-Hermes-Event": event,
"X-Hermes-Delivery": uuid.uuid4().hex,
}
if target.secret:
digest = hmac.new(
target.secret.encode("utf-8"), body, hashlib.sha256
).hexdigest()
headers["X-Hermes-Signature-256"] = f"sha256={digest}"
return {
"url": target.url,
"label": target.label,
"event": event,
"body": body,
"headers": headers,
"timeout": target.timeout,
}
def _enqueue(delivery: Dict[str, Any]) -> None:
_ensure_worker()
try:
_delivery_queue.put_nowait(delivery)
except queue.Full:
logger.warning(
"outbound webhook queue full (%d pending) — dropping %s event "
"for %s", QUEUE_MAX_SIZE, delivery["event"], delivery["label"],
)
def _ensure_worker() -> None:
global _worker
if _worker is not None and _worker.is_alive():
return
with _worker_lock:
if _worker is not None and _worker.is_alive():
return
_worker = threading.Thread(
target=_worker_loop, name="outbound-webhooks", daemon=True,
)
_worker.start()
def _worker_loop() -> None:
while True:
delivery = _delivery_queue.get()
try:
if delivery is not None:
_deliver(delivery)
except Exception: # pragma: no cover — defensive
logger.warning(
"outbound webhook delivery crashed (target=%s)",
delivery.get("label") if isinstance(delivery, dict) else "?",
exc_info=True,
)
finally:
_delivery_queue.task_done()
def _deliver(delivery: Dict[str, Any]) -> None:
"""POST with bounded retries. Retries on connection errors and 5xx;
4xx is the receiver telling us the request itself is wrong no retry."""
last_error = ""
for attempt in range(1, MAX_DELIVERY_ATTEMPTS + 1):
req = urlrequest.Request(
delivery["url"],
data=delivery["body"],
headers=delivery["headers"],
method="POST",
)
try:
with urlrequest.urlopen(req, timeout=delivery["timeout"]) as resp:
status = getattr(resp, "status", 200)
if 200 <= status < 300:
logger.debug(
"outbound webhook delivered: %s -> %s (HTTP %d)",
delivery["event"], delivery["label"], status,
)
return
last_error = f"HTTP {status}"
except urlerror.HTTPError as exc:
last_error = f"HTTP {exc.code}"
if 400 <= exc.code < 500:
logger.warning(
"outbound webhook rejected (event=%s target=%s): %s"
"not retrying", delivery["event"], delivery["label"],
last_error,
)
return
except Exception as exc:
last_error = str(exc) or type(exc).__name__
if attempt < MAX_DELIVERY_ATTEMPTS:
time.sleep(RETRY_BACKOFF_SECONDS * attempt)
logger.warning(
"outbound webhook delivery failed after %d attempt(s) (event=%s "
"target=%s): %s",
MAX_DELIVERY_ATTEMPTS, delivery["event"], delivery["label"], last_error,
)

View file

@ -318,8 +318,9 @@ def _parse_hooks_block(hooks_cfg: Any) -> List[ShellHookSpec]:
for event_name, entries in hooks_cfg.items():
# Reserved sub-keys that aren't event names — skip silently. These
# are config sub-sections nested under `hooks:` for related
# functionality (e.g. output-spill budgets).
if event_name in ("output_spill",):
# functionality (e.g. output-spill budgets, outbound webhooks —
# the latter parsed by agent/outbound_webhooks.py).
if event_name in ("output_spill", "outbound"):
continue
if event_name not in VALID_HOOKS:
suggestion = difflib.get_close_matches(

9
cli.py
View file

@ -1032,7 +1032,14 @@ def _prepare_deferred_agent_startup() -> None:
from agent.shell_hooks import register_from_config
from hermes_cli.config import load_config
register_from_config(load_config(), accept_hooks=_accept_hooks)
_hooks_cfg = load_config()
register_from_config(_hooks_cfg, accept_hooks=_accept_hooks)
from agent.outbound_webhooks import (
register_from_config as register_outbound_webhooks,
)
register_outbound_webhooks(_hooks_cfg)
except Exception:
logger.debug(
"shell-hook registration failed at deferred CLI startup",

View file

@ -7653,7 +7653,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
try:
from hermes_cli.config import load_config
from agent.shell_hooks import register_from_config
register_from_config(load_config(), accept_hooks=False)
_hooks_cfg = load_config()
register_from_config(_hooks_cfg, accept_hooks=False)
from agent.outbound_webhooks import (
register_from_config as register_outbound_webhooks,
)
register_outbound_webhooks(_hooks_cfg)
except Exception:
logger.debug(
"shell-hook registration failed at gateway startup",

View file

@ -50,53 +50,71 @@ def hooks_command(args) -> None:
def _cmd_list(_args) -> None:
from hermes_cli.config import load_config
from agent import shell_hooks
from agent import outbound_webhooks, shell_hooks
specs = shell_hooks.iter_configured_hooks(load_config())
cfg = load_config()
specs = shell_hooks.iter_configured_hooks(cfg)
outbound = outbound_webhooks.iter_configured_targets(cfg)
if not specs:
print("No shell hooks configured in ~/.hermes/config.yaml.")
if not specs and not outbound:
print("No shell hooks or outbound webhooks configured in ~/.hermes/config.yaml.")
print("See `hermes hooks --help` or")
print(" website/docs/user-guide/features/hooks.md")
print("for the config schema and worked examples.")
return
by_event: Dict[str, List] = {}
for spec in specs:
by_event.setdefault(spec.event, []).append(spec)
if not specs:
print("No shell hooks configured in ~/.hermes/config.yaml.")
else:
by_event: Dict[str, List] = {}
for spec in specs:
by_event.setdefault(spec.event, []).append(spec)
allowlist = shell_hooks.load_allowlist()
approved = {
(e.get("event"), e.get("command"))
for e in allowlist.get("approvals", [])
if isinstance(e, dict)
}
allowlist = shell_hooks.load_allowlist()
approved = {
(e.get("event"), e.get("command"))
for e in allowlist.get("approvals", [])
if isinstance(e, dict)
}
print(f"Configured shell hooks ({len(specs)} total):\n")
print(f"Configured shell hooks ({len(specs)} total):\n")
for event in sorted(by_event.keys()):
print(f" [{event}]")
for spec in by_event[event]:
is_approved = (spec.event, spec.command) in approved
status = "✓ allowed" if is_approved else "✗ not allowlisted"
matcher_part = f" matcher={spec.matcher!r}" if spec.matcher else ""
for event in sorted(by_event.keys()):
print(f" [{event}]")
for spec in by_event[event]:
is_approved = (spec.event, spec.command) in approved
status = "✓ allowed" if is_approved else "✗ not allowlisted"
matcher_part = f" matcher={spec.matcher!r}" if spec.matcher else ""
print(
f" - {spec.command}{matcher_part} "
f"(timeout={spec.timeout}s, {status})"
)
if is_approved:
entry = shell_hooks.allowlist_entry_for(spec.event, spec.command)
if entry and entry.get("approved_at"):
print(f" approved_at: {entry['approved_at']}")
mtime_now = shell_hooks.script_mtime_iso(spec.command)
mtime_at = entry.get("script_mtime_at_approval")
if mtime_now and mtime_at and mtime_now > mtime_at:
print(
f" ⚠ script modified since approval "
f"(was {mtime_at}, now {mtime_now}) — "
f"run `hermes hooks doctor` to re-validate"
)
print()
if outbound:
print(f"Configured outbound webhooks ({len(outbound)} total):\n")
for target in outbound:
signed = "signed" if target.secret else "UNSIGNED"
matcher_part = f" matcher={target.matcher!r}" if target.matcher else ""
print(f" - {target.label}")
print(f" url: {target.url}")
print(
f" - {spec.command}{matcher_part} "
f"(timeout={spec.timeout}s, {status})"
f" events: {', '.join(target.events)}{matcher_part} "
f"(timeout={target.timeout}s, {signed})"
)
if is_approved:
entry = shell_hooks.allowlist_entry_for(spec.event, spec.command)
if entry and entry.get("approved_at"):
print(f" approved_at: {entry['approved_at']}")
mtime_now = shell_hooks.script_mtime_iso(spec.command)
mtime_at = entry.get("script_mtime_at_approval")
if mtime_now and mtime_at and mtime_now > mtime_at:
print(
f" ⚠ script modified since approval "
f"(was {mtime_at}, now {mtime_now}) — "
f"run `hermes hooks doctor` to re-validate"
)
print()

View file

@ -13608,7 +13608,14 @@ def _prepare_agent_startup(args) -> None:
from hermes_cli.config import load_config
from agent.shell_hooks import register_from_config
register_from_config(load_config(), accept_hooks=_accept_hooks)
_hooks_cfg = load_config()
register_from_config(_hooks_cfg, accept_hooks=_accept_hooks)
from agent.outbound_webhooks import (
register_from_config as register_outbound_webhooks,
)
register_outbound_webhooks(_hooks_cfg)
except Exception:
logger.debug(
"shell-hook registration failed at CLI startup",

View file

@ -0,0 +1,437 @@
"""Tests for the outbound webhook dispatcher (agent.outbound_webhooks).
Covers config parsing, matcher behaviour, HMAC signing, payload shape,
idempotent registration on the plugin manager, and end-to-end delivery
against a real local HTTP server (no mocks on the network path).
"""
from __future__ import annotations
import hashlib
import hmac
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import pytest
from agent import outbound_webhooks
# ── helpers ───────────────────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def _reset_registration_state():
_strip_outbound_callbacks()
outbound_webhooks.reset_for_tests()
yield
_strip_outbound_callbacks()
outbound_webhooks.reset_for_tests()
def _strip_outbound_callbacks():
"""Remove outbound-webhook callbacks from the shared plugin manager.
``reset_for_tests`` clears the idempotence set but the manager singleton
keeps previously-registered callbacks; without this, a target registered
in one test would fire (real network!) in every later test in this file.
"""
from hermes_cli.plugins import get_plugin_manager
manager = get_plugin_manager()
for event, callbacks in list(manager._hooks.items()):
manager._hooks[event] = [
cb for cb in callbacks
if not getattr(cb, "__name__", "").startswith("outbound_webhook[")
]
class _CapturingHandler(BaseHTTPRequestHandler):
"""Records every POST (path, headers, body) on the server instance."""
def do_POST(self): # noqa: N802 — http.server naming
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
self.server.captured.append( # type: ignore[attr-defined]
{
"path": self.path,
"headers": dict(self.headers),
"body": body,
}
)
status = getattr(self.server, "respond_status", 200)
self.send_response(status)
self.end_headers()
def log_message(self, format, *args): # noqa: A002 — http.server naming
pass
@pytest.fixture()
def http_server():
server = HTTPServer(("127.0.0.1", 0), _CapturingHandler)
server.captured = [] # type: ignore[attr-defined]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
yield server
server.shutdown()
server.server_close()
def _url(server: HTTPServer, path: str = "/hook") -> str:
return f"http://127.0.0.1:{server.server_address[1]}{path}"
def _cfg(*entries):
return {"hooks": {"outbound": list(entries)}}
# ── config parsing ────────────────────────────────────────────────────────
class TestParseConfig:
def test_missing_block_is_empty(self):
assert outbound_webhooks.iter_configured_targets({}) == []
assert outbound_webhooks.iter_configured_targets(None) == []
assert outbound_webhooks.iter_configured_targets({"hooks": {}}) == []
assert outbound_webhooks.iter_configured_targets(
{"hooks": "not-a-dict"}
) == []
def test_non_list_outbound_is_empty(self):
assert outbound_webhooks.iter_configured_targets(
{"hooks": {"outbound": {"url": "https://x"}}}
) == []
def test_valid_entry_parses(self):
targets = outbound_webhooks.iter_configured_targets(
_cfg(
{
"url": "https://example.com/hook",
"events": ["on_session_end"],
"name": "ci",
"timeout": 5,
}
)
)
assert len(targets) == 1
t = targets[0]
assert t.url == "https://example.com/hook"
assert t.events == ["on_session_end"]
assert t.name == "ci"
assert t.timeout == 5
def test_missing_url_skipped(self):
assert outbound_webhooks.iter_configured_targets(
_cfg({"events": ["on_session_end"]})
) == []
def test_non_http_url_skipped(self):
assert outbound_webhooks.iter_configured_targets(
_cfg({"url": "ftp://example.com", "events": ["on_session_end"]})
) == []
assert outbound_webhooks.iter_configured_targets(
_cfg({"url": "file:///etc/passwd", "events": ["on_session_end"]})
) == []
def test_unknown_events_filtered_known_kept(self):
targets = outbound_webhooks.iter_configured_targets(
_cfg(
{
"url": "https://example.com/hook",
"events": ["on_session_end", "not_a_real_event"],
}
)
)
assert len(targets) == 1
assert targets[0].events == ["on_session_end"]
def test_all_unknown_events_skips_entry(self):
assert outbound_webhooks.iter_configured_targets(
_cfg({"url": "https://example.com", "events": ["bogus"]})
) == []
def test_empty_events_skips_entry(self):
assert outbound_webhooks.iter_configured_targets(
_cfg({"url": "https://example.com", "events": []})
) == []
def test_timeout_clamped(self):
targets = outbound_webhooks.iter_configured_targets(
_cfg(
{
"url": "https://example.com",
"events": ["on_session_end"],
"timeout": 9999,
}
)
)
assert targets[0].timeout == outbound_webhooks.MAX_TIMEOUT_SECONDS
def test_bad_timeout_falls_back_to_default(self):
targets = outbound_webhooks.iter_configured_targets(
_cfg(
{
"url": "https://example.com",
"events": ["on_session_end"],
"timeout": "soon",
}
)
)
assert targets[0].timeout == outbound_webhooks.DEFAULT_TIMEOUT_SECONDS
def test_matcher_dropped_for_non_tool_events(self):
targets = outbound_webhooks.iter_configured_targets(
_cfg(
{
"url": "https://example.com",
"events": ["on_session_end"],
"matcher": "terminal",
}
)
)
assert targets[0].matcher is None
def test_matcher_kept_for_tool_events(self):
targets = outbound_webhooks.iter_configured_targets(
_cfg(
{
"url": "https://example.com",
"events": ["post_tool_call"],
"matcher": "terminal|delegate_task",
}
)
)
assert targets[0].matcher == "terminal|delegate_task"
def test_secret_env_wins_over_literal(self, monkeypatch):
monkeypatch.setenv("MY_HOOK_SECRET", "from-env")
targets = outbound_webhooks.iter_configured_targets(
_cfg(
{
"url": "https://example.com",
"events": ["on_session_end"],
"secret_env": "MY_HOOK_SECRET",
"secret": "literal",
}
)
)
assert targets[0].secret == "from-env"
def test_unset_secret_env_means_unsigned(self, monkeypatch):
monkeypatch.delenv("MISSING_SECRET_VAR", raising=False)
targets = outbound_webhooks.iter_configured_targets(
_cfg(
{
"url": "https://example.com",
"events": ["on_session_end"],
"secret_env": "MISSING_SECRET_VAR",
}
)
)
assert targets[0].secret is None
# ── matcher behaviour ─────────────────────────────────────────────────────
class TestMatcher:
def _target(self, matcher):
return outbound_webhooks.WebhookTarget(
url="https://example.com",
events=["post_tool_call"],
matcher=matcher,
)
def test_no_matcher_matches_everything(self):
assert self._target(None).matches_tool("terminal")
assert self._target(None).matches_tool(None)
def test_regex_fullmatch(self):
t = self._target("terminal|delegate_task")
assert t.matches_tool("terminal")
assert t.matches_tool("delegate_task")
assert not t.matches_tool("terminal_extra")
assert not t.matches_tool(None)
def test_invalid_regex_falls_back_to_equality(self):
t = self._target("terminal(")
assert t.matches_tool("terminal(")
assert not t.matches_tool("terminal")
# ── payload shape ─────────────────────────────────────────────────────────
class TestPayload:
def test_top_level_shape_matches_shell_hooks_wire(self):
body = outbound_webhooks._serialize_payload(
"post_tool_call",
{
"tool_name": "terminal",
"args": {"command": "ls"},
"session_id": "sess_1",
"status": "ok",
"duration_ms": 42,
},
)
payload = json.loads(body)
assert payload["hook_event_name"] == "post_tool_call"
assert payload["tool_name"] == "terminal"
assert payload["tool_input"] == {"command": "ls"}
assert payload["session_id"] == "sess_1"
assert payload["extra"]["status"] == "ok"
assert payload["extra"]["duration_ms"] == 42
assert payload["delivery_id"]
assert payload["timestamp"].endswith("Z")
def test_unserialisable_values_stringified(self):
body = outbound_webhooks._serialize_payload(
"on_session_end", {"weird": object()}
)
payload = json.loads(body)
assert isinstance(payload["extra"]["weird"], str)
# ── registration ──────────────────────────────────────────────────────────
class TestRegistration:
def test_registration_is_idempotent(self):
cfg = _cfg(
{"url": "https://example.com/hook", "events": ["on_session_end"]}
)
first = outbound_webhooks.register_from_config(cfg)
second = outbound_webhooks.register_from_config(cfg)
assert len(first) == 1
assert second == []
def test_safe_mode_skips_registration(self, monkeypatch):
monkeypatch.setenv("HERMES_SAFE_MODE", "1")
cfg = _cfg(
{"url": "https://example.com/hook", "events": ["on_session_end"]}
)
assert outbound_webhooks.register_from_config(cfg) == []
def test_callbacks_never_return_directives(self, http_server):
"""Outbound webhooks are notify-only — invoke_hook must see None."""
cfg = _cfg({"url": _url(http_server), "events": ["pre_tool_call"]})
outbound_webhooks.register_from_config(cfg)
from hermes_cli.plugins import get_plugin_manager
results = get_plugin_manager().invoke_hook(
"pre_tool_call", tool_name="terminal", args={"command": "ls"},
)
assert outbound_webhooks.flush()
# No block/context directives from the webhook callback.
assert results == []
assert len(http_server.captured) == 1
# ── E2E delivery against a real HTTP server ──────────────────────────────
class TestDelivery:
def test_delivery_with_hmac_signature(self, http_server):
secret = "s3cret"
cfg = _cfg(
{
"url": _url(http_server),
"events": ["on_session_end"],
"secret": secret,
"name": "e2e",
}
)
registered = outbound_webhooks.register_from_config(cfg)
assert len(registered) == 1
from hermes_cli.plugins import get_plugin_manager
get_plugin_manager().invoke_hook(
"on_session_end",
session_id="sess_e2e",
completed=True,
interrupted=False,
model="test-model",
platform="cli",
)
assert outbound_webhooks.flush()
assert len(http_server.captured) == 1
req = http_server.captured[0]
payload = json.loads(req["body"])
assert payload["hook_event_name"] == "on_session_end"
assert payload["session_id"] == "sess_e2e"
assert payload["extra"]["completed"] is True
assert payload["extra"]["model"] == "test-model"
assert req["headers"]["X-Hermes-Event"] == "on_session_end"
assert req["headers"]["X-Hermes-Delivery"]
expected = hmac.new(
secret.encode(), req["body"], hashlib.sha256
).hexdigest()
assert req["headers"]["X-Hermes-Signature-256"] == f"sha256={expected}"
def test_unsigned_delivery_has_no_signature_header(self, http_server):
cfg = _cfg({"url": _url(http_server), "events": ["on_session_end"]})
outbound_webhooks.register_from_config(cfg)
from hermes_cli.plugins import get_plugin_manager
get_plugin_manager().invoke_hook("on_session_end", session_id="s")
assert outbound_webhooks.flush()
assert len(http_server.captured) == 1
assert "X-Hermes-Signature-256" not in http_server.captured[0]["headers"]
def test_matcher_filters_tool_events(self, http_server):
cfg = _cfg(
{
"url": _url(http_server),
"events": ["post_tool_call"],
"matcher": "terminal",
}
)
outbound_webhooks.register_from_config(cfg)
from hermes_cli.plugins import get_plugin_manager
manager = get_plugin_manager()
manager.invoke_hook(
"post_tool_call", tool_name="web_search", args={}, status="ok",
)
manager.invoke_hook(
"post_tool_call", tool_name="terminal", args={}, status="ok",
)
assert outbound_webhooks.flush()
assert len(http_server.captured) == 1
payload = json.loads(http_server.captured[0]["body"])
assert payload["tool_name"] == "terminal"
def test_4xx_not_retried(self, http_server):
http_server.respond_status = 400
target = outbound_webhooks.WebhookTarget(
url=_url(http_server), events=["on_session_end"],
)
delivery = outbound_webhooks._build_delivery(
"on_session_end", target, b"{}",
)
outbound_webhooks._deliver(delivery)
assert len(http_server.captured) == 1
def test_connection_error_does_not_raise(self):
target = outbound_webhooks.WebhookTarget(
# Port 9 (discard) — nothing listening.
url="http://127.0.0.1:9/unreachable",
events=["on_session_end"],
timeout=1,
)
delivery = outbound_webhooks._build_delivery(
"on_session_end", target, b"{}",
)
# Must swallow the failure (logged), never raise into the agent loop.
outbound_webhooks._deliver(delivery)

View file

@ -46,7 +46,7 @@ class TestHooksList:
def test_empty_config(self, tmp_path):
with patch("hermes_cli.config.load_config", return_value={}):
out = _run(SimpleNamespace(hooks_action="list"))
assert "No shell hooks configured" in out
assert "No shell hooks or outbound webhooks configured" in out
def test_shows_configured_and_consent_status(self, tmp_path):
script = _hook_script(

View file

@ -6,15 +6,16 @@ description: "Run custom code at key lifecycle points — log activity, send ale
# Event Hooks
Hermes has three hook systems that run custom code at key lifecycle points:
Hermes has four hook systems that run custom code at key lifecycle points:
| System | Registered via | Runs in | Use case |
|--------|---------------|---------|----------|
| **[Gateway hooks](#gateway-event-hooks)** | `HOOK.yaml` + `handler.py` in `~/.hermes/hooks/` | Gateway only | Logging, alerts, webhooks |
| **[Plugin hooks](#plugin-hooks)** | `ctx.register_hook()` in a [plugin](/user-guide/features/plugins) | CLI + Gateway | Tool interception, metrics, guardrails |
| **[Shell hooks](#shell-hooks)** | `hooks:` block in `~/.hermes/config.yaml` pointing at shell scripts | CLI + Gateway | Drop-in scripts for blocking, auto-formatting, context injection |
| **[Outbound webhooks](#outbound-webhooks)** | `hooks.outbound:` list in `~/.hermes/config.yaml` | CLI + Gateway | Push signed lifecycle events to external HTTP endpoints — CI, dashboards, other agents |
All three systems are non-blocking — errors in any hook are caught and logged, never crashing the agent.
All four systems are non-blocking — errors in any hook are caught and logged, never crashing the agent.
## Gateway Event Hooks
@ -1495,3 +1496,83 @@ Shell hooks run with **your full user credentials** — same trust boundary as a
### Ordering and precedence
Both Python plugin hooks and shell hooks flow through the same `invoke_hook()` dispatcher. Python plugins are registered first (`discover_and_load()`), shell hooks second (`register_from_config()`), so Python `pre_tool_call` block decisions take precedence in tie cases. The first valid block wins — the aggregator returns as soon as any callback produces `{"action": "block", "message": str}` with a non-empty message.
## Outbound Webhooks
Outbound webhooks are the push-side mirror of the [inbound webhook platform](/user-guide/messaging/webhooks): inbound webhooks wake Hermes when the world changes; outbound webhooks tell the world when Hermes does something. Configure a list of HTTP endpoints and the lifecycle events they care about, and Hermes POSTs a signed JSON payload to each endpoint whenever a matching event fires — no polling on the receiving end.
Typical uses:
- Notify a CI system or dashboard when an agent turn finishes (`on_session_end`)
- Track subagent completions across a fleet (`subagent_stop`)
- Feed tool activity into external monitoring (`post_tool_call` with a `matcher`)
- Wake *another* Hermes instance: point the URL at that instance's inbound webhook
### Configuration
Add a `hooks.outbound:` list to `~/.hermes/config.yaml`:
```yaml
hooks:
outbound:
- name: ci-notify # optional label for logs
url: https://ci.example.com/hermes-events
events: [on_session_end, subagent_stop]
secret_env: HERMES_OUTBOUND_WEBHOOK_SECRET # env var holding the HMAC secret
timeout: 10 # per-attempt seconds (160)
- name: tool-monitor
url: https://metrics.example.com/hooks/hermes
events: [post_tool_call]
matcher: "terminal|delegate_task" # regex, tool-scoped events only
```
Any event from the plugin-hook set is valid (`pre_tool_call`, `post_tool_call`, `pre_llm_call`, `post_llm_call`, `on_session_start`, `on_session_end`, `subagent_start`, `subagent_stop`, ...). Malformed entries warn and are skipped — a broken webhook never crashes the agent. Changes take effect on the next CLI session / gateway restart.
Secrets: prefer `secret_env` (the name of an environment variable, typically set in `~/.hermes/.env`) over an inline `secret:` literal, so the config file stays free of credentials. Entries without a secret are delivered unsigned (flagged as `UNSIGNED` by `hermes hooks list`).
### Wire format
Each firing POSTs a JSON body with the same top-level shape as shell hooks' stdin, plus delivery metadata:
```json
{
"hook_event_name": "on_session_end",
"tool_name": null,
"tool_input": null,
"session_id": "sess_abc123",
"cwd": "/home/user/project",
"extra": {"completed": true, "interrupted": false, "model": "...", "platform": "cli"},
"delivery_id": "3f2c9a...",
"timestamp": "2026-07-22T14:00:00Z"
}
```
Headers:
| Header | Value |
|--------|-------|
| `Content-Type` | `application/json` |
| `X-Hermes-Event` | The hook event name |
| `X-Hermes-Delivery` | Unique id per delivery (for idempotency on the receiver) |
| `X-Hermes-Signature-256` | `sha256=<hex>` — HMAC-SHA256 of the raw body, GitHub-style; only present when a secret is configured |
Verify the signature exactly as you would a GitHub webhook:
```python
import hashlib, hmac
def verify(body: bytes, header: str, secret: str) -> bool:
expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, header)
```
### Delivery semantics
- **Fire-and-forget, off the hot path.** Events are serialized and queued instantly; a single background thread performs the HTTP POSTs. A slow or dead endpoint can never stall a tool call or an agent turn.
- **Notify-only.** Unlike shell hooks, outbound webhooks cannot block tool calls or inject context — the response body is ignored. They observe, never steer.
- **Bounded retries.** Connection errors and 5xx responses are retried once with backoff; 4xx responses are not retried (the receiver said the request itself is wrong). Failures are logged and dropped — delivery is best-effort, not guaranteed.
- **Bounded queue.** If the queue backs up (dead endpoint, event storm), new events are dropped with a warning rather than consuming unbounded memory.
- **No consent prompt.** Outbound targets execute no code on your machine — they receive data at a URL you configured. `HERMES_SAFE_MODE=1` still skips registration, same as plugins and shell hooks. Note that payloads include tool inputs and event metadata, so only point targets at endpoints you trust, and prefer `https://`.
`hermes hooks list` shows configured outbound targets alongside shell hooks, including whether each target is signed.