API SMS w Pythonie. Kompletny przewodnik z requests, httpx i webhookami

Zespół Przypominamy.com · 17 maja 2026 · 10 min czytania

Przypomnienia, kody 2FA i alerty z Pythona to zwykle requests lub httpx (albo paczka przypominamy) i POST /v1/messages. Poniżej pierwsza wysyłka, bulk, błędy HTTP, webhook DLR w Flasku i wariant asyncio.

1. Setup: jedna biblioteka, kilka sekund

Masz dwie drogi. Oficjalna paczka przypominamy (zero zależności, standardowa biblioteka) daje gotowe metody, wyjątek z kodem błędu i weryfikację webhooków:

pip install przypominamy

import os
from przypominamy import Przypominamy

sms = Przypominamy(os.environ["PRZYPOMINAMY_API_KEY"])
msg = sms.send("+48600123456", "Cześć! Przypominamy o wizycie jutro o 14:00.", reference="wizyta-4521")
print(msg["id"], msg["status"], msg["cost_grosze"])

Albo - bo API jest na tyle proste - requests (synchroniczny) lub httpx (synchroniczny + asynchroniczny). Reszta wpisu pokazuje tę drugą drogę, żebyś widział dokładnie, co leci po sieci:

pip install requests
# lub, gdy potrzebujesz async:
pip install httpx

Token API trzymaj w zmiennej środowiskowej - nigdy nie commituj go do gita:

# .env (nie commituj!)
PRZYPOMINAMY_API_KEY=pk_live_a1b2c3d4e5f6...

# w kodzie:
import os
TOKEN = os.environ["PRZYPOMINAMY_API_KEY"]
BASE_URL = "https://api.przypominamy.com"

Dla aplikacji produkcyjnej rozważ secrets managera (AWS Secrets Manager, GCP Secret Manager, HashiCorp Vault) zamiast pliku .env.

2. Pierwszy SMS w 10 liniach

Najprostszy POST na /v1/messages:

import os, requests

resp = requests.post(
    "https://api.przypominamy.com/v1/messages",
    headers={"Authorization": f"Bearer {os.environ['PRZYPOMINAMY_API_KEY']}"},
    json={
        "to": "+48600123456",
        "text": "Cześć! Przypominamy o wizycie jutro o 14:00.",
        "from": "FIRMA",
    },
    timeout=10,
)
resp.raise_for_status()
print(resp.json())

Odpowiedź (HTTP 201):

{
  "id": "msg_qY08hZ2mmDXAazdRTytS",
  "status": "queued",
  "to": "+48600123456",
  "from": "FIRMA",
  "text": "Cześć! Przypominamy o wizycie jutro o 14:00.",
  "parts": 1,
  "cost_grosze": 10,
  "reference": null,
  "send_at": null,
  "delivered_at": null,
  "error": null,
  "created_at": "2026-09-07T10:30:00.000Z",
  "updated_at": "2026-09-07T10:30:00.000Z"
}

Pole status przechodzi przez queued (przyjęta), sent (przekazana do sieci), delivered (potwierdzone doręczenie) albo kończy jako undelivered, failed, expired lub rejected. Aktualizacje statusu otrzymasz przez webhook (sekcja 6) albo odczytasz z GET /v1/messages/{id}.

Parametry warte zapamiętania

• from - nadpis, czyli własna nazwa nadawcy (do 11 znaków). Musi być na liście GET /v1/senders; domyślny ustawisz przez PATCH /v1/account.
• send_at - planowana wysyłka (ISO 8601, do 90 dni), np. "2026-09-18T14:00:00Z". Wiadomość ma wtedy status scheduled.
• reference - Twój identyfikator (np. id wizyty). Wraca w odpowiedzi, w webhookach i jako filtr historii.
• nagłówek Idempotency-Key - chroni przed podwójną wysyłką przy retry: ta sama treść zwraca pierwotną odpowiedź, inna dostaje 409.
• Polskie znaki - dozwolone; wiadomość idzie wtedy jako UCS-2 (70 znaków na część zamiast 160). Liczbę części i koszt zwraca odpowiedź (parts, cost_grosze).

3. Bulk: wysyłka do wielu odbiorców

Jeśli wszyscy mają dostać tę samą treść (kampania, promocja), podaj w to tablicę - maksymalnie 500 numerów jednym żądaniem, duplikaty są usuwane:

resp = requests.post(
    f"{BASE_URL}/v1/messages",
    headers={"Authorization": f"Bearer {TOKEN}"},
    json={
        "to": ["+48600123456", "+48600234567", "+48600345678"],
        "text": "Promocja -20% tylko do niedzieli! Kod: SMS20",
        "from": "SKLEP",
        "reference": "promo-2026-09",
    },
    timeout=10,
)
result = resp.json()
print(result["accepted"], "z", result["count"], "przyjęto, koszt:", result["total_cost_grosze"] / 100, "PLN")
for msg in result["messages"]:
    print(msg["to"], "→", msg["status"], msg["error"] or "")

