Python

python3 -m venv .venv
# macOS / Linux
source .venv/bin/activate
# Windows PowerShell : .venv\Scripts\Activate.ps1
python -m pip install requests

import os
import uuid
import requests


def send_sms(to, message, sender_name, idempotency_key, client_reference):
    api_key = os.environ['COUSSEMA_API_KEY'].strip()
    if not api_key or not idempotency_key.strip():
        raise ValueError('Clé API et Idempotency-Key requises.')

    # Ces valeurs doivent rester identiques pour reprendre le même envoi.
    payload = {
        'to': to,
        'message': message,
        'senderName': sender_name,
        'clientReference': client_reference,
    }
    try:
        response = requests.post(
            'https://api.coussema.com/v1/sms/send',
            headers={
                'Authorization': f'Bearer {api_key}',
                'Idempotency-Key': idempotency_key,
                'X-Request-Id': str(uuid.uuid4()),
            },
            json=payload,
            timeout=(5, 30),
            allow_redirects=False,
        )
    except requests.exceptions.RequestException as exc:
        raise RuntimeError(
            'Résultat réseau incertain. Conserver la même clé et le même '
            'contenu pour une reprise contrôlée, sans nouvel envoi automatique.'
        ) from exc

    request_id = response.headers.get('x-request-id', '')
    if not 200 <= response.status_code < 300:
        # Ne pas journaliser la clé API, les headers ou le contenu du SMS.
        raise RuntimeError(
            f'HTTP {response.status_code}; request_id={request_id}; '
            f'Retry-After={response.headers.get("Retry-After", "")}. '
            'Consulter les règles de reprise avant de retenter.'
        )
    try:
        result = response.json()
    except ValueError as exc:
        raise RuntimeError(
            f'Réponse illisible; request_id={request_id}. '
            'Ne pas créer une nouvelle Idempotency-Key.'
        ) from exc
    if (
        not isinstance(result, dict)
        or result.get('success') is not True
        or result.get('queued') is not True
        or result.get('status') != 'accepted'
        or not isinstance(result.get('message'), dict)
        or not result['message'].get('id')
    ):
        raise RuntimeError(
            f'Réponse inattendue; request_id={request_id}. '
            'Vérifier le résultat avant toute reprise.'
        )
    return result


if __name__ == '__main__':
    # Configurer ces variables sur le serveur avant de lancer le script.
    # La clé d'idempotence est créée ET enregistrée par votre application
    # avant le premier appel, puis réutilisée pour cette même opération.
    result = send_sms(
        to=os.environ['COUSSEMA_SMS_TO'],
        message=os.environ['COUSSEMA_SMS_MESSAGE'],
        sender_name=os.environ['COUSSEMA_SENDER_NAME'],
        idempotency_key=os.environ['COUSSEMA_IDEMPOTENCY_KEY'],
        client_reference=os.environ['COUSSEMA_CLIENT_REFERENCE'],
    )
    print({
        'message_id': result['message']['id'],
        'status': result['status'],
        'remaining_quota': result['remainingQuota'],
    })

{
  "success": true,
  "count": 1,
  "queued": true,
  "status": "accepted",
  "remainingQuota": 999,
  "message": {
    "id": "a2f7e5cb-bbe9-4d7d-bd17-80d1d7a5e5af",
    "clientReference": "commande-2026-0001"
  },
  "provider": { "code": "coussema", "messageId": null, "httpStatus": 202 }
}

import hashlib
import hmac
import json
from datetime import datetime, timezone


def verify_webhook(raw_body, headers, signing_secret):
    # raw_body : bytes reçus AVANT tout décodage JSON.
    headers = {key.lower(): value for key, value in headers.items()}
    timestamp = headers.get('x-coussema-timestamp', '')
    signature = headers.get('x-coussema-signature', '')
    if not signing_secret or not timestamp or not signature:
        raise ValueError('Signature ou secret manquant.')
    try:
        sent_at = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
        if sent_at.tzinfo is None:
            raise ValueError('Fuseau horaire manquant.')
        age = abs((datetime.now(timezone.utc) - sent_at).total_seconds())
    except (ValueError, OverflowError) as exc:
        raise ValueError('Horodatage invalide.') from exc
    # Tolérance choisie pour cet exemple : cinq minutes.
    if age > 300:
        raise ValueError('Signature expirée.')
    digest = hmac.new(
        signing_secret.encode('utf-8'),
        timestamp.encode('utf-8') + b'.' + raw_body,
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(
        ('v1=' + digest).encode('ascii'), signature.encode('utf-8')
    ):
        raise ValueError('Signature invalide.')
    return json.loads(raw_body)