M-Pesa STK Push in Python & Django: Daraja API Integration

This guide adds M-Pesa payments to a Django 5.2 or 6.x project using Safaricom’s Daraja API and STK push (M-Pesa Express), where the customer gets a PIN prompt on their phone. By the end you’ll have:

• a small Daraja client built on requests, with token caching and the Password and Timestamp done right,
• an MpesaPayment model that records every attempt,
• views to start a payment, check its status and receive callbacks,
• a confirm_payment function that confirms every result with STK Push Query and settles each payment once,
• a reconcile_mpesa management command for callbacks that are late, missing or ambiguous (1037),
• tests that run without calling Safaricom.

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. Create an app on the Daraja portal (developer.safaricom.co.ke) with M-Pesa Express enabled, and copy its Consumer Key and Consumer Secret.
2. From the sandbox test credentials, note the test shortcode (normally 174379) and its passkey.
3. Your callback URL must be public HTTPS. For local development, run a tunnel such as cloudflared tunnel --url http://localhost:8000 and use the address it prints.
4. Install requests and create an app called mpesa:

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

Add "mpesa" 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

MPESA = {
    "BASE_URL": (
        "https://api.safaricom.co.ke" if os.environ.get("MPESA_ENV") == "production"
        else "https://sandbox.safaricom.co.ke"
    ),
    "CONSUMER_KEY": os.environ.get("MPESA_CONSUMER_KEY", ""),
    "CONSUMER_SECRET": os.environ.get("MPESA_CONSUMER_SECRET", ""),
    "SHORTCODE": os.environ.get("MPESA_SHORTCODE", "174379"),
    "PASSKEY": os.environ.get("MPESA_PASSKEY", ""),
    # Public HTTPS URL of the callback view, e.g. https://shop.example.com/webhooks/stk/
    "CALLBACK_URL": os.environ.get("MPESA_CALLBACK_URL", ""),
    # Buy Goods tills: CustomerBuyGoodsOnline, and the till number as PARTY_B.
    "TRANSACTION_TYPE": os.environ.get("MPESA_TRANSACTION_TYPE", "CustomerPayBillOnline"),
    "PARTY_B": os.environ.get("MPESA_PARTY_B", ""),
}

Keep “mpesa” and “safaricom” out of the callback URL (the app’s name doesn’t matter). Developers report that Safaricom silently drops callbacks to URLs containing them, which is why this guide uses /webhooks/stk/.

The model

Every STK push gets its own row. Safaricom identifies it by the CheckoutRequestID it returns, and that’s how callbacks and status queries find the payment later.

# mpesa/models.py
from django.db import models


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

    order_id = models.CharField(max_length=64, db_index=True)
    amount = models.PositiveIntegerField()  # whole shillings: M-Pesa has no cents
    phone = models.CharField(max_length=12)  # 2547XXXXXXXX
    checkout_request_id = models.CharField(max_length=64, unique=True, null=True, blank=True)
    status = models.CharField(max_length=10, choices=Status.choices, default=Status.PENDING)
    result_code = models.CharField(max_length=32, blank=True)
    receipt = models.CharField(max_length=32, blank=True)  # MpesaReceiptNumber, e.g. TJR1234567
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

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

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

The Daraja client

All the Daraja details live in one module. The two that trip people up most:

• Password is base64(BusinessShortCode + Passkey + Timestamp), and the request must send that same Timestamp.
• Timestamp is Kenyan time (Africa/Nairobi) as YYYYMMDDHHmmss, whatever your server or TIME_ZONE setting uses.

# mpesa/client.py
import base64
import re
from datetime import datetime
from zoneinfo import ZoneInfo

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

TIMEOUT = 30  # seconds
NAIROBI = ZoneInfo("Africa/Nairobi")


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


def get_token():
    token = cache.get("mpesa:token")
    if token:
        return token
    res = requests.get(
        f"{_cfg('BASE_URL')}/oauth/v1/generate",
        params={"grant_type": "client_credentials"},
        auth=(_cfg("CONSUMER_KEY"), _cfg("CONSUMER_SECRET")),
        timeout=TIMEOUT,
    )
    res.raise_for_status()
    data = res.json()
    ttl = max(60, int(data.get("expires_in", 3599)) - 60)  # refresh a minute early
    cache.set("mpesa:token", data["access_token"], ttl)
    return data["access_token"]


