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.