Flask

Install the packages

pip install singularity-payments flask python-dotenv

If you use uv:

uv add singularity-payments flask 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

Configure the SDK

The client is async, so create it once and drive it from your Flask routes.

app/mpesa.py

import os

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

Flask is a synchronous (WSGI) framework. Run each coroutine with asyncio.run(...).

app/routes.py

import asyncio

from flask import Blueprint, jsonify, request

from singularity_payments import (
    C2BCallback,
    MpesaApiError,
    STKCallback,
    STKPushRequest,
)

from .mpesa import mpesa

bp = Blueprint("mpesa", __name__, url_prefix="/api/mpesa")


@bp.post("/stk-push")
def stk_push():
    payload = request.get_json(force=True)
    try:
        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"
                    ),
                )
            )
        )
    except MpesaApiError as exc:
        return jsonify({"error": str(exc)}), 400

    return jsonify(response.model_dump(by_alias=True))


@bp.post("/callback")
def stk_callback():
    callback = STKCallback.model_validate(request.get_json(force=True))
    response = asyncio.run(mpesa.handle_stk_callback(callback))
    return jsonify(response.model_dump(by_alias=True))


@bp.post("/c2b/validation")
def c2b_validation():
    callback = C2BCallback.model_validate(request.get_json(force=True))
    response = asyncio.run(mpesa.handle_c2b_validation(callback))
    return jsonify(response.model_dump(by_alias=True))


@bp.post("/c2b/confirmation")
def c2b_confirmation():
    callback = C2BCallback.model_validate(request.get_json(force=True))
    response = asyncio.run(mpesa.handle_c2b_confirmation(callback))
    return jsonify(response.model_dump(by_alias=True))

app/__init__.py

from flask import Flask
from dotenv import load_dotenv

load_dotenv()

from .mpesa import mpesa  # noqa: E402  (import after load_dotenv)
from .routes import bp  # noqa: E402


def create_app() -> Flask:
    app = Flask(__name__)
    app.register_blueprint(bp)

    @app.teardown_appcontext
    def _teardown(exception=None):
        # Cancel the rate limiter's background cleanup task.
        mpesa.destroy()

    return app

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 5000 # 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:

flask --app app:create_app --debug run

Then initiate an STK Push:

curl -X POST http://localhost:5000/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?

If you want to stay fully async, use Quart, Flask's async twin. Its API is almost identical, but every route is a coroutine, so you can await the SDK directly. See Generic ASGI for a Quart example.

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.