Webhook Subscriptions

Webhook Subscriptions

Create dynamic webhook subscriptions so external services (GitHub, GitLab, Stripe, CI/CD, IoT sensors, monitoring tools) can trigger Hermes agent runs by POSTing events to a URL.

Setup (Required First)

The webhook platform must be enabled before subscriptions can be created. Check with:

hermes webhook list

If it says "Webhook platform is not enabled", set it up:

Option 1: Setup wizard

hermes gateway setup

Option 2: Manual config

platforms:
  webhook:
    enabled: true
    extra:
      host: "0.0.0.0"
      port: 8644
      secret: "generate-a-strong-secret-here"

Option 3: Environment variables

WEBHOOK_ENABLED=true
WEBHOOK_PORT=8644
WEBHOOK_SECRET=generate-a-strong-secret-here

After configuration, start (or restart) the gateway:

hermes gateway run
Or if using systemd:

systemctl --user restart hermes-gateway

Verify it's running:

curl http://localhost:8644/health

Commands

All management is via the hermes webhook CLI command:

Create a subscription

hermes webhook subscribe  \
  --prompt "Prompt template with {payload.fields}" \
  --events "event1,event2" \
  --description "What this does" \
  --skills "skill1,skill2" \
  --deliver telegram \
  --deliver-chat-id "12345" \
  --secret "optional-custom-secret"

Returns the webhook URL and HMAC secret. The user configures their service to POST to that URL.

List subscriptions

hermes webhook list

Remove a subscription

hermes webhook remove 

Test a subscription

hermes webhook test 
hermes webhook test  --payload '{"key": "value"}'

Prompt Templates

Prompts support {dot.notation} for accessing nested payload fields:

If no prompt is specified, the full JSON payload is dumped into the agent prompt.

Common Patterns

GitHub: new issues

hermes webhook subscribe github-issues \
  --events "issues" \
  --prompt "New GitHub issue #{issue.number}: {issue.title}\n\nAction: {action}\nAuthor: {issue.user.login}\nBody:\n{issue.body}\n\nPlease triage this issue." \
  --deliver telegram \
  --deliver-chat-id "-100123456789"

Then in GitHub repo Settings → Webhooks → Add webhook:

GitHub: PR reviews

hermes webhook subscribe github-prs \
  --events "pull_request" \
  --prompt "PR #{pull_request.number} {action}: {pull_request.title}\nBy: {pull_request.user.login}\nBranch: {pull_request.head.ref}\n\n{pull_request.body}" \
  --skills "github-code-review" \
  --deliver github_comment

Stripe: payment events

hermes webhook subscribe stripe-payments \
  --events "payment_intent.succeeded,payment_intent.payment_failed" \
  --prompt "Payment {data.object.status}: {data.object.amount} cents from {data.object.receipt_email}" \
  --deliver telegram \
  --deliver-chat-id "-100123456789"

CI/CD: build notifications

hermes webhook subscribe ci-builds \
  --events "pipeline" \
  --prompt "Build {object_attributes.status} on {project.name} branch {object_attributes.ref}\nCommit: {commit.message}" \
  --deliver discord \
  --deliver-chat-id "1234567890"

Generic monitoring alert

hermes webhook subscribe alerts \
  --prompt "Alert: {alert.name}\nSeverity: {alert.severity}\nMessage: {alert.message}\n\nPlease investigate and suggest remediation." \
  --deliver origin

Direct delivery (no agent, zero LLM cost)

For use cases where you just want to push a notification through to a user's chat — no reasoning, no agent loop — add --deliver-only. The rendered --prompt template becomes the literal message body and is dispatched directly to the target adapter.

Use this for:

hermes webhook subscribe antenna-matches \
  --deliver telegram \
  --deliver-chat-id "123456789" \
  --deliver-only \
  --prompt "🎉 New match: {match.user_name} matched with you!" \
  --description "Antenna match notifications"

The POST returns 200 OK on successful delivery, 502 on target failure — so upstream services can retry intelligently. HMAC auth, rate limits, and idempotency still apply.

