Connect Your Own Server to Avox

Build a backend that sends WhatsApp messages through Avox - Express, Next.js, FastAPI or Flask, with retries, idempotency, logging and testing.

This guide builds a small, production-ready backend that sends WhatsApp messages through a number connected to Avox. Pick your stack - Express, Next.js, FastAPI or Flask - and copy the code. Every version does the same things:

sends one Idempotency-Key per message, so a retry never sends twice,
retries only the errors that are safe to retry, honouring Retry-After,
backs off with jitter between attempts,
logs the X-Request-Id of every call, and
reads error.code to decide what happens next.

Your app → your server (this guide) → Avox API → Meta Cloud API → Customer's WhatsApp
Customer replies → Avox Inbox and Automation → (message forwarding) → your server
Delivery statuses → Avox Logs

Before you start#

A WhatsApp number connected to Avox with Connected status. See Getting started.
An active plan. See Plans.
The Admin role in your Switchboard, to create the API key.
One of: Node.js 18 or later, or Python 3.9 or later.
At least one approved template, for messages your server starts. See Templates.

Step 1: Create an API key#

1
Open Settings > API Keys
In the dashboard, go to Settings and open the API Keys tab.
2
Choose the kind of key and the number
Under Create a new key, pick Live key or Test key, then the number in Choose a connection.
3
Click Create live key (or Create test key)
The dialog Your new API key opens with the full key, starting avx_live_ (or avx_test_).
4
Copy it, then click Done
This is the only time the full key is shown. If you lose it, rotate the key or revoke it and create another.
5
Click Test
Next to the new key, click Test. "Working - routes to" and your connection's name means it's ready. No message is sent.

You can also create a key from the Outbound API keys section on the connection's own page. Every detail of both screens is on API keys.

Build and run the code below with a test key first. It goes through every check a live send does, answers with a pretend result, and never messages anyone. Swap in a live key when you're ready. See Test mode.

Step 2: Store the key safely#

The key can send WhatsApp messages from your number, so treat it like a password.

Do keep it in an environment variable (AVOX_API_KEY) or a secrets manager, and load it on the server only.
Do give each system its own key, so you can revoke one without touching the others.
Don't put it in a browser page, a mobile app, a public repository, or a shared automation template. Anything shipped to a device can be extracted.

# .env - never commit this file
AVOX_API_KEY=avx_live_your_key_here
# A secret your own apps send to this server, so only they can trigger messages
INTERNAL_TOKEN=a-long-random-string

Step 3: The Avox client#

One function that sends a message, retries what's safe to retry, and explains failures. The retry rules follow Which errors to retry:

