Tích Hợp SMS API với Python: Hướng Dẫn Từng Bước Có Code Ví Dụ

Tích Hợp SMS API với Python: Hướng Dẫn Hoàn Chỉnh với Code Ví Dụ Thực Tế 2026

Tích hợp SMS API với Python là bài toán mà bất kỳ developer nào cũng gặp ít nhất một lần: gửi OTP xác thực, thông báo đơn hàng, cảnh báo hệ thống, hay chiến dịch bulk SMS. Với Python, toàn bộ luồng - từ gọi REST API, xử lý response, retry khi lỗi đến nhận webhook Delivery Report - có thể hoàn thành trong 200-300 dòng code sạch. Bài này hướng dẫn từng bước, từ setup môi trường đến production-ready code, với đầy đủ ví dụ cho requests, Flask, Django và asyncio/aiohttp.

Python là ngôn ngữ được nhiều backend team lựa chọn cho SMS integration vì ecosystem phong phú: requests cho HTTP đơn giản, aiohttp cho async throughput cao, pyotp cho OTP generation, và Celery cho task queue bất đồng bộ. Dù bạn đang build Django monolith hay FastAPI microservice, mẫu code trong bài này đều áp dụng được ngay.

Lưu ý trước khi bắt đầu: bài này tập trung vào kỹ thuật tích hợp và không ràng buộc với nhà cung cấp SMS cụ thể nào. Mọi ví dụ đều dùng REST API generic với authentication bằng API key - chuẩn mà hầu hết các nhà cung cấp SMS tại Việt Nam đều hỗ trợ. Để biết thêm về SMS API là gì và cách nó hoạt động, bạn có thể đọc bài tổng quan trước.

1. Prerequisites và Setup Môi Trường

Trước khi viết một dòng code, cần chuẩn bị 4 thứ: Python 3.8+, API key từ nhà cung cấp SMS, endpoint URL và thư viện HTTP. Phần setup này mất khoảng 5 phút.

Cài đặt thư viện

Với bài này bạn cần 3 thư viện core. Chạy lệnh pip sau để cài đủ một lần:

pip install requests httpx aiohttp pyotp python-dotenv

Dùng file .env để lưu credentials - không bao giờ hard-code API key vào source code:

# .env
SMS_API_KEY=your_api_key_here
SMS_API_URL=https://api.your-sms-provider.com/v1/sms/send
SMS_SENDER=YourBrand

# config.py
from dotenv import load_dotenv
import os

load_dotenv()

SMS_API_KEY = os.getenv("SMS_API_KEY")
SMS_API_URL = os.getenv("SMS_API_URL")
SMS_SENDER  = os.getenv("SMS_SENDER", "Notify")

Với developer backend Python cần biết

Với developer đang tích hợp SMS API Python lần đầu, 3 điều quan trọng nhất trước khi bắt đầu:

• Hầu hết SMS REST API xác thực qua HTTP header (Authorization: Bearer <key> hoặc X-API-Key: <key>) - không phải query string
• Response thường trả về JSON với message_id - lưu lại để đối chiếu Delivery Report sau này
• Rate limit thường là 100-500 request/giây - bulk SMS phải dùng batch hoặc async, không vòng lặp tuần tự

2. Gửi SMS Đơn với Python requests

Đây là building block cơ bản nhất. Một function send_sms() chuẩn gồm 3 phần: build payload, gọi API, xử lý response. Viết đúng từ đầu giúp tái sử dụng ở mọi nơi trong project.

import requests
import logging
from config import SMS_API_KEY, SMS_API_URL, SMS_SENDER

logger = logging.getLogger(__name__)

