Setup Guide

$
curl -X POST 'https://api.airweave.ai/webhooks/subscriptions' \
>
  -H 'x-api-key: YOUR_API_KEY' \
>
  -H 'Content-Type: application/json' \
>
  -d '{
>
    "url": "https://your-server.com/webhooks/airweave",
>
    "event_types": ["sync.completed", "sync.failed"]
>
  }'

1
from fastapi import FastAPI, Request, HTTPException
2
3
app = FastAPI()
4
5
@app.post("/webhooks/airweave")
6
async def handle_webhook(request: Request):
7
    event = await request.json()
8
9
    match event["event_type"]:
10
        case "sync.completed":
11
            job_id = event["job_id"]
12
            collection = event["collection_name"]
13
            print(f"Sync completed: {collection} (job: {job_id})")
14
            # Trigger your downstream workflow here
15
16
        case "sync.failed":
17
            error = event.get("error", "Unknown error")
18
            collection = event["collection_name"]
19
            print(f"Sync failed: {collection} - {error}")
20
            # Send alert or log the failure
21
22
        case "sync.running":
23
            print(f"Sync started: {event['collection_name']}")
24
25
    return {"status": "ok"}

1
import hmac
2
import hashlib
3
import base64
4
5
def verify_signature(payload: bytes, headers: dict, secret: str) -> bool:
6
    msg_id = headers.get("svix-id")
7
    timestamp = headers.get("svix-timestamp")
8
    signature = headers.get("svix-signature")
9
10
    if not all([msg_id, timestamp, signature]):
11
        return False
12
13
    # Decode the secret (remove whsec_ prefix)
14
    secret_bytes = base64.b64decode(secret.replace("whsec_", ""))
15
16
    # Build the signed content
17
    signed_content = f"{msg_id}.{timestamp}.{payload.decode()}"
18
19
    # Compute expected signature
20
    expected = hmac.new(
21
        secret_bytes,
22
        signed_content.encode(),
23
        hashlib.sha256
24
    ).digest()
25
    expected_b64 = base64.b64encode(expected).decode()
26
27
    # Compare against provided signatures
28
    for sig in signature.split(","):
29
        sig_value = sig.split(" ")[-1]  # Handle "v1,..." format
30
        if hmac.compare_digest(expected_b64, sig_value):
31
            return True
32
33
    return False

$
curl -X PATCH 'https://api.airweave.ai/webhooks/subscriptions/{subscription_id}' \
>
  -H 'x-api-key: YOUR_API_KEY' \
>
  -H 'Content-Type: application/json' \
>
  -d '{
>
    "event_types": ["sync.completed", "sync.failed", "sync.running"],
>
    "disabled": false
>
  }'