Webhooks

from kallima import KallimaClient

client = KallimaClient(api_key)

webhook = client.webhooks.create(
    url="https://your-server.example.com/kallima-events",
    events=["job.completed", "job.failed"],
    description="CI pipeline listener",
)

# Save the secret — it's only shown once
print(webhook["secret"])   # whsec_...

from kallima import WebhookVerifier, WebhookSignatureError

verifier = WebhookVerifier(secret="whsec_...")

# In your request handler (framework-agnostic):
def handle_webhook(headers: dict, raw_body: bytes) -> None:
    try:
        event = verifier.verify(
            payload=raw_body,
            signature=headers["X-Kallima-Signature"],
            timestamp=headers["X-Kallima-Timestamp"],
        )
    except WebhookSignatureError:
        return  # reject — invalid or replayed request

    if event["type"] == "job.completed":
        job_id = event["data"]["id"]
        print(f"Job {job_id} finished")

delivery = client.webhooks.test(webhook_id)
print(delivery["response_status"])   # should be 200

deliveries = client.webhooks.list_deliveries(webhook_id, limit=20)
for d in deliveries:
    print(d["event_type"], d["response_status"], d["attempts"])

client.webhooks.retry_delivery(delivery_id)

updated = client.webhooks.update(webhook_id, rotate_secret=True)
new_secret = updated["secret"]

# Pause without deleting
client.webhooks.update(webhook_id, enabled=False)

# Permanent removal
client.webhooks.delete(webhook_id)