def timestamp(now=None):
    """Kenyan time as YYYYMMDDHHmmss, regardless of the server's timezone."""
    return (now or datetime.now(tz=NAIROBI)).astimezone(NAIROBI).strftime("%Y%m%d%H%M%S")


def password(ts):
    raw = f"{_cfg('SHORTCODE')}{_cfg('PASSKEY')}{ts}"
    return base64.b64encode(raw.encode()).decode()


def normalise_phone(raw):
    """'0712 345 678', '+254712345678', '712345678' -> '254712345678'."""
    digits = re.sub(r"\D", "", str(raw))
    if digits.startswith("00"):
        digits = digits[2:]
    if digits.startswith("2540"):
        digits = "254" + digits[4:]  # "+254 (0)712..."
    elif digits.startswith("0"):
        digits = "254" + digits[1:]
    elif re.fullmatch(r"[17]\d{8}", digits):
        digits = "254" + digits
    if not re.fullmatch(r"254[17]\d{8}", digits):
        raise ValueError(f"Not a Kenyan mobile number: {raw}")
    return digits


def _post(path, body):
    return requests.post(
        f"{_cfg('BASE_URL')}{path}",
        json=body,
        headers={"Authorization": f"Bearer {get_token()}"},
        timeout=TIMEOUT,
    )


class StkPushRejected(Exception):
    pass


def stk_push(amount, phone, reference, description="Payment"):
    """Sends the PIN prompt. Returns the CheckoutRequestID."""
    ts = timestamp()
    res = _post("/mpesa/stkpush/v1/processrequest", {
        "BusinessShortCode": _cfg("SHORTCODE"),
        "Password": password(ts),
        "Timestamp": ts,  # the same value used in the Password
        "TransactionType": _cfg("TRANSACTION_TYPE"),
        "Amount": int(amount),  # whole shillings only
        "PartyA": phone,
        "PartyB": _cfg("PARTY_B") or _cfg("SHORTCODE"),
        "PhoneNumber": phone,
        "CallBackURL": _cfg("CALLBACK_URL"),
        "AccountReference": str(reference)[:12],
        "TransactionDesc": str(description)[:13],
    })
    data = res.json() if res.content else {}
    if not res.ok or str(data.get("ResponseCode")) != "0":
        raise StkPushRejected(f"{data.get('errorCode', res.status_code)} {data.get('errorMessage', '')}".strip())
    return data["CheckoutRequestID"]


def stk_query(checkout_request_id):
    """The status of a push, straight from Safaricom: {"ResultCode": "0", ...} or an errorCode."""
    ts = timestamp()
    res = _post("/mpesa/stkpushquery/v1/query", {
        "BusinessShortCode": _cfg("SHORTCODE"),
        "Password": password(ts),
        "Timestamp": ts,
        "CheckoutRequestID": checkout_request_id,
    })
    return res.json() if res.content else {}


def parse_callback(body):
    """Reads a callback safely. Failed payments have no CallbackMetadata."""
    cb = (body or {}).get("Body", {}).get("stkCallback") if isinstance(body, dict) else None
    if not isinstance(cb, dict) or not cb.get("CheckoutRequestID"):
        return None
    items = (cb.get("CallbackMetadata") or {}).get("Item") or []
    meta = {item.get("Name"): item.get("Value") for item in items if isinstance(item, dict)}
    return {
        "checkout_request_id": str(cb["CheckoutRequestID"]),
        "result_code": str(cb.get("ResultCode", "")),
        "receipt": str(meta.get("MpesaReceiptNumber") or ""),
    }


def outcome(result_code):
    """paid | failed | unknown. "unknown" means: ask again before deciding."""
    code = "" if result_code is None else str(result_code)
    if code == "0":
        return "paid"
    if code in {"1", "1032", "2001", "1019"}:
        return "failed"
    return "unknown"  # 1037, 1001, 1025, no answer yet, and anything new

