waba-sdk 1.0.0

Async Python SDK for the WhatsApp Business Cloud API

waba-sdk

An async Python SDK for the WhatsApp Business Cloud API. Send messages, handle webhooks, upload and download media — all with typed pydantic models and an API designed for ergonomics.

import asyncio
from waba_sdk import WhatsApp

async def main():
    async with WhatsApp.from_env() as client:
        await client.send_text("[phone]", "Hello from waba-sdk!")

asyncio.run(main())

Table of contents

• Installation
• Configuration
• Quickstart
• Sending messages
  • Text
  • Media
  • Location
  • Contacts
  • Reaction
  • Reply (in-thread)
  • Template
  • Interactive — buttons
  • Interactive — list
  • Interactive — CTA URL
  • Interactive — flow
  • Interactive — products & catalog
  • Interactive — location request
• Mark as read
• Media (upload & download)
• Webhooks
• Error handling
• Development
• License

---

Installation

uv add git+https://github.com/asimzz/waba-sdk.git
# or with pip:
pip install git+https://github.com/asimzz/waba-sdk.git

Requires Python 3.9+.

For the optional FastAPI helper (mount_webhook):

pip install "waba-sdk[fastapi] @ git+https://github.com/asimzz/waba-sdk.git"

Configuration

The SDK reads credentials from environment variables (or a .env in the working directory) when you call WhatsApp.from_env(). Direct construction (WhatsApp(token=..., phone_number_id=...)) doesn't read env vars at all.

Variable	Required	Description
WABA_ACCESS_TOKEN	yes (for from_env())	Permanent or system-user access token.
WABA_NUMBER_ID	yes (for from_env())	WhatsApp phone number ID from the Meta dashboard.
WABA_API_VERSION	no	Graph API version. Defaults to v21.0.
WABA_BASE_URL	no	Base URL. Defaults to https://graph.facebook.com.
WABA_TIMEOUT	no	HTTP timeout in seconds. Defaults to 30.0.
WABA_MAX_RETRIES	no	Max retries on 429/5xx. Defaults to 2.
WABA_ID	no	WhatsApp Business Account (WABA) ID.
WABA_BUSINESS_ID	no	Facebook Business Manager ID.
FACEBOOK_VERIFY_TOKEN	webhook only	Token echoed during the Meta webhook verification handshake.

The composed Graph base URL is ${WABA_BASE_URL}/${WABA_API_VERSION} — bumping the API version no longer requires a code change to the SDK.

Quickstart

import asyncio
from waba_sdk import WhatsApp

async def main():
    async with WhatsApp.from_env() as client:
        await client.send_text("[phone]", "Hello from waba-sdk!")
        await client.send_image(
            "[phone]",
            url="https://example.com/cat.jpg",
            caption="A very good cat",
        )
        await client.send_buttons(
            "[phone]",
            body="Did this help?",
            buttons=[("yes", "Yes"), ("no", "No")],
        )

asyncio.run(main())

Or construct directly without env vars:

client = WhatsApp(token="EAAG...", phone_number_id="1234567890")

Phone numbers accept any reasonable format ([phone], [phone], [phone]) and are normalized to digits-only internally.

---

Sending messages

Two equivalent styles for every message type:

• Convenience methods on the client — best for the common case.
• Typed messages passed to client.send(message) — best when you need every field, when you build messages elsewhere, or when you reuse them.

Each client.send_* helper accepts an optional reply_to=<message_id> keyword to send the message in-thread.

Text

from waba_sdk import TextMessage

# Convenience
await client.send_text("[phone]", "https://example.com check this out", preview_url=True)

# Typed equivalent
await client.send(TextMessage(
    to="[phone]",
    body="https://example.com check this out",
    preview_url=True,
))

Media

One method per media type. For each call, supply exactly one of url= or media_id= — the SDK enforces this at validation time.

