MTN MoMo Collections API in Python & Django: Full Integration

This guide adds MTN Mobile Money payments to a Django 5.2 or 6.x project. By the end you’ll have:

• a small MoMo client built on requests, with token caching,
• a Payment model that records every attempt,
• views to start a payment, check its status and receive callbacks,
• an idempotent confirm_payment function that never trusts the callback body,
• a reconcile_momo management command for callbacks that never arrive,
• tests that run without calling MTN.

Every code block in this guide was run in fresh Django 5.2 and 6.1 projects, and its tests pass on both.

Before you start

1. Sign up at momodeveloper.mtn.com and subscribe to the Collections product.
2. Copy your primary subscription key from your profile.
3. Create a sandbox API user and API key. Steps 1 and 2 of the MoMo Request Builder generate the cURL commands for you.
4. Install requests and create an app called payments:

pip install "django>=5.2" requests
python manage.py startapp payments

Add "payments" to INSTALLED_APPS.

Settings

Keep credentials in environment variables and read them once in settings.py:

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

MOMO = {
    "BASE_URL": os.environ.get("MOMO_BASE_URL", "https://sandbox.momodeveloper.mtn.com"),
    "TARGET_ENV": os.environ.get("MOMO_TARGET_ENV", "sandbox"),
    "CURRENCY": os.environ.get("MOMO_CURRENCY", "EUR"),
    "SUBSCRIPTION_KEY": os.environ.get("MOMO_SUBSCRIPTION_KEY", ""),
    "API_USER": os.environ.get("MOMO_API_USER", ""),
    "API_KEY": os.environ.get("MOMO_API_KEY", ""),
    # Public HTTPS address of this site, e.g. "https://shop.example.com".
    # Leave empty to build it from the incoming request.
    "CALLBACK_BASE_URL": os.environ.get("MOMO_CALLBACK_BASE_URL", ""),
}

Set MOMO_CALLBACK_BASE_URL in production. Behind a load balancer or proxy, the request’s own host can be an internal name, and MTN only sends callbacks to the host registered for your API user.

The Payment model

Every attempt gets its own row, keyed by the UUID you send to MTN as X-Reference-Id. That row is how you match callbacks, avoid fulfilling an order twice, and answer support questions.

# payments/models.py
import uuid

from django.db import models


