> ## Documentation Index
> Fetch the complete documentation index at: https://docs.limitguard.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Webhooks
> Real-time event notifications with HMAC signature verification
Webhooks let you subscribe to LimitGuard events and receive real-time HTTP POST notifications to your server the moment something changes — a trust score update, a new compliance alert, or a certificate issuance. Instead of polling the API, your system reacts instantly.
Webhooks require an API key with the `starter` tier or above. Sandbox webhooks fire with simulated payloads and do not count against your quota.
## How It Works
Every time an event occurs in the LimitGuard platform, we send an HTTP `POST` request to each registered endpoint with a JSON payload describing what happened. Your server responds with `HTTP 200` to acknowledge receipt. If delivery fails, we retry with exponential backoff.
Each request includes an `X-LimitGuard-Signature` header containing an HMAC-SHA256 signature computed from the raw request body using the webhook secret returned at registration. **Always verify this signature before processing the payload.**
## Setup
Send a `POST /v1/webhooks` request with your HTTPS URL and the event types you want to receive.
```bash curl theme={null}
curl -X POST https://api.limitguard.ai/v1/webhooks \
-H "X-API-Key: lg_live_xxxxxxxxxxxxxxxxxxxx" \
-H "Content-Type: application/json" \
-d '{
"url": "https://your-app.com/webhooks/limitguard",
"events": ["entity.checked", "score.updated", "compliance.alert"]
}'
```
```python Python theme={null}
import httpx
response = httpx.post(
"https://api.limitguard.ai/v1/webhooks",
headers={"X-API-Key": "lg_live_xxxxxxxxxxxxxxxxxxxx"},
json={
"url": "https://your-app.com/webhooks/limitguard",
"events": ["entity.checked", "score.updated", "compliance.alert"],
},
)
webhook = response.json()
print(webhook["webhook_id"]) # wh_abc123
print(webhook["secret"]) # whsec_xxxxxxxxxxxxxxxx — store this securely
```
```javascript JavaScript theme={null}
const response = await fetch("https://api.limitguard.ai/v1/webhooks", {
method: "POST",
headers: {
"X-API-Key": "lg_live_xxxxxxxxxxxxxxxxxxxx",
"Content-Type": "application/json",
},
body: JSON.stringify({
url: "https://your-app.com/webhooks/limitguard",
events: ["entity.checked", "score.updated", "compliance.alert"],
}),
});
const webhook = await response.json();
console.log(webhook.webhook_id); // wh_abc123
console.log(webhook.secret); // whsec_xxxxxxxxxxxxxxxx — store this securely
```
```json Response (201 Created) theme={null}
{
"webhook_id": "wh_abc123def456",
"url": "https://your-app.com/webhooks/limitguard",
"events": ["entity.checked", "score.updated", "compliance.alert"],
"secret": "whsec_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"status": "active",
"created_at": "2026-02-20T10:00:00Z"
}
```
The `secret` is returned **once only** at creation and is never shown again. Store it immediately in a secure secret manager (e.g., AWS Secrets Manager, HashiCorp Vault, or a `.env` file that is never committed to source control).
Before processing any webhook payload, verify that it genuinely came from LimitGuard by computing the expected HMAC-SHA256 signature and comparing it to the value in the `X-LimitGuard-Signature` header.
```python Python theme={null}
import hashlib
import hmac
def verify_signature(payload: bytes, signature_header: str, secret: str) -> bool:
"""
Returns True if the payload matches the LimitGuard signature.
Always use this before processing any webhook event.
"""
expected = hmac.new(
secret.encode("utf-8"),
payload,
hashlib.sha256,
).hexdigest()
# Use hmac.compare_digest to prevent timing attacks
return hmac.compare_digest(f"sha256={expected}", signature_header)
# --- FastAPI example ---
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
WEBHOOK_SECRET = "whsec_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # from env
@app.post("/webhooks/limitguard")
async def handle_webhook(request: Request):
payload = await request.body()
signature = request.headers.get("X-LimitGuard-Signature", "")
if not verify_signature(payload, signature, WEBHOOK_SECRET):
raise HTTPException(status_code=401, detail="Invalid signature")
event = await request.json()
# Hand off to async worker — respond 200 immediately
await process_event(event)
return {"received": True}
```
```javascript JavaScript theme={null}
import crypto from "crypto";
import express from "express";
const WEBHOOK_SECRET = process.env.LIMITGUARD_WEBHOOK_SECRET;
function verifySignature(payload, signatureHeader, secret) {
const expected = `sha256=${crypto
.createHmac("sha256", secret)
.update(payload)
.digest("hex")}`;
// Use timingSafeEqual to prevent timing attacks
return crypto.timingSafeEqual(
Buffer.from(expected),
Buffer.from(signatureHeader)
);
}
const app = express();
app.post(
"/webhooks/limitguard",
// Use raw body — JSON parsing breaks the HMAC
express.raw({ type: "application/json" }),
(req, res) => {
const signature = req.headers["x-limitguard-signature"] ?? "";
if (!verifySignature(req.body, signature, WEBHOOK_SECRET)) {
return res.status(401).json({ error: "Invalid signature" });
}
const event = JSON.parse(req.body);
// Acknowledge immediately, process asynchronously
res.json({ received: true });
processEvent(event).catch(console.error);
}
);
```
Always verify signatures using a **constant-time comparison** (`hmac.compare_digest` in Python, `crypto.timingSafeEqual` in Node.js). Standard string equality (`===`) is vulnerable to timing attacks.
Each webhook payload follows a consistent envelope structure. Dispatch on `event.type`:
```json Payload Envelope theme={null}
{
"id": "evt_01JMVQX3KZPB4Y8N9C2WD7FGH1",
"type": "score.updated",
"api_version": "2026-02-01",
"created_at": "2026-02-20T14:32:07.419Z",
"livemode": true,
"data": {
"entity_id": "ent_acme_corp_bv_nl",
"entity_name": "Acme Corp BV",
"country": "NL",
"previous_score": 74,
"new_score": 87,
"trust_level": "high",
"recommendation": "proceed",
"changed_sources": ["kvk", "sanctions"],
"processing_time_ms": 312
}
}
```
```python Python theme={null}
async def process_event(event: dict):
event_type = event["type"]
data = event["data"]
if event_type == "entity.checked":
await on_entity_checked(data)
elif event_type == "score.updated":
await on_score_updated(data)
elif event_type == "compliance.alert":
await on_compliance_alert(data)
elif event_type == "certificate.issued":
await on_certificate_issued(data)
elif event_type == "certificate.revoked":
await on_certificate_revoked(data)
else:
# Forward-compatible: ignore unknown event types
pass
```
```javascript JavaScript theme={null}
async function processEvent(event) {
const { type, data } = event;
switch (type) {
case "entity.checked":
await onEntityChecked(data);
break;
case "score.updated":
await onScoreUpdated(data);
break;
case "compliance.alert":
await onComplianceAlert(data);
break;
case "certificate.issued":
await onCertificateIssued(data);
break;
case "certificate.revoked":
await onCertificateRevoked(data);
break;
default:
// Forward-compatible: ignore unknown event types
break;
}
}
```
Use the event `id` field (`evt_...`) as an idempotency key. Store processed event IDs in your database to safely deduplicate retried deliveries.
## Event Types
| Event | Trigger | Key Payload Fields |
| --------------------- | ---------------------------------------------------------------- | ------------------------------------------------------------------------------ |
| `entity.checked` | Any `POST /v1/entity/check` call completes | `entity_id`, `trust_score`, `trust_level`, `recommendation`, `sources_checked` |
| `score.updated` | A monitored entity's trust score changes by ≥5 points | `entity_id`, `previous_score`, `new_score`, `changed_sources` |
| `compliance.alert` | New sanctions match, FATF listing, or adverse media flag | `entity_id`, `alert_type`, `severity`, `source`, `description` |
| `certificate.issued` | A new trust certificate is generated for an entity | `entity_id`, `certificate_id`, `valid_until`, `score_at_issuance` |
| `certificate.revoked` | A trust certificate is revoked (score dropped or entity flagged) | `entity_id`, `certificate_id`, `revocation_reason` |
### Subscribe to All Events
Pass `"events": ["*"]` to receive every event type, including future additions.
```json theme={null}
{
"url": "https://your-app.com/webhooks/limitguard",
"events": ["*"]
}
```
New event types may be added as LimitGuard adds data sources. Subscribing to `["*"]` is the easiest way to stay current, but ensure your handler ignores unknown types gracefully.
## Signature Verification
Every webhook request includes two headers:
| Header | Example Value | Description |
| ------------------------ | ------------------- | ------------------------------------ |
| `X-LimitGuard-Signature` | `sha256=a3f8c...` | HMAC-SHA256 of the raw request body |
| `X-LimitGuard-Event` | `score.updated` | The event type (convenience header) |
| `X-LimitGuard-Delivery` | `evt_01JMVQX3KZ...` | Unique delivery ID for deduplication |
| `X-LimitGuard-Timestamp` | `1740054727` | Unix timestamp of delivery attempt |
### Computing the Signature
```
HMAC-SHA256(key=webhook_secret, message=raw_request_body)
```
The `X-LimitGuard-Signature` header value is formatted as `sha256=`.
### Replay Attack Prevention
Compare the `X-LimitGuard-Timestamp` header against your server's current time. Reject deliveries older than 5 minutes to prevent replay attacks:
```python Python theme={null}
import time
def verify_webhook(payload: bytes, headers: dict, secret: str) -> bool:
timestamp = int(headers.get("X-LimitGuard-Timestamp", "0"))
signature = headers.get("X-LimitGuard-Signature", "")
# Reject stale deliveries (older than 5 minutes)
if abs(time.time() - timestamp) > 300:
return False
expected = f"sha256={hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()}"
return hmac.compare_digest(expected, signature)
```
## Retry Logic
If your endpoint returns anything other than a `2xx` status code, or if the connection times out, LimitGuard retries delivery with exponential backoff:
| Attempt | Delay | Cumulative Time |
| ----------- | ------ | --------------- |
| 1 (initial) | — | 0s |
| 2 | 30s | 30s |
| 3 | 5 min | \~5.5 min |
| 4 (final) | 30 min | \~36 min |
After 4 failed attempts, the event is marked as undelivered and no further retries occur. You can replay undelivered events from the [dashboard](https://limitguard.ai/dashboard/webhooks) or by sending a test event.
**Timeout:** LimitGuard waits up to **10 seconds** for your server to respond. Respond with `200` immediately and process the payload asynchronously.
If your endpoint consistently fails (>10 consecutive failures), the webhook is automatically deactivated to protect system stability. You'll receive an email notification. Re-enable it from the dashboard or by POSTing to `POST /v1/webhooks/{webhook_id}/enable`.
## Managing Webhooks
### List Active Webhooks
```bash curl theme={null}
curl -X GET https://api.limitguard.ai/v1/webhooks \
-H "X-API-Key: lg_live_xxxxxxxxxxxxxxxxxxxx"
```
```python Python theme={null}
response = httpx.get(
"https://api.limitguard.ai/v1/webhooks",
headers={"X-API-Key": "lg_live_xxxxxxxxxxxxxxxxxxxx"},
)
webhooks = response.json()["webhooks"]
for wh in webhooks:
print(f"{wh['webhook_id']} → {wh['url']} ({wh['status']})")
```
```javascript JavaScript theme={null}
const response = await fetch("https://api.limitguard.ai/v1/webhooks", {
headers: { "X-API-Key": "lg_live_xxxxxxxxxxxxxxxxxxxx" },
});
const { webhooks } = await response.json();
webhooks.forEach((wh) => {
console.log(`${wh.webhook_id} → ${wh.url} (${wh.status})`);
});
```
### Remove a Webhook
```bash curl theme={null}
curl -X DELETE https://api.limitguard.ai/v1/webhooks/wh_abc123def456 \
-H "X-API-Key: lg_live_xxxxxxxxxxxxxxxxxxxx"
```
Returns `204 No Content` on success. The endpoint immediately stops receiving deliveries.
## Testing
Use the test endpoint to fire a simulated event to your webhook without triggering a real check:
```bash curl theme={null}
curl -X POST https://api.limitguard.ai/v1/webhooks/wh_abc123def456/test \
-H "X-API-Key: lg_live_xxxxxxxxxxxxxxxxxxxx" \
-H "Content-Type: application/json" \
-d '{"event": "entity.checked"}'
```
```python Python theme={null}
response = httpx.post(
"https://api.limitguard.ai/v1/webhooks/wh_abc123def456/test",
headers={"X-API-Key": "lg_live_xxxxxxxxxxxxxxxxxxxx"},
json={"event": "entity.checked"},
)
print(response.json())
# {"delivery_id": "evt_test_...", "status": "delivered", "response_code": 200}
```
```javascript JavaScript theme={null}
const response = await fetch(
"https://api.limitguard.ai/v1/webhooks/wh_abc123def456/test",
{
method: "POST",
headers: {
"X-API-Key": "lg_live_xxxxxxxxxxxxxxxxxxxx",
"Content-Type": "application/json",
},
body: JSON.stringify({ event: "entity.checked" }),
}
);
const result = await response.json();
console.log(result.status); // "delivered"
console.log(result.response_code); // 200
```
```json Response theme={null}
{
"delivery_id": "evt_test_01JMVR2KXPB4Y8N9C2WD7FGH1",
"event": "entity.checked",
"status": "delivered",
"response_code": 200,
"response_time_ms": 43,
"delivered_at": "2026-02-20T14:32:07.419Z"
}
```
Test events include a realistic synthetic payload using the `Acme Corp BV` sandbox entity. The signature is computed using your real webhook secret, so your full verification stack is exercised.
Run the test endpoint from your CI/CD pipeline after deployment to verify your webhook handler is reachable and returning `200` before going live.
## Best Practices
### Respond Immediately, Process Asynchronously
Acknowledge the delivery within 10 seconds by returning `200 OK` before doing any meaningful work. Offload processing to a background queue (Celery, BullMQ, SQS, etc.):
```python Python theme={null}
# GOOD — respond immediately, process in background
@app.post("/webhooks/limitguard")
async def handle_webhook(request: Request):
payload = await request.body()
if not verify_signature(payload, request.headers.get("X-LimitGuard-Signature", ""), WEBHOOK_SECRET):
raise HTTPException(status_code=401, detail="Invalid signature")
event = await request.json()
await queue.enqueue("process_limitguard_event", event) # async, non-blocking
return {"received": True} # Return 200 immediately
# BAD — slow processing blocks the response
@app.post("/webhooks/limitguard")
async def handle_webhook_slow(request: Request):
event = await request.json()
await run_heavy_database_operations(event) # Will time out for slow ops
return {"received": True}
```
### Implement Idempotency
LimitGuard may deliver the same event more than once on retry. Use the `id` field as an idempotency key:
```python Python theme={null}
async def process_event(event: dict):
event_id = event["id"]
# Check if already processed
if await db.event_processed(event_id):
return # Safe to ignore — already handled
# Mark as processed atomically before side effects
async with db.transaction():
await db.mark_event_processed(event_id)
await apply_business_logic(event)
```
### Forward-Compatible Event Handling
New event types will be introduced as LimitGuard adds capabilities. Always handle unknown types gracefully rather than raising an exception:
```python Python theme={null}
KNOWN_EVENTS = {"entity.checked", "score.updated", "compliance.alert",
"certificate.issued", "certificate.revoked"}
async def process_event(event: dict):
if event["type"] not in KNOWN_EVENTS:
logger.info(f"Ignoring unknown event type: {event['type']}")
return # Do not raise — return 200 to prevent retries
```
### Validate Event Data
Even after signature verification, validate required fields before using them:
```python Python theme={null}
def validate_score_updated(data: dict) -> bool:
required = {"entity_id", "previous_score", "new_score", "trust_level"}
return required.issubset(data.keys())
```
### Use HTTPS with a Valid Certificate
LimitGuard only delivers to `https://` endpoints with a valid TLS certificate. Self-signed certificates and `http://` URLs are rejected at registration time.
```json Error (400 Bad Request) theme={null}
{
"error": "invalid_webhook_url",
"message": "Webhook URL must use HTTPS with a valid certificate."
}
```
## Webhook Payload Reference
All event payloads share the same outer envelope:
| Field | Type | Description |
| ------------- | --------- | ------------------------------------------------------------- |
| `id` | `string` | Globally unique event ID (`evt_...`) — use as idempotency key |
| `type` | `string` | Event type (e.g., `score.updated`) |
| `api_version` | `string` | API version at time of delivery (e.g., `2026-02-01`) |
| `created_at` | `string` | ISO 8601 timestamp of the event |
| `livemode` | `boolean` | `false` for sandbox events |
| `data` | `object` | Event-specific payload (varies by type) |
## Next Steps
Interactive playground for all four webhook endpoints
Understand quota limits and how they apply to webhook-triggering calls
Complete error catalog — including webhook delivery failures
Test your webhook handler with deterministic sandbox payloads