def send_sms(phone: str, message: str) -> dict:
    """
    Gửi SMS đơn lẻ qua REST API.

    Args:
        phone: Số điện thoại nhận, định dạng 84xxxxxxxxx (không dấu +)
        message: Nội dung tin nhắn, tối đa 160 ký tự ASCII hoặc 70 ký tự Unicode

    Returns:
        dict: {"success": bool, "message_id": str | None, "error": str | None}
    """
    headers = {
        "Authorization": f"Bearer {SMS_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "to":      phone,
        "from":    SMS_SENDER,
        "message": message,
    }

    try:
        response = requests.post(
            SMS_API_URL,
            json=payload,
            headers=headers,
            timeout=10,  # timeout 10 giây - không để mặc định vô hạn
        )
        response.raise_for_status()  # raise HTTPError nếu 4xx/5xx

        data = response.json()
        message_id = data.get("message_id") or data.get("id")
        logger.info("SMS sent to %s, message_id=%s", phone, message_id)
        return {"success": True, "message_id": message_id, "error": None}

    except requests.exceptions.Timeout:
        logger.error("SMS API timeout sending to %s", phone)
        return {"success": False, "message_id": None, "error": "timeout"}

    except requests.exceptions.HTTPError as e:
        logger.error("SMS API HTTP error %s for %s: %s", e.response.status_code, phone, e.response.text)
        return {"success": False, "message_id": None, "error": str(e)}

    except requests.exceptions.RequestException as e:
        logger.error("SMS API request failed for %s: %s", phone, str(e))
        return {"success": False, "message_id": None, "error": str(e)}

Hai điểm quan trọng trong code trên: timeout=10 là bắt buộc - SMS API đôi khi chậm do tải, không set timeout sẽ treo request vô thời hạn. Và raise_for_status() giúp bắt lỗi HTTP 4xx/5xx thành exception thay vì phải kiểm tra status code thủ công.

3. Gửi SMS Hàng Loạt với Batch Processing

Bulk SMS cho chiến dịch marketing hay thông báo hàng loạt cần xử lý khác hoàn toàn. Vòng lặp tuần tự với 10.000 số điện thoại chạy mất hàng giờ và dễ bị rate limit. Giải pháp: batch + ThreadPoolExecutor để song song hóa, hoặc async (xem phần 7).

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
from typing import List, Dict

def send_bulk_sms(
    recipients: List[Dict[str, str]],
    batch_size: int = 50,
    max_workers: int = 5,
    delay_between_batches: float = 0.5,
) -> List[dict]:
    """
    Gửi SMS hàng loạt với batching và rate limit control.

    Args:
        recipients: List[{"phone": "84xxx", "message": "..."}]
        batch_size:  Số SMS mỗi batch (mặc định 50)
        max_workers: Số thread song song (mặc định 5)
        delay_between_batches: Delay giữa các batch (giây)

    Returns:
        List kết quả tương ứng với từng recipient
    """
    results = []
    total = len(recipients)

    # Chia thành các batch
    batches = [recipients[i:i + batch_size] for i in range(0, total, batch_size)]
    logger.info("Sending %d SMS in %d batches", total, len(batches))

    for batch_idx, batch in enumerate(batches):
        batch_results = []

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_recipient = {
                executor.submit(send_sms, r["phone"], r["message"]): r
                for r in batch
            }
            for future in as_completed(future_to_recipient):
                recipient = future_to_recipient[future]
                try:
                    result = future.result()
                    result["phone"] = recipient["phone"]
                    batch_results.append(result)
                except Exception as e:
                    batch_results.append({
                        "phone":      recipient["phone"],
                        "success":    False,
                        "message_id": None,
                        "error":      str(e),
                    })

        results.extend(batch_results)
        success_count = sum(1 for r in batch_results if r["success"])
        logger.info(
            "Batch %d/%d done: %d/%d success",
            batch_idx + 1, len(batches), success_count, len(batch)
        )

        # Delay để tránh rate limit, bỏ qua batch cuối
        if batch_idx < len(batches) - 1:
            time.sleep(delay_between_batches)

    return results

Với chiến dịch 10.000 tin nhắn, cấu hình batch_size=50, max_workers=5, delay=0.2s cho throughput khoảng 200-250 SMS/giây mà không vượt rate limit phổ biến. Lưu toàn bộ message_id vào database để đối chiếu DLR sau.

4. OTP Generation và Gửi Qua SMS

OTP qua SMS là luồng quan trọng nhất với ứng dụng có xác thực người dùng. Có 2 cách: OTP ngẫu nhiên đơn giản (6 chữ số, hết hạn sau 5 phút) và TOTP chuẩn RFC 6238 (tương thích Google Authenticator). Cho SMS thông thường, OTP ngẫu nhiên là đủ và dễ implement hơn.

import secrets
import hashlib
import time
from typing import Optional

# Lưu OTP tạm trong memory (production nên dùng Redis với TTL)
_otp_store: dict = {}  # {"phone": {"otp": "123456", "expires_at": timestamp, "attempts": 0}}

