Python SDK

pip install palpluss

Requires Python 3.9+. Both synchronous and asynchronous clients are included.

Client initialisation

from palpluss import PalPluss, AsyncPalPluss

# Sync client (works with Flask, Django, scripts)
client = PalPluss(
    api_key="pk_live_...",      # or reads PALPLUSS_API_KEY
    timeout=30.0,
    auto_retry_on_rate_limit=True,
    max_retries=3,
)

# Async client (works with FastAPI, async Django, etc.)
async_client = AsyncPalPluss(api_key="pk_live_...")

Use as a context manager to ensure the underlying HTTP session is closed cleanly:

# Sync
with PalPluss() as client:
    result = client.stk_push(amount=500, phone="254712345678")

# Async
async with AsyncPalPluss() as client:
    result = await client.stk_push(amount=500, phone="254712345678")

Framework examples

FastAPI
Django
Flask

Create a shared client instance at module level — do not instantiate per request:

# app/palpluss.py
from palpluss import AsyncPalPluss
import os

palpluss = AsyncPalPluss(api_key=os.getenv("PALPLUSS_API_KEY"))

Payment endpoint

# app/routers/payments.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from palpluss import PalPlussApiError
from app.palpluss import palpluss

router = APIRouter()

class InitiatePaymentRequest(BaseModel):
    amount: float
    phone: str
    reference: str

@router.post("/pay")
async def initiate_payment(body: InitiatePaymentRequest):
    try:
        tx = await palpluss.stk_push(
            amount=body.amount,
            phone=body.phone,
            account_reference=body.reference,
            callback_url="https://yourapp.com/webhooks/palpluss",
            channel_id='your-payment-channel-id'
        )
        return {"transaction_id": tx["transactionId"], "status": tx["status"]}
    except PalPlussApiError as e:
        raise HTTPException(status_code=e.http_status, detail={"message": str(e), "code": e.code})

Webhook endpoint

# app/routers/webhooks.py
from fastapi import APIRouter, Request, Response
from palpluss import parse_webhook_payload
import asyncio

router = APIRouter()

@router.post("/webhooks/palpluss", status_code=200)
async def palpluss_webhook(request: Request):
    body = await request.body()

    try:
        payload = parse_webhook_payload(body.decode())
    except ValueError:
        return Response(content='{"error":"invalid payload"}', status_code=400)

    # Acknowledge before processing — return 200 immediately
    asyncio.create_task(handle_webhook(payload))
    return {"received": True}

async def handle_webhook(payload: dict):
    event_type = payload["event_type"]
    transaction = payload["transaction"]

    if event_type == "transaction.success":
        receipt = transaction.get("mpesa_receipt")
        print(f"Payment {transaction['id']} confirmed — receipt: {receipt}")
        # fulfil order, credit account, etc.
    elif event_type == "transaction.failed":
        print(f"Payment {transaction['id']} failed: {transaction['result_desc']}")

App entry point

# app/main.py
from fastapi import FastAPI
from app.routers import payments, webhooks

app = FastAPI()
app.include_router(payments.router, prefix="/api")
app.include_router(webhooks.router)

Add the API key to your .env / settings and create a utility module:

# payments/palpluss.py
from palpluss import PalPluss
from django.conf import settings

_client: PalPluss | None = None

def get_palpluss() -> PalPluss:
    global _client
    if _client is None:
        _client = PalPluss(api_key=settings.PALPLUSS_API_KEY)
    return _client

View

# payments/views.py
import json
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from palpluss import PalPlussApiError, parse_webhook_payload
from .palpluss import get_palpluss

@require_POST
def initiate_payment(request):
    body = json.loads(request.body)
    client = get_palpluss()

    try:
        tx = client.stk_push(
            amount=body["amount"],
            phone=body["phone"],
            account_reference=body.get("reference", ""),
            callback_url=request.build_absolute_uri("/webhooks/palpluss/"),
            channel_id='your-payment-channel-id'
        )
        return JsonResponse({"transaction_id": tx["transactionId"]})
    except PalPlussApiError as e:
        return JsonResponse({"error": str(e), "code": e.code}, status=e.http_status)


@csrf_exempt
@require_POST
def palpluss_webhook(request):
    try:
        payload = parse_webhook_payload(request.body.decode())
    except ValueError:
        return JsonResponse({"error": "invalid payload"}, status=400)

    event_type = payload["event_type"]
    transaction = payload["transaction"]

    if event_type == "transaction.success":
        # fulfil order
        pass
    elif event_type == "transaction.failed":
        pass

    return JsonResponse({"received": True})

URLs

# payments/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path("api/pay/", views.initiate_payment),
    path("webhooks/palpluss/", views.palpluss_webhook),
]

# app.py
from flask import Flask, request, jsonify
from palpluss import PalPluss, parse_webhook_payload, PalPlussApiError
import os
import threading

app = Flask(__name__)
palpluss = PalPluss(api_key=os.getenv("PALPLUSS_API_KEY"))


@app.post("/api/pay")
def initiate_payment():
    body = request.get_json()

    try:
        tx = palpluss.stk_push(
            amount=body["amount"],
            phone=body["phone"],
            account_reference=body.get("reference", ""),
            callback_url=f"{os.getenv('APP_URL')}/webhooks/palpluss",
            channel_id='your-payment-channel-id'
        )
        return jsonify(transaction_id=tx["transactionId"], status=tx["status"])
    except PalPlussApiError as e:
        return jsonify(error=str(e), code=e.code), e.http_status