Gdy każdy odbiorca ma dostać spersonalizowaną wiadomość (np. przypomnienia o wizytach), wysyłaj osobne żądania - z reference per wizyta i Idempotency-Key, żeby retry nie zdublował SMS-a:

wizyty = [
    {"id": 4521, "to": "+48600123456", "imie": "Anna",  "godzina": "14:00"},
    {"id": 4522, "to": "+48600234567", "imie": "Piotr", "godzina": "15:30"},
    {"id": 4523, "to": "+48600345678", "imie": "Maria", "godzina": "16:45"},
]

for w in wizyty:
    resp = requests.post(
        f"{BASE_URL}/v1/messages",
        headers={
            "Authorization": f"Bearer {TOKEN}",
            "Idempotency-Key": f"wizyta-{w['id']}-przypomnienie",
        },
        json={
            "to": w["to"],
            "text": f"Cześć {w['imie']}, wizyta jutro o {w['godzina']}. Potwierdź odpisując TAK.",
            "reference": f"wizyta-{w['id']}",
        },
        timeout=10,
    )
    print(w["to"], "→", resp.json()["status"])

Limit to 120 żądań na minutę. Dla większych wolumenów użyj kolejki i respektuj Retry-After (sekcja 4).

4. Obsługa błędów

API zwraca standardowe kody HTTP. Zawsze sprawdzaj status_code przed parsowaniem JSON-a:

class SmsGatewayError(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message
        super().__init__(f"[{code}] {message}")

def send_sms(to: str, message: str, **kwargs) -> dict:
    resp = requests.post(
        f"{BASE_URL}/v1/messages",
        headers={"Authorization": f"Bearer {TOKEN}"},
        json={"to": to, "text": message, **kwargs},
        timeout=10,
    )
    if resp.status_code == 201:
        return resp.json()
    err = resp.json().get("error", {})
    raise SmsGatewayError(resp.status_code, err.get("message", "Unknown error"))

Kody, które warto rozróżnić:

Obsługa rate limit (429) z biblioteką tenacity

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=4, max=60),
    retry=retry_if_exception_type(SmsGatewayError),
)
def send_with_retry(to: str, message: str, **kwargs):
    return send_sms(to, message, **kwargs)

Dla 429 możesz też ręcznie odczytać Retry-After i poczekać dokładnie tyle sekund - to przyjemniejsze dla API niż exponential backoff.

5. Async: równoległa wysyłka z httpx

Gdy masz np. 50 niezależnych SMS-ów do wysłania, asynchroniczny httpx jest znacznie szybszy od sekwencyjnego requests - wszystkie żądania lecą równolegle, a Ty czekasz tylko na najwolniejsze.

import asyncio, httpx, os

TOKEN = os.environ["PRZYPOMINAMY_API_KEY"]

async def send_one(client: httpx.AsyncClient, to: str, message: str):
    resp = await client.post(
        "/v1/messages",
        json={"to": to, "text": message},
    )
    resp.raise_for_status()
    return resp.json()

async def send_many(messages: list[dict]):
    async with httpx.AsyncClient(
        base_url="https://api.przypominamy.com",
        headers={"Authorization": f"Bearer {TOKEN}"},
        timeout=10.0,
    ) as client:
        tasks = [send_one(client, m["to"], m["text"]) for m in messages]
        return await asyncio.gather(*tasks, return_exceptions=True)

# użycie:
messages = [
    {"to": "+48600123456", "text": "SMS 1"},
    {"to": "+48600234567", "text": "SMS 2"},
    # ... 48 więcej
]
results = asyncio.run(send_many(messages))

Uwaga na rate limit: 120 żądań/min to twardy limit. Dla 100+ wiadomości równolegle dodaj asyncio.Semaphore(20), a jeśli treść jest ta sama, wyślij ją jedną tablicą to (sekcja 3).

6. Webhooki statusów w Flasku

Po wysłaniu SMS-a chcesz wiedzieć, czy faktycznie dotarł. Webhook wysyła POST application/json na Twój URL przy każdej zmianie statusu, podpisany HMAC-SHA256.

Adres ustawiasz jednym wywołaniem; w odpowiedzi dostajesz sekret do weryfikacji podpisu:

resp = requests.put(
    f"{BASE_URL}/v1/account/webhook",
    headers={"Authorization": f"Bearer {TOKEN}"},
    json={"url": "https://twojadomena.pl/webhook/sms"},
    timeout=10,
)
WEBHOOK_SECRET = resp.json()["webhook_secret"]  # zapisz w sekretach

Handler we Flasku z weryfikacją podpisu:

import hmac, hashlib, os, re, time, logging
from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["PRZYPOMINAMY_WEBHOOK_SECRET"]

