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

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

Jeśli budujesz w Pythonie aplikację, która ma wysyłać SMS-y — przypomnienia o wizytach, kody 2FA, powiadomienia transakcyjne — najprostszą drogą jest REST API z biblioteką requests lub jej asynchronicznym odpowiednikiem httpx. W tym przewodniku pokażemy, jak od zera zintegrować Pythona z REST API Przypominamy.com — od pierwszego SMS-a, przez bulk i obsługę błędów, po webhooki DLR w Flasku i asynchroniczne wysyłki przez asyncio.

1. Setup: jedna biblioteka, kilka sekund

API jest na tyle proste, że nie potrzebujesz dedykowanego SDK. Wystarczy requests (synchroniczny) lub httpx (synchroniczny + asynchroniczny):

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/sms:

import os, requests

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

Odpowiedź:

{
  "success": true,
  "data": {
    "count": 1,
    "messages": [{
      "id": "msg_abc123",
      "to": "+48600123456",
      "status": "queued",
      "parts": 1,
      "cost": 0.10,
      "sent_at": "2026-05-17T10:30:00.000Z"
    }]
  }
}

Pole status może przyjąć wartości queued (w kolejce), sent (wysłany do sieci), delivered (potwierdzone doręczenie), failed, expired, not_found, rejected i kilka innych. Aktualizacje statusu otrzymasz przez webhook DLR (sekcja 6).

Parametry warte zapamiętania

• from — własna nazwa nadawcy (do 11 znaków, wymaga rejestracji w panelu).
• date — planowana wysyłka (ISO 8601), np. "2026-05-18T14:00:00Z".
• flash — Flash SMS, wyświetlany na ekranie bez otwierania.
• idx — Twój identyfikator (idempotency key). Wraca w webhookach DLR — pozwala skojarzyć raport z encją w Twojej bazie.
• encoding — "utf-8" (domyślnie, polskie znaki, 70 znaków/część) lub "gsm" (bez polskich znaków, 160 znaków/część).

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

Dwa tryby. Jeśli wszyscy mają dostać tę samą treść (kampania, promocja), użyj recipients + message — maksymalnie 10 000 numerów jednym żądaniem:

resp = requests.post(
    f"{BASE_URL}/v1/sms/bulk",
    headers={"Authorization": f"Bearer {TOKEN}"},
    json={
        "recipients": ["+48600123456", "+48600234567", "+48600345678"],
        "message": "Promocja -20% tylko do niedzieli! Kod: SMS20",
        "from": "SKLEP",
    },
)

Gdy każdy odbiorca ma dostać spersonalizowaną wiadomość (np. przypomnienia o wizytach), użyj messages — do 100 wiadomości jednym żądaniem:

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

resp = requests.post(
    f"{BASE_URL}/v1/sms/bulk",
    headers={"Authorization": f"Bearer {TOKEN}"},
    json={
        "messages": [
            {
                "to": w["to"],
                "message": f"Cześć {w['imie']}, wizyta jutro o {w['godzina']}. Potwierdź odpisując TAK.",
                "idx": f"appointment-{w['to']}",
            }
            for w in wizyty
        ]
    },
)
for msg in resp.json()["data"]["messages"]:
    print(msg["to"], "→", msg["status"])

Gdy masz więcej niż 100 wiadomości, podziel na chunks i wyślij sekwencyjnie z respektowaniem rate limitu (sekcja 4).

4. Obsługa błędów

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

class SmsApiError(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/sms",
        headers={"Authorization": f"Bearer {TOKEN}"},
        json={"to": to, "message": message, **kwargs},
        timeout=10,
    )
    if resp.status_code == 200:
        return resp.json()
    err = resp.json().get("error", {})
    raise SmsApiError(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(SmsApiError),
)
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/sms",
        json={"to": to, "message": 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["message"]) for m in messages]
        return await asyncio.gather(*tasks, return_exceptions=True)

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

Uwaga na rate limit: 100 req/min to twardy limit. Dla 100+ wiadomości równolegle dodaj asyncio.Semaphore(50) żeby ograniczyć liczbę jednoczesnych połączeń.

6. Webhooki DLR w Flasku

Po wysłaniu SMS-a chcesz wiedzieć, czy faktycznie dotarł. Webhook DLR (Delivery Report) wysyła POST application/json na Twój skonfigurowany URL przy każdej zmianie statusu.

W panelu klienta ustaw webhook_url wskazujący np. na https://twojadomena.pl/webhook/dlr. Najprostszy handler we Flasku:

from flask import Flask, request, jsonify
import logging

app = Flask(__name__)

@app.route("/webhook/dlr", methods=["POST"])
def dlr_webhook():
    payload = request.get_json(silent=True) or {}
    event = payload.get("event")

    if event == "delivery_report":
        msg_id = payload.get("message_id")
        status = payload.get("status")
        idx    = payload.get("idx")
        # zapisz do bazy (najlepiej w tle przez Celery/RQ)
        update_sms_status.delay(msg_id, status, idx)
        logging.info("DLR: %s → %s (idx=%s)", msg_id, status, idx)

    elif event == "incoming_sms":
        # klient odpowiedział na Twoją wiadomość
        from_ = payload.get("from")
        text  = payload.get("message")
        handle_incoming.delay(from_, text)

    return jsonify({"received": True}), 200

Krytyczne: odpowiedz HTTP 200 jak najszybciej (< 2 sekundy). Ciężka logika (zapis do bazy, wysłanie powiadomienia) musi iść do background workera (Celery, RQ, asyncio task). Platforma nie ponawia DLR-ów — jeśli Twój endpoint zwróci 5xx lub timeout, raport zostanie zgubiony.

Payload DLR:

{
  "event": "delivery_report",
  "message_id": "msg_abc123",
  "to": "+48600123456",
  "status": "delivered",
  "sent_at": "2026-05-17T10:30:00.000Z",
  "done_at": "2026-05-17T10:30:04.000Z",
  "idx": "appointment-12345"
}

Pole idx to wartość, którą podałeś przy wysyłce — łatwo skojarzysz raport z encją (np. wizytą, zamówieniem, transakcją) w Twojej bazie.

7. Najlepsze praktyki produkcyjne

• Idempotency przez idx: zawsze podawaj unikalne idx per wiadomość (np. f"order-{order_id}"). Jeśli kod retry'uje wysyłkę z tym samym idx, łatwiej zorientujesz się w logach że to duplikat.
• 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 message ani numerów w logach produkcyjnych (RODO). Loguj message_id, status, kod błędu.
• Walidacja numeru przez HLR przed dużą kampanią: GET /v1/hlr?number=+48... sprawdza, czy numer jest aktywny — eliminuje koszt wysyłki na martwe numery.
• Pula tokenów dla różnych environmentów: produkcja, staging, testy każde ma własny token. Łatwiej rotować 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 SmsApiError(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 == 200:
                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 SmsApiError(resp.status_code, err.get("message", "Unknown"))
        raise SmsApiError(429, "Rate limit po 3 próbach")

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

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

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


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

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

FAQ