Requires --deliver to be a real target (telegram, discord, slack, github_comment, etc.) — --deliver log is rejected because log-only direct delivery is pointless.

Security

How It Works

Troubleshooting

If webhooks aren't working:

Advanced Techniques (2025-2026)

Idempotency dedup store (Redis)

Webhook senders often retry delivery on timeout (e.g., GitHub retries up to 3 times over 24 hours). Without deduplication, the same event triggers multiple agent runs. Use Redis as a simple dedup store:

import redis, hashlib, json
r = redis.Redis(host="localhost", port=6379, db=0)
def is_duplicate(payload: dict, event_type: str) -> bool:
    """Check if we've already processed this exact event."""
    payload_hash = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
    key = f"webhook_dedup:{event_type}:{payload_hash}"
    if r.exists(key):
        return True
    # Mark as seen, TTL = 24h (covers GitHub's full retry window)
    r.setex(key, 86400, "1")
    return False

This requires a running Redis instance (apt-get install redis-server). The HMAC secret from the webhook subscription ensures authenticity; the dedup key ensures exactly-once processing.

HMAC timing-safe comparison (hmac.compare_digest)

Never compare webhook signatures using == — it's vulnerable to timing attacks that can leak the secret byte-by-byte. Always use hmac.compare_digest:

import hmac, hashlib
def verify_signature(payload_body: bytes, signature_header: str, secret: str) -> bool:
    """Verify HMAC-SHA256 signature from GitHub/Stripe-style webhooks."""
    expected = "sha256=" + hmac.new(
        secret.encode(), payload_body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature_header)

Python's == short-circuits on the first differing byte, leaking timing information. hmac.compare_digest runs in constant time regardless of how many bytes match.

Health monitoring cron

Set up a cron job to verify the webhook gateway is alive and can process events:

# /etc/cron.d/webhook-healthcheck
/5    * root curl -sf http://localhost:8644/health || (echo "Webhook gateway down at $(date)" | /usr/local/bin/notify-alerts.sh)

For deeper checks, create a dedicated "ping" subscription and POST to it periodically:

# Create a health-probe webhook (once)
hermes webhook subscribe health-probe \
  --events "ping" \
  --prompt "Health probe received" \
  --deliver log
Cron: send a test event every 5 minutes
/5    * root curl -sf -X POST http://localhost:8644/webhook/health-probe -H "Content-Type: application/json" -d '{"event":"ping","ts":"$(date -Iseconds)"}' || echo "Webhook gateway unreachable at $(date)" >> /var/log/webhook-health.log

Cloudflare/ngrok tunnels for local testing

To test webhooks from external services against a local development instance:

ngrok (quick start):

ngrok http 8644
Returns a public URL like https://abc123.ngrok.io

Use that URL as the webhook endpoint in GitHub/Stripe/etc.

Cloudflare Tunnel (persistent, no signup needed if you have cloudflared):

cloudflared tunnel --url http://localhost:8644
Returns a public URL like https://random-words.trycloudflare.com

Both forward external HTTPS traffic to your local webhook port. Remember to update the webhook URL in the external service when the tunnel URL changes.

jq for template testing

Test prompt templates with sample payloads using jq to verify field extraction before creating the subscription:

# Sample GitHub issue payload
curl -s https://api.github.com/repos/org/repo/issues/1 | jq '{
  title: .title,
  author: .user.login,
  body: .body,
  number: .number,
  action: "opened"
}'
Verify your template resolves correctly
echo '{"issue":{"number":42,"title":"Bug found","user":{"login":"alice"},"body":"Details here"},"action":"opened"}' | \
  jq 'New GitHub issue #' + (.issue.number|tostring) + ': ' + .issue.title

This catches template errors (wrong field names, missing nesting) before you wire up the full webhook pipeline.

Production Webhook Design (2025-2026)

Seven design rules for building production-grade webhook systems, with Python examples for each.

#### 1. Sign every payload with HMAC-SHA256

Every outbound webhook payload must be cryptographically signed. The sender computes an HMAC-SHA256 over the raw body bytes using a shared secret; the receiver verifies before processing. Never use plain tokens or skip verification.

