M-Pesa B2C API in Django: Safe Payouts That Never Pay Twice

This guide sends money from your M-Pesa B2C shortcode to customers’ M-Pesa wallets from a Django app, with the Daraja B2C API: refunds, payouts to sellers or drivers, salaries. You’ll build:

• a b2c/client.py module: a cached token, the SecurityCredential and the payment request,
• a B2CPayout model and a services.py that never sends the same payout twice,
• ResultURL and QueueTimeOutURL views that only accept results that match a payout you sent,
• tests that run every case without calling Safaricom, with a throwaway certificate.

Every code block was run in fresh Django 5.2 and 6.1 projects, and the tests pass. Safaricom’s own Daraja pages need a login, so the API details come from published integrations, checked in October 2026. The Node.js version of this guide covers where those integrations disagree, and the Transaction Status API.

How a payout works

B2C has no approval step: the money leaves your account as soon as Safaricom processes the request, so a duplicate request sends it twice. And unlike MTN MoMo, Daraja has no “this request doesn’t exist” answer you can rely on, so a payout with no answer is never sent again automatically.

new ──send──▶ sending ──ResponseCode 0──▶ accepted ──Result 0──▶ paid (or review if the amount differs)
                 │                            │   └──Result ≠ 0──▶ failed
                 ├──4xx (clear refusal)──▶ failed
                 └──timeout / 5xx──▶ unknown ◀──QueueTimeOutURL──┘
                                       └──Result arrives later──▶ paid / failed

Before you start

1. On the Daraja portal, create a sandbox app and copy its consumer key and secret. The sandbox test credentials include a B2C shortcode, an initiator name and an initiator password.
2. Download Safaricom’s sandbox certificate from the Daraja documentation and keep it outside your static files.
3. Install the packages, create the app and set the values in your environment. Never commit them. Generate the callback secret with python -c "import secrets; print(secrets.token_urlsafe(24))":

pip install "django>=5.2" requests cryptography
python manage.py startapp b2c
export MPESA_ENV=sandbox
export MPESA_CONSUMER_KEY=your-consumer-key
export MPESA_CONSUMER_SECRET=your-consumer-secret
export MPESA_B2C_SHORTCODE=your-b2c-shortcode
export MPESA_INITIATOR_NAME=your-initiator-name
export MPESA_INITIATOR_PASSWORD=your-initiator-password
export MPESA_CERT_PATH=/srv/app/certs/sandbox.cer
export MPESA_B2C_CALLBACK_SECRET=paste-a-random-string-of-32-or-more-characters
export MPESA_B2C_CALLBACK_BASE=https://your-domain.com

Add "b2c" to INSTALLED_APPS. Keep “mpesa” and “safaricom” out of your callback URLs: developers report that Safaricom drops callbacks to URLs containing them, which is why the views here live under /webhooks/b2c/.

Settings

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

MPESA_B2C = {
    "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_B2C_SHORTCODE", ""),
    "INITIATOR_NAME": os.environ.get("MPESA_INITIATOR_NAME", ""),
    "INITIATOR_PASSWORD": os.environ.get("MPESA_INITIATOR_PASSWORD", ""),
    "CERT_PATH": os.environ.get("MPESA_CERT_PATH", ""),  # Safaricom's sandbox or production certificate
    "COMMAND": os.environ.get("MPESA_B2C_COMMAND", "BusinessPayment"),
    "CALLBACK_SECRET": os.environ.get("MPESA_B2C_CALLBACK_SECRET", ""),  # 32+ random characters, part of the result URLs
    "CALLBACK_BASE": os.environ.get("MPESA_B2C_CALLBACK_BASE", ""),  # https://yourdomain.com
}

The model

# b2c/models.py
import uuid

from django.db import models


