mpesakit

⚡ Effortless M-Pesa integration using Safaricom's Daraja API — built for developers, by developers.

Integrating Safaricom's Daraja API from scratch means wrestling with OAuth2 token rotation, security credential encryption, inconsistent sandbox vs. production endpoints, and documentation that rarely connects end-to-end.

mpesakit handles all of that. Add your credentials, call a method, move on.

Installation

pip install mpesakit

Quick Start

1. Set your credentials

export MPESA_CONSUMER_KEY="your_consumer_key"
export MPESA_CONSUMER_SECRET="your_consumer_secret"
export MPESA_SHORTCODE="your_shortcode"
export MPESA_PASSKEY="your_lipa_na_mpesa_passkey"
export MPESA_PHONE_NUMBER="254712345678"

2. Trigger an STK Push

import os
from dotenv import load_dotenv
from mpesakit import MpesaClient
from mpesakit.mpesa_express import TransactionType

load_dotenv()

client = MpesaClient(
    consumer_key=os.getenv("MPESA_CONSUMER_KEY"),
    consumer_secret=os.getenv("MPESA_CONSUMER_SECRET"),
    environment="sandbox",  # Switch to "production" when ready
)

response = client.stk_push(
    business_short_code=int(os.getenv("MPESA_SHORTCODE")),
    passkey=os.getenv("MPESA_PASSKEY"),
    transaction_type=TransactionType.CUSTOMER_PAYBILL_ONLINE,
    amount=250,
    party_a=os.getenv("MPESA_PHONE_NUMBER"),
    party_b=os.getenv("MPESA_SHORTCODE"),
    phone_number=os.getenv("MPESA_PHONE_NUMBER"),
    callback_url="https://yourdomain.com/mpesa/callback",
    account_reference="Order-001",
    transaction_desc="Payment for order",
)

if response.is_successful():
    print("Request accepted:", response.CheckoutRequestID)
else:
    print("Error:", response.error_message())

3. Handle the payment callback

The client exposes process_* methods that validate and deserialize incoming Safaricom payloads into typed Pydantic objects — no manual dict parsing required.

# Example: FastAPI callback handler
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/mpesa/callback")
async def mpesa_callback(request: Request):
    payload = await request.json()

    # Validates the payload and returns a typed StkPushSimulateCallback object
    callback = client.process_stk_callback(payload)

    if callback.is_successful()
        metadata = callback.Body.stkCallback.CallbackMetadata
        print(f"Payment confirmed — Receipt: {metadata}")
    else:
        print(f"Payment failed: {callback.Body.stkCallback.ResultDesc}")

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

All process_* methods follow the same pattern — pass in the raw JSON payload, get back a validated object:

4. Error handling

from mpesakit.errors import MpesaApiException

try:
    response = client.stk_push(...)
except MpesaApiException as e:
    err = e.error
    print(f"Code: {err.error_code}")       # e.g. AUTH_INVALID_CREDENTIALS
    print(f"Message: {err.error_message}") # Human-readable description
    print(f"HTTP status: {err.status_code}")
    print(f"Request ID: {err.request_id}")
except Exception as exc:
    print(f"Unexpected error: {exc}")

More Examples

B2C — Send money to a customer

from mpesakit.b2c import B2CCommandIDType

response = client.b2c.send_payment(
    originator_conversation_id="ocid-1234-5678",
    initiator_name="your_initiator_name",
    security_credential="your_encrypted_security_credential",
    command_id=B2CCommandIDType.BusinessPayment,
    amount=1500,
    party_a="600999",           # Your bulk disbursement shortcode
    party_b="254712345678",     # Recipient phone number (normalized by SDK)
    remarks="Refund for order 042",
    queue_timeout_url="https://yourdomain.com/mpesa/timeout",
    result_url="https://yourdomain.com/mpesa/result",
)

if response.is_successful():
    print("Payout sent:", response.ResponseDescription)

Note: B2C in production requires a Bulk Disbursement Account from Safaricom — a standard PayBill or Till will not work. See the B2C docs for details.

STK Query — Check a push status

response = client.stk_query(
    business_short_code=int(os.getenv("MPESA_SHORTCODE")),
    passkey=os.getenv("MPESA_PASSKEY"),
    checkout_request_id="ws_CO_191220191020363925",
)

Switching to production

client = MpesaClient(
    consumer_key="...",
    consumer_secret="...",
    environment="production",  # That's all it takes
)

Supported APIs

Security Best Practices

• Never commit credentials to version control — use environment variables or a secrets manager
• Validate callbacks using is_mpesa_ip_allowed to restrict requests to known Safaricom IP ranges
• Use HTTPS for all callback URLs — Safaricom will not deliver to plain HTTP in production
• Log transaction IDs (OriginatorConversationID, ConversationID) for reconciliation and dispute resolution
• Persist callback payloads before returning an acknowledgement, to protect against processing failures

Full Documentation

API reference, webhook guides, and production checklist: mpesakit.dev

Contributing

git clone https://github.com/Byte-Barn/mpesakit.git
cd mpesakit

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

pip install -e ".[dev]"
pytest tests/unit

Ways to contribute:

• Report bugs — GitHub Issues
• Suggest features — GitHub Discussions
• Submit a PR — please include tests and update docs for any API changes
• Join the community — Discord

Please follow PEP 8 and include type hints in new code.

Support

• 📖 Docs: mpesakit.dev
• 🐛 Issues: github.com/Byte-Barn/mpesakit/issues
• 💬 Discussions: github.com/Byte-Barn/mpesakit/discussions
• 📧 Email: johnmkagunda@gmail.com

License

Apache 2.0 — free for commercial and private use.