OTP_EXPIRY_SECONDS = 300   # 5 phút
OTP_MAX_ATTEMPTS   = 3     # Tối đa 3 lần nhập sai

def generate_otp(length: int = 6) -> str:
    """Tạo OTP số ngẫu nhiên an toàn (dùng secrets, không phải random)."""
    return "".join([str(secrets.randbelow(10)) for _ in range(length)])

def send_otp_sms(phone: str) -> dict:
    """
    Tạo OTP và gửi qua SMS. Mỗi phone chỉ được gửi 1 lần/phút (rate limit).

    Returns:
        dict: {"success": bool, "message": str}
    """
    # Rate limit: kiểm tra OTP cũ có còn trong 60 giây không
    existing = _otp_store.get(phone)
    if existing:
        remaining = existing["expires_at"] - time.time()
        if remaining > (OTP_EXPIRY_SECONDS - 60):  # Mới gửi trong vòng 60 giây
            wait = int(OTP_EXPIRY_SECONDS - 60 - (OTP_EXPIRY_SECONDS - remaining))
            return {"success": False, "message": f"Vui lòng đợi {wait} giây trước khi gửi lại"}

    otp = generate_otp()
    expires_at = time.time() + OTP_EXPIRY_SECONDS

    # Lưu OTP (hash để bảo mật hơn, tránh lộ trong log)
    _otp_store[phone] = {
        "otp_hash":   hashlib.sha256(otp.encode()).hexdigest(),
        "expires_at": expires_at,
        "attempts":   0,
    }

    message = f"Ma xac thuc cua ban la: {otp}. Het han sau 5 phut. Khong chia se ma nay voi bat ky ai."
    result = send_sms(phone, message)

    if not result["success"]:
        del _otp_store[phone]  # Xóa nếu gửi thất bại
        return {"success": False, "message": "Không gửi được OTP, vui lòng thử lại"}

    return {"success": True, "message": "OTP đã được gửi"}


def verify_otp(phone: str, otp_input: str) -> dict:
    """
    Xác minh OTP người dùng nhập.

    Returns:
        dict: {"valid": bool, "message": str}
    """
    record = _otp_store.get(phone)

    if not record:
        return {"valid": False, "message": "OTP không tồn tại hoặc đã hết hạn"}

    if time.time() > record["expires_at"]:
        del _otp_store[phone]
        return {"valid": False, "message": "OTP đã hết hạn"}

    if record["attempts"] >= OTP_MAX_ATTEMPTS:
        del _otp_store[phone]
        return {"valid": False, "message": "Đã nhập sai quá 3 lần, vui lòng yêu cầu OTP mới"}

    input_hash = hashlib.sha256(otp_input.encode()).hexdigest()
    if input_hash != record["otp_hash"]:
        _otp_store[phone]["attempts"] += 1
        remaining = OTP_MAX_ATTEMPTS - record["attempts"]
        return {"valid": False, "message": f"OTP không đúng. Còn {remaining} lần thử"}

    del _otp_store[phone]  # OTP dùng một lần - xóa sau khi xác thực thành công
    return {"valid": True, "message": "Xác thực thành công"}

Lưu ý production: thay _otp_store dictionary bằng Redis với TTL. Dictionary Python sẽ mất dữ liệu khi restart process và không hoạt động với multi-instance deployment. Redis TTL tự động xóa OTP hết hạn, không cần cleanup job riêng. Xem thêm pattern tích hợp SMS OTP API đầy đủ với Redis và rate limiting nâng cao.

5. Flask Integration: Endpoint Nhận Request và Gửi SMS

Flask phù hợp cho microservice SMS gọn nhẹ hoặc API gateway xử lý SMS request từ frontend. Hai endpoint thực tế nhất: POST /send-otp cho xác thực người dùng và POST /send-notification cho thông báo từ hệ thống khác.

from flask import Flask, request, jsonify
from functools import wraps
import hmac
import hashlib

app = Flask(__name__)

# --- Middleware xác thực request nội bộ ---
INTERNAL_SECRET = os.getenv("INTERNAL_API_SECRET", "change-me-in-production")

def require_internal_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get("X-Internal-Token")
        if not token or not hmac.compare_digest(token, INTERNAL_SECRET):
            return jsonify({"error": "Unauthorized"}), 401
        return f(*args, **kwargs)
    return decorated


