grubgenie-waba-sdk

Python SDK for the GrubGenie WABA middleware. Send WhatsApp messages, manage templates, handle media, and receive inbound webhooks.

Full integration guide (API routes, webhook format, migration from direct Meta API): docs/integration-guide.md

Install

pip install grubgenie-waba-sdk

With FastAPI webhook support:

pip install grubgenie-waba-sdk[fastapi]

Quick Start

import os
from waba_sdk import WabaClient

waba = WabaClient(
    base_url="https://dev-wabamiddleware.grubgenie.ai",
    service_key=os.environ["WABA_SERVICE_KEY"],
)

Messages

# Text message
await waba.messages.send_text("phone_number_id", "919999999999", "Hello!")

# Template message
await waba.messages.send_template(
    "phone_number_id", "919999999999",
    template_name="hello_world",
    language_code="en_US",
)

# Media message (image, video, document, audio)
await waba.messages.send_media(
    "phone_number_id", "919999999999",
    media_type="image",
    media_url="https://example.com/photo.jpg",
    caption="Check this out",
)

# Interactive message (buttons, lists)
await waba.messages.send_interactive("phone_number_id", "919999999999", {
    "type": "button",
    "body": {"text": "Pick one"},
    "action": {
        "buttons": [
            {"type": "reply", "reply": {"id": "yes", "title": "Yes"}},
            {"type": "reply", "reply": {"id": "no", "title": "No"}},
        ]
    },
})

# Mark message as read
await waba.messages.mark_as_read("phone_number_id", "wamid.xxx")

# Raw send (full control over body)
await waba.messages.send("phone_number_id", {
    "messaging_product": "whatsapp",
    "to": "919999999999",
    "type": "text",
    "text": {"body": "raw"},
})

Templates

WABA_ID = "your_waba_id"
PN_ID = "phone_number_id"

# List templates (single page)
page = await waba.templates.list(PN_ID, WABA_ID)

# List all templates (auto-paginates)
all_templates = await waba.templates.list_all(PN_ID, WABA_ID)

# Get template by name
template = await waba.templates.get(PN_ID, WABA_ID, "hello_world")

# Get template by ID
template = await waba.templates.get_by_id(PN_ID, WABA_ID, "123456789")

# Create template
await waba.templates.create(PN_ID, WABA_ID, {
    "name": "order_update",
    "language": "en_US",
    "category": "UTILITY",
    "components": [{"type": "BODY", "text": "Your order {{1}} is ready."}],
})

# Update template
await waba.templates.update(PN_ID, WABA_ID, "123456789", {
    "components": [{"type": "BODY", "text": "Order {{1}} updated."}],
})

# Delete template
await waba.templates.delete(PN_ID, WABA_ID, "order_update")

Media

# Upload
result = await waba.media.upload("phone_number_id", file_bytes, "photo.png", "image/png")
media_id = result["id"]

# Download
data = await waba.media.download(media_id, "phone_number_id")

Phone Numbers

# List phone numbers for a WABA
numbers = await waba.phone_numbers.list("phone_number_id", "waba_id")

# Register a phone number
await waba.phone_numbers.register("phone_number_id", pin="123456")

Timeouts and errors

client = WabaClient(base_url, service_key, timeout=15.0)   # seconds; default 30, None disables (httpx default is 5 s)

await client.messages.send_text(pn_id, to, "hi", idempotency_key=job_id, timeout=30.0)   # per-call override
await client.accounts.unsubscribe(waba_id=waba_id, timeout=5.0)

• Non-2xx raises WabaApiError (a RuntimeError): status_code, code, outcome, body, retry_after_ms.
• A timeout raises WabaTimeoutError (a TimeoutError), not WabaApiError. For a send the outcome is unknown, so retry with the same idempotency key. httpx applies the value per phase (connect, write, read), not as a total deadline.
• templates.list_all pages by cursor through the middleware and raises if any page fails; timeout applies per page.

Cloud API Proxy

Direct access to any WhatsApp Cloud API endpoint through the middleware:

response = await waba.proxy("phone_number_id", "GET", "{waba_id}/phone_numbers")
print(response.json())

Inbound Webhooks (FastAPI)

from fastapi import FastAPI, Request
from waba_sdk import WabaWebhookHandler

app = FastAPI()

async def on_event(event):
    context = event["context"]  # { partnerId, phoneNumberId, guest? }
    raw = event["raw"]          # original Meta webhook body
    # guest is present only when the sender matched a Guest for the partner
    print(f"Message from partner {context['partnerId']}, guest: {context.get('guest')}")

handler = WabaWebhookHandler(
    secret=os.environ["WABA_WEBHOOK_SECRET"],
    on_event=on_event,
)

@app.post("/webhooks/waba")
async def waba_webhook(request: Request):
    return await handler.handle(request)

WabaWebhookHandler raises ValueError on an empty secret and takes an optional on_error callback (the response is still a plain 500).

Signature Verification

from waba_sdk import verify_signature

is_valid = verify_signature(payload_bytes, signature_header, secret)

Requirements

• Python >= 3.10
• httpx >= 0.27
• fastapi >= 0.115 (optional, for webhook handler)