`error.code`	What the client does
`rate_limited`, `idempotency_in_progress`	Waits `Retry-After` seconds, retries with the same `Idempotency-Key`.
`internal_error`, network errors and timeouts	Backs off, retries with the same key.
`meta_rate_limited`	Waits `Retry-After`, retries with a new key (WhatsApp's refusal is stored against the old one).
`meta_unavailable`	Doesn't retry automatically - the message may already be out. Your code decides.
Everything else	Stops and raises the error with its code, details and request id.

// avox.js - Node.js 18+ (built-in fetch)
import { randomUUID } from "node:crypto";

const AVOX_URL = "https://api.avox.com/api/messages/send";
const RETRY_SAME_KEY = new Set(["rate_limited", "idempotency_in_progress", "internal_error"]);
const RETRY_NEW_KEY = new Set(["meta_rate_limited"]);
const MAX_ATTEMPTS = 5;

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const backoffMs = (attempt) => {
  const base = Math.min(30_000, 1_000 * 2 ** (attempt - 1)); // 1s, 2s, 4s ... 30s
  return base / 2 + Math.random() * (base / 2);
};

export class AvoxError extends Error {
  constructor({ status, code, message, requestId, details }) {
    super(`${code}: ${message}`);
    Object.assign(this, { status, code, requestId, details });
  }
}

export async function sendWhatsApp(message, idempotencyKey = randomUUID()) {
  let key = idempotencyKey;

  for (let attempt = 1; ; attempt++) {
    let res;
    try {
      res = await fetch(AVOX_URL, {
        method: "POST",
        headers: {
          Authorization: `Bearer ${process.env.AVOX_API_KEY}`,
          "Content-Type": "application/json",
          "Idempotency-Key": key,
        },
        body: JSON.stringify(message),
        signal: AbortSignal.timeout(30_000),
      });
    } catch (networkError) {
      if (attempt >= MAX_ATTEMPTS) throw networkError;
      console.warn(`[avox] network error on attempt ${attempt}, retrying`);
      await sleep(backoffMs(attempt));
      continue;
    }

    const requestId = res.headers.get("X-Request-Id");
    const data = await res.json().catch(() => null);

    if (data?.ok) {
      console.info(`[avox] sent message_id=${data.message_id} request_id=${requestId}`);
      return data;
    }

    const code = data?.error?.code ?? `http_${res.status}`;
    console.warn(`[avox] ${res.status} ${code} request_id=${requestId} attempt=${attempt}`);

    const retryable = RETRY_SAME_KEY.has(code) || RETRY_NEW_KEY.has(code);
    if (!retryable || attempt >= MAX_ATTEMPTS) {
      throw new AvoxError({
        status: res.status,
        code,
        message: data?.error?.message ?? "Unexpected response",
        requestId: data?.error?.request_id ?? requestId,
        details: data?.error?.details,
      });
    }
    if (RETRY_NEW_KEY.has(code)) key = randomUUID();

    const retryAfter = Number(res.headers.get("Retry-After"));
    await sleep(retryAfter > 0 ? retryAfter * 1000 : backoffMs(attempt));
  }
}

// lib/avox.ts - Next.js (App Router), server only
import "server-only";
import { randomUUID } from "node:crypto";

const AVOX_URL = "https://api.avox.com/api/messages/send";
const RETRY_SAME_KEY = new Set(["rate_limited", "idempotency_in_progress", "internal_error"]);
const RETRY_NEW_KEY = new Set(["meta_rate_limited"]);
const MAX_ATTEMPTS = 5;

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
const backoffMs = (attempt: number) => {
  const base = Math.min(30_000, 1_000 * 2 ** (attempt - 1));
  return base / 2 + Math.random() * (base / 2);
};

export type AvoxSuccess = { ok: true; request_id: string; message_id: string | null; result: unknown };

export class AvoxError extends Error {
  constructor(
    public status: number,
    public code: string,
    message: string,
    public requestId: string | null,
    public details?: { path: string; message: string }[]
  ) {
    super(`${code}: ${message}`);
  }
}

export async function sendWhatsApp(message: Record<string, unknown>, idempotencyKey: string = randomUUID()): Promise<AvoxSuccess> {
  let key = idempotencyKey;

  for (let attempt = 1; ; attempt++) {
    let res: Response;
    try {
      res = await fetch(AVOX_URL, {
        method: "POST",
        headers: {
          Authorization: `Bearer ${process.env.AVOX_API_KEY}`,
          "Content-Type": "application/json",
          "Idempotency-Key": key,
        },
        body: JSON.stringify(message),
        cache: "no-store",
        signal: AbortSignal.timeout(30_000),
      });
    } catch (networkError) {
      if (attempt >= MAX_ATTEMPTS) throw networkError;
      await sleep(backoffMs(attempt));
      continue;
    }

    const requestId = res.headers.get("X-Request-Id");
    const data = await res.json().catch(() => null);

    if (data?.ok) {
      console.info(`[avox] sent message_id=${data.message_id} request_id=${requestId}`);
      return data as AvoxSuccess;
    }

    const code: string = data?.error?.code ?? `http_${res.status}`;
    console.warn(`[avox] ${res.status} ${code} request_id=${requestId} attempt=${attempt}`);

    const retryable = RETRY_SAME_KEY.has(code) || RETRY_NEW_KEY.has(code);
    if (!retryable || attempt >= MAX_ATTEMPTS) {
      throw new AvoxError(res.status, code, data?.error?.message ?? "Unexpected response", data?.error?.request_id ?? requestId, data?.error?.details);
    }
    if (RETRY_NEW_KEY.has(code)) key = randomUUID();

    const retryAfter = Number(res.headers.get("Retry-After"));
    await sleep(retryAfter > 0 ? retryAfter * 1000 : backoffMs(attempt));
  }
}

# avox.py - FastAPI (async, httpx)
import asyncio
import logging
import os
import random
import uuid

import httpx

log = logging.getLogger("avox")
AVOX_URL = "https://api.avox.com/api/messages/send"
RETRY_SAME_KEY = {"rate_limited", "idempotency_in_progress", "internal_error"}
RETRY_NEW_KEY = {"meta_rate_limited"}
MAX_ATTEMPTS = 5


class AvoxError(Exception):
    def __init__(self, status, code, message, request_id, details=None):
        super().__init__(f"{code}: {message}")
        self.status, self.code, self.request_id, self.details = status, code, request_id, details


def backoff(attempt):
    base = min(30.0, 2 ** (attempt - 1))  # 1s, 2s, 4s ... 30s
    return base / 2 + random.random() * base / 2


async def send_whatsapp(client: httpx.AsyncClient, message: dict, idempotency_key: str | None = None) -> dict:
    key = idempotency_key or str(uuid.uuid4())
    attempt = 0
    while True:
        attempt += 1
        try:
            res = await client.post(
                AVOX_URL,
                headers={"Authorization": f"Bearer {os.environ['AVOX_API_KEY']}", "Idempotency-Key": key},
                json=message,
                timeout=30,
            )
        except httpx.TransportError:
            if attempt >= MAX_ATTEMPTS:
                raise
            await asyncio.sleep(backoff(attempt))
            continue

        request_id = res.headers.get("X-Request-Id")
        try:
            data = res.json()
        except ValueError:
            data = None

        if data and data.get("ok"):
            log.info("sent message_id=%s request_id=%s", data.get("message_id"), request_id)
            return data

        error = (data or {}).get("error") or {}
        code = error.get("code") or f"http_{res.status_code}"
        log.warning("%s %s request_id=%s attempt=%s", res.status_code, code, request_id, attempt)

        if code not in RETRY_SAME_KEY | RETRY_NEW_KEY or attempt >= MAX_ATTEMPTS:
            raise AvoxError(res.status_code, code, error.get("message", "Unexpected response"),
                            error.get("request_id", request_id), error.get("details"))
        if code in RETRY_NEW_KEY:
            key = str(uuid.uuid4())

        retry_after = int(res.headers.get("Retry-After") or 0)
        await asyncio.sleep(retry_after if retry_after > 0 else backoff(attempt))

# avox.py - Flask (sync, requests)
import logging
import os
import random
import time
import uuid

import requests

log = logging.getLogger("avox")
AVOX_URL = "https://api.avox.com/api/messages/send"
RETRY_SAME_KEY = {"rate_limited", "idempotency_in_progress", "internal_error"}
RETRY_NEW_KEY = {"meta_rate_limited"}
MAX_ATTEMPTS = 5
session = requests.Session()


class AvoxError(Exception):
    def __init__(self, status, code, message, request_id, details=None):
        super().__init__(f"{code}: {message}")
        self.status, self.code, self.request_id, self.details = status, code, request_id, details


def backoff(attempt):
    base = min(30.0, 2 ** (attempt - 1))
    return base / 2 + random.random() * base / 2


def send_whatsapp(message, idempotency_key=None):
    key = idempotency_key or str(uuid.uuid4())
    attempt = 0
    while True:
        attempt += 1
        try:
            res = session.post(
                AVOX_URL,
                headers={"Authorization": f"Bearer {os.environ['AVOX_API_KEY']}", "Idempotency-Key": key},
                json=message,
                timeout=30,
            )
        except requests.RequestException:
            if attempt >= MAX_ATTEMPTS:
                raise
            time.sleep(backoff(attempt))
            continue

        request_id = res.headers.get("X-Request-Id")
        try:
            data = res.json()
        except ValueError:
            data = None

        if data and data.get("ok"):
            log.info("sent message_id=%s request_id=%s", data.get("message_id"), request_id)
            return data

        error = (data or {}).get("error") or {}
        code = error.get("code") or f"http_{res.status_code}"
        log.warning("%s %s request_id=%s attempt=%s", res.status_code, code, request_id, attempt)

        if code not in RETRY_SAME_KEY | RETRY_NEW_KEY or attempt >= MAX_ATTEMPTS:
            raise AvoxError(res.status_code, code, error.get("message", "Unexpected response"),
                            error.get("request_id", request_id), error.get("details"))
        if code in RETRY_NEW_KEY:
            key = str(uuid.uuid4())

        retry_after = int(res.headers.get("Retry-After") or 0)
        time.sleep(retry_after if retry_after > 0 else backoff(attempt))

Step 4: A route your apps call#

Your own apps (a website backend, a job, an admin tool) call this route instead of the Avox API directly, so the key lives in one place. The caller passes a reference - an id from your own system, like an order id - which becomes the Idempotency-Key, so the same notification is never sent twice even if the caller retries.

// server.js - npm install express
import express from "express";
import { timingSafeEqual } from "node:crypto";
import { sendWhatsApp, AvoxError } from "./avox.js";

const app = express();
app.use(express.json({ limit: "100kb" }));

function authorized(req) {
  const given = Buffer.from(req.get("x-internal-token") ?? "");
  const expected = Buffer.from(process.env.INTERNAL_TOKEN ?? "");
  return expected.length > 0 && given.length === expected.length && timingSafeEqual(given, expected);
}

// POST /notify { "to": "15551234567", "template": "order_shipped", "params": ["A-1042"], "reference": "order-A-1042-shipped" }
app.post("/notify", async (req, res) => {
  if (!authorized(req)) return res.sendStatus(401);
  const { to, template, params = [], language = "en_US", reference } = req.body ?? {};
  if (!to || !template || !reference) return res.status(400).json({ error: "to, template and reference are required" });

  try {
    const result = await sendWhatsApp(
      {
        to,
        type: "template",
        template: {
          name: template,
          language: { code: language },
          components: params.length ? [{ type: "body", parameters: params.map((text) => ({ type: "text", text: String(text) })) }] : undefined,
        },
        biz_opaque_callback_data: reference,
      },
      reference
    );
    return res.json({ ok: true, message_id: result.message_id });
  } catch (err) {
    if (err instanceof AvoxError) {
      return res.status(err.status >= 500 ? 502 : 422).json({ ok: false, code: err.code, request_id: err.requestId, details: err.details });
    }
    console.error(err);
    return res.status(502).json({ ok: false, code: "network_error" });
  }
});

app.get("/healthz", (_req, res) => res.send("ok"));
app.listen(process.env.PORT ?? 3000, () => console.log("Listening"));

// app/api/notify/route.ts - Next.js (App Router)
import { timingSafeEqual } from "node:crypto";
import { NextResponse, type NextRequest } from "next/server";
import { AvoxError, sendWhatsApp } from "@/lib/avox";

export const runtime = "nodejs";

function authorized(request: NextRequest) {
  const given = Buffer.from(request.headers.get("x-internal-token") ?? "");
  const expected = Buffer.from(process.env.INTERNAL_TOKEN ?? "");
  return expected.length > 0 && given.length === expected.length && timingSafeEqual(given, expected);
}

export async function POST(request: NextRequest) {
  if (!authorized(request)) return NextResponse.json({ error: "unauthorized" }, { status: 401 });

  const body = await request.json().catch(() => null);
  const { to, template, params = [], language = "en_US", reference } = body ?? {};
  if (!to || !template || !reference) {
    return NextResponse.json({ error: "to, template and reference are required" }, { status: 400 });
  }

  try {
    const result = await sendWhatsApp(
      {
        to,
        type: "template",
        template: {
          name: template,
          language: { code: language },
          ...(params.length
            ? { components: [{ type: "body", parameters: params.map((text: unknown) => ({ type: "text", text: String(text) })) }] }
            : {}),
        },
        biz_opaque_callback_data: reference,
      },
      reference
    );
    return NextResponse.json({ ok: true, message_id: result.message_id });
  } catch (err) {
    if (err instanceof AvoxError) {
      return NextResponse.json(
        { ok: false, code: err.code, request_id: err.requestId, details: err.details },
        { status: err.status >= 500 ? 502 : 422 }
      );
    }
    console.error(err);
    return NextResponse.json({ ok: false, code: "network_error" }, { status: 502 });
  }
}

# main.py - pip install fastapi uvicorn httpx ; run: uvicorn main:app --port 3000
import hmac
import os
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from avox import AvoxError, send_whatsapp


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient()
    yield
    await app.state.http.aclose()


app = FastAPI(lifespan=lifespan)


class Notify(BaseModel):
    to: str
    template: str
    reference: str
    params: list[str] = []
    language: str = "en_US"


@app.post("/notify")
async def notify(body: Notify, x_internal_token: str = Header(default="")):
    if not hmac.compare_digest(x_internal_token, os.environ.get("INTERNAL_TOKEN", "")) or not x_internal_token:
        raise HTTPException(status_code=401)

    template = {"name": body.template, "language": {"code": body.language}}
    if body.params:
        template["components"] = [{"type": "body", "parameters": [{"type": "text", "text": p} for p in body.params]}]

    try:
        result = await send_whatsapp(
            app.state.http,
            {"to": body.to, "type": "template", "template": template, "biz_opaque_callback_data": body.reference},
            idempotency_key=body.reference,
        )
        return {"ok": True, "message_id": result.get("message_id")}
    except AvoxError as err:
        status = 502 if err.status >= 500 else 422
        return JSONResponse({"ok": False, "code": err.code, "request_id": err.request_id, "details": err.details}, status_code=status)
    except httpx.TransportError:
        return JSONResponse({"ok": False, "code": "network_error"}, status_code=502)


@app.get("/healthz")
async def healthz():
    return "ok"

# app.py - pip install flask requests ; run: flask --app app run --port 3000
import hmac
import os

import requests
from flask import Flask, jsonify, request

from avox import AvoxError, send_whatsapp

app = Flask(__name__)


def authorized():
    given = request.headers.get("x-internal-token", "")
    expected = os.environ.get("INTERNAL_TOKEN", "")
    return bool(expected) and hmac.compare_digest(given, expected)


@app.post("/notify")
def notify():
    if not authorized():
        return "", 401
    body = request.get_json(silent=True) or {}
    to, template_name, reference = body.get("to"), body.get("template"), body.get("reference")
    if not (to and template_name and reference):
        return jsonify(error="to, template and reference are required"), 400

    template = {"name": template_name, "language": {"code": body.get("language", "en_US")}}
    params = body.get("params") or []
    if params:
        template["components"] = [{"type": "body", "parameters": [{"type": "text", "text": str(p)} for p in params]}]

    try:
        result = send_whatsapp(
            {"to": to, "type": "template", "template": template, "biz_opaque_callback_data": reference},
            idempotency_key=reference,
        )
        return jsonify(ok=True, message_id=result.get("message_id"))
    except AvoxError as err:
        status = 502 if err.status >= 500 else 422
        return jsonify(ok=False, code=err.code, request_id=err.request_id, details=err.details), status
    except requests.RequestException:
        return jsonify(ok=False, code="network_error"), 502


@app.get("/healthz")
def healthz():
    return "ok"

Call it from any of your apps:

curl -X POST http://localhost:3000/notify \
  -H "x-internal-token: $INTERNAL_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"to":"15551234567","template":"order_shipped","params":["A-1042"],"reference":"order-A-1042-shipped"}'

The route sends templates because your server usually starts the conversation. To send other types (text, media, buttons) to someone who messaged you in the last 24 hours, pass a different body to sendWhatsApp / send_whatsapp - see Message types.

For high volume, put messages on a queue and let a few workers call the client, instead of sending from inside a web request. The reference still makes every retry safe. See Rate limits.

Step 5: Receiving events#

What happens after a send is handled inside Avox:

Event	Where it goes
Customer replies, button taps, new conversations	The Inbox, where your team answers, and Automation, which can answer automatically. With message forwarding on, also your own server - see Receive messages.
Delivery statuses (`sent`, `delivered`, `read`, `failed`) of your API messages	Logs, each with its full payload. Match them by `message_id`, or by the `reference` you sent as `biz_opaque_callback_data`.
The result of each API call	The Sent tab on the API logs page (and the key's Recent requests in Settings > API Keys). See API logs.
Each message forwarded to your server	The Received tab on the API logs page, and Recent deliveries on the connection's page.

See Statuses and replies for the payload shapes.

Receive messages#

To get each new customer message on your own server too, turn on Message forwarding on the connection's page (Admins). Avox then POSTs one signed message.received event per incoming customer message to your Endpoint URL. Only incoming customer messages are forwarded - statuses, calls and messages your team sends stay in Avox.

What's sent. A small JSON event - the message only, never tokens or account ids:

{
  "id": "evt_5f0c9a7e1b2d4c6e8f00a1b2c3d4e5f6",
  "type": "message.received",
  "api_version": "2026-10-01",
  "created_at": "2026-10-10T10:00:02.000Z",
  "connection_id": "0b1c2d3e-4f50-4a6b-8c7d-9e0f1a2b3c4d",
  "data": {
    "message_id": "wamid.HBgLMTU1NTEyMzQ1NjcVAgASGBQzQTdC",
    "from": "+15551234567",
    "contact_name": "Maria Lopez",
    "timestamp": "2026-10-10T10:00:00.000Z",
    "type": "image",
    "media": { "id": "1203949581720283", "mime_type": "image/jpeg", "caption": "My receipt" }
  }
}

data.type is text, image, audio, video, document, sticker, location, contacts, interactive, button or reaction; an example of each is on What's sent.

Headers. Avox-Signature: t=<unix>,v1=<hex>, Avox-Event-Id, Avox-Event-Type: message.received and User-Agent: Avox-Webhooks/1.0.

Check the signature. Each v1 is a hex HMAC-SHA256 of <t>.<raw body>, keyed with your whole signing secret (whsec_...) as text. Read the raw body, refuse a t more than 5 minutes from your clock, and compare in constant time. Then answer 2xx within 10 seconds and do your work afterwards. Add this route next to /notify, with the secret in AVOX_WEBHOOK_SECRET:

// add to server.js
import { createHmac } from "node:crypto"; // timingSafeEqual is already imported above

function verifyAvoxSignature(rawBody, header) {
  const secret = process.env.AVOX_WEBHOOK_SECRET;
  if (!header || !secret) return false;
  let t = null;
  const signatures = [];
  for (const part of header.split(",")) {
    const [key, value = ""] = part.trim().split("=");
    if (key === "t" && /^\d+$/.test(value)) t = Number(value);
    if (key === "v1") signatures.push(value);
  }
  if (t === null || Math.abs(Date.now() / 1000 - t) > 300) return false;
  const expected = createHmac("sha256", secret).update(`${t}.`).update(rawBody).digest();
  return signatures.some((sig) => {
    const given = Buffer.from(sig, "hex");
    return given.length === expected.length && timingSafeEqual(given, expected);
  });
}

const handled = new Set(); // use your database in production

// Registered with express.raw so the body stays byte-for-byte as sent.
app.post("/avox/messages", express.raw({ type: "application/json", limit: "1mb" }), (req, res) => {
  if (!verifyAvoxSignature(req.body, req.get("Avox-Signature"))) return res.sendStatus(401);
  res.sendStatus(200); // answer fast

  const event = JSON.parse(req.body.toString("utf8"));
  if (event.test || handled.has(event.id)) return; // test events and repeats
  handled.add(event.id);
  console.log(`[avox] ${event.data.type} from ${event.data.from}`);
});

// app/avox/messages/route.ts
import { createHmac, timingSafeEqual } from "node:crypto";
import { after } from "next/server";

export const runtime = "nodejs";

function verifyAvoxSignature(rawBody: string, header: string | null): boolean {
  const secret = process.env.AVOX_WEBHOOK_SECRET;
  if (!header || !secret) return false;
  let t: number | null = null;
  const signatures: string[] = [];
  for (const part of header.split(",")) {
    const [key, value = ""] = part.trim().split("=");
    if (key === "t" && /^\d+$/.test(value)) t = Number(value);
    if (key === "v1") signatures.push(value);
  }
  if (t === null || Math.abs(Date.now() / 1000 - t) > 300) return false;
  const expected = createHmac("sha256", secret).update(`${t}.${rawBody}`, "utf8").digest();
  return signatures.some((sig) => {
    const given = Buffer.from(sig, "hex");
    return given.length === expected.length && timingSafeEqual(given, expected);
  });
}

export async function POST(request: Request) {
  const rawBody = await request.text();
  if (!verifyAvoxSignature(rawBody, request.headers.get("avox-signature"))) {
    return new Response("invalid signature", { status: 401 });
  }
  const event = JSON.parse(rawBody);
  after(async () => {
    if (event.test) return;
    // dedupe on event.id in your database, then handle event.data
    console.info(`[avox] ${event.data.type} from ${event.data.from}`);
  });
  return new Response(null, { status: 200 });
}

# add to main.py
import hashlib, json, time
from fastapi import BackgroundTasks, Request, Response


def verify_avox_signature(raw: bytes, header: str | None) -> bool:
    secret = os.environ.get("AVOX_WEBHOOK_SECRET", "").encode()
    if not header or not secret:
        return False
    t, signatures = None, []
    for part in header.split(","):
        key, _, value = part.strip().partition("=")
        if key == "t" and value.isdigit():
            t = int(value)
        elif key == "v1":
            signatures.append(value)
    if t is None or abs(time.time() - t) > 300:
        return False
    expected = hmac.new(secret, f"{t}.".encode() + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, s) for s in signatures)


def handle_event(event: dict) -> None:
    if event.get("test"):
        return
    # dedupe on event["id"] in your database, then handle event["data"]
    print("avox:", event["data"]["type"], "from", event["data"]["from"])


@app.post("/avox/messages")
async def avox_messages(request: Request, background: BackgroundTasks, avox_signature: str | None = Header(default=None)):
    raw = await request.body()
    if not verify_avox_signature(raw, avox_signature):
        return Response(status_code=401)
    background.add_task(handle_event, json.loads(raw))
    return Response(status_code=200)

# add to app.py
import hashlib, json, time
from concurrent.futures import ThreadPoolExecutor

pool = ThreadPoolExecutor(max_workers=4)


def verify_avox_signature(raw: bytes, header: str | None) -> bool:
    secret = os.environ.get("AVOX_WEBHOOK_SECRET", "").encode()
    if not header or not secret:
        return False
    t, signatures = None, []
    for part in header.split(","):
        key, _, value = part.strip().partition("=")
        if key == "t" and value.isdigit():
            t = int(value)
        elif key == "v1":
            signatures.append(value)
    if t is None or abs(time.time() - t) > 300:
        return False
    expected = hmac.new(secret, f"{t}.".encode() + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, s) for s in signatures)


