Webhooks

Polling:  Client repeatedly asks "anything new?" every N seconds
          GET /v1/events?since=2026-04-10T00:00:00Z (wasteful, delayed)

Webhook:  Server POSTs to client URL when event occurs
          POST https://your-app.example.com/webhooks/events (instant, efficient)

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()

@app.post("/webhooks/github")
async def github_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Hub-Signature-256", "")

    # Verify signature
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail="Invalid signature")

    payload = await request.json()
    event = request.headers.get("X-GitHub-Event")
    print(f"Received {event}: {payload.get('action', 'N/A')}")
    return {"status": "ok"}

# Listen on port 9999, print incoming webhook payload
while true; do
    echo -e "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK" | \
      nc -l -p 9999 | tee -a webhook_log.txt
    echo "---" >> webhook_log.txt
done

# Verify a GitHub webhook payload
body='{"action":"push","ref":"refs/heads/main"}'
secret="whsec_abc123"
expected=$(echo -n "$body" | openssl dgst -sha256 -hmac "$secret" | awk '{print $2}')
received="abc123def456..."  # from X-Hub-Signature-256 header

if [[ "sha256=$expected" == "$received" ]]; then
    echo "Signature valid"
else
    echo "Signature INVALID -- reject" >&2
fi

{
  "version": "0.1",
  "sharedSecret": "secret123",
  "sentAt": "2026-04-10T12:00:00Z",
  "organizationId": "12345",
  "organizationName": "DomusDigitalis",
  "networkId": "L_123",
  "alertType": "Motion detected",
  "alertData": {}
}

gh api repos/owner/repo/hooks \
  -f 'name=web' \
  -f 'config[url]=https://your-app.example.com/webhooks/github' \
  -f 'config[content_type]=json' \
  -f 'config[secret]=your-webhook-secret' \
  -f 'events[]=push' \
  -f 'events[]=pull_request' \
  -f 'active=true'

gh api repos/owner/repo/hooks --jq '.[] | {id, url: .config.url, events, active}'

gh api repos/owner/repo/hooks/HOOK_ID/pings -X POST

from fastapi import BackgroundTasks

@app.post("/webhooks/events")
async def receive_event(request: Request, bg: BackgroundTasks):
    payload = await request.json()
    bg.add_task(process_event, payload)  # async processing
    return {"status": "accepted"}        # return 200 immediately

Provider        Retry Policy
GitHub          3 retries over ~1 hour, exponential backoff
Meraki          Unknown retries, shared secret for verification
Custom          Implement your own: queue failed deliveries, retry with backoff

processed_ids: set[str] = set()  # use Redis/DB in production

@app.post("/webhooks/events")
async def receive_event(request: Request):
    payload = await request.json()
    event_id = request.headers.get("X-Delivery-ID", "")

    if event_id in processed_ids:
        return {"status": "already processed"}

    processed_ids.add(event_id)
    await process_event(payload)
    return {"status": "ok"}

# ngrok -- tunnel to local FastAPI
ngrok http 8000
# Copy the https://xxx.ngrok.io URL to webhook config

# Tailscale Funnel -- if on Tailscale network
tailscale funnel 8000