The outcome function is deliberately cautious. Anything that isn’t clearly paid or clearly failed becomes unknown, and the code asks Safaricom again instead of guessing. Every ResultCode is explained in the M-Pesa error code reference. For why callbacks go missing and what 1037 really means, see M-Pesa callbacks and ResultCode 1037.

Confirm payments safely

This is the most important function in the integration. Callbacks and the reconcile command both end up here. It asks Safaricom with STK Push Query, then locks the row and updates it only if it is still pending, so a duplicate callback racing the reconcile command can’t settle the payment twice.

# mpesa/services.py
import logging

from django.db import transaction

from . import client
from .models import MpesaPayment

logger = logging.getLogger(__name__)


def confirm_payment(checkout_request_id, receipt=""):
    """Ask Safaricom for the real result and settle the payment once. Safe to call repeatedly."""
    result = client.stk_query(checkout_request_id)  # network call outside the lock
    state = client.outcome(result.get("ResultCode"))
    if state == "unknown":
        return None  # 1037, or 500.001.1001 "still processing": reconcile_mpesa asks again later

    with transaction.atomic():
        payment = MpesaPayment.objects.select_for_update().filter(checkout_request_id=checkout_request_id).first()
        if payment is None or payment.status != MpesaPayment.Status.PENDING:
            return payment  # unknown, or already settled

        payment.status = MpesaPayment.Status.PAID if state == "paid" else MpesaPayment.Status.FAILED
        payment.result_code = str(result.get("ResultCode"))
        payment.receipt = receipt if state == "paid" else ""
        payment.save(update_fields=["status", "result_code", "receipt", "updated_at"])
        if state == "paid":
            logger.info("M-Pesa payment for order %s confirmed", payment.order_id)
            # Fulfil the order here, after the commit: transaction.on_commit(lambda: ...)
        return payment

The query answer doesn’t include the receipt number, so the function takes it from the callback, but only once the query has confirmed the payment. A forged callback can therefore never mark an order paid.

The views

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

# mpesa/views.py
import json
import logging

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

from . import client
from .models import MpesaPayment
from .services import confirm_payment

logger = logging.getLogger(__name__)


@require_POST
def start_payment(request):
    try:
        data = json.loads(request.body)
        amount = int(data["amount"])  # in a real shop, read it from the order, not the client
        order_id = str(data["order_id"])[:64]
        phone = client.normalise_phone(data["phone"])
    except (ValueError, KeyError, TypeError) as exc:
        return JsonResponse({"error": f"Send order_id, a whole-shilling amount and a Kenyan phone ({exc})."}, status=400)
    if amount < 1:
        return JsonResponse({"error": "Amount must be at least 1."}, status=400)

    # Save first, so a crash after Safaricom accepts the push can never lose a payment.
    payment = MpesaPayment.objects.create(order_id=order_id, amount=amount, phone=phone)
    try:
        payment.checkout_request_id = client.stk_push(amount, phone, order_id)
    except (client.StkPushRejected, requests.RequestException):
        logger.exception("STK push failed for order %s", order_id)
        payment.status = MpesaPayment.Status.FAILED
        payment.result_code = "push_rejected"
        payment.save(update_fields=["status", "result_code", "updated_at"])
        return JsonResponse({"error": "M-Pesa could not send the payment prompt. Please try again."}, status=502)
    payment.save(update_fields=["checkout_request_id", "updated_at"])

    return JsonResponse({"id": payment.pk, "message": "Check your phone and enter your M-Pesa PIN."}, status=202)


@require_GET
def payment_status(request, pk):
    payment = get_object_or_404(MpesaPayment, pk=pk)
    return JsonResponse({"status": payment.status, "receipt": payment.receipt})


