Webhooks

result = client.webhooks.list()
result.data  # list of Webhook objects

webhook = client.webhooks.create(
    url="https://yourapp.com/webhooks/whatsrb",
    events=["message.received", "message.status", "message.failed"],
)

webhook.id        # 1
webhook.url       # "https://yourapp.com/webhooks/whatsrb"
webhook.events    # ["message.received", "message.status", "message.failed"]
webhook.secret    # "whsec_xxx" — only present on create, store it now!
webhook.is_active # True

webhook = client.webhooks.retrieve(1)

webhook = client.webhooks.update(1,
    url="https://yourapp.com/new-path",
    active=False,
    events=["message.failed"],
)

client.webhooks.delete(1)  # True

from whatsrb_cloud import WebhookSignature

is_valid = WebhookSignature.verify(
    payload=request_body,
    secret="whsec_xxx",
    signature=request.headers["X-Webhook-Signature"],
    timestamp=request.headers.get("X-Webhook-Timestamp"),
)

from fastapi import FastAPI, Request, HTTPException
from whatsrb_cloud import WebhookSignature, EventRegistry
import os

app = FastAPI()
events = EventRegistry()

@app.post("/webhooks/whatsrb")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Webhook-Signature", "")
    timestamp = request.headers.get("X-Webhook-Timestamp")

    if not WebhookSignature.verify(
        payload=body.decode(),
        secret=os.environ["WHATSRB_WEBHOOK_SECRET"],
        signature=signature,
        timestamp=timestamp,
    ):
        raise HTTPException(status_code=401)

    payload = await request.json()
    events.dispatch(payload)
    return {"ok": True}

from flask import Flask, request, abort
from whatsrb_cloud import WebhookSignature
import os

app = Flask(__name__)

@app.post("/webhooks/whatsrb")
def handle_webhook():
    if not WebhookSignature.verify(
        payload=request.get_data(as_text=True),
        secret=os.environ["WHATSRB_WEBHOOK_SECRET"],
        signature=request.headers.get("X-Webhook-Signature", ""),
        timestamp=request.headers.get("X-Webhook-Timestamp"),
    ):
        abort(401)

    payload = request.get_json()
    events.dispatch(payload)
    return "", 200