# By public URL
await client.send_image("[phone]", url="https://example.com/cat.jpg", caption="hi")

# By previously uploaded media_id
await client.send_image("[phone]", media_id="123456789012345")

await client.send_video("[phone]", url="https://example.com/clip.mp4", caption="watch")
await client.send_audio("[phone]", url="https://example.com/voice.mp3")
await client.send_document(
    "[phone]",
    url="https://example.com/invoice.pdf",
    filename="invoice.pdf",
    caption="your invoice",
)
await client.send_sticker("[phone]", media_id="987654321")

AudioMessage and StickerMessage reject caption (Graph API does too).

Typed:

from waba_sdk import ImageMessage

await client.send(ImageMessage(
    to="[phone]",
    link="https://example.com/cat.jpg",
    caption="hi",
))

Location

await client.send_location(
    "[phone]",
    latitude=37.7749,
    longitude=-122.4194,
    name="San Francisco",
    address="San Francisco, CA",
)

Contacts

from waba_sdk import Contact, ContactName, ContactPhone

await client.send_contacts(
    "[phone]",
    contacts=[
        Contact(
            name=ContactName(formatted_name="Ada Lovelace"),
            phones=[ContactPhone(phone="[phone]", type="WORK", wa_id="15551112222")],
        )
    ],
)

send_contacts also accepts plain dict objects — they're validated into Contact models for you.

Reaction

# React
await client.react("[phone]", "wamid.HBg...", "🎉")

# Remove the reaction
await client.react("[phone]", "wamid.HBg...", "")

Reply (in-thread)

# Sugar over send_text(..., reply_to=...)
await client.reply("[phone]", "wamid.HBg...", "thanks!")

# Or any send_* method:
await client.send_image(
    "[phone]",
    url="https://example.com/yes.png",
    reply_to="wamid.HBg...",
)

Template

A single TemplateMessage covers every shape. Parameters are a discriminated union — use the right subclass per parameter type instead of leaving five fields as None.

from waba_sdk import (
    TemplateMessage, BodyComponent, HeaderComponent, ButtonComponent,
    TextParameter, CurrencyParameter, CurrencyValue, ImageParameter,
    ButtonParameter,
)

await client.send(TemplateMessage(
    to="[phone]",
    name="order_confirmation",
    language="en_US",
    components=[
        HeaderComponent(parameters=[
            ImageParameter(link="https://example.com/order-header.jpg"),
        ]),
        BodyComponent(parameters=[
            TextParameter(text="Ada"),
            TextParameter(text="#A1024"),
            CurrencyParameter(currency=CurrencyValue(
                fallback_value="$29.00", code="USD", amount_1000=29000,
            )),
        ]),
        ButtonComponent(
            sub_type="quick_reply",
            index=0,
            parameters=[ButtonParameter(type="payload", payload="track_A1024")],
        ),
    ],
))

For simple templates, the convenience method is shorter:

await client.send_template("[phone]", "hello_world")

Interactive — buttons

buttons accepts a list of Button(...) models, ("id", "title") tuples, or {"id": ..., "title": ...} dicts.

await client.send_buttons(
    "[phone]",
    body="Did this help?",
    buttons=[("yes", "Yes"), ("no", "No")],
    header="Quick check",     # str → text header shortcut
    footer="You can change this later",
)

Typed:

from waba_sdk import ButtonsMessage, TextHeader

await client.send(ButtonsMessage(
    to="[phone]",
    body="Did this help?",
    buttons=[("yes", "Yes"), ("no", "No")],
    header=TextHeader(text="Quick check"),
    footer="You can change this later",
))

Interactive — list

sections accepts dicts or ListSection models; rows accept dicts or ListRow models.

await client.send_list(
    "[phone]",
    body="Pick a plan",
    button_text="View plans",
    sections=[
        {
            "title": "Monthly",
            "rows": [
                {"id": "basic", "title": "Basic", "description": "$9/mo"},
                {"id": "pro", "title": "Pro", "description": "$29/mo"},
            ],
        },
    ],
)