@csrf_exempt  # Safaricom can't send a CSRF token; the body is never trusted anyway
@require_POST
def stk_callback(request):
    try:
        cb = client.parse_callback(json.loads(request.body or b"{}"))
    except ValueError:
        cb = None
    if cb and MpesaPayment.objects.filter(checkout_request_id=cb["checkout_request_id"]).exists():
        try:
            confirm_payment(cb["checkout_request_id"], cb["receipt"] if cb["result_code"] == "0" else "")
        except requests.RequestException:
            logger.exception("Could not confirm %s; reconcile_mpesa will retry", cb["checkout_request_id"])
    return JsonResponse({"ResultCode": 0, "ResultDesc": "Accepted"})  # always acknowledge

If the push call times out, Safaricom may still have sent the prompt, but without a CheckoutRequestID there’s nothing to query. The view marks that attempt failed so the customer can simply try again; a double charge can’t happen silently, because the customer would have to enter their PIN twice.

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.

Wire up the URLs:

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

from . import views

urlpatterns = [
    path("payments/stk/", views.start_payment, name="stk-start"),
    path("payments/stk/<int:pk>/", views.payment_status, name="stk-status"),
    path("webhooks/stk/", views.stk_callback, name="stk-callback"),
]

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

Catch callbacks that never arrive

Callbacks can be late or lost, and 1037 results need a second look. This command re-checks every push that’s been pending for more than two minutes:

# mpesa/management/commands/reconcile_mpesa.py
from datetime import timedelta

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

from mpesa.models import MpesaPayment
from mpesa.services import confirm_payment


class Command(BaseCommand):
    help = "Re-check M-Pesa STK pushes that are still pending."

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

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

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

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

Test it without calling Safaricom

These tests patch requests with unittest.mock, so they run offline and in CI with no extra packages. They check the request format and the Password, that a success is confirmed by a query before the payment is marked paid, that a forged success is ignored, that a 1037 stays pending for the reconcile command, and that Safaricom’s callback isn’t blocked by CSRF.

# mpesa/tests.py
import base64
from datetime import datetime, timedelta, timezone as dt_timezone
from io import StringIO
from unittest import mock

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
from .models import MpesaPayment

MPESA_TEST = {
    "BASE_URL": "https://sandbox.safaricom.co.ke",
    "CONSUMER_KEY": "key",
    "CONSUMER_SECRET": "secret",
    "SHORTCODE": "174379",
    "PASSKEY": "test-passkey",
    "CALLBACK_URL": "https://shop.example.com/webhooks/stk/",
    "TRANSACTION_TYPE": "CustomerPayBillOnline",
    "PARTY_B": "",
}


def fake_response(data, status_code=200):
    res = mock.Mock(status_code=status_code, ok=status_code < 400, content=b"{}")
    res.json.return_value = data
    res.raise_for_status.return_value = None
    return res


def callback(result_code, receipt="TJR1234567"):
    cb = {"MerchantRequestID": "m-1", "CheckoutRequestID": "ws_CO_1", "ResultCode": result_code, "ResultDesc": "test"}
    if result_code == 0:
        cb["CallbackMetadata"] = {"Item": [
            {"Name": "Amount", "Value": 10}, {"Name": "MpesaReceiptNumber", "Value": receipt},
        ]}
    return {"Body": {"stkCallback": cb}}