@app.route("/send-otp", methods=["POST"])
@require_internal_auth
def api_send_otp():
    """Gửi OTP xác thực đến số điện thoại."""
    data = request.get_json(silent=True)
    if not data or "phone" not in data:
        return jsonify({"error": "Thiếu trường 'phone'"}), 400

    phone = data["phone"].strip()
    # Chuẩn hóa số: 0912345678 -> 84912345678
    if phone.startswith("0"):
        phone = "84" + phone[1:]

    result = send_otp_sms(phone)
    status = 200 if result["success"] else 429
    return jsonify(result), status


@app.route("/send-notification", methods=["POST"])
@require_internal_auth
def api_send_notification():
    """Gửi SMS thông báo (đơn hàng, lịch hẹn, cảnh báo)."""
    data = request.get_json(silent=True)
    required = {"phone", "message"}
    if not data or not required.issubset(data.keys()):
        return jsonify({"error": f"Thiếu các trường: {required - set(data or {})}"}), 400

    phone   = data["phone"].strip()
    message = data["message"].strip()[:160]  # Giới hạn 160 ký tự

    if phone.startswith("0"):
        phone = "84" + phone[1:]

    result = send_sms(phone, message)
    status = 200 if result["success"] else 503
    return jsonify(result), status


@app.route("/webhook/dlr", methods=["POST"])
def webhook_dlr():
    """Nhận Delivery Report từ SMS provider. Xem Section 8 để biết chi tiết."""
    payload = request.get_json(silent=True) or {}
    message_id = payload.get("message_id") or payload.get("id")
    status     = payload.get("status")
    phone      = payload.get("to") or payload.get("recipient")

    logger.info("DLR received: message_id=%s, status=%s, phone=%s", message_id, status, phone)

    # TODO: cập nhật trạng thái trong database
    # SMSLog.objects.filter(message_id=message_id).update(status=status)

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


if __name__ == "__main__":
    app.run(debug=False, host="0.0.0.0", port=5000)

Điểm đáng chú ý: endpoint /webhook/dlr phải trả về HTTP 200 ngay lập tức, dù xử lý chưa xong - SMS provider có thể retry nếu không nhận được 2xx trong vài giây. Mọi logic database update nên chạy background (Celery hoặc asyncio task).

6. Django Integration: Signal và Celery Task

Django project thường gửi SMS khi có event cụ thể: đơn hàng tạo mới, user đăng ký, trạng thái thay đổi. Hai pattern phù hợp nhất: Django Signal để tự động trigger và Celery Task để xử lý bất đồng bộ, tránh chặn request/response cycle.

Cài đặt Celery task gửi SMS

# tasks.py (Celery)
from celery import shared_task
import logging

logger = logging.getLogger(__name__)

@shared_task(
    bind=True,
    autoretry_for=(Exception,),   # Tự retry mọi exception
    retry_backoff=True,           # Exponential backoff: 1s, 2s, 4s, 8s...
    retry_backoff_max=60,         # Tối đa 60 giây mỗi lần chờ
    max_retries=3,                # Tối đa 3 lần retry
    default_retry_delay=5,        # Delay mặc định 5 giây
)
def send_sms_task(self, phone: str, message: str, metadata: dict = None):
    """
    Celery task gửi SMS bất đồng bộ với auto-retry.

    Dùng .delay() hoặc .apply_async() để gửi:
        send_sms_task.delay("84912345678", "Nội dung tin nhắn")
        send_sms_task.apply_async(args=[phone, msg], countdown=10)  # delay 10 giây
    """
    result = send_sms(phone, message)

    if not result["success"]:
        # Raise exception để trigger auto-retry
        raise Exception(f"SMS send failed for {phone}: {result['error']}")

    # Lưu kết quả vào database nếu cần
    if metadata and "order_id" in metadata:
        from orders.models import Order
        Order.objects.filter(id=metadata["order_id"]).update(
            sms_message_id=result["message_id"],
            sms_sent=True
        )

    logger.info("SMS task done: phone=%s, message_id=%s", phone, result["message_id"])
    return result

Django Signal tự động gửi SMS khi đơn hàng tạo mới

# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order
from .tasks import send_sms_task