Interactive — CTA URL

await client.send_cta_url(
    "[phone]",
    body="Your order is ready",
    button_text="Track shipment",
    url="https://example.com/track/123",
)

Interactive — flow

await client.send_flow(
    "[phone]",
    body="Complete your profile",
    flow_id="FLOW_ID",
    flow_cta="Start",
    flow_action="navigate",
    mode="published",
    screen="WELCOME",
    data={"user_id": "42"},
)

flow_token is optional; pass it when Meta requires correlation between flow runs.

Interactive — products & catalog

# Single product card
await client.send_single_product(
    "[phone]",
    catalog_id="CATALOG_ID",
    product_retailer_id="SKU-123",
    body="Check this out",
)

# Multi-product list
await client.send_multi_product(
    "[phone]",
    body="Featured items",
    catalog_id="CATALOG_ID",
    sections=[
        {"title": "Best sellers", "product_retailer_ids": ["SKU-1", "SKU-2", "SKU-3"]},
    ],
)

# Full catalog
await client.send_catalog(
    "[phone]",
    body="Browse our catalog",
    thumbnail_product_retailer_id="SKU-1",
)

Interactive — location request

await client.send_location_request(
    "[phone]",
    body="Where would you like delivery?",
)

---

Mark as read

# Just mark read
await client.mark_read("wamid.HBg...")

# Mark read + show typing indicator
await client.mark_read("wamid.HBg...", typing=True)

Media (upload & download)

client.media exposes three helpers:

# Upload from a file path or raw bytes
media_id = await client.media.upload("photo.jpg")  # mime guessed from filename
media_id = await client.media.upload(open("voice.ogg", "rb").read(), mime_type="audio/ogg")

# Resolve a media_id (e.g. from an inbound webhook) to a CDN URL + metadata
info = await client.media.get_url(media_id)
# info.url, info.mime_type, info.sha256, info.file_size

# Download — accepts a media_id or a direct CDN URL
download = await client.media.download(media_id)
# or:
download = await client.media.download(info.url)
with open("photo.jpg", "wb") as f:
    f.write(download.content)
print(download.content_type)   # "image/jpeg"

MediaInfo is a typed pydantic model; MediaDownload is a frozen dataclass with .content: bytes and .content_type: str.

Webhooks

WebhookHandler is framework-agnostic. It exposes three methods:

• verify(query) — validates Meta's GET handshake. Accepts a dict, query-string, or iterable of (k, v) pairs.
• parse(payload) — pure: validates the envelope and returns a list of typed events (MessageEvent for inbound messages, StatusEvent for sent/delivered/read/failed updates).
• handle(payload, *, auto_mark_read=False) — calls on_message/on_status/on_error callbacks. Optionally marks each inbound message as read.

from fastapi import FastAPI, Request
from waba_sdk import WhatsApp
from waba_sdk.webhook import (
    WebhookHandler, MessageEvent, StatusEvent, IncomingTextMessage,
)

app = FastAPI()
client = WhatsApp.from_env()

async def on_message(event: MessageEvent) -> None:
    msg = event.message
    if isinstance(msg, IncomingTextMessage):
        await client.send_text(event.contact_wa_id, f"You said: {msg.text.body}")

async def on_status(event: StatusEvent) -> None:
    print(event.status.id, event.status.status)  # delivered/read/failed

handler = WebhookHandler(
    verify_token="<FACEBOOK_VERIFY_TOKEN value>",
    client=client,
    on_message=on_message,
    on_status=on_status,
)

@app.get("/webhook")
async def verify(request: Request):
    challenge = handler.verify(dict(request.query_params))
    return challenge if challenge else ("forbidden", 403)

@app.post("/webhook")
async def receive(request: Request):
    body = await request.json()
    await handler.handle(body, auto_mark_read=True)
    return {"ok": True}