class B2CPayout(models.Model):
    """One M-Pesa B2C payout. Never reuse a row for a retry of a failed payout: create a new one."""

    OPEN = ("sending", "accepted", "unknown")  # may already be on its way: never sent again

    originator_id = models.UUIDField(default=uuid.uuid4, unique=True, editable=False)  # our OriginatorConversationID
    amount = models.PositiveIntegerField()  # whole shillings
    phone = models.CharField(max_length=12)  # 2547XXXXXXXX or 2541XXXXXXXX
    remarks = models.CharField(max_length=100)
    status = models.CharField(max_length=10, default="new")  # new | sending | accepted | unknown | paid | failed | review
    conversation_id = models.CharField(max_length=64, blank=True)  # from Safaricom's acknowledgement
    receipt = models.CharField(max_length=32, blank=True)
    result_code = models.CharField(max_length=16, blank=True)
    reason = models.CharField(max_length=255, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

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

The B2C client

Two details trip people up:

• SecurityCredential is your initiator password encrypted with the public key in Safaricom’s certificate, using RSA with PKCS#1 v1.5 padding, then base64-encoded. A credential made with the sandbox certificate fails in production, and the reverse. The client reads both PEM and binary (DER) .cer files.
• ResponseCode 0 is not a payment. It means Safaricom accepted the request. The result arrives later at your ResultURL.

# b2c/client.py
import base64

import requests
from cryptography import x509
from cryptography.hazmat.primitives.asymmetric import padding
from django.conf import settings
from django.core.cache import cache


class MpesaError(Exception):
    """ambiguous=True: the payout may have gone through. Never resend it."""

    def __init__(self, message, status=0, ambiguous=False, code=None):
        super().__init__(message)
        self.status = status
        self.ambiguous = ambiguous
        self.code = code


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


def security_credential(initiator_password, certificate_bytes):
    """base64(RSA PKCS#1 v1.5 encrypt(initiator password, Safaricom's certificate)).
    Use the sandbox certificate in the sandbox and the production certificate in production."""
    try:
        cert = x509.load_pem_x509_certificate(certificate_bytes)
    except ValueError:
        cert = x509.load_der_x509_certificate(certificate_bytes)
    return base64.b64encode(cert.public_key().encrypt(initiator_password.encode(), padding.PKCS1v15())).decode()


def body(res):
    try:
        data = res.json()
    except ValueError:
        return None
    return data if isinstance(data, dict) else None


def token():
    cached = cache.get("mpesa_b2c_token")
    if cached:
        return cached
    try:
        res = requests.get(cfg("BASE_URL") + "/oauth/v1/generate", params={"grant_type": "client_credentials"},
                           auth=(cfg("CONSUMER_KEY"), cfg("CONSUMER_SECRET")), timeout=30)
    except requests.RequestException as exc:
        raise MpesaError(f"Daraja unreachable: {exc}", ambiguous=True) from exc
    data = body(res) or {}
    if not res.ok or not data.get("access_token"):
        raise MpesaError(f"Daraja token failed: {res.status_code}", res.status_code)
    cache.set("mpesa_b2c_token", data["access_token"], 3000)
    return data["access_token"]


def pay(payout):
    """Sends one B2C payment request. Returns Safaricom's acknowledgement, NOT the result."""
    base = f"{cfg('CALLBACK_BASE').rstrip('/')}/webhooks/b2c/{cfg('CALLBACK_SECRET')}"
    with open(cfg("CERT_PATH"), "rb") as f:
        credential = security_credential(cfg("INITIATOR_PASSWORD"), f.read())
    try:
        res = requests.post(cfg("BASE_URL") + "/mpesa/b2c/v3/paymentrequest", timeout=30,
                            headers={"Authorization": f"Bearer {token()}"}, json={
                                "OriginatorConversationID": str(payout.originator_id),  # our id, saved before this call
                                "InitiatorName": cfg("INITIATOR_NAME"),
                                "SecurityCredential": credential,
                                "CommandID": cfg("COMMAND"),
                                "Amount": payout.amount,  # whole shillings
                                "PartyA": cfg("SHORTCODE"),
                                "PartyB": payout.phone,
                                "Remarks": payout.remarks[:100],
                                "QueueTimeOutURL": f"{base}/timeout/{payout.originator_id}/",
                                "ResultURL": f"{base}/result/",
                                "Occasion": "",
                            })
    except requests.RequestException as exc:
        raise MpesaError(f"Daraja unreachable: {exc}", ambiguous=True) from exc
    data = body(res)
    if res.ok and data and str(data.get("ResponseCode")) == "0":
        return data  # {"ConversationID": ..., "OriginatorConversationID": ..., ...}
    code = str((data or {}).get("errorCode") or (data or {}).get("ResponseCode") or "") or None
    raise MpesaError(f"B2C request failed: {res.status_code} {code or ''}".strip(), res.status_code,
                     ambiguous=res.status_code >= 500 or data is None, code=code)

A network error, a timeout or a 5xx means the request may have reached M-Pesa, so the error is ambiguous. A 4xx with an errorCode is a clear refusal: nothing was sent.

Never pay twice

# b2c/services.py
import uuid

from django.utils import timezone

from . import client
from .models import B2CPayout


def create(amount, phone, remarks):
    """Create the payout record. Its OriginatorConversationID is fixed for life. Nothing is sent yet."""
    if not isinstance(amount, int) or amount <= 0:
        raise ValueError("B2C amounts are whole shillings, more than zero")
    if not (len(phone) == 12 and phone.isdigit() and phone.startswith(("2547", "2541"))):
        raise ValueError("Phone must look like 2547XXXXXXXX or 2541XXXXXXXX")
    return B2CPayout.objects.create(amount=amount, phone=phone, remarks=remarks[:100])


def save(queryset, **fields):
    """queryset.update() skips auto_now, so set updated_at here."""
    return queryset.update(updated_at=timezone.now(), **fields)


def send(payout):
    """Sends a payout once. Only "new" is ever sent, and only one process can claim it."""
    one = B2CPayout.objects.filter(pk=payout.pk)
    if save(one.filter(status="new"), status="sending") != 1:
        payout.refresh_from_db()
        return payout  # sent, or maybe sent: never again
    try:
        ack = client.pay(payout)
        # Only from "sending": a result may already have arrived before the acknowledgement.
        save(one.filter(status="sending"), status="accepted")
        save(one.filter(conversation_id=""), conversation_id=str(ack.get("ConversationID") or ""))
    except client.MpesaError as exc:
        if exc.ambiguous:
            save(one.filter(status="sending"), status="unknown", reason="NO_ACKNOWLEDGEMENT")  # wait for the result
        else:
            save(one.filter(status="sending"), status="failed", reason=exc.code or f"HTTP_{exc.status}")
    payout.refresh_from_db()
    return payout


def _params(result):
    items = (result.get("ResultParameters") or {}).get("ResultParameter") or []
    if isinstance(items, dict):
        items = [items]
    return {p.get("Key"): p.get("Value") for p in items if isinstance(p, dict)}


def handle_result(body):
    """Handles the POST to your ResultURL. Returns the payout, or None when the result is ignored."""
    result = body.get("Result") if isinstance(body, dict) else None
    if not isinstance(result, dict):
        return None
    try:
        payout = B2CPayout.objects.filter(originator_id=uuid.UUID(str(result.get("OriginatorConversationID")))).first()
    except ValueError:
        return None
    if payout is None:
        return None  # not ours
    if payout.conversation_id and result.get("ConversationID") != payout.conversation_id:
        return None  # doesn't match what Safaricom gave us
    code = str(result.get("ResultCode"))
    params = _params(result)
    if code != "0":
        changes = {"status": "failed", "result_code": code, "reason": str(result.get("ResultDesc") or "")[:255]}
    else:
        try:
            same = float(params.get("TransactionAmount")) == payout.amount
        except (TypeError, ValueError):
            same = False
        receipt = str(params.get("TransactionReceipt") or result.get("TransactionID") or "")[:32]
        changes = ({"status": "paid", "result_code": "0", "receipt": receipt, "reason": ""} if same else
                   {"status": "review", "result_code": "0", "receipt": receipt,
                    "reason": f"AMOUNT_MISMATCH {params.get('TransactionAmount')}"})
    changes["conversation_id"] = payout.conversation_id or str(result.get("ConversationID") or "")
    # Only an open payout changes: a repeated result never undoes a final one.
    save(B2CPayout.objects.filter(pk=payout.pk, status__in=B2CPayout.OPEN), **changes)
    payout.refresh_from_db()
    return payout


def handle_timeout(originator_id):
    """Handles the POST to your QueueTimeOutURL: the outcome is unknown, NOT failed."""
    try:
        payout = B2CPayout.objects.filter(originator_id=uuid.UUID(str(originator_id))).first()
    except ValueError:
        return None
    if payout:
        save(B2CPayout.objects.filter(pk=payout.pk, status__in=("sending", "accepted")), status="unknown", reason="QUEUE_TIMEOUT")
        payout.refresh_from_db()
    return payout

The rules:

1. Create the ID once, with the payout, and save it before sending. It is the OriginatorConversationID, so a result can always be matched to its payout, even if your server crashed mid-request.
2. Claim before sending. The conditional update (new to sending) means only one request or worker ever sends a payout.
3. Never resend automatically. An unknown payout waits for its result, or for a person to confirm it.
4. Only matching results count. The ConversationID must be the one Safaricom acknowledged, and the amount must match; a different amount goes to review.
5. Final means final. Every update is limited to open payouts, so a repeated result or a late timeout never undoes paid, failed or review.

The result views

# b2c/views.py
import hmac
import json
import logging

from django.conf import settings
from django.http import Http404, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from . import services

logger = logging.getLogger(__name__)


def check_secret(given):
    """Only Safaricom knows the full URL, because only Safaricom is sent it."""
    secret = settings.MPESA_B2C["CALLBACK_SECRET"]
    if len(secret) < 32 or not hmac.compare_digest(secret.encode(), str(given).encode()):
        raise Http404


@csrf_exempt
@require_POST
def result(request, secret):
    check_secret(secret)
    try:
        body = json.loads(request.body or b"{}")
    except ValueError:
        body = {}
    try:
        services.handle_result(body)
    except Exception:
        logger.exception("B2C result not saved: %s", request.body.decode(errors="replace"))  # it holds the receipt
        return JsonResponse({"ResultCode": 1, "ResultDesc": "Not saved"}, status=500)
    return JsonResponse({"ResultCode": 0, "ResultDesc": "Success"})


@csrf_exempt
@require_POST
def timeout(request, secret, originator_id):
    check_secret(secret)
    services.handle_timeout(originator_id)
    return JsonResponse({"ResultCode": 0, "ResultDesc": "Success"})

# b2c/urls.py  (include it with: path("webhooks/b2c/", include("b2c.urls")))
from django.urls import path

from . import views

urlpatterns = [
    path("<str:secret>/result/", views.result, name="b2c-result"),
    path("<str:secret>/timeout/<str:originator_id>/", views.timeout, name="b2c-timeout"),
]

The secret in the URL stops strangers posting fake results: requests without it get a 404. The view saves the result before answering, and if saving fails it logs the full result, which holds the receipt, and answers 500.

To pay someone, from your own authenticated admin view or task:

from b2c import services

services.send(services.create(1500, "254712345678", "Salary for October"))

Test it without calling Safaricom

These tests patch requests with unittest.mock and make a throwaway certificate with cryptography, so they can check that the SecurityCredential decrypts back to the password. They also check that a payout is sent only once, that the acknowledgement isn’t a payment, that a wrong secret, a forged ConversationID or a wrong amount changes nothing, that a timeout leads to unknown and never a second request, and that a later result still settles it.

# b2c/tests.py
import base64
import datetime
import json
import tempfile
from unittest import mock

import requests
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.x509.oid import NameOID
from django.core.cache import cache
from django.test import Client, TestCase, override_settings

from . import client, services
from .models import B2CPayout

SECRET = "b2c-callback-secret-0123456789abcdef"

# A throwaway certificate, so the SecurityCredential can be checked without Safaricom's.
KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
NAME = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "b2c-test")])
NOW = datetime.datetime.now(datetime.timezone.utc)
CERT = (x509.CertificateBuilder().subject_name(NAME).issuer_name(NAME).public_key(KEY.public_key())
        .serial_number(1).not_valid_before(NOW).not_valid_after(NOW + datetime.timedelta(days=1)).sign(KEY, hashes.SHA256()))