import hmac, hashlib
def sign_payload(payload_bytes: bytes, secret: str) -> str:
    """Sign raw payload bytes with HMAC-SHA256. Returns 'sha256='."""
    mac = hmac.new(secret.encode(), payload_bytes, hashlib.sha256)
    return f"sha256={mac.hexdigest()}"
def verify_signature(payload_bytes: bytes, signature: str, secret: str) -> bool:
    """Constant-time comparison — never use == for signature checks."""
    expected = sign_payload(payload_bytes, secret)
    return hmac.compare_digest(expected, signature)

Why: Prevents tampering and replay from untrusted sources. Always verify the raw body bytes — never re-serialize from parsed JSON, as key ordering and whitespace differ.

#### 2. Design for idempotency with event IDs

Every event must carry a unique, immutable ID (e.g., evt_1a2b3c). Receivers store processed IDs in a dedup store (Redis, DB) and skip duplicates on sight. This makes retry delivery safe — the same event processed twice yields the same result.

import redis, json
r = redis.Redis(host="localhost", port=6379, db=0)
def process_webhook(payload: dict) -> None:
    event_id = payload.get("id")  # e.g., "evt_1a2b3c"
    if not event_id:
        raise ValueError("Missing event ID")
dedup_key = f"webhook:processed:{event_id}"
    if r.exists(dedup_key):
        return  # Already processed — skip
# ... business logic here ...
    result = handle_event(payload)
# Mark as processed, TTL covers retry window (e.g., 72h)
    r.setex(dedup_key, 259200, json.dumps({"status": "ok", "result": result}))

Why: Distributed systems retry on timeout. Without idempotency, a retried payment event creates double charges. With it, retries are harmless.

#### 3. Retry with exponential backoff + jitter

When delivery fails, retry with increasing delays to避免 overwhelming the receiver. Add random jitter to prevent thundering-herd synchronization across all senders retrying at the same instant.

import random, time
def retry_webhook(url: str, payload: bytes, headers: dict,
                  max_retries: int = 5, base_delay: float = 1.0) -> bool:
    """Send webhook with exponential backoff + full jitter."""
    for attempt in range(max_retries):
        try:
            resp = httpx.post(url, content=payload, headers=headers, timeout=10)
            if resp.status_code in (200, 201, 202):
                return True
            if resp.status_code in (400, 401, 403, 404, 410):
                return False  # Client error — don't retry
        except (httpx.TimeoutException, httpx.ConnectError):
            pass
# Exponential backoff with full jitter
        max_backoff = min(base_delay  (2 * attempt), 3600)  # cap at 1 hour
        delay = random.uniform(0, max_backoff)
        time.sleep(delay)
return False  # Exhausted retries

Why: Fixed-interval retries amplify traffic spikes. Exponential backoff with jitter is the industry standard (AWS, Stripe, GitHub all use it). Cap the max delay to avoid multi-day retry loops.

#### 4. Version your schemas

API payloads evolve. Add an explicit schema_version field (e.g., "v1", "v2") and a X-Schema-Version HTTP header so receivers can handle multiple versions concurrently. Never break old versions — deprecate on a published timeline.

SCHEMA_VERSIONS = {
    "v1": {
        "transform": lambda payload: {
            "id": payload["id"],
            "status": payload["status"],
            "amount": payload["amount"] / 100,  # v1: cents → dollars
        },
    },
    "v2": {
        "transform": lambda payload: {
            "id": payload["id"],
            "status": payload["status"],
            "amount_cents": payload["amount"],  # v2: raw cents
            "currency": payload.get("currency", "USD"),
        },
    },
}
def send_webhook_v2(url: str, payload: dict, version: str = "v2") -> None:
    transformer = SCHEMA_VERSIONS[version]["transform"]
    body = json.dumps(transformer(payload)).encode()
    headers = {
        "Content-Type": "application/json",
        "X-Schema-Version": version,
        "X-Hub-Signature-256": sign_payload(body, WEBHOOK_SECRET),
    }
    retry_webhook(url, body, headers)

Why: Schema changes without versioning break every consumer simultaneously. Versioned schemas let you ship changes incrementally with zero-downtime migration.

