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. Deliveries are durable, retried with backoff, and every attempt is recorded in a log you can inspect and replay.

The endpoint must be public HTTPS: private, loopback, and metadata addresses are rejected at registration time. Every management call (create, update, delete, test, replay, rotate) requires an admin or owner API key.

Python and the CLI address an endpoint by its registered URL, matched exactly (an id also works). REST addresses it by the endpoint’s org-unique name, which defaults to the URL’s host, or by its id.

event_types

The event types you can subscribe to. An endpoint with an empty subscription receives all of them, including ones added later.

Source: Webhooks.event_types

print(client.webhooks.event_types())
# ['workflow_run.completed', 'workflow_run.failed']

pictograph webhooks event-types

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

Returns list[str]

create

Register an endpoint. Returns CreatedWebhookEndpoint - an endpoint plus the signing secret. It is DERIVED, not stored, so it can always be shown again from Settings (use rotate_secret to mint a new one).

Source: Webhooks.create

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

pictograph webhooks create https://example.com/hooks/pictograph \
  --description "Prod ingestion" \
  --event workflow_run.completed --event workflow_run.failed

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

class CreatedWebhookEndpoint(BaseModel):
    """Create / rotate response - carries the one-time signing secret."""
    endpoint: WebhookEndpoint
    secret: str

Omit event_types to receive every event.

list

Every endpoint in your organization. No secret material is returned, only a display secret_prefix.

Source: Webhooks.list

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

pictograph webhooks list

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

Returns Sequence[WebhookEndpoint]

class WebhookEndpoint(BaseModel):
    """A registered outbound webhook destination."""
    id: str
    organization_id: str
    url: str
    description: str | None = None
    event_types: list[str] = []
    enabled: bool = True
    secret_version: int = 1
    secret_prefix: str | None = None
    consecutive_failures: int = 0
    disabled_reason: str | None = None
    auth_header_names: list[str] | None = None
    last_delivery_at: datetime | None = None
    created_at: datetime | None = None

get

Fetch one endpoint.

Source: Webhooks.get

ep = client.webhooks.get(
    endpoint="https://example.com/hooks/pictograph",
)
print(ep.url, ep.enabled)

pictograph webhooks get https://example.com/hooks/pictograph

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

Returns WebhookEndpoint

class WebhookEndpoint(BaseModel):
    """A registered outbound webhook destination."""
    id: str
    organization_id: str
    url: str
    description: str | None = None
    event_types: list[str] = []
    enabled: bool = True
    secret_version: int = 1
    secret_prefix: str | None = None
    consecutive_failures: int = 0
    disabled_reason: str | None = None
    auth_header_names: list[str] | None = None
    last_delivery_at: datetime | None = None
    created_at: datetime | None = None

update

Change the url, description, subscribed events, or enabled flag. Only the fields you pass change; re-enabling clears the failure health state.

Source: Webhooks.update

ep = client.webhooks.update(
    endpoint="https://example.com/hooks/pictograph",
    url="https://example.com/hooks/v2",
    event_types=["workflow_run.completed"],
    enabled=True,
)

pictograph webhooks update https://example.com/hooks/pictograph \
  --url https://example.com/hooks/v2 \
  --event workflow_run.completed \
  --enabled

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

Returns WebhookEndpoint

class WebhookEndpoint(BaseModel):
    """A registered outbound webhook destination."""
    id: str
    organization_id: str
    url: str
    description: str | None = None
    event_types: list[str] = []
    enabled: bool = True
    secret_version: int = 1
    secret_prefix: str | None = None
    consecutive_failures: int = 0
    disabled_reason: str | None = None
    auth_header_names: list[str] | None = None
    last_delivery_at: datetime | None = None
    created_at: datetime | None = None

rotate_secret

Mint a new signing secret. The plaintext is returned once; the previous secret stays valid through a short grace window, and deliveries carry both signatures, so you can roll receivers without dropping anything.

Source: Webhooks.rotate_secret

rotated = client.webhooks.rotate_secret(
    endpoint="https://example.com/hooks/pictograph",
)
print(rotated.secret)   # also revealable from Settings

pictograph webhooks rotate-secret https://example.com/hooks/pictograph

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

Returns CreatedWebhookEndpoint

class CreatedWebhookEndpoint(BaseModel):
    """Create / rotate response - carries the one-time signing secret."""
    endpoint: WebhookEndpoint
    secret: str

test

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

Source: Webhooks.test

result = client.webhooks.test(
    endpoint="https://example.com/hooks/pictograph",
)
print(result["delivered"], result["status_code"])

pictograph webhooks test https://example.com/hooks/pictograph

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

Returns dict[str, Any]

deliveries

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

Source: Webhooks.deliveries

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)

pictograph webhooks deliveries --status failed --limit 20

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

Returns Sequence[WebhookDelivery]

class WebhookDelivery(BaseModel):
    """One delivery attempt-set for an event to an endpoint (durable, retried)."""
    id: str
    endpoint_id: str
    organization_id: str
    event_type: str
    delivery_id: str
    status: Literal['pending', 'delivered', 'failed', 'dead_letter']
    attempts: int = 0
    last_status_code: int | None = None
    last_error: str | None = None
    next_retry_at: datetime | None = None
    created_at: datetime | None = None
    delivered_at: datetime | None = None

replay

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

Source: Webhooks.replay

client.webhooks.replay(
    delivery_id="f3a4b5c6-7d8e-4f92-a013-b4c5d6e7f809",
)

pictograph webhooks replay f3a4b5c6-7d8e-4f92-a013-b4c5d6e7f809

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

Returns None

delete

Delete an endpoint and its delivery history.

Source: Webhooks.delete

client.webhooks.delete(
    endpoint="https://example.com/hooks/pictograph",
)

pictograph webhooks delete https://example.com/hooks/pictograph --yes

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

Returns None

Verifying a delivery

Every delivery carries X-Pictograph-Signature shaped t=<unix_ts>,v1=<hex>. Compute HMAC-SHA256 over "{timestamp}.{raw_body}" with your signing secret and compare in constant time. Reject deliveries older than five minutes to close the replay window. During a rotation the header carries several v1= values, so accept 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)

Retries and auto-disable

A non-2xx response or a connection failure is retried on a backoff schedule (roughly 30s, 2m, 10m, 1h, 6h). Once the budget is exhausted the delivery is marked dead_letter rather than dropped, so you can replay it after the endpoint recovers. An endpoint that fails many deliveries in a row is auto-disabled; update(..., enabled=True) resumes it. Return a 2xx quickly and do heavy processing asynchronously: a slow handler looks like a failure and triggers retries.