def handle_event(event: dict) -> None:
    if event.get("test"):
        return
    # dedupe on event["id"] in your database, then handle event["data"]
    print("avox:", event["data"]["type"], "from", event["data"]["from"])


@app.post("/avox/messages")
def avox_messages():
    raw = request.get_data()
    if not verify_avox_signature(raw, request.headers.get("Avox-Signature")):
        return "", 401
    pool.submit(handle_event, json.loads(raw))
    return "", 200

In Express, register /avox/messages with express.raw as shown. The app.use(express.json()) from Step 4 parses every route's body first, so either register this route before it, or limit express.json() to your other routes - otherwise the raw bytes are gone and every signature fails.

Skip repeats. Retries and resends keep the same Avox-Event-Id, so store the ids you've handled.

Retries. If your endpoint doesn't answer 2xx within 10 seconds (redirects count as failures), Avox retries after about 1 minute, 5 minutes, 30 minutes, 2 hours, 6 hours and 12 hours, with a little randomness - about 21 hours in all - then marks the delivery Failed.

Test events. Send test event on the connection's page sends a signed sample with "test": true right away and shows the result. It isn't retried.

Rotating the secret. Rotate secret shows a new secret once. For 24 hours each request carries two v1 signatures (new and old), so code that accepts any matching v1 keeps working while you deploy the new one.

