waotomatis 0.3.0

Official Python SDK for WAOtomatis — headless WhatsApp (WABA Cloud API).

waotomatis

Official Python SDK for WAOtomatis — headless WhatsApp infrastructure on the WhatsApp Business Platform (WABA Cloud API). Send messages, upload media, list chats and contacts, register webhooks, and verify webhook signatures.

• Zero dependencies. Pure standard library (urllib, hmac, hashlib).
• Python 3.8+. Type-hinted, py.typed.
• Idiomatic. snake_case methods, keyword arguments, a typed error hierarchy.

Install

pip install waotomatis

Quickstart

import os
from waotomatis import Waotomatis

wao = Waotomatis(api_key=os.environ["WAO_API_KEY"])

msg = wao.sessions("sess_123").messages.send(
    to="628123456789",
    type="text",
    text="Halo dari WAOtomatis 👋",
)

print(msg["id"])  # msg_abc123

Waotomatis defaults to https://api.waotomatis.com; pass base_url=... to override. The API key is sent as Authorization: Bearer <api_key>.

Sending messages

session = wao.sessions("sess_123")

# Text (optionally with a link preview)
session.messages.send(to="628123456789", type="text", text="Hi", preview_url=True)

# Image by public link, or by an uploaded media id
session.messages.send(to="628...", type="image", link="https://example.com/a.jpg",
                      caption="Hello")
session.messages.send(to="628...", type="image", media_id="media_abc")

# Document with a filename
session.messages.send(to="628...", type="document", media_id="media_abc",
                      file_name="invoice.pdf")

# Audio as a voice note
session.messages.send(to="628...", type="audio", media_id="media_abc", voice=True)

# Idempotent send (safe to retry — the same key returns the original result)
session.messages.send(to="628...", type="text", text="hi",
                      idempotency_key="order-42")

# Mark an inbound message read by its provider wamid
session.messages.mark_read("wamid.HBg...")

Interactive messages

Typed helpers assemble the interactive payload for you — no hand-built dicts. Each accepts the same optional header_text, footer_text, reply_to, and idempotency_key as send.

session = wao.sessions("sess_123")

# Reply buttons (max 3) — pass (id, title) tuples or {"id", "title"} dicts
session.messages.send_buttons(
    to="628...",
    body_text="Confirm your order?",
    buttons=[("yes", "Yes"), ("no", "No")],
    footer_text="WAOtomatis",
)

# List menu
session.messages.send_list(
    to="628...",
    body_text="Pick a service",
    list_button="View menu",
    sections=[
        {"title": "Services", "rows": [
            {"id": "svc_1", "title": "Consultation", "description": "30 min"},
            {"id": "svc_2", "title": "Repair"},
        ]},
    ],
)

# Call-to-action URL button
session.messages.send_cta_url(
    to="628...",
    body_text="See our full catalog",
    cta_display_text="Open catalog",
    cta_url="https://example.com/catalog",
)

There are also send_flow(...) (WhatsApp Flows), send_product(...), and send_product_list(...) (catalog commerce) — same shape, snake_case in.

Media

session = wao.sessions("sess_123")

# Upload raw bytes
with open("photo.jpg", "rb") as f:
    res = session.media.upload(f.read(), file_name="photo.jpg", mime_type="image/jpeg")
print(res["mediaId"])

# Or upload a local file by path
res = session.media.upload_file("photo.jpg", mime_type="image/jpeg")

# Or upload by URL
res = session.media.upload_from_url("https://example.com/photo.jpg")

# Download inbound media bytes
data, mime_type = session.media.download("media_abc")

Sessions, chats, and contacts

# List sessions (one page)
page = wao.list_sessions()
for s in page:
    print(s["id"], s["status"])

session = wao.sessions("sess_123")
session.get()
session.delete()  # disconnect

# Chats and contacts auto-paginate — iterate every item across all pages
for chat in session.chats.list():
    print(chat["chatId"], chat.get("lastText"))

for message in session.chats.history("628123456789"):
    print(message["direction"], message["type"])

for contact in session.contacts.list():
    print(contact["waId"], contact.get("name"))

# Or grab just one page
first = session.contacts.list(limit=50).first_page()
print(len(first), first.has_more, first.cursor)

contact = session.contacts.get("628123456789")

Webhooks

Register a webhook — the signing secret is returned once:

hook = wao.sessions("sess_123").webhooks.create(
    url="https://example.com/webhook",
    events=["message.received", "message.updated", "session.status"],
)
secret = hook["secret"]  # store this

Verify and parse incoming deliveries. The server signs the exact raw request body with HMAC-SHA256 and sends it in the X-Wao-Signature: sha256=<hex> header. Verify against the raw bytes you received — never a re-serialized object:

from waotomatis import construct_event, verify_webhook, WaotomatisError

# Just verify
ok = verify_webhook(raw_body, request.headers.get("X-Wao-Signature"), secret)

# Verify + parse (raises on a bad signature or unparseable body)
try:
    event = construct_event(raw_body, request.headers.get("X-Wao-Signature"), secret)
except WaotomatisError:
    return ("", 401)

if event["event"] == "message.received":
    print(event["data"]["text"])

Flask example

import os
from flask import Flask, request, abort
from waotomatis import construct_event, WaotomatisError, WEBHOOK_SIGNATURE_HEADER

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["WAO_WEBHOOK_SECRET"]

@app.post("/webhook")
def webhook():
    try:
        event = construct_event(
            request.get_data(),  # raw bytes
            request.headers.get(WEBHOOK_SIGNATURE_HEADER),
            WEBHOOK_SECRET,
        )
    except WaotomatisError:
        abort(401)
    # handle event...
    return ("", 200)

Errors

Every failure raises a subclass of WaotomatisError, carrying the stable code, message, request_id, and HTTP status from the server's uniform error model ({"error": {"code", "message", "requestId"}}).

from waotomatis import (
    Waotomatis, WaotomatisError,
    AuthenticationError, PermissionError, NotFoundError,
    ValidationError, RateLimitError, ApiError,
    ConnectionError, TimeoutError,
)
import time

try:
    wao.sessions("sess_123").messages.send(to="628...", type="text", text="hi")
except RateLimitError as e:
    time.sleep(e.retry_after or 1)
except WaotomatisError as e:
    if e.code == "session_disconnected":
        ...
    print(e.code, e.status, e.request_id)

Exception	HTTP
AuthenticationError	401
PermissionError	403
NotFoundError	404
TimeoutError	408 / local
ValidationError	409 / 422
RateLimitError	429
ApiError	5xx
ConnectionError	network

Transient failures (408/429/5xx/network) on idempotent verbs — or any call given an idempotency_key — are retried automatically with exponential backoff and jitter, honoring Retry-After. Tune with max_retries= and timeout= on the constructor.

License

MIT