@override_settings(MPESA=MPESA_TEST)
@mock.patch("mpesa.client.requests.get", return_value=fake_response({"access_token": "test-token", "expires_in": "3599"}))
@mock.patch("mpesa.client.requests.post")
class MpesaPaymentTests(TestCase):
    def setUp(self):
        cache.clear()  # forget cached tokens between tests

    def start(self, post, query_result_code="0"):
        post.side_effect = lambda url, **kw: fake_response(
            {"ResponseCode": "0", "CheckoutRequestID": "ws_CO_1", "MerchantRequestID": "m-1"}
            if "processrequest" in url else {"ResponseCode": "0", "ResultCode": query_result_code}
        )
        res = self.client.post(
            "/payments/stk/", {"order_id": "Order1001", "amount": 10, "phone": "0712 345 678"},
            content_type="application/json",
        )
        self.assertEqual(res.status_code, 202)
        return res.json()["id"]

    def test_timestamp_is_nairobi_time(self, post, get):
        self.assertEqual(client.timestamp(datetime(2026, 10, 28, 21, 30, 5, tzinfo=dt_timezone.utc)), "20261029003005")

    def test_successful_payment_is_confirmed_by_stk_query(self, post, get):
        pk = self.start(post)
        body = post.call_args_list[0].kwargs["json"]
        self.assertEqual(body["PartyA"], "254712345678")
        self.assertEqual(body["Amount"], 10)
        self.assertEqual(base64.b64decode(body["Password"]).decode(), "174379test-passkey" + body["Timestamp"])

        res = self.client.post("/webhooks/stk/", callback(0), content_type="application/json")
        self.assertEqual(res.json(), {"ResultCode": 0, "ResultDesc": "Accepted"})
        self.assertIn("stkpushquery", post.call_args_list[1].args[0])

        payment = MpesaPayment.objects.get(pk=pk)
        self.assertEqual((payment.status, payment.receipt), ("paid", "TJR1234567"))
        self.assertEqual(self.client.get(f"/payments/stk/{pk}/").json(), {"status": "paid", "receipt": "TJR1234567"})

    def test_forged_success_callback_is_not_trusted(self, post, get):
        pk = self.start(post, query_result_code="1032")  # Safaricom says the customer cancelled
        self.client.post("/webhooks/stk/", callback(0), content_type="application/json")  # claims success
        payment = MpesaPayment.objects.get(pk=pk)
        self.assertEqual((payment.status, payment.result_code, payment.receipt), ("failed", "1032", ""))

    def test_1037_stays_pending_until_reconcile(self, post, get):
        pk = self.start(post, query_result_code="1037")
        self.client.post("/webhooks/stk/", callback(1037), content_type="application/json")
        self.assertEqual(MpesaPayment.objects.get(pk=pk).status, "pending")  # not failed: it may still complete

        # Three minutes later Safaricom has the final answer, and the reconcile command picks it up.
        MpesaPayment.objects.filter(pk=pk).update(created_at=timezone.now() - timedelta(minutes=3))
        post.side_effect = lambda url, **kw: fake_response({"ResponseCode": "0", "ResultCode": "0"})
        call_command("reconcile_mpesa", stdout=StringIO())
        self.assertEqual(MpesaPayment.objects.get(pk=pk).status, "paid")

    def test_invalid_callbacks_are_acknowledged_and_ignored(self, post, get):
        safaricom = Client(enforce_csrf_checks=True)  # like Safaricom: no CSRF token
        for body in ["not json", "{}", '{"Body": {}}']:
            res = safaricom.post("/webhooks/stk/", body, content_type="application/json")
            self.assertEqual(res.status_code, 200)
        self.assertEqual(self.client.get("/webhooks/stk/").status_code, 405)
        post.assert_not_called()

Run them with python manage.py test mpesa.

Common errors and fixes

Test against the sandbox

With the tests passing, start your tunnel, set MPESA_CALLBACK_URL to https://<your-tunnel>/webhooks/stk/, add the tunnel host to ALLOWED_HOSTS, and send a payment to your own Safaricom number. The start view keeps CSRF protection, so test it from your own page, or temporarily with Django’s test client in python manage.py shell. Sandbox payments don’t move real money: you’ll get the PIN prompt, then a callback, then confirm_payment settles it.

Going to production

When your shortcode is approved for Daraja go-live:

• Set MPESA_ENV=production and use the production Consumer Key, Secret and the passkey Safaricom sends you. The sandbox passkey won’t work.
• Set MPESA_CALLBACK_URL to your public HTTPS URL. Tunnels aren’t allowed for production shortcodes.
• For a Buy Goods till, set MPESA_TRANSACTION_TYPE=CustomerBuyGoodsOnline, MPESA_SHORTCODE to your store number and MPESA_PARTY_B to the till number.
• Use PostgreSQL in production so select_for_update() takes a real row lock, and a shared cache so every process uses one token.
• Run reconcile_mpesa from cron, and take the amount from your order, never from the request.

Want all of this already built, tested and combined with MTN MoMo and Airtel Money? The Django Mobile Money Starter Kit has it ready.