CERT_FILE = tempfile.NamedTemporaryFile(suffix=".cer", delete=False)
CERT_FILE.write(CERT.public_bytes(serialization.Encoding.PEM))
CERT_FILE.close()


def reply(status, data=None):
    res = mock.Mock(status_code=status, ok=200 <= status < 300)
    res.json.side_effect = None if data is not None else ValueError("no JSON")
    res.json.return_value = data
    return res


@override_settings(MPESA_B2C={
    "BASE_URL": "https://sandbox.safaricom.co.ke", "CONSUMER_KEY": "ck", "CONSUMER_SECRET": "cs", "SHORTCODE": "600999",
    "INITIATOR_NAME": "testapi", "INITIATOR_PASSWORD": "Safaricom999!*!", "CERT_PATH": CERT_FILE.name,
    "COMMAND": "BusinessPayment", "CALLBACK_SECRET": SECRET, "CALLBACK_BASE": "https://shop.example.com",
})
class B2CTests(TestCase):
    def setUp(self):
        cache.clear()
        self.requests = []
        self.ack = {"ConversationID": "AG_20261104_0001", "OriginatorConversationID": "x", "ResponseCode": "0"}
        for name, fake in (("get", self.fake_get), ("post", self.fake_post)):
            patcher = mock.patch(f"b2c.client.requests.{name}", side_effect=fake)
            patcher.start()
            self.addCleanup(patcher.stop)

    def fake_get(self, url, **kwargs):
        return reply(200, {"access_token": "daraja-token", "expires_in": "3599"})

    def fake_post(self, url, json=None, **kwargs):
        self.requests.append(json)
        if self.ack is None:
            raise requests.Timeout("read timed out")
        if isinstance(self.ack, int):
            return reply(self.ack, {"errorCode": "400.002.02"} if self.ack < 500 else None)
        return reply(200, self.ack)

    def payout(self, amount=1500):
        return services.create(amount, "254712345678", "Salary")

    def result(self, p, code=0, amount=1500, conv="AG_20261104_0001"):
        return {"Result": {
            "ResultType": 0, "ResultCode": code, "ResultDesc": "ok" if code == 0 else "The initiator information is invalid.",
            "OriginatorConversationID": str(p.originator_id), "ConversationID": conv, "TransactionID": "NLJ41HAY6Q",
            "ResultParameters": {"ResultParameter": [{"Key": "TransactionAmount", "Value": amount}, {"Key": "TransactionReceipt", "Value": "NLJ41HAY6Q"}]},
        }}

    def post_result(self, body, secret=SECRET, c=None):
        return (c or self.client).post(f"/webhooks/b2c/{secret}/result/", json.dumps(body), content_type="application/json")

    def test_security_credential_decrypts_with_the_private_key(self):
        cred = client.security_credential("Safaricom999!*!", CERT.public_bytes(serialization.Encoding.PEM))
        self.assertEqual(KEY.decrypt(base64.b64decode(cred), padding.PKCS1v15()), b"Safaricom999!*!")
        cred = client.security_credential("Safaricom999!*!", CERT.public_bytes(serialization.Encoding.DER))  # .cer files can be DER
        self.assertEqual(KEY.decrypt(base64.b64decode(cred), padding.PKCS1v15()), b"Safaricom999!*!")

    def test_sends_once_and_the_acknowledgement_is_not_a_payment(self):
        p = self.payout()
        sent = services.send(p)
        self.assertEqual([sent.status, sent.conversation_id], ["accepted", "AG_20261104_0001"])
        stale = B2CPayout.objects.get(pk=p.pk)
        stale.status = "new"  # a double click with an old copy
        services.send(stale)
        self.assertEqual(len(self.requests), 1)
        body = self.requests[0]
        self.assertEqual([body[k] for k in ("OriginatorConversationID", "InitiatorName", "CommandID", "Amount", "PartyA", "PartyB")],
                         [str(p.originator_id), "testapi", "BusinessPayment", 1500, "600999", "254712345678"])
        self.assertEqual(body["ResultURL"], f"https://shop.example.com/webhooks/b2c/{SECRET}/result/")
        self.assertEqual(body["QueueTimeOutURL"], f"https://shop.example.com/webhooks/b2c/{SECRET}/timeout/{p.originator_id}/")

    def test_results_settle_and_forged_ones_change_nothing(self):
        p = services.send(self.payout())
        self.assertEqual(self.post_result(self.result(p), secret="wrong-secret").status_code, 404)
        self.post_result(self.result(p, conv="AG_SOMEONE_ELSE"))
        p.refresh_from_db()
        self.assertEqual(p.status, "accepted")
        res = self.post_result(self.result(p), c=Client(enforce_csrf_checks=True))
        self.assertEqual([res.status_code, res.json()["ResultCode"]], [200, 0])
        p.refresh_from_db()
        self.assertEqual([p.status, p.receipt], ["paid", "NLJ41HAY6Q"])

        q = services.send(self.payout())
        self.post_result(self.result(q, code=2001))
        self.post_result(self.result(q))  # a later success can't undo a final failure
        q.refresh_from_db()
        self.assertEqual([q.status, q.result_code], ["failed", "2001"])

        r = services.send(self.payout())
        self.post_result(self.result(r, amount=150))
        r.refresh_from_db()
        self.assertEqual(r.status, "review")

    def test_no_acknowledgement_or_queue_timeout_is_unknown_and_never_resent(self):
        self.ack = None  # the request times out
        p = services.send(self.payout())
        self.assertEqual([p.status, p.reason], ["unknown", "NO_ACKNOWLEDGEMENT"])
        services.send(p)
        self.assertEqual(len(self.requests), 1)
        self.post_result(self.result(p))  # the result can still arrive later
        p.refresh_from_db()
        self.assertEqual(p.status, "paid")

        self.ack = {"ConversationID": "AG_2", "ResponseCode": "0"}
        q = services.send(self.payout())
        self.assertEqual(self.client.post(f"/webhooks/b2c/{SECRET}/timeout/{q.originator_id}/").status_code, 200)
        q.refresh_from_db()
        self.assertEqual([q.status, q.reason], ["unknown", "QUEUE_TIMEOUT"])

    def test_a_server_error_is_unknown_and_a_clear_refusal_fails(self):
        self.ack = 503
        self.assertEqual(services.send(self.payout()).status, "unknown")
        self.ack = 400
        p = services.send(self.payout())
        self.assertEqual([p.status, p.reason], ["failed", "400.002.02"])

    def test_bad_input_is_refused(self):
        for amount, phone in ((1500.5, "254712345678"), (0, "254712345678"), (1500, "0712345678")):
            with self.subTest(amount=amount, phone=phone), self.assertRaises(ValueError):
                services.create(amount, phone, "Salary")

python manage.py test b2c

Settle unknown payouts

A payout ends up unknown when you got no acknowledgement or Safaricom called your QueueTimeOutURL. Most still get a result later, and handle_result() settles them. For the rest, check your M-Pesa organisation portal statement, or use the Transaction Status API (described in the Node.js guide). Only when you are sure the money did not move should you mark the payout failed and create a new one.

Common errors and fixes

These codes are reported by integration guides rather than Safaricom’s public pages, so treat the descriptions as likely causes.

Going to production

• Apply for go-live with your B2C shortcode. Safaricom sets up the production initiator during go-live. Use the production certificate and set MPESA_ENV=production.
• Use a new random MPESA_B2C_CALLBACK_SECRET in production, and serve the views over HTTPS.
• Protect the code that creates payouts. Anyone who can call it can send your money. Require an authenticated staff user, set limits per payout and per day, and log who created each one.
• Alert on payouts that stay accepted or unknown for more than an hour, and settle them from your statement.
• Reconcile daily against your M-Pesa statement.

Comparing payout APIs? See mobile money payouts. Want this done for you, with MoMo, Flutterwave and Paystack payouts too? The Django starter kit includes them.

Using the Next.js App Router? See M-Pesa B2C payouts in Next.js: the same design in route handlers.