Webhooks
Event-driven integrations for async operations
2 min read
Webhooks are currently in beta. The webhook URL parameter is available on async fetch operations. A dedicated client.webhooks namespace for registration and management is planned for a future SDK release.
KLOAKD can push events to your server as async operations complete by passing a webhook_url when starting an async job.
Receive fetch completion via webhook
Pass a webhook_url to evadr.fetch_async() — KLOAKD will POST the result to your endpoint when the fetch completes:
job_id = client.evadr.fetch_async(
"https://example.com",
webhook_url="https://your-server.com/webhooks/kloakd",
)
const jobId = await client.evadr.fetchAsync('https://example.com', {
webhookUrl: 'https://your-server.com/webhooks/kloakd',
});
Webhook payload
When an async operation completes, KLOAKD sends a POST request to your webhook URL with a JSON body:
{
"job_id": "evadr-fetch-abc123",
"status": "completed",
"url": "https://example.com",
"success": true,
"artifact_id": "art_xyz789",
"tier_used": 2,
"timestamp": "2026-08-29T06:00:00Z"
}
Event types
| Event | Trigger | Method |
|-------|---------|--------|
| fetch.complete | Async fetch finished | evadr.fetch_async(webhook_url=...) |
| fetch.failed | Async fetch failed | evadr.fetch_async(webhook_url=...) |
For crawl operations, use client.crawl_stream() or client.webgrph.crawl_stream() for real-time SSE progress instead of webhooks. Streaming gives you per-page events as they happen.
Verify signatures
Webhook requests include an X-Kloakd-Signature header. Verify it with HMAC-SHA256:
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)
Responding to webhooks
Return HTTP 200 OK within 30 seconds. KLOAKD retries failed deliveries with exponential backoff (up to 3 attempts).
# FastAPI example
from fastapi import FastAPI, Request, Header
app = FastAPI()
@app.post("/webhooks/kloakd")
async def kloakd_webhook(request: Request, x_kloakd_signature: str = Header(None)):
body = await request.body()
# verify_webhook(body, x_kloakd_signature, "your-signing-secret")
event = await request.json()
if event["status"] == "completed":
print(f"Fetch done: {event['url']} (tier {event['tier_used']})")
return {"ok": True}