SMS z Pythonu: requests, idempotence, webhooky (FastAPI) a testy

TopSMS má obyčejné JSON REST API, takže v Pythonu nepotřebujete žádnou knihovnu, stačí requests (nebo httpx). Níže je klient, který byste jinak psali sami: správná autentizace, opakování při chybě bez zdvojené SMS, hromadné odeslání a příjem doručenek s ověřením podpisu. Všechny ukázky jsou kompletní a spustitelné.

Příprava

1. V dashboardu TopSMS vytvořte API klíč se scopes send a read. Pro vývoj použijte klíč v testovacím režimu (nic se neodesílá ani neúčtuje, odpovědi jsou stejné).
2. Klíč dejte do proměnných prostředí, nikdy do kódu:

export TOPSMS_CLIENT_ID="..."
export TOPSMS_SECRET="..."
export TOPSMS_WEBHOOK_SECRET="..."   # u klíče, jen pokud používáte webhooky

pip install requests

1. Odeslání jedné SMS

import os, uuid, time, requests

API = "https://www.topsms.cz/api"
AUTH = f"Bearer {os.environ['TOPSMS_CLIENT_ID']}:{os.environ['TOPSMS_SECRET']}"

class TopSmsError(Exception):
    def __init__(self, status, payload):
        super().__init__(f"{status}: {payload.get('error')}")
        self.status, self.code = status, payload.get("code")

def send_sms(to: str, text: str, sender: str | None = None, idempotency_key: str | None = None, retries: int = 3) -> dict:
    headers = {"Authorization": AUTH, "Idempotency-Key": idempotency_key or str(uuid.uuid4())}
    body = {"to": to, "text": text}
    if sender:
        body["from"] = sender
    for attempt in range(retries):
        try:
            r = requests.post(f"{API}/sms/send", json=body, headers=headers, timeout=15)
        except requests.RequestException:
            time.sleep(2 ** attempt)          # timeout / síť: zopakovat se STEJNÝM klíčem
            continue
        if r.status_code in (200, 202):
            return r.json()
        if r.status_code == 409 or r.status_code >= 500:
            time.sleep(2 ** attempt)          # ještě se zpracovává, nebo dočasná chyba
            continue
        raise TopSmsError(r.status_code, r.json())   # 400/401/402/403/429: opakování nepomůže
    raise TopSmsError(503, {"error": "TopSMS neodpovídá, zkuste později"})

res = send_sms("+420601234567", "Vase objednavka 1234 je pripravena k vyzvednuti.", idempotency_key="order-1234-ready")
print(res["id"], res["smsCount"], res["price"])

Co se tu děje:

• Autentizace je Authorization: Bearer CLIENT_ID:SECRET.
• Idempotency‑Key zajistí, že opakovaný požadavek (timeout, restart) SMS nepošle znovu; TopSMS vrátí původní odpověď s hlavičkou Idempotent-Replayed: true. Klíč odvozujte od události (order-1234-ready), ne od času.
• Odpověď obsahuje id (uložte si ho ke své entitě), smsCount (skutečný počet částí) a price v Kč. Kredit se strhává při přijetí; když operátor zprávu nepřijme, dostanete chybu a kredit se vrátí.
• 402 znamená nedostatek kreditu, 429 překročený rate limit klíče (výchozí 60 požadavků/min; pro větší objem použijte hromadné odeslání níže).

Pole from použijte až po schválení vlastního Sender ID; do té doby zpráva odejde pod odesílatelem TopSMS.

2. Hromadné odeslání (až 500 příjemců)

Pro notifikace více zákazníkům najednou neposílejte 500 jednotlivých požadavků:

def send_bulk(recipients: list[str], text: str, sender: str | None = None, key: str | None = None) -> dict:
    headers = {"Authorization": AUTH, "Idempotency-Key": key or str(uuid.uuid4())}
    body = {"to": recipients, "text": text}
    if sender:
        body["from"] = sender
    r = requests.post(f"{API}/sms/send-bulk", json=body, headers=headers, timeout=60)
    r.raise_for_status()
    return r.json()

out = send_bulk(["+420601234567", "+420777123456"], "Zitra od 8 do 12 bude odstavka vody.", key="odstavka-2026-10-17")
for item in out["results"]:
    print(item["to"], item["status"], item["id"], item.get("error"))

Dávky do 50 příjemců se odesílají hned, větší se zařadí do fronty (HTTP 202 s jobId, průběh přes GET /api/sms/jobs/{jobId}). Neplatná čísla dostanou status: "failed" a neúčtují se, ostatní odejdou.

3. Stav zprávy

def status(message_id: str) -> dict:
    r = requests.get(f"{API}/sms/status/{message_id}", headers={"Authorization": AUTH}, timeout=15)
    r.raise_for_status()
    return r.json()
# -> {"id": "...", "status": "delivered", "deliveredAt": "...", "errorCode": None, ...}

Stavy: sent (přijato, čeká na doručenku), delivered, failed (s errorCode: undelivered, rejected, deleted, unknown), expired.

4. Doručenky webhookem (FastAPI)

U API klíče nastavte Webhook URL. TopSMS tam při změně stavu pošle POST s JSON tělem a hlavičkou X-TopSMS-Signature: sha256=<HMAC‑SHA256 těla>:

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

app = FastAPI()
SECRET = os.environ["TOPSMS_WEBHOOK_SECRET"].encode()

@app.post("/webhooks/topsms")
async def topsms_webhook(request: Request):
    raw = await request.body()                       # podpis se počítá z PŮVODNÍHO těla, ne z JSON po parsování
    expected = "sha256=" + hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    received = request.headers.get("x-topsms-signature", "")
    if not hmac.compare_digest(expected, received):
        raise HTTPException(status_code=401, detail="bad signature")
    event = await request.json()
    # event = {"event": "sms.status", "id": "...", "to": "+420...", "status": "delivered", "errorCode": None, ...}
    # TODO: aktualizovat stav u své entity podle event["id"]; zpracování musí být idempotentní (webhook může přijít opakovaně)
    return {"ok": True}

Odpovídejte rychle (do 5 s) a stavem 2xx; při chybě to TopSMS zkusí znovu po 30 s a po 2 minutách. Endpoint musí být veřejně dostupný bez přihlášení (ne za Basic Auth ani Cloudflare Access), bezpečnost zajišťuje podpis.

5. Testování bez odesílání

S klíčem v testovacím režimu funguje celý kód beze změny: číslo končící 0000 simuluje nedoručení, 0001 expiraci, ostatní doručení za 2–4 s, webhooky chodí s příznakem test: true. V pytestu tak otestujete i větev „nedoručeno“:

def test_failed_delivery():
    res = send_sms("+420600000000", "test", idempotency_key=f"t-{uuid.uuid4()}")
    assert res["test"] is True
    time.sleep(5)
    assert status(res["id"])["status"] == "failed"

Kam dál

• Dokumentace REST API a strojově čitelná OpenAPI specifikace, ze které si vygenerujete typovaného klienta.
• Verify API: ověřovací kódy (OTP) pro přihlášení a registraci, kód generujeme a hlídáme my.
• 7 pravidel, jak zabezpečit SMS účet a API klíč.