Envoyer un SMS en Python : API REST, Django et FastAPI

import os

HEADERS = {
    "X-APP-ID":  os.environ["CONEXTEO_APP_ID"],
    "X-API-KEY": os.environ["CONEXTEO_API_KEY"],
}

Le piège qui coûte le plus d'heures : intervertir les deux valeurs. L'API répond alors 401 Unauthorized — exactement le même code que si la clé était fausse ou révoquée. On cherche donc du côté de la clé, on en régénère une, et le problème persiste. Le moyen mnémotechnique : APP identifie l'application, KEY est le secret. Si vous obtenez un 401 avec des identifiants fraîchement copiés, échangez les deux en-têtes avant toute autre hypothèse.

import requests

response = requests.post(
    "https://api.conexteo.com/messages/sms",
    headers=HEADERS,
    json={
        "content": "Votre code de connexion : 847291",
        "sender": "MaBoutique",
        "recipients": ["+33612345678"],
    },
    timeout=10,
)

print(response.status_code, response.json())
# 200 {'success': True, 'message': 'message created',
#      'message_id': 123456, 'credits_used': 1}

Le paramètre est json=, pas data=. C'est l'erreur la plus fréquente sur cette première requête. Avec data={...}, requests encode le corps en formulaire au lieu de JSON, et l'API répond 400 sans que rien n'indique la cause dans votre code. Le paramètre json= sérialise correctement et pose l'en-tête Content-Type tout seul.

Second détail : le timeout. Sans lui, requests attend indéfiniment. Sur un envoi déclenché depuis une requête web, cela suffit à bloquer un worker.

import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

BASE = "https://api.conexteo.com"


def build_session() -> requests.Session:
    session = requests.Session()
    session.headers.update({
        "X-APP-ID":  os.environ["CONEXTEO_APP_ID"],
        "X-API-KEY": os.environ["CONEXTEO_API_KEY"],
    })
    # On ne rejoue que les erreurs serveur et réseau, jamais un 4xx.
    retry = Retry(total=3, backoff_factor=1,
                  status_forcelist=[500, 502, 503, 504],
                  allowed_methods=["POST"])
    session.mount("https://", HTTPAdapter(max_retries=retry))
    return session


SESSION = build_session()


def send_sms(recipients: list[str], content: str,
             external_id: str | None = None) -> dict | None:
    payload = {
        "content": content,
        "sender": os.environ.get("CONEXTEO_SENDER", "MaBoutique"),
        "recipients": recipients,
    }
    if external_id:
        payload["external_id"] = external_id

    response = SESSION.post(f"{BASE}/messages/sms", json=payload, timeout=10)

    # 409 = message déjà soumis avec cet external_id. Ce n'est pas une erreur.
    if response.status_code == 409:
        return None

    response.raise_for_status()
    return response.json()

Le 409 n'est pas une panne, c'est une protection — et raise_for_status() le traite à l'envers. Le champ external_id est votre identifiant de déduplication. Si un message a déjà été soumis avec la même valeur, l'API répond 409 Conflict — « External ID already processed » — sans renvoyer le message d'origine.

Or raise_for_status() lève une exception sur un 409. Le scénario qui piège : votre appel expire, votre tâche est rejouée, l'API répond 409, l'exception remonte, et vous concluez que le client n'a rien reçu. Il a reçu le message. D'où le test explicite avant l'appel. Sans external_id, le même incident produit deux SMS facturés au même destinataire.

# Django + Celery
@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def send_order_sms(self, order_id: int) -> None:
    order = Order.objects.get(pk=order_id)
    try:
        send_sms(
            [order.phone],
            f"Votre commande #{order.id} vient d'être expédiée.",
            external_id=f"order-{order.id}-shipped",
        )
    except requests.HTTPError as exc:
        # Le rejeu est sûr : l'external_id empêche le doublon.
        raise self.retry(exc=exc)


# FastAPI, pour un volume modéré
@app.post("/orders/{order_id}/ship")
async def ship(order_id: int, background: BackgroundTasks):
    background.add_task(send_order_sms_sync, order_id)
    return {"status": "ok"}

from datetime import datetime
from zoneinfo import ZoneInfo

# Un rappel la veille à 10 h, heure de Paris.
local = datetime(2026, 9, 15, 10, 0, tzinfo=ZoneInfo("Europe/Paris"))
utc   = local.astimezone(ZoneInfo("UTC"))

payload = {
    "content": "Rappel : votre rendez-vous est demain à 14h30.",
    "sender": "MonCabinet",
    "recipients": [phone],
    "external_id": f"rdv-{appointment_id}-j1",
    "scheduleAt": utc.strftime("%Y-%m-%dT%H:%M:%SZ"),
}

Deux façons de se tromper sur ce champ, et Python les offre toutes les deux. datetime.now() renvoie une heure locale sans fuseau : la campagne part avec une à deux heures d'avance selon la saison, et une campagne prévue à 8 h partirait à 6 h — hors des plages autorisées pour un message promotionnel. Et .isoformat() sur un objet conscient du fuseau produit +00:00 et non le Z attendu : d'où le strftime explicite ci-dessus.

import hmac, hashlib, json, os
from fastapi import FastAPI, Request, Response, HTTPException

app = FastAPI()
SECRET = os.environ["CONEXTEO_WEBHOOK_SECRET"].encode()   # whsec_…
# Le nom exact de l'en-tête de signature figure dans votre espace
# Mon Compte / API / Webhook, à côté du secret.
SIGNATURE_HEADER = os.environ["CONEXTEO_SIGNATURE_HEADER"]


@app.post("/webhooks/conexteo")
async def conexteo_webhook(request: Request):
    raw = await request.body()          # corps brut, avant tout parsing

    attendue = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    recue    = request.headers.get(SIGNATURE_HEADER, "")

    if not hmac.compare_digest(attendue, recue):
        raise HTTPException(status_code=401)

    # Idempotence : le même identifiant revient à chaque tentative.
    event_id = request.headers.get("X-Conexteo-Webhook-Id")
    if event_id and already_processed(event_id):
        return Response(status_code=204)

    event = json.loads(raw)
    enqueue(event)                      # traitement lourd hors requête

    return Response(status_code=204)    # 204 = livré

Toute réponse 2xx vaut accusé de réception. Tout autre code, un timeout ou une erreur réseau compte comme une tentative en échec. Répondez donc avant de faire le travail lourd : accusez réception, puis traitez en file. Un webhook qui attend la fin d'un traitement de huit secondes finit par être considéré en échec et rejoué.

Deux erreurs classiques sur la vérification de signature : calculer l'empreinte sur le dictionnaire déjà décodé — json.loads puis json.dumps réordonne les clés et change le résultat —, et comparer avec == au lieu de hmac.compare_digest, qui compare en temps constant.

Votre premier SMS en Python dans dix minutes

Inscription gratuite, 5 messages de test offerts, sans carte bancaire. L'App ID et la clé API sont disponibles immédiatement après la création du compte.