HMAC-signed run lifecycle callbacks
Webhooks
Webhooks, at a glance
HMAC-signed run lifecycle callbacks
Coasty → your HTTPS webhook- 01Eventrun state committed
- 02Signtimestamp + HMAC
- 03Deliverbounded retry
- 04Dedupereceiver records event ID
Pass a webhook_url (https only) when you create a run and we POST a signed callback on lifecycle transitions. The response to your create call includes a webhook_secret only in the create response or an exact bounded Idempotency-Key replay: store it, because GET/list never return it and every callback is signed with it. Each request carries a Coasty-Signature header of the form t=<unix_ts>,v1=<hex>. Each terminal callback also carries Coasty-Delivery: <event_uuid>; the JSON body's id is that same UUID and its logical delivered_at timestamp remains stable across retries.
To verify, build the signed payload as "<t>." + raw_request_body, compute HMAC-SHA256 over it keyed by the webhook_secret, and compare against v1 with a constant-time check. Always hash the raw body bytes, before any JSON re-serialisation. Agent Run terminal delivery is durably queued and at least once, with at most three durably recorded delivery attempts. A worker crash after sending but before recording the outcome can cause additional duplicate HTTP sends after lease recovery. Recorded retryable failures become eligible again after about 30 seconds; redirects and 4xx responses are terminal. Atomically insert Coasty-Delivery into a uniquely constrained receiver table before applying side effects, and return success without repeating work when that UUID already exists. The non-terminal run.awaiting_human callback has up to three in-process attempts but no durable outbox or delivery UUID, so a worker failure can miss it; dedupe that event on (run.id, event, run.awaiting_human_since) and poll the run state.
import hashlib, hmac, os, requests
BASE = "https://coasty.ai/v1"
HEADERS = {"X-API-Key": os.environ["COASTY_API_KEY"]}
# 1. Create a run with a webhook_url. Create/exact-replay responses return webhook_secret; GET/list do not.
run = requests.post(
f"{BASE}/runs",
headers=HEADERS,
json={
"machine_id": "mch_test_0123456789abcdef",
"task": "Reconcile the invoice against the order",
"webhook_url": "https://example.com/hooks/coasty",
},
timeout=30,
).json()
webhook_secret = run["webhook_secret"] # persist this securely
# 2. In your webhook handler, verify the Coasty-Signature header.
def verify(raw_body: bytes, signature_header: str, secret: str) -> bool:
parts = dict(p.split("=", 1) for p in signature_header.split(","))
signed = f"{parts['t']}.".encode() + raw_body
expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, parts["v1"])
# Example (your framework supplies the raw body + header):
# ok = verify(request.body, request.headers["Coasty-Signature"], webhook_secret)