volga-sdk 1.2.1

Official Python SDK for the Volga Public API (conversations, messages, outbound webhooks).

volga

Official Python SDK for the Volga Public API — conversations, messages, and outbound webhooks.

• Zero dependencies (standard library only)
• Automatic retries (429 + 5xx + network) with Retry-After support
• Cursor auto-pagination via generators
• Built-in webhook signature verification
• Python 3.8+

Install

pip install volga-sdk

(The distribution is volga-sdk; you still import volga.)

Quickstart

import os
from volga import VolgaClient

volga = VolgaClient(api_key=os.environ["VOLGA_API_KEY"])

# List conversations
page = volga.conversations.list(channel="whatsapp", limit=25)

# Send a reply (idempotent retries)
import uuid
volga.messages.send(
    conversation_id=page["data"][0]["id"],
    text="Thanks for reaching out!",
    idempotency_key=str(uuid.uuid4()),
)

Pagination

# One page at a time
page = volga.messages.list(conversation_id="c_123")
print(page["data"], page["has_more"], page["next_cursor"])

# Or auto-paginate every item
for message in volga.messages.iterate(conversation_id="c_123"):
    print(message["id"], message["text"])

Webhooks

Register an endpoint (the signing secret is returned once):

endpoint = volga.webhook_endpoints.create(
    url="https://example.com/volga/webhooks",
    event_types=["message.received", "conversation.created"],
)
# store endpoint["secret"] securely

Verify and parse incoming deliveries (pass the raw request body):

from volga import construct_event, VolgaSignatureVerificationError

@app.post("/volga/webhooks")
def handle(request):
    try:
        event = construct_event(
            secret=os.environ["VOLGA_WEBHOOK_SECRET"],
            payload=request.get_data(),  # raw bytes, not parsed JSON
            signature_header=request.headers.get("Volga-Signature"),
        )
    except VolgaSignatureVerificationError:
        return "", 400
    # handle event["type"] / event["data"] ...
    return "", 200

Deliveries are at-least-once — deduplicate on the event id (or the Volga-Delivery-Id header).

WhatsApp templates

Manage each tenant's own WhatsApp message templates and send approved ones. WhatsApp is contact-initiated, so a business-initiated message must use an approved template — typically a UTILITY template for transactional content like receipts.

Scopes: listing/retrieving needs templates:read (granted by default); creating/deleting needs templates:write, which is opt-in — grant it to the key in the Volga dashboard (Settings → API keys).

End-to-end flow — create → wait for approval → send:

# 1) Create a template (it starts PENDING; Meta reviews it).
volga.templates.create(
    name="comprobante_pago",          # lowercase, digits, underscores
    language="es_AR",
    category="UTILITY",
    body_text="Hola {{1}}, tu comprobante por {{2}} está listo.",
    body_samples=["Ana", "$1.500"],   # one sample per {{n}} placeholder
)

# 2a) Poll its status…
tpl = volga.templates.retrieve("comprobante_pago", language="es_AR")
print(tpl["status"])  # PENDING | APPROVED | REJECTED | PAUSED | DISABLED

# 2b) …or (recommended) subscribe to the webhook and react to approval:
#     event_types=["whatsapp.template.status_updated"]
#     event["data"] -> {account_id, waba_id, name, language, status, reason?}

# 3) Once APPROVED, send it by opening a WhatsApp conversation:
volga.conversations.create(
    channel="whatsapp",
    phone="[phone]",
    template={
        "name": "comprobante_pago",
        "language": "es_AR",
        "components": [
            {"type": "body", "parameters": [
                {"type": "text", "text": "Ana"},
                {"type": "text", "text": "$1.500"},
            ]},
        ],
    },
)

List, retrieve and delete:

for t in volga.templates.iterate():
    print(t["name"], t["status"])

volga.templates.retrieve("comprobante_pago")
volga.templates.delete("comprobante_pago")

Pass account_id (a WhatsApp channel id) on any template call when the tenant has more than one connected WhatsApp account.

Errors

from volga import VolgaRateLimitError, VolgaPermissionError, VolgaApiError

try:
    volga.messages.send(conversation_id=cid, text=text)
except VolgaRateLimitError as e:
    time.sleep(e.retry_after_sec or 1)
except VolgaPermissionError:
    ...  # the key is missing a scope
except VolgaApiError as e:
    print(e.status, e.code, e.message, e.trace_id)

Class	Status
VolgaInvalidRequestError	400 / 422
VolgaAuthenticationError	401
VolgaPaymentRequiredError	402
VolgaPermissionError	403
VolgaNotFoundError	404
VolgaConflictError	409
VolgaRateLimitError	429
VolgaServerError	5xx
VolgaConnectionError / VolgaTimeoutError	no response

Configuration

VolgaClient(
    api_key="vk_live_…",
    base_url="https://hooks.volga-ai.com/v1",  # default
    timeout=30.0,        # default
    max_retries=2,       # default
    retry_base_delay=0.5,
    retry_max_delay=8.0,
)

License

MIT