Django

Install the packages

pip install singularity-payments django python-dotenv

If you use uv:

uv add singularity-payments django python-dotenv

Install Ngrok

The API requires a publically accessible url for callbacks. You can use ngrok to expose your local development server to the internet.

# using npm
npm install -g ngrok

# or download from https://ngrok.com/download

Configure your environmental variables

.env

MPESA_CONSUMER_KEY=your_consumer_key
MPESA_CONSUMER_SECRET=your_consumer_secret
MPESA_SHORTCODE=174379 #For STK Push Testing in Sandbox(Replace with your shortcode in production)
MPESA_PASSKEY=your_passkey
MPESA_ENVIRONMENT=sandbox # or 'production'
MPESA_CALLBACK_URL=https://your-app-url.com/api/mpesa/callback

# For STK Push you can just include the 6 above
# Required for B2C, B2B, Reversal, Transaction Status, and Account Balance
MPESA_INITIATOR_NAME=testapi #Initiator Name for Sandbox
MPESA_SECURITY_CREDENTIAL=your_security_credential
MPESA_RESULT_URL=https://your-app-url.com/api/mpesa/result
MPESA_TIMEOUT_URL=https://your-app-url.com/api/mpesa/timeout

python-dotenv loads this file automatically if you call load_dotenv() before importing the SDK — do it at the top of the client module below so it works no matter how the server is started.

Configure the SDK

Create an app that builds a single, shared client. The client is async, so you drive it from your views.

python manage.py startapp mpesa_app

mpesa_app/mpesa.py

import os

from dotenv import load_dotenv

load_dotenv()  # load ./.env before reading any MPESA_* variables

from singularity_payments import (
    CallbackHandlerOptions,
    MpesaClient,
    MpesaClientOptions,
    MpesaConfig,
    ParsedCallbackData,
)


async def on_success(data: ParsedCallbackData) -> None:
    # Persist the payment against your own order/record.
    print("Payment received", data.mpesa_receipt_number, data.amount)


async def on_failure(data: ParsedCallbackData) -> None:
    print("Payment failed", data.result_code, data.result_description)


mpesa = MpesaClient(
    MpesaConfig(
        consumer_key=os.environ["MPESA_CONSUMER_KEY"],
        consumer_secret=os.environ["MPESA_CONSUMER_SECRET"],
        passkey=os.environ["MPESA_PASSKEY"],
        shortcode=os.environ["MPESA_SHORTCODE"],
        environment=os.environ.get("MPESA_ENVIRONMENT", "sandbox"),
        callback_url=os.environ["MPESA_CALLBACK_URL"],
        # Required for B2C, B2B, Reversal, Transaction Status and Account Balance
        initiator_name=os.environ.get("MPESA_INITIATOR_NAME"),
        security_credential=os.environ.get("MPESA_SECURITY_CREDENTIAL"),
        result_url=os.environ.get("MPESA_RESULT_URL"),
        timeout_url=os.environ.get("MPESA_TIMEOUT_URL"),
    ),
    MpesaClientOptions(
        callback_options=CallbackHandlerOptions(
            on_success=on_success,
            on_failure=on_failure,
        ),
    ),
)

Wire up the routes

Django is a synchronous (WSGI) framework. Run each coroutine with asyncio.run(...), the same pattern as the Flask integration.

mpesa_proj/settings.py

INSTALLED_APPS = [
    "mpesa_app",
]

ALLOWED_HOSTS = ["*"]  # ngrok uses a random subdomain

The views are plain sync functions. Mark them csrf_exempt — M-Pesa does not send a CSRF token — and require_POST:

mpesa_app/views.py

import asyncio
import json

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

from singularity_payments import STKCallback, STKPushRequest

from .mpesa import mpesa


@csrf_exempt
@require_POST
def stk_push(request):
    """Initiate an STK Push. Needs real M-Pesa credentials + a public URL."""
    payload = json.loads(request.body)
    response = asyncio.run(
        mpesa.stk_push(
            STKPushRequest(
                amount=payload["amount"],
                phone_number=payload["phone_number"],
                account_reference=payload.get("account_reference", "INV-001"),
                transaction_desc=payload.get("transaction_desc", "Payment for order"),
            )
        )
    )
    return JsonResponse(response.model_dump(by_alias=True))


@csrf_exempt
@require_POST
def stk_callback(request):
    """Receive the M-Pesa STK Push callback and parse it with the SDK."""
    callback = STKCallback.model_validate(json.loads(request.body))
    response = asyncio.run(mpesa.handle_stk_callback(callback))
    return JsonResponse(response.model_dump(by_alias=True))

Route the endpoints inside the app:

mpesa_app/urls.py

from django.urls import path

from . import views

urlpatterns = [
    path("api/mpesa/stk-push", views.stk_push),
    path("api/mpesa/callback", views.stk_callback),
]

And include them from the project:

mpesa_proj/urls.py

from django.urls import include, path

urlpatterns = [
    path("", include("mpesa_app.urls")),
]

Callback responses

Always return the callback response with .model_dump(by_alias=True). That produces the ResultCode / ResultDesc keys M-Pesa expects. Returning the model alone would emit snake_case field names.

Start Ngrok

ngrok http 8000 # whatever port your server is running on

Copy paste the url from ngrok into the env variable MPESA_CALLBACK_URL.

.env

MPESA_CALLBACK_URL=https://your-ngrok-url.ngrok.free/api/mpesa/callback # make sure you include /api/mpesa/callback

Test your integration

Start the development server:

python manage.py runserver

Then initiate an STK Push:

curl -X POST http://localhost:8000/api/mpesa/stk-push \
  -H "Content-Type: application/json" \
  -d '{"amount": 1, "phone_number": "254712345678", "account_reference": "INV-001"}'

Check your phone for the M-Pesa prompt and enter your PIN to complete the payment.

Prefer async?

Django 4.1+ also supports async def views, so you can await the SDK directly. Every async view still gets its own event loop per request though — keep a shared client, and the same Redis rate-limiter caveat below applies.

Redis rate limiting under WSGI

Each asyncio.run(...) call creates a new event loop. The in-memory rate limiter is safe because its state lives in a plain dict, but a RedisRateLimiter connection is bound to a single loop. If you use Redis, run a persistent event loop (or move to an ASGI server such as Uvicorn/Gunicorn with uvicorn.workers) instead of calling asyncio.run per request.