The delivery log. Recent deliveries on the connection's page shows each message's status, HTTP code, attempts and payload. Resend queues one again with the same event id. The Received tab on the API logs page lists the same deliveries across all your numbers.

Auto-pause. After 5 deliveries in a row fail all their retries, Avox pauses forwarding and emails your Admins. Fix the endpoint, send a test event, turn forwarding back on, and resend what you missed.

Troubleshooting. A 401 from your own route almost always means the body was parsed before the signature check, or the wrong secret. More fixes are on Troubleshooting, and every detail of the section is on Message forwarding.

Step 6: Test locally#

1
Run your server
Start it on port 3000 with AVOX_API_KEY and INTERNAL_TOKEN set.
2
Open a 24-hour window (optional)
Send any WhatsApp message from your personal phone to the connected number, so non-template messages work too.
3
Call your route
Run the curl above with your personal number. Expect {"ok":true,"message_id":"wamid..."} and the message on your phone.
4
Call it again with the same reference
You get the same message_id, and no second message arrives - that's the idempotency key at work.
5
Check Avox
Open API logs in the dashboard: the Sent tab shows both calls, the second marked replayed. Open Logs to see the sent, delivered and read statuses.
6
Receive a reply (optional)
With message forwarding on and a tunnel (below), reply from your phone. The message reaches /avox/messages and shows Delivered under Recent deliveries on the connection's page and on the Received tab of API logs.