#### 5. Sequence IDs for ordering

When events must be processed in order (e.g., state machine transitions), include a monotonically increasing sequence ID. Receivers detect gaps and out-of-order delivery, then request retransmission.

class SequenceTracker:
    """Track last-processed sequence ID per source."""
def __init__(self, redis_client):
        self.r = redis_client
def check_and_advance(self, source: str, seq_id: int, payload: dict) -> bool:
        key = f"webhook:seq:{source}"
        last = int(self.r.get(key) or 0)
if seq_id <= last:
            # Duplicate or reordered — skip or reprocess
            return False
if seq_id > last + 1:
            # Gap detected — fetch missing events
            missing = range(last + 1, seq_id)
            for s in missing:
                fetch_and_reprocess(source, s)
# Advance watermark
        self.r.set(key, seq_id)
        return True

Why: Network reordering is real. Without sequence IDs, a "state=B" arriving before "state=A" leaves the system in an inconsistent state. Sequence IDs make ordering violations detectable and recoverable.

#### 6. Rate limiting for flood protection

Protect downstream receivers from event floods (e.g., a batch import triggers 10,000 events in 60 seconds). Implement per-event-type and per-destination rate limits. Return 429 Too Many Requests with a Retry-After header when exceeded.

from datetime import datetime, timedelta
RATE_LIMITS = {
    "default": (100, 60),      # 100 events per minute
    "payment.*": (30, 60),     # 30 payment events per minute
    "alert.critical": (5, 60),  # 5 critical alerts per minute
}
def check_rate_limit(event_type: str, destination: str) -> tuple[bool, int]:
    """Returns (allowed, retry_after_seconds)."""
    key = f"webhook:ratelimit:{destination}:{event_type}"
    limit, window = RATE_LIMITS.get(event_type, RATE_LIMITS["default"])
    current = int(r.get(key) or 0)
if current >= limit:
        return False, window  # Retry-After header value
pipe = r.pipeline()
    pipe.incr(key)
    pipe.expire(key, window)
    pipe.execute()
    return True, 0

Why: Without rate limiting, one misconfigured sender can DOS your receivers. Rate limits also protect your own sender from runaway loops (e.g., a webhook that triggers another webhook recursively).

#### 7. Full observability with logging, monitoring, and alerting

Every webhook delivery must be logged with: event ID, schema version, sequence ID, destination, HTTP status, response time, retry count. Surface this in metrics dashboards and set alerts on sustained failure rates.

import structlog, prometheus_client as prom
logger = structlog.get_logger()
Prometheus metrics
webhooks_sent = prom.Counter(
    "webhooks_sent_total", "Webhooks sent", ["destination", "event_type", "status"]
)
webhooks_latency = prom.Histogram(
    "webhooks_delivery_duration_seconds", "Delivery latency",
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)
def deliver_with_observability(url, payload, headers, event_type, event_id):
    start = time.monotonic()
    try:
        resp = httpx.post(url, content=payload, headers=headers, timeout=10)
        latency = time.monotonic() - start
        webhooks_sent.labels(url, event_type, resp.status_code).inc()
        webhooks_latency.observe(latency)
        logger.info("webhook_delivered",
                     event_id=event_id, destination=url,
                     status=resp.status_code, latency_ms=latency * 1000)
        return resp
    except Exception as e:
        latency = time.monotonic() - start
        webhooks_sent.labels(url, event_type, "error").inc()
        logger.error("webhook_failed",
                      event_id=event_id, destination=url,
                      error=str(e), latency_ms=latency * 1000)
        raise
Alert rule (for Prometheus Alertmanager):
- alert: WebhookDeliveryFailures
  expr: sum(rate(webhooks_sent_total{status=~"5..|error"}[5m])) > 0.1
  for: 5m
  labels: { severity: critical }
  annotations: { summary: "Sustained webhook delivery failures" }

Why: Silent failures are the norm in webhook systems — the sender has no user-facing UI. Without observability, you discover delivery failures weeks later when a customer reports missing data. Log everything, metric it, and alert on failure rate spikes.

Edge Cases / Pitfalls