Server Examples

Ready-to-use receivers for SuperWaba webhook events. Each verifies the X-Webhook-Secret header and returns a 2xx within the required window. See the overview for the envelope and headers, and the event pages for individual payloads.

---

Full example: cURL test

You can test your webhook endpoint with this cURL command:

curl -X POST https://your-endpoint.com/webhook \
  -H "Content-Type: application/json" \
  -H "X-Webhook-Event: message.received" \
  -H "X-Webhook-Delivery-Id: test-$(uuidgen)" \
  -H "X-Webhook-Secret: your-secret-here" \
  -d '{
    "event": "message.received",
    "timestamp": "2026-05-16T12:00:00.000Z",
    "webhook_id": "test-webhook-id",
    "data": {
      "message_id": "wamid.test123",
      "from": "+918245678901",
      "contact_name": "Test Customer",
      "type": "text",
      "content": {
        "text": "Hello, I need help!"
      },
      "timestamp": "2026-05-16T12:00:00Z"
    }
  }'

---

Full example: Express.js webhook server

import express from 'express';
import crypto from 'crypto';

const app = express();
app.use(express.json());

const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET!;

// Verify webhook signature
function verifySignature(req: express.Request): boolean {
  const secret = req.headers['x-webhook-secret'] as string;
  return secret === WEBHOOK_SECRET;
}

app.post('/webhook', (req, res) => {
  // 1. Verify signature
  if (!verifySignature(req)) {
    return res.status(401).send('Invalid signature');
  }

  const { event, data, timestamp } = req.body;
  const deliveryId = req.headers['x-webhook-delivery-id'];

  console.log(`[${timestamp}] Event: ${event} (delivery: ${deliveryId})`);

  // 2. Handle events
  switch (event) {
    case 'message.received':
      console.log(`Message from ${data.contact_name}: ${data.content?.text}`);
      // Save to your database, forward to Slack, trigger a workflow, etc.
      break;

    case 'message.sent':
      console.log(`Message sent to ${data.to}: ${data.content?.text}`);
      if (data.is_ai) {
        console.log(`  (sent by AI agent ${data.agent_id})`);
      }
      break;

    case 'message.delivered':
      console.log(`Message ${data.message_id} delivered to ${data.recipient_id}`);
      break;

    case 'message.read':
      console.log(`Message ${data.message_id} read by ${data.recipient_id}`);
      break;

    case 'message.failed':
      console.error(`Message ${data.message_id} failed: ${data.error?.message}`);
      // Alert your team, retry logic, etc.
      break;

    case 'contact.created':
      console.log(`New contact: ${data.name} (${data.phone})`);
      // Sync to your CRM
      break;

    case 'conversation.opened':
      console.log(`New conversation with ${data.contact_name}`);
      break;

    case 'conversation.closed':
      console.log(`Conversation closed: ${data.summary}`);
      break;

    default:
      console.log(`Unhandled event: ${event}`);
  }

  // 3. Always return 200
  res.status(200).json({ received: true });
});

app.listen(3000, () => {
  console.log('Webhook server listening on port 3000');
});

---

Full example: Python Flask webhook server

from flask import Flask, request, jsonify
import os

app = Flask(__name__)
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "")

@app.route("/webhook", methods=["POST"])
def webhook():
    # Verify secret
    if request.headers.get("X-Webhook-Secret") != WEBHOOK_SECRET:
        return "Unauthorized", 401

    payload = request.json
    event = payload.get("event")
    data = payload.get("data", {})

    if event == "message.received":
        print(f"Message from {data.get('contact_name')}: {data.get('content', {}).get('text')}")

    elif event == "message.sent":
        print(f"Message sent to {data.get('to')}")

    elif event == "message.failed":
        error = data.get("error", {})
        print(f"Message failed: {error.get('code')} - {error.get('message')}")

    elif event == "conversation.closed":
        print(f"Conversation closed: {data.get('summary')}")

    return jsonify({"received": True}), 200

if __name__ == "__main__":
    app.run(port=3000)