Skip to main content
Kallima sends a signed POST request to your HTTPS endpoint whenever a subscribed event fires. This lets your pipeline react to job completion without polling.

​
Register an endpoint

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_...
The secret field is only returned on creation. Store it in your secrets manager.

​
Verify incoming requests

Use WebhookVerifier to validate the X-Kallima-Signature header before processing events: 
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")
Signatures are HMAC-SHA256 over timestamp.body. Requests older than 5 minutes are rejected automatically to prevent replay attacks.

​
Event types

EventWhen it fires
job.completedAny pipeline job reaches status=completed
job.failedAny pipeline job reaches status=failed
webhook.testFired by client.webhooks.test(webhook_id)

​
Test an endpoint

Fire a webhook.test event immediately to verify your URL is reachable: 
delivery = client.webhooks.test(webhook_id)
print(delivery["response_status"])   # should be 200

​
Inspect delivery history

deliveries = client.webhooks.list_deliveries(webhook_id, limit=20)
for d in deliveries:
    print(d["event_type"], d["response_status"], d["attempts"])
Retry a failed delivery: 
client.webhooks.retry_delivery(delivery_id)

​
Rotate the signing secret

updated = client.webhooks.update(webhook_id, rotate_secret=True)
new_secret = updated["secret"]
The old secret is invalidated immediately. Update your server before rotating.

​
Disable or delete

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

# Permanent removal
client.webhooks.delete(webhook_id)