Metadata-Version: 2.4
Name: waba-sdk
Version: 1.1.0
Summary: Async Python SDK for the WhatsApp Business Cloud API
Author-email: Asim Mohamed <amohamed@aimsammi.org>
License: MIT License
        
        Copyright (c) 2026 Asim Mohamed
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Project-URL: Homepage, https://github.com/asimzz/waba-sdk
Project-URL: Source, https://github.com/asimzz/waba-sdk
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Framework :: AsyncIO
Classifier: Topic :: Communications :: Chat
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiohttp>=3.8.1
Requires-Dist: pydantic>=2.0
Requires-Dist: pydantic-settings>=2.0
Provides-Extra: test
Requires-Dist: pytest>=8.0; extra == "test"
Requires-Dist: pytest-asyncio>=0.23; extra == "test"
Requires-Dist: aioresponses>=0.7.6; extra == "test"
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100; extra == "fastapi"
Dynamic: license-file

# waba-sdk

An async Python SDK for the [WhatsApp Business Cloud API](https://developers.facebook.com/docs/whatsapp/cloud-api). Send messages, handle webhooks, upload and download media — all with typed pydantic models and an API designed for ergonomics.

```python
import asyncio
from waba_sdk import WhatsApp

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

asyncio.run(main())
```

## Table of contents

- [Installation](#installation)
- [Configuration](#configuration)
- [Quickstart](#quickstart)
- [Sending messages](#sending-messages)
  - [Text](#text)
  - [Media](#media)
  - [Location](#location)
  - [Contacts](#contacts)
  - [Reaction](#reaction)
  - [Reply (in-thread)](#reply-in-thread)
  - [Template](#template)
  - [Interactive — buttons](#interactive--buttons)
  - [Interactive — list](#interactive--list)
  - [Interactive — CTA URL](#interactive--cta-url)
  - [Interactive — flow](#interactive--flow)
  - [Interactive — products & catalog](#interactive--products--catalog)
  - [Interactive — location request](#interactive--location-request)
- [Mark as read](#mark-as-read)
- [Media (upload & download)](#media-upload--download)
- [Template management](#template-management)
- [Webhooks](#webhooks)
- [Webhook override](#webhook-override)
- [Error handling](#error-handling)
- [Development](#development)
- [License](#license)

---

## Installation

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

Requires Python **3.9+**.

For the optional FastAPI helper (`mount_webhook`):

```bash
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

```python
import asyncio
from waba_sdk import WhatsApp

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

asyncio.run(main())
```

Or construct directly without env vars:

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

Phone numbers accept any reasonable format (`+15551234567`, `+1 555 123 4567`, `+1-(555)-123-4567`) 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

```python
from waba_sdk import TextMessage

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

# Typed equivalent
await client.send(TextMessage(
    to="+15551234567",
    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.

```python
# By public URL
await client.send_image("+15551234567", url="https://example.com/cat.jpg", caption="hi")

# By previously uploaded media_id
await client.send_image("+15551234567", media_id="123456789012345")

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

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

Typed:

```python
from waba_sdk import ImageMessage

await client.send(ImageMessage(
    to="+15551234567",
    link="https://example.com/cat.jpg",
    caption="hi",
))
```

### Location

```python
await client.send_location(
    "+15551234567",
    latitude=37.7749,
    longitude=-122.4194,
    name="San Francisco",
    address="San Francisco, CA",
)
```

### Contacts

```python
from waba_sdk import Contact, ContactName, ContactPhone

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

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

### Reaction

```python
# React
await client.react("+15551234567", "wamid.HBg...", "🎉")

# Remove the reaction
await client.react("+15551234567", "wamid.HBg...", "")
```

### Reply (in-thread)

```python
# Sugar over send_text(..., reply_to=...)
await client.reply("+15551234567", "wamid.HBg...", "thanks!")

# Or any send_* method:
await client.send_image(
    "+15551234567",
    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`.

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

await client.send(TemplateMessage(
    to="+15551234567",
    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:

```python
await client.send_template("+15551234567", "hello_world")
```

### Interactive — buttons

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

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

Typed:

```python
from waba_sdk import ButtonsMessage, TextHeader

await client.send(ButtonsMessage(
    to="+15551234567",
    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.

```python
await client.send_list(
    "+15551234567",
    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

```python
await client.send_cta_url(
    "+15551234567",
    body="Your order is ready",
    button_text="Track shipment",
    url="https://example.com/track/123",
)
```

### Interactive — flow

```python
await client.send_flow(
    "+15551234567",
    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

```python
# Single product card
await client.send_single_product(
    "+15551234567",
    catalog_id="CATALOG_ID",
    product_retailer_id="SKU-123",
    body="Check this out",
)

# Multi-product list
await client.send_multi_product(
    "+15551234567",
    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(
    "+15551234567",
    body="Browse our catalog",
    thumbnail_product_retailer_id="SKU-1",
)
```

### Interactive — location request

```python
await client.send_location_request(
    "+15551234567",
    body="Where would you like delivery?",
)
```

---

## Mark as read

```python
# 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:

```python
# 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.

```python
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.

```python
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="+15551112222"),
        ]),
    ],
))

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`:

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

Named-parameter bodies are supported too:

```python
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

```python
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

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

### Edit a template

```python
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.

```python
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:

```python
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.

```python
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:

```python
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:

```python
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):

```python
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](#webhooks) above).

```python
# 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.

```python
from waba_sdk import RateLimitError, AuthenticationError, WhatsAppError

try:
    await client.send_text("+15551234567", "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

```bash
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:

```bash
uv add <package>
```

## License

MIT — see [pyproject.toml](pyproject.toml).
