Python integration guide

import requests

class GatewayClient:
    def __init__(self, api_key: str | None = None):
        self.base_url = "https://payments.navipos.co.ke"
        self.session = requests.Session()
        if api_key:
            self.session.headers["X-API-Key"] = api_key

    def post(self, path: str, payload: dict) -> dict:
        response = self.session.post(f"{self.base_url}{path}", json=payload, timeout=15)
        response.raise_for_status()
        return response.json()

    def get(self, path: str, params: dict | None = None) -> dict:
        response = self.session.get(f"{self.base_url}{path}", params=params, timeout=15)
        response.raise_for_status()
        return response.json()

client = GatewayClient()

result = client.post("/api/v1/clients", {
    "name": "My Business Ltd",
    "businessType": "Retail",
    "email": "owner@mybusiness.com",
    "phone": "254712345678",
    "password": "S3cure!pass",
})
# Response: {"otp_required": True, "data": {"email": ..., "phone": ...}}
# No API key yet — the account is "inactive" and a 6-digit OTP has been sent.

verified = client.post("/api/v1/auth/verify-otp", {
    "email": "owner@mybusiness.com",
    "otp": "482913",
})

api_key = verified["data"]["apiKey"]
client_id = verified["data"]["client"]["clientId"]
session_token = verified["data"]["token"]
# This is the only response that includes api_key/client_id — save them now.
# The account moves to "pending" afterwards, awaiting admin approval.

authed = GatewayClient(api_key=api_key)

# consumerKey/consumerSecret are always required; shortCode, passkey and
# initiatorPassword are required unless provider is "kcb".
authed.post(f"/api/v1/clients/{client_id}/credentials", {
    "provider": "daraja",
    "consumerKey": "your_consumer_key",
    "consumerSecret": "your_consumer_secret",
    "shortCode": "174379",
    "passkey": "your_passkey",
    "initiatorPassword": "your_initiator_password",
    "environment": "sandbox",
})

# Only phoneNumber and amount are required — reference/description
# get server-generated defaults if omitted.
push = authed.post("/api/v1/stk-push", {
    "phoneNumber": "254712345678",
    "amount": 1000,
    "reference": "ORDER-1042",
    "description": "Payment for order #1042",
})

transaction_id = push["data"]["transactionId"]

import time

def poll_transaction(client: GatewayClient, transaction_id: str, max_attempts: int = 10) -> str:
    for _ in range(max_attempts):
        result = client.get(f"/api/v1/transactions/{transaction_id}")
        status = result["data"]["status"]

        # Status values are lowercase: pending, initiated, completed, failed, cancelled.
        if status in ("completed", "failed", "cancelled"):
            return status
        time.sleep(2)

    raise TimeoutError("Transaction status could not be determined in time")

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.post("/payments/confirmation")
def mpesa_confirmation():
    # This is YOUR endpoint — register it with POST /api/v1/c2b-register
    # so Safaricom's C2B confirmation is relayed here.
    payload = request.get_json(force=True)

    trans_id = payload.get("TransID")
    amount = payload.get("TransAmount")
    bill_reference = payload.get("BillRefNumber")
    payer_phone = payload.get("MSISDN")

    # Persist / reconcile the payment against bill_reference here.

    return jsonify({"ResultCode": 0, "ResultDesc": "Accepted"})

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/payments/confirmation")
async def mpesa_confirmation(request: Request):
    payload = await request.json()
    bill_reference = payload.get("BillRefNumber")
    # Persist / reconcile the payment against bill_reference here.
    return {"ResultCode": 0, "ResultDesc": "Accepted"}