Webhooks

Event-driven integrations for async operations

1 min read

KLOAKD pushes events to your server as async operations complete.

Register a webhook

webhook = client.webhooks.create(
    url="https://your-server.com/webhook",
    events=["webgrph.crawl.complete", "kolektr.extract.complete"],
    secret="your-signing-secret",
)
print(webhook.webhook_id)

Event types

| Event | Trigger | |-------|---------| | webgrph.crawl.complete | Site crawl finished | | webgrph.crawl.failed | Site crawl failed | | kolektr.extract.complete | Extraction job complete | | nexus.strategy.complete | Strategy execution complete | | fetchyr.workflow.complete | RPA workflow finished |

Verify signatures

import hmac, hashlib

def verify_webhook(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)