Skip to content
Oxtro
Documentation menu

Examples

Send WhatsApp messages and OTP codes from Python

Uses the requests library. The webhook example uses Flask.

You need an API key and the ID of a connected number. The quickstart covers both.

Install
pip install requests flask

Send a message

The API answers 202 with status: "queued". The Idempotency-Key header makes a retry safe. Full reference: Messages API.

Python
import os
import requests

res = requests.post(
    "https://api.getoxtro.com/v1/messages",
    headers={
        "Authorization": f"Bearer {os.environ['OXTRO_API_KEY']}",
        "Idempotency-Key": "order-4821-confirmation",
    },
    json={
        "connection_id": "conn_123",
        "to": "+213555123456",
        "text": "Your order ORD-4821 has been confirmed.",
    },
    timeout=10,
)

body = res.json()
if not res.ok:
    raise RuntimeError(f"{body['error']['code']}: {body['error']['message']}")

print(body["id"], body["status"])  # msg_123 queued
202 Accepted
{
  "id": "msg_123",
  "status": "queued",
  "queued_at": "2026-10-10T09:41:07.000Z"
}

Send a one-time code

Oxtro generates the code and sends it on WhatsApp. You only keep the request_id. Full reference: OTP API.

Python
import os
import requests

res = requests.post(
    "https://api.getoxtro.com/v1/otp/send",
    headers={"Authorization": f"Bearer {os.environ['OXTRO_API_KEY']}"},
    json={"connection_id": "conn_123", "to": "+213555123456"},
    timeout=10,
)

body = res.json()
if not res.ok:
    raise RuntimeError(f"{body['error']['code']}: {body['error']['message']}")

# Keep request_id with the user's session. The code itself never reaches your server.
request_id = body["request_id"]
200 OK
{
  "request_id": "otp_123",
  "expires_in": 300,
  "status": "sent"
}

Verify the code

A correct code returns 200. A wrong, expired or locked code returns 400 with the code OTP_INVALID, OTP_EXPIRED or OTP_MAX_ATTEMPTS.

Python
import os
import requests

REJECTED = {"OTP_INVALID", "OTP_EXPIRED", "OTP_MAX_ATTEMPTS"}


def verify_code(request_id: str, code: str) -> bool:
    res = requests.post(
        "https://api.getoxtro.com/v1/otp/verify",
        headers={"Authorization": f"Bearer {os.environ['OXTRO_API_KEY']}"},
        json={"request_id": request_id, "code": code},
        timeout=10,
    )
    if res.ok:
        return True

    # A wrong, expired or locked code is an HTTP 400 with one of these codes.
    error = res.json()["error"]
    if error["code"] in REJECTED:
        return False
    raise RuntimeError(f"{error['code']}: {error['message']}")
200 OK
{
  "verified": true,
  "request_id": "otp_123",
  "phone_number": "+213555123456",
  "verified_at": "2026-10-10T09:42:31.000Z"
}

Verify a webhook signature

Oxtro signs every webhook with HMAC-SHA256 over <timestamp>.<raw body>. Check it before you trust the event. Full reference: Webhooks.

Python
import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)


def signature_is_valid(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    parts = dict(pair.split("=", 1) for pair in header.split(",") if "=" in pair)
    timestamp, signature = parts.get("t", ""), parts.get("v1", "")
    if not timestamp.isdigit() or not signature:
        return False
    if abs(time.time() - int(timestamp)) > tolerance:
        return False

    expected = hmac.new(
        secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)


@app.post("/webhooks/oxtro")
def oxtro_webhook():
    # get_data() is the raw body, the exact bytes Oxtro signed.
    header = request.headers.get("X-Webhook-Signature", "")
    if not signature_is_valid(request.get_data(), header, os.environ["OXTRO_WEBHOOK_SECRET"]):
        abort(401)

    event = request.get_json()
    if event["event"] == "message.received":
        print(event["data"]["from"], event["data"]["content"])

    return "", 200