Airtel Money Collections API in Django: USSD Push to Callback

This guide adds Airtel Money payments to a Django app with Airtel’s Open API. The customer gets a USSD prompt on their phone and approves the payment with their PIN. By the end you’ll have:

• a client.py with token caching, the payment request and the status check,
• an AirtelPayment model that records every attempt, saved before Airtel is called,
• views to start a payment, poll it, and receive Airtel’s callback,
• a callback that confirms every payment with the status API and settles it once, inside a row lock,
• handling for TIP (in progress) and TA (ambiguous) that never fails a payment early,
• a check_airtel_payments command for callbacks that never arrive,
• tests that run the whole flow without calling Airtel.

Every code block was run in fresh Django 5.2 and 6.1 projects, and the tests at the end pass. The Laravel version of this guide has the same design.

Airtel’s public documentation is thinner than MTN’s or Safaricom’s, and some details vary by country. Where this guide relies on behaviour reported by developers rather than documented by Airtel, it says so, and the code is written to stay safe either way.

Before you start

1. Create an application in the Airtel developer portal (developers.airtel.africa) for your country, with the Collection product.
2. Copy the client ID and client secret from Key Management. UAT (sandbox) and production have separate credentials, and production needs Airtel to approve your business.
3. Register your callback URL on the application, for example https://your-domain/airtel/callback/. It must be public HTTPS; for local development, use a tunnel.
4. Install requests, create the app and set the credentials in your environment. Never commit them:

pip install "django>=5.2" requests
python manage.py startapp airtel
mkdir -p airtel/management/commands
touch airtel/management/__init__.py airtel/management/commands/__init__.py
export AIRTEL_CLIENT_ID=your-client-id
export AIRTEL_CLIENT_SECRET=your-client-secret

Add "airtel" to INSTALLED_APPS, and the settings:

# settings.py (add to your existing settings)
import os

AIRTEL = {
    "BASE_URL": "https://openapi.airtel.africa" if os.environ.get("AIRTEL_ENV") == "production" else "https://openapiuat.airtel.africa",
    "CLIENT_ID": os.environ.get("AIRTEL_CLIENT_ID", ""),
    "CLIENT_SECRET": os.environ.get("AIRTEL_CLIENT_SECRET", ""),
    "COUNTRY": os.environ.get("AIRTEL_COUNTRY", "UG"),
    "CURRENCY": os.environ.get("AIRTEL_CURRENCY", "UGX"),
    "COUNTRY_CODE": os.environ.get("AIRTEL_COUNTRY_CODE", "256"),
}

The model

With Airtel, you choose the transaction ID and send it with the payment request. That ID is how callbacks and status checks find the payment, so it is unique.

# airtel/models.py
import secrets

from django.db import models


def new_transaction_id():
    return secrets.token_hex(10)  # 20 characters, unique per attempt, never reused


