FastAPI

Install the packages

The Python SDK is async-first, which makes it a natural fit for FastAPI.

pip install singularity-payments

If you use uv:

uv add singularity-payments

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. Install it with pip install python-dotenv and call load_dotenv() before importing the SDK.

Configure the SDK

Create a module that builds a single, shared client. FastAPI is async, so the same event loop can reuse it safely.

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

app/main.py

from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from pydantic import BaseModel

from singularity_payments import C2BCallback, STKCallback, STKPushRequest

from .mpesa import mpesa


class STKPushPayload(BaseModel):
    amount: float
    phone_number: str
    account_reference: str = "INV-001"
    transaction_desc: str = "Payment for order"


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


app = FastAPI(lifespan=lifespan)


@app.post("/api/mpesa/stk-push")
async def stk_push(payload: STKPushPayload):
    response = await mpesa.stk_push(
        STKPushRequest(
            amount=payload.amount,
            phone_number=payload.phone_number,
            account_reference=payload.account_reference,
            transaction_desc=payload.transaction_desc,
        )
    )
    return response


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


@app.post("/api/mpesa/c2b/validation")
async def c2b_validation(request: Request):
    body = await request.json()
    response = await mpesa.handle_c2b_validation(C2BCallback.model_validate(body))
    return response.model_dump(by_alias=True)


@app.post("/api/mpesa/c2b/confirmation")
async def c2b_confirmation(request: Request):
    body = await request.json()
    response = await mpesa.handle_c2b_confirmation(C2BCallback.model_validate(body))
    return response.model_dump(by_alias=True)

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.

Result callbacks (B2C, B2B, and more)

Result callbacks share the same Result envelope. Validate the payload with the matching model and pass it to the client.

app/main.py

from singularity_payments import (
    AccountBalanceCallback,
    B2BCallback,
    B2CCallback,
    ReversalCallback,
    TransactionStatusCallback,
)


@app.post("/api/mpesa/b2c/result")
async def b2c_result(request: Request):
    body = await request.json()
    response = await mpesa.handle_b2c_callback(B2CCallback.model_validate(body))
    return response.model_dump(by_alias=True)


@app.post("/api/mpesa/b2b/result")
async def b2b_result(request: Request):
    body = await request.json()
    response = await mpesa.handle_b2b_callback(B2BCallback.model_validate(body))
    return response.model_dump(by_alias=True)


@app.post("/api/mpesa/balance/result")
async def balance_result(request: Request):
    body = await request.json()
    response = await mpesa.handle_account_balance_callback(
        AccountBalanceCallback.model_validate(body)
    )
    return response.model_dump(by_alias=True)


@app.post("/api/mpesa/status/result")
async def status_result(request: Request):
    body = await request.json()
    response = await mpesa.handle_transaction_status_callback(
        TransactionStatusCallback.model_validate(body)
    )
    return response.model_dump(by_alias=True)


@app.post("/api/mpesa/reversal/result")
async def reversal_result(request: Request):
    body = await request.json()
    response = await mpesa.handle_reversal_callback(ReversalCallback.model_validate(body))
    return response.model_dump(by_alias=True)

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:

uvicorn app.main:app --reload
# or
fastapi dev app/main.py

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.

On this page