@receiver(post_save, sender=Order)
def order_created_sms(sender, instance, created, **kwargs):
    """Gửi SMS xác nhận khi đơn hàng được tạo mới."""
    if not created:
        return  # Chỉ trigger khi CREATE, không phải UPDATE

    if not instance.customer_phone:
        return  # Bỏ qua nếu không có số điện thoại

    message = (
        f"Don hang #{instance.order_number} da duoc xac nhan. "
        f"Tong tien: {instance.total_amount:,.0f}d. "
        f"Du kien giao: {instance.estimated_delivery}."
    )

    # Dùng .delay() để gửi bất đồng bộ - không chặn HTTP response
    send_sms_task.delay(
        phone=instance.customer_phone,
        message=message,
        metadata={"order_id": instance.id}
    )


# apps.py - đăng ký signal
from django.apps import AppConfig

class OrdersConfig(AppConfig):
    name = "orders"

    def ready(self):
        import orders.signals  # noqa: F401 - import để register signal

Với Django integration, không bao giờ gọi SMS API trực tiếp trong view hoặc signal handler. Luôn dùng Celery task để tránh làm chậm HTTP response. Nếu SMS provider phản hồi chậm 3-5 giây, người dùng sẽ cảm nhận rõ ràng nếu gọi đồng bộ.

7. Async SMS với asyncio và aiohttp

Khi cần throughput cao - gửi hàng nghìn SMS trong vài giây - asyncio + aiohttp là lựa chọn tốt hơn ThreadPoolExecutor. Aiohttp nhanh hơn requests khoảng 10 lần ở concurrency cao vì dùng non-blocking I/O, không tạo thread mới cho mỗi request.

So sánh 3 HTTP library phổ biến cho SMS integration

Async bulk SMS với aiohttp và asyncio.Semaphore

import asyncio
import aiohttp
from typing import List, Dict

async def send_sms_async(
    session: aiohttp.ClientSession,
    phone: str,
    message: str,
    semaphore: asyncio.Semaphore,
) -> dict:
    """Gửi 1 SMS bất đồng bộ với rate limit qua Semaphore."""
    async with semaphore:  # Giới hạn số request đồng thời
        headers = {
            "Authorization": f"Bearer {SMS_API_KEY}",
            "Content-Type":  "application/json",
        }
        payload = {"to": phone, "from": SMS_SENDER, "message": message}

        try:
            async with session.post(
                SMS_API_URL,
                json=payload,
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=10),
            ) as response:
                response.raise_for_status()
                data = await response.json()
                return {
                    "phone":      phone,
                    "success":    True,
                    "message_id": data.get("message_id"),
                    "error":      None,
                }
        except aiohttp.ClientError as e:
            return {"phone": phone, "success": False, "message_id": None, "error": str(e)}


