Skip to content

Python / FastAPI

There is no Python SDK yet; the API is plain HTTPS + JSON. These examples use httpx.

pip install httpx fastapi uvicorn

Client helper

import os, time, uuid
import httpx

WT = os.environ.get("WHATSTIG_BASE_URL", "https://api.whatstig.com")
client = httpx.Client(
    base_url=WT,
    headers={"Authorization": f"Bearer {os.environ['WHATSTIG_API_KEY']}"},
    timeout=30,
)

class WhatsTigError(Exception):
    def __init__(self, status: int, code: str, message: str, request_id: str | None):
        super().__init__(f"{status} {code}: {message}")
        self.status, self.code, self.request_id = status, code, request_id

def call(method: str, path: str, *, json=None, params=None, idempotency_key: str | None = None, retries: int = 2):
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    safe = method == "GET" or idempotency_key is not None
    for attempt in range(retries + 1):
        try:
            r = client.request(method, path, json=json, params=params, headers=headers)
        except httpx.TransportError:
            if safe and attempt < retries:
                time.sleep(0.5 * 2 ** attempt)
                continue
            raise
        if r.status_code in (429, 502, 503, 504) and safe and attempt < retries:
            time.sleep(float(r.headers.get("Retry-After", 0.5 * 2 ** attempt)))
            continue
        if r.status_code == 204:
            return None
        body = r.json()
        if r.is_error:
            e = body.get("error", {})
            raise WhatsTigError(r.status_code, e.get("code", "ERROR"), e.get("message", ""), e.get("requestId"))
        return body

Instances and messages

def ensure_instance(tenant_id: str, name: str) -> dict:
    ref = f"tenant-{tenant_id}"
    found = call("GET", "/v1/instances", params={"externalReference": ref})["data"]
    if found:
        return found[0]
    return call("POST", "/v1/instances", json={"name": name, "externalReference": ref}, idempotency_key=f"create-{ref}")

inst = ensure_instance("101", "School A WhatsApp")
call("POST", f"/v1/instances/{inst['id']}/connect", json={"method": "qr"})
time.sleep(2)
qr = call("GET", f"/v1/instances/{inst['id']}/qr")  # qr["qrDataUrl"] -> <img src=...>

accepted = call(
    "POST", "/v1/messages",
    json={"instanceId": inst["id"], "recipient": "923001234567", "type": "text",
          "content": {"text": "Your child was marked absent today."}, "clientReference": "attendance-987"},
    idempotency_key="attendance-987",
)
print(accepted["id"], accepted["status"])  # QUEUED

Webhooks with FastAPI

import hashlib, hmac, json, os, time
from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()
SECRET = os.environ["WHATSTIG_WEBHOOK_SECRET"]

def verify_whatstig(raw_body: bytes, header: str | None, secret: str, tolerance: int = 300) -> dict:
    if not header:
        raise ValueError("missing signature")
    parts = dict(p.split("=", 1) for p in header.split(","))
    t = int(parts["t"])
    if abs(time.time() - t) > tolerance:
        raise ValueError("stale signature")
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, parts.get("v1", "")):
        raise ValueError("bad signature")
    return json.loads(raw_body)

@app.post("/whatstig/webhooks")
async def whatstig_webhook(request: Request, whatstig_signature: str | None = Header(default=None)):
    raw = await request.body()  # raw bytes, before JSON parsing
    try:
        event = verify_whatstig(raw, whatstig_signature, SECRET)
    except (ValueError, KeyError):
        raise HTTPException(status_code=401)
    # dedupe on event["id"], then hand off to a background task / queue
    if event["type"] == "message.received":
        print(event["externalReference"], event["data"]["sender"], event["data"]["content"])
    return {"ok": True}

The verify_whatstig helper is identical to the one in Webhooks and was checked against signatures produced by WhatsTIG's signer.