class AirtelPayment(models.Model):
    transaction_id = models.CharField(max_length=32, unique=True, default=new_transaction_id, editable=False)  # sent as transaction.id
    order_id = models.CharField(max_length=64, db_index=True)
    amount = models.PositiveBigIntegerField()  # UGX, KES, TZS...: whole units
    currency = models.CharField(max_length=3)
    msisdn = models.CharField(max_length=15)  # without the country code, e.g. 752123456
    status = models.CharField(max_length=10, default="pending")  # pending | paid | failed
    airtel_money_id = models.CharField(max_length=64, blank=True)  # Airtel's reference once paid
    failure_reason = models.CharField(max_length=190, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        indexes = [models.Index(fields=["status", "created_at"])]

Run python manage.py makemigrations airtel and python manage.py migrate.

The Airtel client

The client handles the three calls you need: get a token, request a payment, and check its status. Every call carries the X-Country and X-Currency headers for your market. Network failures and 5xx answers raise an ambiguous error: Airtel may have received the request.

# airtel/client.py
import re
from urllib.parse import quote

import requests
from django.conf import settings
from django.core.cache import cache


class AirtelError(Exception):
    """ambiguous=True: Airtel may have received the request. Keep the payment pending and check it."""

    def __init__(self, message, ambiguous=False, code=""):
        super().__init__(message)
        self.ambiguous = ambiguous
        self.code = code


def cfg(key):
    return settings.AIRTEL[key]


def call(method, path, **kwargs):
    try:
        res = requests.request(method, cfg("BASE_URL").rstrip("/") + path, timeout=30, **kwargs)
    except requests.RequestException as exc:
        raise AirtelError(f"Airtel unreachable: {exc}", ambiguous=True) from exc
    try:
        data = res.json()
    except ValueError:
        data = {}
    return res, data if isinstance(data, dict) else {}


def token():
    cached = cache.get("airtel_token")
    if cached:
        return cached
    res, data = call("POST", "/auth/oauth2/token", json={
        "client_id": cfg("CLIENT_ID"), "client_secret": cfg("CLIENT_SECRET"), "grant_type": "client_credentials",
    }, headers={"Accept": "*/*"})
    if not res.ok or not data.get("access_token"):
        raise AirtelError(f"Airtel token failed: {res.status_code}", ambiguous=res.status_code >= 500)
    try:
        ttl = int(data.get("expires_in") or 180)
    except (TypeError, ValueError):
        ttl = 180
    cache.set("airtel_token", data["access_token"], max(60, ttl - 60))  # refresh a minute early
    return data["access_token"]


def headers():
    return {"Authorization": f"Bearer {token()}", "Accept": "*/*", "X-Country": cfg("COUNTRY"), "X-Currency": cfg("CURRENCY")}


def msisdn(raw, country_code=None):
    """Airtel wants the number without the country code or leading 0: "0752 123 456" -> "752123456"."""
    country_code = country_code or cfg("COUNTRY_CODE")
    digits = re.sub(r"\D", "", str(raw))
    if digits.startswith("00"):
        digits = digits[2:]
    if digits.startswith(country_code):
        digits = digits[len(country_code):]
    if digits.startswith("0"):
        digits = digits[1:]
    if not 8 <= len(digits) <= 10:
        raise ValueError(f"Not a valid mobile number: {raw}")
    return digits


def request_payment(transaction_id, amount, number, reference):
    """Sends the USSD prompt. A success only means Airtel received the request."""
    res, data = call("POST", "/merchant/v1/payments/", headers=headers(), json={
        "reference": str(reference)[:64],
        "subscriber": {"country": cfg("COUNTRY"), "currency": cfg("CURRENCY"), "msisdn": number},
        "transaction": {"amount": amount, "country": cfg("COUNTRY"), "currency": cfg("CURRENCY"), "id": transaction_id},
    })
    status = data.get("status") if isinstance(data.get("status"), dict) else {}
    if res.status_code >= 500:
        raise AirtelError(f"Airtel HTTP {res.status_code}", ambiguous=True)
    if not res.ok or status.get("success") is False:
        code = status.get("response_code") or status.get("code") or res.status_code
        raise AirtelError(f"Airtel rejected the payment: {code} {status.get('message') or ''}".strip(), code=str(code))


def payment_status(transaction_id):
    """The source of truth: {"status": "TS" | "TF" | "TIP" | "TA" | "", "airtel_money_id": ..., "code": ...}."""
    res, data = call("GET", f"/standard/v1/payments/{quote(transaction_id, safe='')}", headers=headers())
    if res.status_code >= 500:
        raise AirtelError(f"Airtel HTTP {res.status_code}", ambiguous=True)
    txn = (data.get("data") or {}).get("transaction") if isinstance(data.get("data"), dict) else None
    txn = txn if isinstance(txn, dict) else {}
    status = data.get("status") if isinstance(data.get("status"), dict) else {}
    return {
        "status": str(txn.get("status") or "").upper(),
        "airtel_money_id": str(txn.get("airtel_money_id") or ""),
        "code": str(status.get("response_code") or status.get("code") or txn.get("message") or ""),
    }

Change AIRTEL_COUNTRY_CODE to your market’s calling code, for example 254 for Kenya. Not sure of the format? Check the number in the African Phone Number Formatter.

Start a payment and settle it once

# airtel/services.py
import logging

from django.conf import settings
from django.db import transaction

from . import client
from .models import AirtelPayment

logger = logging.getLogger(__name__)


def start(order_id, amount, phone):
    """Saves the payment, then sends the USSD prompt. Raises ValueError for bad input."""
    if not isinstance(amount, int) or isinstance(amount, bool) or amount <= 0:
        raise ValueError("Amount must be a whole number greater than zero")
    number = client.msisdn(phone)
    payment = AirtelPayment.objects.create(order_id=str(order_id)[:64], amount=amount,
                                           currency=settings.AIRTEL["CURRENCY"], msisdn=number)  # saved BEFORE the call
    try:
        client.request_payment(payment.transaction_id, amount, number, f"Order {payment.order_id}")
    except client.AirtelError as exc:
        if exc.ambiguous:
            logger.warning("Airtel request for %s may have gone through: %s", payment.transaction_id, exc)  # stays pending
        else:
            AirtelPayment.objects.filter(pk=payment.pk, status="pending").update(status="failed", failure_reason=str(exc)[:190])
    payment.refresh_from_db()
    return payment


def confirm(transaction_id):
    """Asks Airtel for the real status and settles the payment once. Safe to call any number of times."""
    if not AirtelPayment.objects.filter(transaction_id=transaction_id, status="pending").exists():
        return AirtelPayment.objects.filter(transaction_id=transaction_id).first()  # unknown, or already settled
    try:
        result = client.payment_status(transaction_id)  # network call outside the lock
    except client.AirtelError as exc:
        logger.warning("Airtel status check failed for %s: %s", transaction_id, exc)
        return AirtelPayment.objects.get(transaction_id=transaction_id)
    with transaction.atomic():
        payment = AirtelPayment.objects.select_for_update().get(transaction_id=transaction_id)
        if payment.status != "pending":
            return payment  # another callback or check settled it first
        if result["status"] == "TS":
            payment.status, payment.airtel_money_id = "paid", result["airtel_money_id"]
            payment.save(update_fields=["status", "airtel_money_id", "updated_at"])
            # Fulfil the order here, e.g. with transaction.on_commit(lambda: fulfil(payment.order_id))
        elif result["status"] == "TF":
            payment.status, payment.failure_reason = "failed", (result["code"] or "TF")[:190]
            payment.save(update_fields=["status", "failure_reason", "updated_at"])
        # TIP, TA or no answer: leave it pending. check_airtel_payments asks again later.
    return payment

What the code does with each status:

Response codes like DP00800001006 are explained in the Airtel Money error code reference.

The views

Developers report callbacks shaped like {"transaction": {"id": "...", "status_code": "TS"}}, but the shape isn’t consistently documented. The callback view only reads transaction.id and asks Airtel, so a different shape, or a fake callback, can’t mark anything paid.

# airtel/views.py
import json

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_GET, require_POST

from . import services
from .models import AirtelPayment


@require_POST
def start_payment(request):
    """POST order_id, amount, phone (from your own checkout page, with its CSRF token)."""
    try:
        amount = int(request.POST.get("amount", ""))  # take it from your order record in a real app
        payment = services.start(request.POST.get("order_id", ""), amount, request.POST.get("phone", ""))
    except ValueError as exc:
        return JsonResponse({"error": str(exc)}, status=422)
    if payment.status == "failed":
        return JsonResponse({"error": "Airtel Money could not send the payment prompt. Please try again."}, status=502)
    return JsonResponse({"transaction_id": payment.transaction_id, "message": "Check your phone and enter your Airtel Money PIN."}, status=202)


@require_GET
def payment_status(request, transaction_id):
    """Your frontend polls this, not Airtel."""
    payment = AirtelPayment.objects.filter(transaction_id=transaction_id).first()
    if payment is None:
        return JsonResponse({"error": "Not found"}, status=404)
    return JsonResponse({"status": payment.status})


@csrf_exempt  # Airtel can't send a CSRF token; the body is never trusted
@require_POST
def callback(request):
    try:
        body = json.loads(request.body or b"{}")
    except ValueError:
        body = {}
    txn = body.get("transaction") if isinstance(body, dict) else None
    transaction_id = txn.get("id") if isinstance(txn, dict) else None
    if isinstance(transaction_id, str) and AirtelPayment.objects.filter(transaction_id=transaction_id).exists():
        services.confirm(transaction_id)  # asks Airtel; the body's status_code is ignored
    return JsonResponse({"ok": True})  # always acknowledge

# airtel/urls.py
from django.urls import path

from . import views

urlpatterns = [
    path("airtel/pay/", views.start_payment),
    path("airtel/payments/<str:transaction_id>/", views.payment_status),
    path("airtel/callback/", views.callback),
]

Include them in your project’s urls.py with path("", include("airtel.urls")).

Catch callbacks that never arrive

Callbacks can be late or lost, and TA payments need another look. This command re-checks every payment that has been pending for more than two minutes, for up to a day:

# airtel/management/commands/check_airtel_payments.py
from datetime import timedelta

from django.core.management.base import BaseCommand
from django.utils import timezone

from airtel import services
from airtel.models import AirtelPayment


class Command(BaseCommand):
    help = "Ask Airtel about pending payments whose callback never arrived. Run it every few minutes."

    def handle(self, *args, **options):
        now = timezone.now()
        ids = AirtelPayment.objects.filter(status="pending", created_at__lt=now - timedelta(minutes=2),
                                           created_at__gt=now - timedelta(days=1)).values_list("transaction_id", flat=True)[:200]
        counts = {}
        for transaction_id in ids:
            status = services.confirm(transaction_id).status
            counts[status] = counts.get(status, 0) + 1
        self.stdout.write(", ".join(f"{n} {s}" for s, n in counts.items()) or "Nothing to check.")

Run it from cron: */5 * * * * cd /srv/app && .venv/bin/python manage.py check_airtel_payments.

Test it without calling Airtel

These tests patch requests with unittest.mock. They check the request format, that a callback claiming success is ignored while Airtel says TIP, that a real TS marks the payment paid, that a settled payment never changes, that TA and an unreachable Airtel leave it pending, and that a timeout on the request leaves it pending instead of failing it.

# airtel/tests.py
import json
from datetime import timedelta
from io import StringIO
from unittest import mock

import requests
from django.core.cache import cache
from django.core.management import call_command
from django.test import Client, TestCase, override_settings
from django.utils import timezone

from . import client, services
from .models import AirtelPayment

AIRTEL = {"BASE_URL": "https://airtel.test", "CLIENT_ID": "id", "CLIENT_SECRET": "secret",
          "COUNTRY": "UG", "CURRENCY": "UGX", "COUNTRY_CODE": "256"}


def reply(status, data=None):
    res = mock.Mock(status_code=status, ok=200 <= status < 300)
    res.json.return_value = data if data is not None else {}
    return res


@override_settings(AIRTEL=AIRTEL)
class AirtelTests(TestCase):
    def setUp(self):
        cache.clear()
        self.airtel_status = "TIP"
        self.calls = []
        self.request_mode = "ok"  # ok, rejected, timeout
        self.status_down = False
        patcher = mock.patch("airtel.client.requests.request", side_effect=self.fake)
        patcher.start()
        self.addCleanup(patcher.stop)

    def fake(self, method, url, timeout=None, headers=None, json=None):
        self.calls.append({"method": method, "url": url, "headers": headers, "json": json})
        if url.endswith("/auth/oauth2/token"):
            return reply(200, {"access_token": "test-token", "expires_in": 3600})
        if url.endswith("/merchant/v1/payments/"):
            if self.request_mode == "timeout":
                raise requests.Timeout("timed out")
            if self.request_mode == "rejected":
                return reply(400, {"status": {"success": False, "response_code": "DP00800001006", "message": "Invalid MSISDN"}})
            return reply(200, {"status": {"success": True, "code": "200", "result_code": "ESB000010"}})
        if "/standard/v1/payments/" in url:
            if self.status_down:
                return reply(503)
            return reply(200, {"data": {"transaction": {"status": self.airtel_status, "airtel_money_id": "MP261106.0741.A00001"}},
                               "status": {"success": True}})
        raise AssertionError(f"unexpected {method} {url}")

    def start(self, phone="0752 123 456", amount="25000"):
        res = self.client.post("/airtel/pay/", {"order_id": "1001", "amount": amount, "phone": phone})
        return res, AirtelPayment.objects.order_by("-id").first()

    def callback(self, body):
        return Client(enforce_csrf_checks=True).post("/airtel/callback/", json.dumps(body), content_type="application/json")

    def test_the_payment_is_confirmed_with_the_status_api_not_the_callback(self):
        res, p = self.start()
        self.assertEqual((res.status_code, res.json()["transaction_id"]), (202, p.transaction_id))
        sent = [c for c in self.calls if c["url"].endswith("/merchant/v1/payments/")][0]
        self.assertEqual((sent["headers"]["X-Country"], sent["headers"]["X-Currency"], sent["headers"]["Authorization"]), ("UG", "UGX", "Bearer test-token"))
        self.assertEqual(sent["json"]["subscriber"], {"country": "UG", "currency": "UGX", "msisdn": "752123456"})
        self.assertEqual(sent["json"]["transaction"], {"amount": 25000, "country": "UG", "currency": "UGX", "id": p.transaction_id})

        self.assertEqual(self.callback({"transaction": {"id": p.transaction_id, "status_code": "TS"}}).status_code, 200)
        self.assertEqual(AirtelPayment.objects.get(pk=p.pk).status, "pending")  # the body says TS, Airtel says TIP
        self.airtel_status = "TS"
        self.callback({"transaction": {"id": p.transaction_id, "status_code": "TS"}})
        p.refresh_from_db()
        self.assertEqual((p.status, p.airtel_money_id), ("paid", "MP261106.0741.A00001"))
        self.assertEqual(self.client.get(f"/airtel/payments/{p.transaction_id}/").json(), {"status": "paid"})

    def test_failed_and_final(self):
        self.airtel_status = "TF"
        _, p = self.start(phone="+256752123456")
        self.callback({"transaction": {"id": p.transaction_id}})
        self.assertEqual(AirtelPayment.objects.get(pk=p.pk).status, "failed")
        self.airtel_status = "TS"
        services.confirm(p.transaction_id)  # a late, different answer
        self.assertEqual(AirtelPayment.objects.get(pk=p.pk).status, "failed")

        self.airtel_status = "TS"
        _, q = self.start()
        services.confirm(q.transaction_id)
        self.airtel_status = "TF"
        self.callback({"transaction": {"id": q.transaction_id}})
        self.assertEqual(AirtelPayment.objects.get(pk=q.pk).status, "paid")

    def test_ta_and_an_unreachable_airtel_leave_it_pending(self):
        _, p = self.start()
        self.airtel_status = "TA"
        self.assertEqual(services.confirm(p.transaction_id).status, "pending")
        self.status_down = True
        self.assertEqual(services.confirm(p.transaction_id).status, "pending")

    def test_a_timeout_keeps_it_pending_and_a_rejection_fails_it(self):
        self.request_mode = "timeout"
        res, p = self.start()
        self.assertEqual((res.status_code, p.status), (202, "pending"))
        self.request_mode = "rejected"
        res, q = self.start()
        self.assertEqual((res.status_code, q.status), (502, "failed"))
        self.assertIn("DP00800001006", q.failure_reason)

    def test_unknown_callbacks_and_bad_input_are_ignored(self):
        self.assertEqual(self.callback({"transaction": {"id": "not-ours", "status_code": "TS"}}).status_code, 200)
        self.assertEqual(self.callback({"hello": "world"}).status_code, 200)
        self.assertFalse(any("/standard/v1/payments/" in c["url"] for c in self.calls))
        self.assertEqual(self.start(phone="12345")[0].status_code, 422)
        self.assertEqual(self.start(amount="12.5")[0].status_code, 422)
        self.assertEqual(AirtelPayment.objects.count(), 0)

    def test_msisdn_formats(self):
        self.assertEqual([client.msisdn("0752 123 456"), client.msisdn("+256752123456"), client.msisdn("00256752123456")], ["752123456"] * 3)
        self.assertEqual(client.msisdn("254733123456", "254"), "733123456")

    def test_the_token_is_cached(self):
        self.start()
        self.start()
        self.assertEqual(len([c for c in self.calls if c["url"].endswith("/auth/oauth2/token")]), 1)

    def test_the_command_checks_payments_whose_callback_never_came(self):
        _, p = self.start()
        _, recent = self.start()
        AirtelPayment.objects.filter(pk=p.pk).update(created_at=timezone.now() - timedelta(minutes=5))
        self.airtel_status = "TS"
        call_command("check_airtel_payments", stdout=StringIO())
        self.assertEqual((AirtelPayment.objects.get(pk=p.pk).status, AirtelPayment.objects.get(pk=recent.pk).status), ("paid", "pending"))

python manage.py test airtel

Common errors and fixes

Going to production

When Airtel approves your application for production:

• Set AIRTEL_ENV=production and use your production client ID and secret.
• Register the production callback URL (HTTPS) on the production application.
• Check whether your app is enabled for /merchant/v1/payments/ or the newer v2 endpoint, which encrypts the payload. This guide covers v1.
• Use a cache all your processes share for the token, and keep check_airtel_payments running.
• Reconcile daily: compare your paid payments with the Airtel Money statement, especially any that went through TA.

Want all of this already built, tested and combined with MTN MoMo, M-Pesa, Flutterwave, Paystack and Pesapal? The Django starter kit has it ready.

Using the Next.js App Router? See Airtel Money collections in Next.js: the same design in route handlers.