Docs · Webhooks

Asynchronous events you don't want to poll for.

HMAC-signed POSTs for batch completions, anomaly detections, and balance thresholds. At-least-once delivery with exponential retry.

Setup

Register an endpoint

Open dashboard → Settings → Webhooks.
Add your endpoint URL (must be HTTPS).
Pick which event types to subscribe to (or all).
Copy the generated webhook signing secret — stored once, shown once. Use it to verify signatures.
Anoman fires a test event immediately so you can confirm your endpoint is reachable.

You can register up to 3 endpoints per customer. Multiple endpoints receive the same event in parallel — useful for separating production from a staging mirror.

Event types

What we fire today

Event type	When
batch.completed	A queued batch job finished. Payload includes cost + savings.
batch.failed	A batch job exhausted retries and won’t recover.
batch.escalated	Job was promoted from batch to realtime to meet SLA. Higher cost.
anomaly.detected	Z-score or ML model flagged unusual behavior. Severity ∈ {low, medium, high}.
balance.threshold	Prepaid balance crossed a configured low-balance threshold.
key.budget_warning	Per-key budget crossed 80% / 90% / 100%.
guardrail.spike	Burst of blocks exceeded baseline — possible attack or misconfigured agent.

Payload shape

Example bodies

Every webhook body is a single JSON object with these top-level keys: event_id, event_type, occurred_at, data, meta.

{
  "event_id": "evt_5kJh4nQp",
  "event_type": "batch.completed",
  "occurred_at": "2026-05-28T14:23:17.412Z",
  "data": {
    "batch_job_id": "job_abc123",
    "api_key_id": "key_xyz789",
    "model": "deepseek-v3",
    "status": "complete",
    "prompt_tokens": 4892,
    "completion_tokens": 1203,
    "cost_usd": "0.0028",
    "savings_usd": "0.0084",
    "poll_url": "/anoman/v1/batch/job_abc123",
    "completed_at": "2026-05-28T14:23:15.001Z"
  },
  "meta": {
    "delivery_attempt": 1
  }
}

Signature verification

Always verify before processing

Every webhook request carries two headers:

x-anoman-timestamp — Unix timestamp the event was dispatched.
x-anoman-signature — hex-encoded HMAC-SHA256 of {timestamp}.{raw_body} using your signing secret.

Always: (1) check the timestamp is within 5 minutes (replay protection), (2) recompute the HMAC over the raw request body, (3) compare in constant time.

import hmac
import hashlib
import os
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
SECRET = os.environ["ANOMAN_WEBHOOK_SECRET"].encode()

@app.post("/webhooks/anoman")
async def receive_webhook(request: Request):
    raw_body = await request.body()
    sig_header = request.headers.get("x-anoman-signature", "")
    timestamp = request.headers.get("x-anoman-timestamp", "")

    # Reject events older than 5 minutes — limits replay window
    import time
    if abs(time.time() - int(timestamp)) > 300:
        raise HTTPException(400, "stale_timestamp")

    # Signature is HMAC-SHA256 over: timestamp + "." + raw_body
    signed_payload = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(SECRET, signed_payload, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(sig_header, expected):
        raise HTTPException(401, "invalid_signature")

    event = await request.json()
    # Process the event...
    return {"received": True}

Don’t parse before verifying. JSON parsing then re-serializing breaks the byte exactness the HMAC depends on. Verify against the raw body bytes; only parse after the signature checks out.

Delivery + retry

At-least-once

We treat any 2xx response as success. Anything else (including timeouts > 10s) triggers a retry on this schedule:

Attempt 1: immediate
Attempt 2: 30 seconds later
Attempt 3: 5 minutes later
Attempt 4: 1 hour later
Attempt 5: 6 hours later (final)

After attempt 5 the event is moved to the failed-deliveries queue (visible in dashboard) and a separate webhook.delivery_exhausted notification is emailed to your account admin.

Each delivery attempt increments meta.delivery_attempt in the payload. Use it to log + debug. Because we retry on any non-2xx, your endpoint must be idempotent:

import sqlite3
from datetime import datetime

# Persist event_ids you've processed. Any DB works — Redis SETNX,
# Postgres UNIQUE constraint, DynamoDB conditional put, etc.
db = sqlite3.connect("webhook_events.db")
db.execute("""
    CREATE TABLE IF NOT EXISTS processed_events (
        event_id TEXT PRIMARY KEY,
        received_at TEXT
    )
""")
db.commit()

def process_event(event):
    event_id = event["event_id"]
    # Atomic: insert-or-fail. If already processed, return 200 OK.
    try:
        db.execute(
            "INSERT INTO processed_events VALUES (?, ?)",
            (event_id, datetime.utcnow().isoformat()),
        )
        db.commit()
    except sqlite3.IntegrityError:
        return  # Already handled; idempotent no-op

    # Now do the real work — guaranteed once-per-event
    if event["event_type"] == "batch.completed":
        notify_user(event["data"]["batch_job_id"])
    elif event["event_type"] == "anomaly.detected":
        alert_oncall(event["data"]["anomaly_id"])

Testing

Replay + manual dispatch

The dashboard webhook page has a “Send test event” button that dispatches a synthetic payload of each event type.
The delivery log shows every attempt with status code + response body. Replay any individual attempt with one click.
For local development, use a tunnel like ngrok or cloudflared tunnel to expose localhost:3000 over HTTPS. The signing secret stays the same.

Set up your first webhook.

Takes 60 seconds. Free test events to verify wiring.

Open dashboard Back to docs