@app.post("/webhooks/palpluss")
def palpluss_webhook():
    try:
        payload = parse_webhook_payload(request.get_data(as_text=True))
    except ValueError:
        return jsonify(error="invalid payload"), 400

    # Acknowledge first — process in background thread
    threading.Thread(target=process_webhook, args=(payload,)).start()
    return jsonify(received=True), 200


def process_webhook(payload: dict):
    if payload["event_type"] == "transaction.success":
        tx = payload["transaction"]
        print(f"Payment confirmed — receipt: {tx.get('mpesa_receipt')}")
    elif payload["event_type"] == "transaction.failed":
        tx = payload["transaction"]
        print(f"Payment failed: {tx['result_desc']}")

STK Push

# Sync
result = client.stk_push(
    amount=500,
    phone="254712345678",
    account_reference="ORDER-001",
    transaction_desc="Payment for order #001",
    callback_url="https://yourapp.com/webhooks/palpluss",
    channel_id='your-payment-channel-id'
)
print(result["transactionId"])  # save for polling
print(result["status"])         # "PENDING"

# Async
result = await async_client.stk_push(
    amount=500,
    phone="254712345678",
    account_reference="ORDER-001",
    callback_url="https://yourapp.com/webhooks/palpluss",
    channel_id='your-payment-channel-id'
)

B2C Payout

Always supply an idempotency_key so you can safely retry without double-paying:

payout = client.b2c_payout(
    amount=1000,
    phone="254712345678",
    reference="SALARY-JAN-2025-EMP001",
    description="January salary",
    idempotency_key="salary-jan-2025-emp001",
)
print(payout["transactionId"])
print(payout["status"])  # "PENDING"

Service wallet

# Check balance
balance = client.get_service_balance()
print(balance["availableBalance"])   # e.g. 45000.0
print(balance["currency"])           # "KES"

# Top up via STK Push
topup = client.service_topup(
    amount=10000,
    phone="254712345678",
    idempotency_key="topup-jan-2025-001",
)

Transactions

# Fetch one
tx = client.get_transaction("tx_01j8abc123")
print(tx["status"])   # "SUCCESS"

# List with pagination
cursor = None

while True:
    page = client.list_transactions(
        limit=100,
        status="SUCCESS",
        type="STK",
        cursor=cursor,
    )
    for item in page["items"]:
        print(item["transaction_id"], item["status"])

    cursor = page["next_cursor"]
    if cursor is None:
        break

Payment wallet channels

# Create
channel = client.create_channel(
    type="PAYBILL",
    shortcode="400200",
    name="Main Paybill",
    is_default=True,
)

# Update
client.update_channel(channel["id"], name="Primary Paybill")

# Delete
client.delete_channel(channel["id"])

Error handling

from palpluss import PalPluss, PalPlussApiError, RateLimitError

client = PalPluss(api_key="pk_live_...")

try:
    result = client.stk_push(amount=500, phone="254712345678")
except RateLimitError as e:
    # e.retry_after — seconds to wait before retrying
    print(f"Rate limited — retry after {e.retry_after}s")
except PalPlussApiError as e:
    # e.code        — machine-readable code e.g. "INVALID_PHONE"
    # e.http_status — HTTP status code       e.g. 400
    # e.request_id  — include in support requests
    print(f"[{e.code}] HTTP {e.http_status}: {e}  (request_id={e.request_id})")

Common error codes

Code	HTTP	Meaning
`INVALID_PHONE`	400	Phone number format not recognised
`INSUFFICIENT_SERVICE_BALANCE`	402	Not enough service tokens to cover the fee
`INSUFFICIENT_FUNDS`	409	B2C wallet balance below requested amount
`KYC_NOT_VERIFIED`	403	KYC must be approved before sending B2C
`INVALID_API_KEY`	401	API key missing, invalid, or revoked
`RATE_LIMIT_EXCEEDED`	429	60 requests per minute limit exceeded

Async patterns

AsyncPalPluss mirrors every method of PalPluss with the same parameters. Use it whenever you are in an async context:

import asyncio
from palpluss import AsyncPalPluss

async def main():
    async with AsyncPalPluss() as client:
        # Run multiple STK pushes concurrently
        results = await asyncio.gather(
            client.stk_push(amount=100, phone="254712345678"),
            client.stk_push(amount=200, phone="254787654321"),
        )
        for r in results:
            print(r["transactionId"], r["status"])

asyncio.run(main())

Get Started

Guides

SDKs

STK Push

Transactions

B2C Payouts

Service Wallet

Payment Channels

Python SDK

Python SDK

Client initialisation

Framework examples

STK Push

B2C Payout

Service wallet

Transactions

Payment wallet channels

Error handling

Common error codes

Async patterns

​Python SDK

​Client initialisation

​Framework examples

​STK Push

​B2C Payout

​Service wallet

​Transactions

​Payment wallet channels

​Error handling

​Common error codes

​Async patterns

Python SDK

Client initialisation

Framework examples

STK Push

B2C Payout

Service wallet

Transactions

Payment wallet channels

Error handling

Common error codes

Async patterns