Exposing localhost over HTTPS#

Message forwarding only accepts a public https:// Endpoint URL. To try it with a server on your own machine, use a tunnel that gives localhost a temporary HTTPS address:

# ngrok
ngrok http 3000

# Cloudflare Tunnel (quick tunnel)
cloudflared tunnel --url http://localhost:3000

Paste the https://... address the tunnel prints, plus /avox/messages, into Endpoint URL, click Save URL, turn forwarding on, and click Send test event. You'll see Delivered - your endpoint answered 200 when the tunnel, your server and your signature check all work. Then send a WhatsApp message to the number from your phone and watch it arrive. Tunnel addresses usually change each time you restart them, so set your real server's URL before going live.

More ways to test, including safe error-path requests, are on Testing.

Step 7: Deploy#

Secrets: AVOX_API_KEY and INTERNAL_TOKEN (and AVOX_WEBHOOK_SECRET, if you receive messages) set in your host's secret store, not in the code or the image.
Live key: production uses an avx_live_... key. A test key answers 200 but never sends - check responses don't carry the Avox-Test-Mode: true header.
Receiving: if you use message forwarding, the Endpoint URL points at your production server and Send test event shows Delivered.
HTTPS: the server runs behind HTTPS, and /notify is only reachable by your own apps.
Timeouts: every Avox call has a timeout (30 seconds above), and your web server allows enough time for a retry or two - or sends from a queue worker instead.
Logs: request_id and error.code are logged for every failed call.
References: every caller sends a stable reference per message.
Templates: every template your server uses is approved, in the languages you send.
Key check: Test on the production key shows "Working - routes to" the right number.
Rotation: you know how to rotate the key - create, deploy, check, revoke. See Authentication.