def verify(raw_body: bytes, header: str) -> bool:
    m = re.match(r"t=(\d+),v1=([0-9a-f]+)", header or "")
    if not m:
        return False
    t, sig = m.group(1), m.group(2)
    expected = hmac.new(WEBHOOK_SECRET.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(sig, expected) and abs(time.time() - int(t)) < 300

@app.route("/webhook/sms", methods=["POST"])
def sms_webhook():
    if not verify(request.get_data(), request.headers.get("X-Przypominamy-Signature")):
        return jsonify({"error": "bad signature"}), 403

    event = request.get_json(silent=True) or {}
    msg = event.get("data", {}).get("message", {})
    # zapisz do bazy (najlepiej w tle przez Celery/RQ)
    update_sms_status.delay(msg.get("id"), msg.get("status"), msg.get("reference"))
    logging.info("%s: %s → %s (reference=%s)", event.get("type"), msg.get("id"), msg.get("status"), msg.get("reference"))
    return jsonify({"received": True}), 200

Ważne: odpowiedz HTTP 2xx jak najszybciej (< 2 sekundy). Ciężka logika (zapis do bazy, wysłanie powiadomienia) musi iść do background workera (Celery, RQ, asyncio task). Przy 5xx lub timeoucie platforma ponawia dostarczenie po 2 s i po 10 s (łącznie 3 próby), potem zdarzenie trafia tylko do historii GET /v1/messages/{id}.

Payload zdarzenia:

{
  "id": "evt_8sK2mQ…",
  "type": "message.delivered",
  "created_at": "2026-09-07T10:30:05.000Z",
  "data": {
    "message": {
      "id": "msg_qY08hZ2mmDXAazdRTytS",
      "status": "delivered",
      "to": "+48600123456",
      "reference": "wizyta-4521",
      "delivered_at": "2026-09-07T10:30:04.000Z",
      ...
    }
  }
}

Typy: message.sent, message.delivered, message.undelivered, message.failed, message.expired. Pole reference to wartość, którą podałeś przy wysyłce - łatwo skojarzysz zdarzenie z encją (wizytą, zamówieniem, transakcją) w Twojej bazie.

7. Najlepsze praktyki produkcyjne

• Idempotencja przez nagłówek Idempotency-Key: podawaj unikalny klucz per wiadomość (np. f"order-{order_id}"). Retry z tym samym kluczem zwraca pierwotną odpowiedź zamiast wysłać SMS drugi raz. Do korelacji w bazie używaj pola reference.
• Timeout 10 sekund: ustaw timeout=10 w każdym żądaniu. API ma fetch timeout 10s; klient też powinien mieć.
• Logowanie nie payloadu, tylko ID: nie loguj treści ani numerów w logach produkcyjnych (RODO). Loguj id wiadomości, status, kod błędu i request_id z odpowiedzi.
• Czyść listę odbiorców przed dużą kampanią: usuwaj numery, które w historii (GET /v1/messages?status=undelivered) regularnie kończą jako niedoręczone - nie płacisz za martwe numery.
• Osobne klucze dla środowisk: produkcja, staging, testy - każde ma własny klucz pk_live_…. Łatwiej odwołać przy wycieku.

8. Pełny gist do skopiowania

Minimalna, produkcyjna klasa do wysyłki SMS - z retry, walidacją i obsługą rate limit:

"""Minimalny klient API SMS dla Przypominamy.com - wersja produkcyjna."""
import os
import time
from typing import Optional
import requests


class SmsGatewayError(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message
        super().__init__(f"[{code}] {message}")


class PrzypominamySmsClient:
    BASE_URL = "https://api.przypominamy.com"

    def __init__(self, token: Optional[str] = None, timeout: float = 10.0):
        self.token = token or os.environ["PRZYPOMINAMY_API_KEY"]
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers["Authorization"] = f"Bearer {self.token}"

    def _post(self, path: str, body: dict, max_retries: int = 3) -> dict:
        for attempt in range(max_retries):
            resp = self.session.post(
                f"{self.BASE_URL}{path}",
                json=body,
                timeout=self.timeout,
            )
            if resp.status_code == 201:
                return resp.json()
            if resp.status_code == 429:
                wait = int(resp.headers.get("Retry-After", 1))
                time.sleep(wait)
                continue
            err = resp.json().get("error", {})
            raise SmsGatewayError(resp.status_code, err.get("message", "Unknown"))
        raise SmsGatewayError(429, "Rate limit po 3 próbach")

    def send(self, to: str, text: str, **kwargs) -> dict:
        return self._post("/v1/messages", {"to": to, "text": text, **kwargs})

    def send_bulk(self, recipients: list[str], text: str, **kwargs) -> dict:
        return self._post("/v1/messages", {"to": recipients, "text": text, **kwargs})

    def status(self, message_id: str) -> dict:
        resp = self.session.get(f"{self.BASE_URL}/v1/messages/{message_id}", timeout=self.timeout)
        resp.raise_for_status()
        return resp.json()

    def balance_pln(self) -> float:
        resp = self.session.get(f"{self.BASE_URL}/v1/account", timeout=self.timeout)
        resp.raise_for_status()
        return resp.json()["balance_grosze"] / 100


if __name__ == "__main__":
    client = PrzypominamySmsClient()
    result = client.send(
        to="+48600123456",
        text="Hello Python!",
        reference="test-001",
    )
    print(result)

~80 linii, zero zewnętrznych zależności poza requests. Skopiuj do swojego projektu i rozszerzaj wedle potrzeb.

FAQ