Webhooks

Webhooks push events to a URL you own instead of making you poll. Register an HTTPS endpoint and Pictograph POSTs a signed JSON body to it whenever a subscribed event fires (for example, when a workflow run finishes). Deliveries are durable and retried with backoff, and every attempt is recorded in a delivery log you can inspect and replay.

Every example below shows the Python SDK call and the equivalent raw REST request. The REST examples authenticate with an X-API-Key header; set PICTOGRAPH_API_KEY in your shell to copy-and-run them.

from pictograph import Client
client = Client()  # reads PICTOGRAPH_API_KEY

Why webhooks

One outbound HTTP sink replaces a pile of polling loops. You wire the last mile (your queue, your data warehouse, Slack, a serverless function) and Pictograph handles signing, retries, and the delivery log. The endpoint must be public HTTPS: private, loopback, and cloud-metadata addresses are rejected at registration time to prevent SSRF.

Event types

Event	Fires when
workflow_run.completed	A workflow run finishes successfully. The payload carries the run id and signed artifact URLs.
workflow_run.failed	A workflow run ends in error.

An endpoint subscribes to one or more event types (default ["workflow_run.completed"]). Management calls (create, delete, test, replay) require an admin/owner API key.

create

Register an endpoint. The response carries the signing secret exactly once. Store it securely: it is never retrievable again (rotate to mint a new one).

created = client.webhooks.create(
    "https://example.com/hooks/pictograph",
    description="Prod ingestion",
    event_types=["workflow_run.completed", "workflow_run.failed"],
)
print(created.endpoint.id)
print(created.secret)   # whsec_… shown once

curl -s -X POST "https://api.pictograph.io/api/v1/developer/webhooks/endpoints" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com/hooks/pictograph", "description": "Prod ingestion", "event_types": ["workflow_run.completed", "workflow_run.failed"]}'

Returns CreatedWebhookEndpoint (an endpoint plus the one-time secret).

list

List every webhook endpoint in your organization. No secret material is ever returned, only a display secret_prefix.

for ep in client.webhooks.list():
    print(ep.id, ep.url, ep.enabled, ep.event_types)

curl -s "https://api.pictograph.io/api/v1/developer/webhooks/endpoints" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns WebhookEndpoint objects.

Field	Meaning
id	Endpoint UUID.
url	The HTTPS destination.
event_types	Subscribed event types.
enabled	Whether deliveries are sent. Auto-disabled after repeated failures.
secret_version	Current signing-secret version (bumped on rotate).
secret_prefix	First characters of the signing secret, for display only.
consecutive_failures	Failed deliveries in a row. Resets to 0 on any success.
disabled_reason	Set when the endpoint was auto-disabled.
last_delivery_at	Timestamp of the most recent delivery attempt.

get

Fetch a single endpoint by UUID.

ep = client.webhooks.get("wh-uuid")
print(ep.url, ep.enabled)

curl -s "https://api.pictograph.io/api/v1/developer/webhooks/endpoints/wh-uuid" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns a WebhookEndpoint.

test

Send a synthetic, signed webhook.test event and get the immediate result. The attempt is recorded in the delivery log.

result = client.webhooks.test("wh-uuid")
print(result["delivered"], result["status_code"])

curl -s -X POST "https://api.pictograph.io/api/v1/developer/webhooks/endpoints/wh-uuid/test" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

deliveries

Inspect the delivery log, optionally filtered by endpoint and status (pending, delivered, failed, dead_letter).

for d in client.webhooks.deliveries(status="failed", limit=20):
    print(d.id, d.event_type, d.attempts, d.last_status_code, d.last_error)

curl -s "https://api.pictograph.io/api/v1/developer/webhooks/deliveries?status=failed&limit=20" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns WebhookDelivery objects.

Field	Meaning
id	Delivery row UUID (use with replay).
delivery_id	Idempotency token sent to your endpoint as X-Pictograph-Delivery-Id.
event_type	The event delivered.
status	pending, delivered, failed, or dead_letter.
attempts	Attempts made so far.
last_status_code	HTTP status of the last attempt.
last_error	Error from the last failed attempt.
next_retry_at	When the next attempt is due (while pending).
delivered_at	When delivery first succeeded.

replay

Re-queue a failed or dead-letter delivery with a fresh retry budget.

client.webhooks.replay("delivery-uuid")

curl -s -X POST "https://api.pictograph.io/api/v1/developer/webhooks/deliveries/delivery-uuid/replay" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

delete

Delete an endpoint and its delivery history.

client.webhooks.delete("wh-uuid")

curl -s -X DELETE "https://api.pictograph.io/api/v1/developer/webhooks/endpoints/wh-uuid" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Verifying a delivery

Every delivery carries an X-Pictograph-Signature header shaped like t=<unix_ts>,v1=<hex> (the same scheme Stripe uses). Compute HMAC-SHA256 over "{timestamp}.{raw_body}" with your signing secret and compare in constant time. Reject deliveries whose timestamp is older than five minutes to close the replay window. During a secret rotation the header carries multiple v1= values (current and previous), so accept the delivery if any one matches.

import hashlib
import hmac
import time


def verify(secret: str, signature_header: str, raw_body: bytes, *, tolerance: int = 300) -> bool:
    timestamp: str | None = None
    signatures: list[str] = []
    for part in signature_header.split(","):
        key, _, value = part.partition("=")
        if key == "t":
            timestamp = value
        elif key == "v1":
            signatures.append(value)
    if timestamp is None or not signatures:
        return False
    if abs(time.time() - int(timestamp)) > tolerance:
        return False  # stale: possible replay
    expected = hmac.new(
        secret.encode(), f"{timestamp}.".encode() + raw_body, hashlib.sha256
    ).hexdigest()
    return any(hmac.compare_digest(sig, expected) for sig in signatures)

Other headers on each delivery:

Header	Meaning
X-Pictograph-Signature	t=<ts>,v1=<hex> (one or more v1 during rotation).
X-Pictograph-Delivery-Id	Stable id for the delivery. Use it to dedupe on your side (deliveries are at-least-once).
X-Pictograph-Event-Type	The event type, for routing.
X-Pictograph-Timestamp	The signed timestamp, mirrored for convenience.

Delivery, retries, and auto-disable

Delivery is at-least-once. A non-2xx response or a connection failure is retried on a backoff schedule (roughly 30s, 2m, 10m, 1h, 6h). After the budget is exhausted the delivery is marked dead_letter rather than dropped, and you can replay it once the endpoint is healthy. An endpoint that fails many deliveries in a row is auto-disabled; re-enable it from Settings to resume.

Always return a 2xx quickly and do heavy processing asynchronously: a slow handler looks like a failure and triggers retries.