Then work through the Go-live checklist.

Troubleshooting#

`error.code`	Likely cause	Fix
`missing_api_key`	`AVOX_API_KEY` isn't loaded, so no header was sent.	Check the environment of the running process.
`invalid_api_key`	Key cut off, quoted, or with a stray space.	Copy it again, or create a new key.
`key_revoked`	The key was revoked.	Create a new key and redeploy.
`key_expired`	The key was rotated and its grace period ended.	Deploy the key that replaced it.
`ip_not_allowed`	The key is limited to certain addresses.	Call from an allowed address, or ask Support to update the list.
`subscription_inactive`	No active plan, or the plan is paused.	Fix it on Billing.
`api_not_in_plan`	Your plan doesn't include API access.	Move to a plan with API access on Billing > Change plan.
`connection_paused`	The number is paused.	Resume it on Connections.
`connection_not_ready`	The number needs reconnecting on Meta's side.	Reconnect it from Connections.
`invalid_recipient`	`to` lacks a country code.	Send the full international number.
`invalid_payload`	A field is missing or too long.	Read `details` - each entry names the field.
`idempotency_conflict`	The same `reference` was used for a different message.	Make references unique per message.
`template_not_found`	Wrong template name or language.	Copy both from Templates.
`template_params_invalid`	Wrong number or format of `params`.	Match the template's placeholders, in order.
`outside_messaging_window`	A non-template message to someone who hasn't written in 24 hours.	Send a template.
`rate_limited`	Too many requests at once.	The client retries; for bulk, queue and pace.
`quota_exceeded`	The key used up its daily allowance.	Wait for midnight UTC, or ask Support about the allowance.
`meta_unavailable`	WhatsApp didn't answer in time.	Check Logs for a status before resending.
`internal_error`	A problem at Avox.	The client retries; if it persists, contact Support with the `request_id`.

Every code is explained on API errors. To see how often each one happens, open API logs and pick it in the error code filter (Any error code), or paste a request_id into the search box.

Last updated 15 October 2026