StatusEvent lets you track delivery / read receipts. The handler emits one event per status update, independently of inbound messages — you'll receive these even when a webhook payload contains no new messages.

FastAPI shortcut

If you're on FastAPI, the same wiring is one call:

from waba_sdk import WhatsApp
from waba_sdk.webhook import MessageEvent, IncomingTextMessage
from waba_sdk.integrations.fastapi import mount_webhook
from fastapi import FastAPI

app = FastAPI()
client = WhatsApp.from_env()

async def on_message(event: MessageEvent) -> None:
    if isinstance(event.message, IncomingTextMessage):
        await client.send_text(event.contact_wa_id, "got it")

mount_webhook(
    app,
    "/webhook",
    client=client,
    verify_token="<FACEBOOK_VERIFY_TOKEN value>",
    on_message=on_message,
    auto_mark_read=True,
)

mount_webhook registers a shutdown hook so the underlying aiohttp session is closed cleanly when FastAPI tears down. Install with pip install waba-sdk[fastapi].

Inbound message types

event.message is a discriminated union; isinstance checks narrow the type:

from waba_sdk.webhook import (
    IncomingTextMessage, IncomingImageMessage, IncomingAudioMessage,
    IncomingVideoMessage, IncomingDocumentMessage, IncomingStickerMessage,
    IncomingLocationMessage, IncomingContactMessage, IncomingReactionMessage,
    IncomingInteractiveMessage, IncomingButtonMessage, IncomingOrderMessage,
    IncomingSystemMessage, IncomingUnknownMessage, IncomingUnsupportedMessage,
)

For interactive replies (button click, list pick, flow response):

from waba_sdk.webhook import ButtonReply, ListReply, NFMReply

if isinstance(event.message, IncomingInteractiveMessage):
    reply = event.message.interactive
    if isinstance(reply, ButtonReply):
        button_id = reply.button_reply.id
    elif isinstance(reply, ListReply):
        row_id = reply.list_reply.id
    elif isinstance(reply, NFMReply):
        flow_payload = reply.nfm_reply.response_json   # already JSON-decoded

Error handling

Every non-2xx response from Graph maps to a typed exception:

Status	Exception	Notes
401	AuthenticationError	Bad / expired token.
429	RateLimitError	.retry_after in seconds (from Retry-After).
4xx	InvalidRequestError	Anything else 4xx (validation, missing field).
5xx	ServerError	Graph API instability.
—	MediaError	Raised by client.media.* on upload/download.
—	WebhookVerificationError	Reserved for handshake failures.
—	WhatsAppError	Base class. All other errors inherit from this.

Every exception carries .status_code, .error_code (Meta's error.code), .error_subcode, .fbtrace_id, and the raw .response dict — so you can log a complete picture without re-parsing.

from waba_sdk import RateLimitError, AuthenticationError, WhatsAppError

try:
    await client.send_text("[phone]", "hi")
except RateLimitError as e:
    print(f"rate limited; retry in {e.retry_after}s; trace: {e.fbtrace_id}")
except AuthenticationError:
    print("token is invalid or expired")
except WhatsAppError as e:
    print(f"send failed [{e.status_code}/{e.error_code}]: {e.message}")

The HTTP layer automatically retries on 429 and 5xx with exponential backoff and jitter, honoring any Retry-After header. Configure via WhatsApp(max_retries=...) (default 2) or WABA_MAX_RETRIES.

Development

git clone https://github.com/asimzz/waba-sdk.git
cd waba-sdk
uv sync --extra test     # install + test deps
uv run pytest -q         # run the test suite

The test suite (66 tests) covers wire-format equivalence per message type, validation rules (media media_id xor link, audio/sticker reject captions, location bounds), webhook parsing, error mapping, and session lifecycle.

To add a new dependency:

uv add <package>

License

MIT — see pyproject.toml.