async def send_bulk_sms_async(
    recipients: List[Dict[str, str]],
    max_concurrent: int = 50,
) -> List[dict]:
    """
    Gửi bulk SMS bất đồng bộ, tối đa max_concurrent request cùng lúc.

    Ví dụ: 1000 SMS với max_concurrent=50 ~ 20 batch song song.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    # Dùng TCPConnector để tái sử dụng connection (connection pooling)
    connector = aiohttp.TCPConnector(limit=max_concurrent)

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [
            send_sms_async(session, r["phone"], r["message"], semaphore)
            for r in recipients
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    # Chuẩn hóa kết quả (xử lý cả trường hợp exception từ gather)
    final = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            final.append({
                "phone":   recipients[i]["phone"],
                "success": False,
                "error":   str(result),
            })
        else:
            final.append(result)

    success_count = sum(1 for r in final if r["success"])
    logger.info("Async bulk done: %d/%d success", success_count, len(final))
    return final


# Dùng từ code sync (Django management command, script):
# results = asyncio.run(send_bulk_sms_async(recipients, max_concurrent=100))

# Dùng từ FastAPI endpoint:
# @app.post("/send-bulk")
# async def api_send_bulk(data: BulkSMSRequest):
#     results = await send_bulk_sms_async(data.recipients)
#     return {"results": results}

Với 10.000 SMS, max_concurrent=100 + aiohttp cho throughput khoảng 500-800 SMS/giây trên server thông thường - gấp 4-5 lần so với ThreadPoolExecutor. Điều chỉnh max_concurrent tùy rate limit của SMS provider bạn đang dùng.

8. Nhận Delivery Report Qua Webhook

Delivery Report (DLR) là cơ chế SMS provider gửi HTTP POST đến server của bạn khi tin nhắn đến tay người nhận (hoặc thất bại). Thiếu DLR, bạn không biết SMS thực sự được giao hay không - chỉ biết đã gửi đi. Đây là phần hay bị bỏ qua nhất khi tích hợp SMS API Python.

Webhook DLR handler hoàn chỉnh với Flask

from flask import Flask, request, jsonify
import hmac
import hashlib
import json
import logging

logger = logging.getLogger(__name__)

# Các trạng thái DLR phổ biến (tùy provider có thể khác tên)
DLR_STATUS_MAP = {
    "DELIVRD":  "delivered",   # Đã giao thành công
    "EXPIRED":  "expired",     # Hết hạn, không giao được
    "DELETED":  "deleted",     # Bị xóa bởi SMSC
    "UNDELIV":  "undelivered", # Không giao được
    "ACCEPTD":  "accepted",    # SMSC đã chấp nhận, chờ giao
    "UNKNOWN":  "unknown",     # Trạng thái không xác định
    "REJECTD":  "rejected",    # Bị từ chối (số không tồn tại, v.v.)
    # Một số provider dùng numeric: "1"=delivered, "2"=failed
    "1": "delivered",
    "2": "undelivered",
}

WEBHOOK_SECRET = os.getenv("SMS_WEBHOOK_SECRET", "")

def verify_webhook_signature(payload_bytes: bytes, signature: str) -> bool:
    """Xác minh chữ ký webhook để tránh giả mạo request."""
    if not WEBHOOK_SECRET:
        return True  # Bỏ qua xác minh nếu chưa cấu hình (không khuyến nghị production)
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)


@app.route("/webhook/sms-dlr", methods=["POST"])
def sms_delivery_report():
    """
    Nhận Delivery Report từ SMS provider.

    Provider sẽ POST JSON hoặc form-data, tùy cấu hình.
    Phải trả về HTTP 200 trong vòng 3-5 giây, xử lý nặng chạy background.
    """
    # Xác minh chữ ký (nếu provider hỗ trợ)
    signature = request.headers.get("X-Signature", "")
    if signature and not verify_webhook_signature(request.data, signature):
        logger.warning("Invalid webhook signature from %s", request.remote_addr)
        return jsonify({"error": "Invalid signature"}), 401

    # Parse payload - hỗ trợ cả JSON và form-data
    if request.is_json:
        payload = request.get_json(silent=True) or {}
    else:
        payload = request.form.to_dict()

    # Chuẩn hóa các field tên khác nhau theo provider
    message_id = (
        payload.get("message_id")
        or payload.get("msgid")
        or payload.get("id")
    )
    raw_status = (
        payload.get("status")
        or payload.get("dlrstatus")
        or payload.get("stat")
        or "UNKNOWN"
    )
    phone = payload.get("to") or payload.get("recipient") or payload.get("msisdn")
    timestamp = payload.get("done_date") or payload.get("timestamp")

    normalized_status = DLR_STATUS_MAP.get(str(raw_status).upper(), "unknown")

    logger.info(
        "DLR received: message_id=%s, phone=%s, status=%s->%s",
        message_id, phone, raw_status, normalized_status
    )

    # Gửi update database sang Celery task để trả về 200 ngay
    if message_id:
        update_sms_status_task.delay(
            message_id=message_id,
            status=normalized_status,
            phone=phone,
            raw_payload=payload,
        )

    # QUAN TRỌNG: Trả về 200 ngay, không chờ database update xong
    return jsonify({"received": True, "message_id": message_id}), 200

Để nhận được DLR, bạn cần đăng ký webhook URL với SMS provider và URL đó phải public (có thể dùng ngrok khi dev local). Với production, đảm bảo endpoint có HTTPS và xử lý idempotency - provider có thể gửi lại DLR nhiều lần cho cùng một message_id.

9. Error Handling, Logging và Retry

SMS API có thể thất bại vì nhiều lý do: timeout, rate limit (HTTP 429), lỗi server provider (5xx), số điện thoại không hợp lệ (400), hoặc mất kết nối mạng. Production code phải xử lý được tất cả trường hợp này một cách graceful.

Retry với exponential backoff

import time
import random
import logging
from typing import Optional

logger = logging.getLogger(__name__)

# Lỗi nên retry (transient) vs lỗi KHÔNG nên retry (permanent)
RETRYABLE_STATUS_CODES    = {429, 500, 502, 503, 504}  # Rate limit, server error
NON_RETRYABLE_STATUS_CODES = {400, 401, 403, 404}       # Bad request, auth error - retry vô ích

def send_sms_with_retry(
    phone: str,
    message: str,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
) -> dict:
    """
    Gửi SMS với retry exponential backoff + jitter.

    Công thức delay: min(base_delay * 2^attempt + random(0, 1), max_delay)
    Ví dụ với base_delay=1: 1.x, 2.x, 4.x, 8.x giây
    """
    last_error = None

    for attempt in range(max_retries + 1):  # attempt 0 là lần đầu, không phải retry
        try:
            headers = {
                "Authorization": f"Bearer {SMS_API_KEY}",
                "Content-Type":  "application/json",
            }
            payload = {"to": phone, "from": SMS_SENDER, "message": message}

            response = requests.post(
                SMS_API_URL,
                json=payload,
                headers=headers,
                timeout=10,
            )

            # Không retry lỗi permanent
            if response.status_code in NON_RETRYABLE_STATUS_CODES:
                logger.error(
                    "SMS permanent error %d for %s: %s",
                    response.status_code, phone, response.text[:200]
                )
                return {
                    "success":    False,
                    "message_id": None,
                    "error":      f"HTTP {response.status_code}: {response.text[:100]}",
                    "retried":    attempt,
                }

            response.raise_for_status()
            data = response.json()
            if attempt > 0:
                logger.info("SMS sent after %d retries to %s", attempt, phone)
            return {
                "success":    True,
                "message_id": data.get("message_id"),
                "error":      None,
                "retried":    attempt,
            }

        except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
            last_error = str(e)
            error_type = "timeout" if isinstance(e, requests.exceptions.Timeout) else "connection"
            logger.warning("SMS %s on attempt %d for %s", error_type, attempt + 1, phone)

        except requests.exceptions.HTTPError as e:
            last_error = str(e)
            status_code = e.response.status_code if e.response else 0
            if status_code not in RETRYABLE_STATUS_CODES:
                return {"success": False, "message_id": None, "error": last_error, "retried": attempt}
            logger.warning("SMS HTTP %d on attempt %d for %s", status_code, attempt + 1, phone)

        # Tính delay với jitter để tránh thundering herd
        if attempt < max_retries:
            delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
            logger.info("Retry %d/%d in %.1fs for %s", attempt + 1, max_retries, delay, phone)
            time.sleep(delay)

    logger.error("SMS failed after %d attempts for %s: %s", max_retries + 1, phone, last_error)
    return {"success": False, "message_id": None, "error": last_error, "retried": max_retries}

Với kỹ sư hệ thống backend cần biết

Với kỹ sư backend quản lý hệ thống SMS ở quy mô production, 3 điều bắt buộc phải làm:

• Structured logging: log đầy đủ phone, message_id, attempt_count, duration_ms vào mỗi SMS event - không dùng print() trong production
• Dead letter queue: SMS thất bại sau hết số retry phải vào DLQ để review và gửi lại thủ công, không được mất im lặng
• Circuit breaker: nếu SMS provider liên tục fail trong 5 phút, tạm ngắt toàn bộ request thay vì tiếp tục retry - tránh flood log và tốn quota

Xem thêm về các pattern xử lý OTP nâng cao trong bài tích hợp SMS OTP API. Nếu hệ thống bạn đang dùng cả Zalo ZNS, bài tích hợp Zalo ZNS API có pattern tương tự nhưng dành cho ZNS với template approval flow khác.

Câu Hỏi Thường Gặp

Tích hợp SMS API Python cần những thứ gì để bắt đầu?

Cần 4 thứ: (1) Python 3.8 trở lên, (2) API key từ nhà cung cấp SMS, (3) endpoint URL của SMS provider, (4) thư viện requests hoặc httpx. Cài đặt bằng pip install requests python-dotenv, lưu API key vào file .env, gọi một POST request là có thể gửi SMS đầu tiên trong vòng 30 phút.

Nên dùng requests, httpx hay aiohttp cho SMS API?

Tuỳ volume và stack. Dùng requests nếu gửi dưới 100 SMS/lần và project là sync (Django, Flask thông thường). Dùng httpx nếu cần cả sync lẫn async hoặc cần HTTP/2. Dùng aiohttp nếu gửi bulk trên 1.000 SMS hoặc project là async-first (FastAPI). Với cùng 1.000 SMS đồng thời, aiohttp nhanh hơn requests khoảng 8-10 lần.

Làm sao tránh bị rate limit khi gửi bulk SMS?

Dùng 3 biện pháp kết hợp: (1) chia danh sách thành batch nhỏ 50-100 tin nhắn, (2) thêm delay 0.2-0.5 giây giữa các batch, (3) dùng asyncio.Semaphore để giới hạn số request đồng thời. Đọc tài liệu provider để biết rate limit cụ thể - thường là 100-500 request/giây. Khi nhận HTTP 429, dừng ngay và chờ ít nhất 1 giây trước khi tiếp tục.

OTP gửi qua SMS nên hết hạn sau bao lâu?

Thời gian hết hạn chuẩn là 5 phút (300 giây) - đủ để người dùng nhận và nhập OTP, nhưng không đủ dài để bị tấn công brute-force. Kết hợp thêm: giới hạn 3 lần nhập sai, rate limit 1 lần gửi/60 giây mỗi số điện thoại, lưu OTP dạng hash (SHA-256) thay vì plaintext, và xóa khỏi Redis ngay sau khi xác thực thành công hoặc hết hạn.

Webhook DLR không nhận được, cần kiểm tra gì?

Kiểm tra 5 điểm theo thứ tự: (1) URL webhook đã đăng ký đúng với SMS provider chưa, (2) server có public IP/domain không (localhost không nhận được), (3) port 80/443 có mở firewall không, (4) endpoint có trả về HTTP 200 không (không phải 301, 404, 500), (5) server có xử lý kịp trong 3-5 giây không. Dùng ngrok trong môi trường dev để test webhook từ provider thực.

Gửi SMS trong Django nên dùng Celery hay gọi trực tiếp?

Luôn dùng Celery cho production. Gọi trực tiếp trong view có thể làm HTTP response chậm 1-5 giây nếu SMS provider phản hồi chậm, hoặc gây timeout toàn bộ request nếu provider down. Celery với Redis broker là pattern chuẩn: view trả về response ngay lập tức, SMS gửi bất đồng bộ background với auto-retry. Chỉ gọi trực tiếp trong management command hoặc script offline.

Cách test SMS integration mà không tốn tiền tin nhắn thật?

Dùng 3 cách: (1) Mock requests.post với unittest.mock.patch trong unit test, (2) dùng SMSC simulator hoặc sandbox environment của SMS provider nếu họ cung cấp, (3) tạo DummySMSBackend chỉ log ra console thay vì gọi API thật - switch giữa backend thật và dummy bằng environment variable SMS_BACKEND=dummy. Cách 1 và 3 không tốn chi phí và chạy được trong CI/CD pipeline.

Tích hợp SMS API với FastAPI khác Flask như thế nào?

FastAPI native async nên dùng trực tiếp async def endpoint + aiohttp hoặc httpx async client, không cần Celery cho bài toán đơn giản. Khác biệt chính: FastAPI có request validation tự động qua Pydantic (khai báo schema một lần), còn Flask phải validate thủ công. Background task trong FastAPI dùng BackgroundTasks built-in cho tác vụ nhẹ, hoặc Celery/ARQ cho task queue phức tạp hơn. Xem thêm hướng dẫn tích hợp SMS API PHP nếu dự án của bạn dùng PHP backend song song.

Kết Luận

Tích hợp SMS API với Python có thể bắt đầu đơn giản chỉ với requests và 30 dòng code, nhưng production-ready cần thêm: retry với exponential backoff, async cho throughput cao, webhook DLR để theo dõi delivery, và Celery để tách SMS task ra khỏi HTTP request cycle. Ba nguyên tắc cốt lõi để tránh lỗi phổ biến nhất: luôn set timeout cho HTTP request, không hard-code API key (dùng biến môi trường), và không gọi SMS API đồng bộ trong view/handler.

Nếu bạn cần tư vấn tích hợp SMS API cho hệ thống Python hoặc cần API key để bắt đầu, liên hệ đội kỹ thuật qua:

• Zalo: 0988 769 317
• Hotline: 0988 769 317