> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kallima.bio/llms.txt
> Use this file to discover all available pages before exploring further.

# Webhooks

> Receive HTTP notifications when jobs complete instead of polling.

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

```python theme={null}
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:

```python theme={null}
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

| Event           | When it fires                               |
| --------------- | ------------------------------------------- |
| `job.completed` | Any pipeline job reaches `status=completed` |
| `job.failed`    | Any pipeline job reaches `status=failed`    |
| `webhook.test`  | Fired by `client.webhooks.test(webhook_id)` |

## Test an endpoint

Fire a `webhook.test` event immediately to verify your URL is reachable:

```python theme={null}
delivery = client.webhooks.test(webhook_id)
print(delivery["response_status"])   # should be 200
```

## Inspect delivery history

```python theme={null}
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:

```python theme={null}
client.webhooks.retry_delivery(delivery_id)
```

## Rotate the signing secret

```python theme={null}
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

```python theme={null}
# Pause without deleting
client.webhooks.update(webhook_id, enabled=False)

# Permanent removal
client.webhooks.delete(webhook_id)
```