waba-sdk 1.1.1

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)
• Template management
• Webhooks
• Webhook override
• Error handling
• Development
• License

---

Installation

uv add waba-sdk
# or with pip:
pip install waba-sdk

Requires Python 3.9+.

For the optional FastAPI helper (mount_webhook):

pip install "waba-sdk[fastapi]"

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	WABA-scoped calls	WABA ID. Required for client.templates.* / webhook_override.*.
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.

Template management

client.templates is a sub-client for managing template definitions on your WABA (create, list, edit, delete). This is distinct from client.send_template(...), which sends a message using an already-approved template.

Template management lives at the WhatsApp Business Account scope, so it needs the WABA ID — pass it to WhatsApp(...) or set WABA_ID in the environment. The token must carry the whatsapp_business_management permission.

client = WhatsApp(token="EAAG...", phone_number_id="...", waba_id="1234567890")
# or:
client = WhatsApp.from_env()   # reads WABA_ID from env / .env

Create a template

Components and buttons are typed pydantic models — discriminated unions validate the format / button-type combos before the request leaves your process.

from waba_sdk import (
    TemplateCreate, TemplateCategory, ParameterFormat,
    HeaderDefinition, HeaderFormat, BodyDefinition, FooterDefinition,
    ButtonsDefinition, QuickReplyButton, UrlButton, PhoneNumberButton,
)

resp = await client.templates.create(TemplateCreate(
    name="order_update",
    language="en_US",
    category=TemplateCategory.UTILITY,
    parameter_format=ParameterFormat.POSITIONAL,
    components=[
        HeaderDefinition(
            format=HeaderFormat.TEXT,
            text="Order {{1}}",
            example={"header_text": ["12345"]},
        ),
        BodyDefinition(
            text="Hi {{1}}, your order {{2}} has shipped.",
            example={"body_text": [["Ada", "12345"]]},
        ),
        FooterDefinition(text="Reply STOP to opt out."),
        ButtonsDefinition(buttons=[
            QuickReplyButton(text="Track order"),
            UrlButton(
                text="View",
                url="https://example.com/orders/{{1}}",
                example=["https://example.com/orders/12345"],
            ),
            PhoneNumberButton(text="Call us", phone_number="[phone]"),
        ]),
    ],
))

print(resp.id, resp.status)   # e.g. "1234567890123456" TemplateStatus.PENDING

Media-header templates use header_handle (from the resumable upload API) instead of header_text:

HeaderDefinition(format=HeaderFormat.IMAGE, example={"header_handle": ["4::aW...=="]})

Named-parameter bodies are supported too:

BodyDefinition(
    text="Hi {{first_name}}, welcome to {{brand}}.",
    example={"body_text_named_params": [
        {"param_name": "first_name", "example": "Ada"},
        {"param_name": "brand", "example": "Acme"},
    ]},
)

List, filter, paginate

from waba_sdk import TemplateStatus, TemplateCategory

resp = await client.templates.list(
    name="order_update",
    category=[TemplateCategory.UTILITY],
    status=[TemplateStatus.APPROVED, TemplateStatus.PENDING],
    language=["en_US", "fr_FR"],
    limit=50,
)
for tpl in resp.data:
    print(tpl.id, tpl.name, tpl.status)

# Pagination cursors are surfaced on the response
next_cursor = (resp.paging or {}).get("cursors", {}).get("after")
if next_cursor:
    resp = await client.templates.list(after=next_cursor)

Get a single template

tpl = await client.templates.get("1234567890123456")
print(tpl.name, tpl.status, tpl.category)

Edit a template

await client.templates.edit(
    "1234567890123456",
    components=[BodyDefinition(text="Updated copy {{1}}")],
    # category=TemplateCategory.MARKETING,
    # message_send_ttl_seconds=3600,
)

Delete a template

Pass template_id= to delete a specific language version, or name= to delete every language version. Exactly one is required.

await client.templates.delete(template_id="1234567890123456")
await client.templates.delete(name="order_update")

Authentication / OTP templates

OTP buttons cover all three Meta flows (COPY_CODE, ONE_TAP, ZERO_TAP). One-tap and zero-tap require Android app metadata — the model rejects the request at validation time if either is missing:

from waba_sdk import OtpButton, OtpType

OtpButton(otp_type=OtpType.COPY_CODE, text="Copy code")

OtpButton(
    otp_type=OtpType.ONE_TAP,
    text="Verify",
    autofill_text="Verify",
    package_name="com.example.app",
    signature_hash="K8a8s...",
)

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

Webhook override

client.webhook_override configures the per-WABA callback URL that Meta delivers messages to, overriding the App-level Dashboard webhook. It wraps the /{waba_id}/subscribed_apps resource — same WABA scope as client.templates, so it needs WABA_ID (and the token needs whatsapp_business_management).

The verify_token you set here must match the one your webhook handler validates against during Meta's GET handshake (see Webhooks above).

# Point this WABA at your own webhook
await client.webhook_override.set(
    "https://example.com/webhook",
    "my-verify-token",
)

# Inspect the current subscription(s)
resp = await client.webhook_override.get()
for app in resp.data:
    print(app.whatsapp_business_api_data.name, app.override_callback_uri)

# Unsubscribe the app entirely (also clears the override)
await client.webhook_override.delete()

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 (94 tests) covers wire-format equivalence per message type and template definition, validation rules (media media_id xor link, audio/sticker reject captions, location bounds, header format constraints, OTP one-tap requirements), webhook parsing, error mapping, and session lifecycle.

To add a new dependency:

uv add <package>

License

MIT — see pyproject.toml.