class Payment(models.Model):
    class Status(models.TextChoices):
        PENDING = "pending"
        PAID = "paid"
        FAILED = "failed"

    reference_id = models.UUIDField(default=uuid.uuid4, unique=True, editable=False)  # X-Reference-Id
    order_id = models.CharField(max_length=64, db_index=True)  # sent as externalId
    amount = models.DecimalField(max_digits=12, decimal_places=2)
    currency = models.CharField(max_length=3)
    phone = models.CharField(max_length=15)
    status = models.CharField(max_length=10, choices=Status.choices, default=Status.PENDING)
    financial_transaction_id = models.CharField(max_length=64, blank=True)
    failure_reason = models.CharField(max_length=64, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return f"{self.order_id} {self.amount} {self.currency} ({self.status})"

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

The MoMo client

Three API calls cover Collections: get a token, request a payment, and check its status. The token is cached in Django’s cache until a minute before it expires, so you don’t fetch a new one for every payment.

# payments/momo.py
import uuid
from decimal import Decimal

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

TIMEOUT = 30  # seconds


def _cfg(key):
    return settings.MOMO[key]


def _headers(**extra):
    return {"Ocp-Apim-Subscription-Key": _cfg("SUBSCRIPTION_KEY"), **extra}


def get_token():
    token = cache.get("momo:token")
    if token:
        return token

    # No body: requests sends Content-Length: 0, which MoMo requires.
    res = requests.post(
        f"{_cfg('BASE_URL')}/collection/token/",
        auth=(_cfg("API_USER"), _cfg("API_KEY")),
        headers=_headers(),
        timeout=TIMEOUT,
    )
    res.raise_for_status()
    data = res.json()
    ttl = max(60, int(data.get("expires_in", 3600)) - 60)  # refresh a minute early
    cache.set("momo:token", data["access_token"], ttl)
    return data["access_token"]


def request_to_pay(amount, phone, order_id, callback_url=None, reference_id=None):
    """Sends a payment prompt to the customer's phone. Returns the X-Reference-Id."""
    reference_id = str(reference_id or uuid.uuid4())
    headers = _headers(**{
        "Authorization": f"Bearer {get_token()}",
        "X-Reference-Id": reference_id,
        "X-Target-Environment": _cfg("TARGET_ENV"),
    })
    if callback_url:
        headers["X-Callback-Url"] = callback_url

    res = requests.post(
        f"{_cfg('BASE_URL')}/collection/v1_0/requesttopay",
        headers=headers,
        json={
            "amount": format_amount(amount),  # a string, no trailing ".00"
            "currency": _cfg("CURRENCY"),
            "externalId": str(order_id),
            "payer": {"partyIdType": "MSISDN", "partyId": normalise_phone(phone)},
            "payerMessage": f"Payment for order {order_id}",
            "payeeNote": f"Order {order_id}",
        },
        timeout=TIMEOUT,
    )
    res.raise_for_status()  # MoMo answers 202 Accepted on success
    return reference_id


def get_status(reference_id):
    """The source of truth for a payment: PENDING, SUCCESSFUL or FAILED."""
    res = requests.get(
        f"{_cfg('BASE_URL')}/collection/v1_0/requesttopay/{reference_id}",
        headers=_headers(**{
            "Authorization": f"Bearer {get_token()}",
            "X-Target-Environment": _cfg("TARGET_ENV"),
        }),
        timeout=TIMEOUT,
    )
    res.raise_for_status()
    return res.json()


def format_amount(amount):
    """Decimal("500.00") -> "500", Decimal("12.50") -> "12.5"."""
    return format(Decimal(str(amount)).normalize(), "f")


def normalise_phone(raw, country_code="256"):
    """'+256 772 123456', '00256772123456', '0772123456' or '772123456' -> '256772123456'.

    Numbers starting with "+" or "00" keep their own country code.

    Only for countries where the leading 0 is a trunk prefix. In Cote d'Ivoire,
    Benin and Congo the 0 is part of the number, so don't use this there.
    """
    digits = "".join(ch for ch in raw if ch.isdigit())
    international = raw.strip().startswith("+") or digits.startswith("00")
    if digits.startswith("00"):
        digits = digits[2:]  # "00256..." international dialling prefix
    if digits.startswith(country_code + "0"):
        return country_code + digits[len(country_code) + 1:]  # "+256 (0)772..."
    if international:
        return digits  # already has a country code, e.g. MTN's "+46733123451" test numbers
    if digits.startswith("0"):
        return country_code + digits[1:]  # local format: swap the 0 for the code
    return digits if digits.startswith(country_code) else country_code + digits

Change the default country_code to match your market, for example "233" for Ghana. Not sure how numbers are written in another country? Check them in the African Phone Number Formatter.

Confirm payments safely

This is the most important function in the integration. Callbacks and your scheduled check both end up here. It locks the row, asks MTN for the real status, checks the amount and currency, and updates the payment only once.

# payments/services.py
import logging
from decimal import Decimal

from django.db import transaction

from . import momo
from .models import Payment

logger = logging.getLogger(__name__)


def confirm_payment(reference_id):
    """Ask MTN for the real status and update the payment. Safe to call repeatedly."""
    with transaction.atomic():
        payment = Payment.objects.select_for_update().filter(reference_id=reference_id).first()
        if payment is None or payment.status != Payment.Status.PENDING:
            return payment  # unknown, or already handled

        result = momo.get_status(payment.reference_id)
        status = str(result.get("status", "")).upper()

        if status == "SUCCESSFUL":
            amount_ok = Decimal(str(result.get("amount", "0"))) == payment.amount
            currency_ok = str(result.get("currency", "")).upper() == payment.currency
            if amount_ok and currency_ok:
                payment.status = Payment.Status.PAID
                payment.financial_transaction_id = result.get("financialTransactionId") or ""
                payment.save(update_fields=["status", "financial_transaction_id", "updated_at"])
                # Fulfil the order here, e.g. send a signal or enqueue an email.
            else:
                logger.error("MoMo amount/currency mismatch for %s: %s", payment.reference_id, result)

        elif status == "FAILED":
            reason = result.get("reason")
            code = reason.get("code") if isinstance(reason, dict) else reason
            payment.status = Payment.Status.FAILED
            payment.failure_reason = str(code or "")[:64]
            payment.save(update_fields=["status", "failure_reason", "updated_at"])

        # PENDING: leave it. A later callback or reconcile_momo will finish it.
        return payment

select_for_update() stops two requests from confirming the same payment at the same moment, for example a duplicate callback racing the reconcile command. The failure reason is stored as MTN’s code, such as APPROVAL_REJECTED or NOT_ENOUGH_FUNDS. The MoMo Callback Validator explains every code.

The views

There are three endpoints: start a payment, let your frontend poll your own database, and receive MTN’s callback.

# payments/views.py
import json
import logging
import uuid
from decimal import Decimal, InvalidOperation

import requests
from django.conf import settings
from django.http import HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404
from django.urls import reverse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_GET, require_http_methods, require_POST

from . import momo
from .models import Payment
from .services import confirm_payment

logger = logging.getLogger(__name__)


@require_POST
def start_payment(request):
    try:
        data = json.loads(request.body)
        amount = Decimal(str(data["amount"]))
        order_id = str(data["order_id"])[:64]
        phone = momo.normalise_phone(str(data["phone"]))
    except (ValueError, KeyError, InvalidOperation):
        return JsonResponse({"error": "Send order_id, amount and phone."}, status=400)
    if amount <= 0:
        return JsonResponse({"error": "Amount must be positive."}, status=400)

    # In a real shop, look the amount up from the order instead of trusting the client.
    payment = Payment.objects.create(
        order_id=order_id, amount=amount, phone=phone, currency=settings.MOMO["CURRENCY"],
    )

    base = settings.MOMO["CALLBACK_BASE_URL"] or request.build_absolute_uri("/").rstrip("/")
    callback_url = f"{base}{reverse('momo-callback')}?ref={payment.reference_id}"

    try:
        momo.request_to_pay(amount, phone, order_id, callback_url, payment.reference_id)
    except requests.RequestException:
        logger.exception("MoMo Request to Pay failed for %s", payment.reference_id)
        payment.status = Payment.Status.FAILED
        payment.failure_reason = "REQUEST_FAILED"
        payment.save(update_fields=["status", "failure_reason", "updated_at"])
        return JsonResponse({"error": "Could not reach MTN MoMo. Please try again."}, status=502)

    return JsonResponse(
        {"reference": str(payment.reference_id), "message": "Check your phone and enter your MoMo PIN."},
        status=202,
    )


@require_GET
def payment_status(request, reference):
    payment = get_object_or_404(Payment, reference_id=reference)
    return JsonResponse({"status": payment.status})


@csrf_exempt  # MTN can't send a CSRF token; the body is never trusted anyway
@require_http_methods(["POST", "PUT"])  # MTN's callback method varies, so accept both
def momo_callback(request):
    try:
        reference = uuid.UUID(request.GET.get("ref", ""))
    except ValueError:
        return HttpResponse(status=200)  # not ours; don't give attackers a signal

    try:
        confirm_payment(reference)
    except requests.RequestException:
        logger.exception("Could not confirm MoMo payment %s; reconcile_momo will retry", reference)
    return HttpResponse(status=200)

start_payment keeps Django’s CSRF protection, so call it from your page with the X-CSRFToken header as usual. Only the callback is exempt, and it never acts on what MTN, or anyone else, sends in the body.

Wire up the URLs:

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

from . import views

urlpatterns = [
    path("payments/momo/", views.start_payment, name="momo-start"),
    path("payments/momo/<uuid:reference>/", views.payment_status, name="momo-status"),
    path("payments/momo/callback/", views.momo_callback, name="momo-callback"),
]

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

Catch callbacks that never arrive

Callbacks can be late or lost, so add a management command that re-checks anything still pending after two minutes:

# payments/management/commands/reconcile_momo.py
from datetime import timedelta

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

from payments.models import Payment
from payments.services import confirm_payment


class Command(BaseCommand):
    help = "Re-check MTN MoMo payments that are still pending."

    def handle(self, *args, **options):
        now = timezone.now()
        pending = Payment.objects.filter(
            status=Payment.Status.PENDING,
            created_at__lt=now - timedelta(minutes=2),
            created_at__gt=now - timedelta(days=1),
        ).values_list("reference_id", flat=True)

        checked = 0
        for reference in pending:
            try:
                confirm_payment(reference)
                checked += 1
            except requests.RequestException as exc:
                self.stderr.write(f"{reference}: {exc}")
        self.stdout.write(f"Checked {checked} pending payment(s).")

Create empty __init__.py files in payments/management/ and payments/management/commands/, then run the command every five minutes from cron:

*/5 * * * * cd /srv/shop && /srv/shop/.venv/bin/python manage.py reconcile_momo

Test it without calling MTN

These tests patch requests with unittest.mock, so they run offline and in CI with no extra packages. They check the happy path, that a fake callback body is ignored, that a duplicate callback doesn’t call MTN again, and that a failure reason is stored.

# payments/tests.py
from unittest import mock

from django.core.cache import cache
from django.test import TestCase, override_settings

from .models import Payment

MOMO_TEST = {
    "BASE_URL": "https://sandbox.momodeveloper.mtn.com",
    "TARGET_ENV": "sandbox",
    "CURRENCY": "EUR",
    "SUBSCRIPTION_KEY": "test-key",
    "API_USER": "test-user",
    "API_KEY": "test-secret",
    "CALLBACK_BASE_URL": "https://shop.example.com",
}


def fake_response(status_code=200, data=None):
    res = mock.Mock(status_code=status_code)
    res.json.return_value = data or {}
    res.raise_for_status.return_value = None
    return res


@override_settings(MOMO=MOMO_TEST)
class MomoPaymentTests(TestCase):
    def setUp(self):
        cache.clear()  # forget cached tokens between tests

    def start(self, post):
        post.side_effect = [
            fake_response(200, {"access_token": "test-token", "expires_in": 3600}),
            fake_response(202),
        ]
        res = self.client.post(
            "/payments/momo/",
            {"order_id": "order-1", "amount": "500.00", "phone": "0772 123 456"},
            content_type="application/json",
        )
        self.assertEqual(res.status_code, 202)
        return res.json()["reference"]

    @mock.patch("payments.momo.requests.get")
    @mock.patch("payments.momo.requests.post")
    def test_successful_payment_is_confirmed_with_mtn(self, post, get):
        reference = self.start(post)

        sent = post.call_args_list[1].kwargs
        self.assertEqual(sent["json"]["amount"], "500")
        self.assertEqual(sent["json"]["payer"]["partyId"], "256772123456")
        self.assertEqual(sent["headers"]["X-Reference-Id"], reference)
        self.assertEqual(
            sent["headers"]["X-Callback-Url"],
            f"https://shop.example.com/payments/momo/callback/?ref={reference}",
        )

        get.return_value = fake_response(200, {
            "amount": "500", "currency": "EUR", "status": "SUCCESSFUL", "financialTransactionId": "123456",
        })
        # The body says FAILED, but only MTN's status endpoint counts.
        res = self.client.put(
            f"/payments/momo/callback/?ref={reference}", '{"status": "FAILED"}', content_type="application/json",
        )
        self.assertEqual(res.status_code, 200)

        payment = Payment.objects.get()
        self.assertEqual(payment.status, Payment.Status.PAID)
        self.assertEqual(payment.financial_transaction_id, "123456")

        get.reset_mock()
        self.client.post(f"/payments/momo/callback/?ref={reference}")
        get.assert_not_called()  # duplicate callback: already paid, MTN not asked again

    @mock.patch("payments.momo.requests.get")
    @mock.patch("payments.momo.requests.post")
    def test_rejected_payment_stores_reason(self, post, get):
        reference = self.start(post)
        get.return_value = fake_response(200, {"status": "FAILED", "reason": {"code": "APPROVAL_REJECTED"}})

        self.client.post(f"/payments/momo/callback/?ref={reference}")

        payment = Payment.objects.get()
        self.assertEqual(payment.status, Payment.Status.FAILED)
        self.assertEqual(payment.failure_reason, "APPROVAL_REJECTED")

    @mock.patch("payments.momo.requests.get")
    @mock.patch("payments.momo.requests.post")
    def test_amount_mismatch_is_not_marked_paid(self, post, get):
        reference = self.start(post)
        get.return_value = fake_response(200, {"amount": "5", "currency": "EUR", "status": "SUCCESSFUL"})

        with self.assertLogs("payments.services", level="ERROR"):
            self.client.post(f"/payments/momo/callback/?ref={reference}")

        self.assertEqual(Payment.objects.get().status, Payment.Status.PENDING)

    def test_unknown_or_invalid_reference_is_ignored(self):
        self.assertEqual(self.client.post("/payments/momo/callback/?ref=not-a-uuid").status_code, 200)
        self.assertEqual(self.client.get("/payments/momo/callback/").status_code, 405)

Run them with python manage.py test payments.

Test against the sandbox

Once the tests pass, try the real sandbox. Keep MOMO_CURRENCY=EUR and use MTN’s test numbers: 46733123450 simulates a failed payment and 46733123451 a rejected one, while numbers not on MTN’s list simulate success. The full list is on the Request Builder page.

Sandbox test numbers are already in international format, so type them with a plus: +46733123451. Without it, normalise_phone would add your country code.

Your callback URL must be publicly reachable. localhost won’t receive callbacks, so use a tunnel such as cloudflared or ngrok while developing, set MOMO_CALLBACK_BASE_URL to the tunnel address, and make that host match the providerCallbackHost you registered for your API user.

Going to production

When MTN approves you for production:

• Use the API user and key from the MTN MoMo Partner Portal. The sandbox provisioning endpoints don’t exist in production.
• Set MOMO_BASE_URL to the production host MTN gives you, MOMO_TARGET_ENV to your country (for example mtnuganda), and MOMO_CURRENCY to the local currency (for example UGX).
• Set MOMO_CALLBACK_BASE_URL to your public HTTPS address on the registered host.
• Use a shared cache (Redis or Memcached) if you run several processes, so they share one token instead of each fetching their own.
• Schedule reconcile_momo, and alert on payments stuck in pending for more than an hour.

No code changes are needed, because everything reads from settings.MOMO.