sage-messaging (Python)

SDK officiel de l'API Sage Messaging : envoi et lecture de messages WhatsApp et SMS, vérification des webhooks.

pip install sage-messaging

Python 3.9+. Client synchrone et asynchrone.

Démarrage

from sage_messaging import SageMessaging

client = SageMessaging(api_key="sk_...")  # ou variable SAGE_MESSAGING_API_KEY

# WhatsApp (canal par défaut)
client.messages.send(phone="+22997000000", message="Bonjour !")

# SMS
client.sms.send(phone="+22997000000", message="Votre code : 4821")

Asynchrone

from sage_messaging import AsyncSageMessaging

async with AsyncSageMessaging() as client:
    await client.messages.send(phone="+22997000000", message="Bonjour !")

Les méthodes sont identiques ; il suffit de les await.

Configuration

Les lectures (GET) sont rejouées en cas d'erreur réseau, de 429 ou de 5xx. Les envois ne sont jamais rejoués : un envoi dont la réponse s'est perdue a pu partir, et le rejouer enverrait le message en double.

Envoi

# Message texte, WhatsApp ou SMS
client.messages.send(phone="+229...", message="Salut", channel="sms")

# SMS différé, depuis un téléphone précis
from datetime import datetime, timedelta, timezone
client.sms.send(
    phone="+229...",
    message="Rappel : RDV demain",
    from_="+229...",                          # voir client.sms.phones()
    send_at=datetime.now(timezone.utc) + timedelta(hours=2),  # 20 jours max
)

# Campagne SMS (1 à 100 destinataires)
client.sms.bulk_send(phones=["+229...", "+229..."], message="Promo -20 %")

# Médias WhatsApp (1 à 10, 16 Mo max chacun), envoyés dans l'ordre
res = client.messages.send_media(
    phone="+229...",
    caption="Votre facture",
    media=[
        {"path": "facture.pdf"},                             # fichier local
        {"url": "https://exemple.com/photo.jpg"},            # URL publique
        {"data": image_bytes, "filename": "capture.png"},    # octets
    ],
)
if res["failed"]:  # succès partiel (HTTP 207)
    print("Échecs :", res["failed"])

Utilisez des datetime avec fuseau horaire pour send_at et since.

Lecture

client.conversations.list(unread_only=True, limit=20)
client.conversations.messages(42, direction="received")
client.conversations.mark_read(42)

client.messages.list(after_id=1000)  # flux global, id croissant

Les réponses paginées ont la forme {"data": [...], "meta": {"current_page", "per_page", "total", "last_page"}}.

Suivre les nouveaux messages sans webhook

for msg in client.messages.poll(direction="received", interval=5):
    print(msg["contact_phone"], msg["body"])

Sans after_id, seuls les messages arrivés après l'appel sont renvoyés. Pour reprendre après un redémarrage, conservez le dernier msg["id"] et passez-le en after_id.

Webhooks

Chaque webhook est signé en HMAC-SHA256 (X-Signature, X-Timestamp). Passez le corps brut de la requête, pas un JSON re-sérialisé.

from sage_messaging import verify_webhook, WebhookVerificationError

# FastAPI
@app.post("/webhooks/sage")
async def sage_webhook(request: Request):
    try:
        event = verify_webhook(await request.body(), request.headers, SECRET)
    except WebhookVerificationError:
        raise HTTPException(400)

    if event["event"] == "message.received":
        msg = event["data"]["message"]
        contact = event["data"]["conversation"]
        ...
    return {"ok": True}

Django : verify_webhook(request.body, request.headers, SECRET) ; Flask : verify_webhook(request.get_data(), request.headers, SECRET).

Les requêtes de plus de 5 minutes sont refusées (tolerance=300) pour empêcher leur rejeu.

Erreurs

from sage_messaging import ValidationError, GatewayError, SageMessagingError

try:
    client.sms.send(phone="+229...", message="...")
except ValidationError as e:     # 422 : champ invalide, canal non provisionné…
    print(e.errors, e.channel)
except GatewayError as e:        # 502 : la passerelle a refusé l'envoi
    print(e.detail)
except SageMessagingError as e:  # toutes les autres erreurs du SDK
    print(e)

Permissions des clés API