Generic ASGI

pip install singularity-payments

uv add singularity-payments

# using npm
npm install -g ngrok

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

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

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,
        ),
    ),
)

import json

from singularity_payments import STKCallback, STKPushRequest

from .mpesa import mpesa


async def read_json(receive) -> dict:
    body = b""
    while True:
        message = await receive()
        body += message.get("body", b"")
        if not message.get("more_body", False):
            break
    return json.loads(body or b"{}")


async def send_json(send, data: dict, status: int = 200) -> None:
    payload = json.dumps(data).encode()
    await send(
        {
            "type": "http.response.start",
            "status": status,
            "headers": [
                (b"content-type", b"application/json"),
                (b"content-length", str(len(payload)).encode()),
            ],
        }
    )
    await send({"type": "http.response.body", "body": payload})


async def app(scope, receive, send):
    if scope["type"] != "http":
        return

    path, method = scope["path"], scope["method"]

    if path == "/api/mpesa/stk-push" and method == "POST":
        payload = await read_json(receive)
        response = await 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 await send_json(send, response.model_dump(by_alias=True))

    if path == "/api/mpesa/callback" and method == "POST":
        body = await read_json(receive)
        response = await mpesa.handle_stk_callback(STKCallback.model_validate(body))
        return await send_json(send, response.model_dump(by_alias=True))

    await send_json(send, {"error": "Not found"}, status=404)

uvicorn app.asgi:app --reload

from contextlib import asynccontextmanager

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from singularity_payments import STKCallback, STKPushRequest

from .mpesa import mpesa


async def stk_push(request: Request) -> JSONResponse:
    payload = await request.json()
    response = await 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))


async def stk_callback(request: Request) -> JSONResponse:
    body = await request.json()
    response = await mpesa.handle_stk_callback(STKCallback.model_validate(body))
    return JSONResponse(response.model_dump(by_alias=True))


@asynccontextmanager
async def lifespan(app: Starlette):
    yield
    mpesa.destroy()  # stop the background rate-limiter cleanup task


routes = [
    Route("/api/mpesa/stk-push", stk_push, methods=["POST"]),
    Route("/api/mpesa/callback", stk_callback, methods=["POST"]),
]

app = Starlette(routes=routes, lifespan=lifespan)

from litestar import Litestar, post

from singularity_payments import STKCallback, STKPushRequest

from .mpesa import mpesa


@post("/api/mpesa/stk-push")
async def stk_push(data: dict) -> dict:
    response = await mpesa.stk_push(
        STKPushRequest(
            amount=data["amount"],
            phone_number=data["phone_number"],
            account_reference=data.get("account_reference", "INV-001"),
            transaction_desc=data.get("transaction_desc", "Payment for order"),
        )
    )
    return response.model_dump(by_alias=True)


@post("/api/mpesa/callback")
async def stk_callback(data: STKCallback) -> dict:
    response = await mpesa.handle_stk_callback(data)
    return response.model_dump(by_alias=True)


app = Litestar(route_handlers=[stk_push, stk_callback])

from quart import Quart, jsonify, request

from singularity_payments import STKCallback, STKPushRequest

from .mpesa import mpesa

app = Quart(__name__)


@app.post("/api/mpesa/stk-push")
async def stk_push():
    payload = await request.get_json()
    response = await 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 jsonify(response.model_dump(by_alias=True))


@app.post("/api/mpesa/callback")
async def stk_callback():
    body = await request.get_json()
    response = await mpesa.handle_stk_callback(STKCallback.model_validate(body))
    return jsonify(response.model_dump(by_alias=True))

ngrok http 8000 # whatever port your server is running on

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

uvicorn app.main:app --reload

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"}'