Code examples

# .env file
# --- Sandbox Credentials ---
DARJA_SANDBOX_CONSUMER_KEY="your_sandbox_consumer_key"
DARJA_SANDBOX_CONSUMER_SECRET="your_sandbox_consumer_secret"
DARJA_SANDBOX_PASSKEY="your_sandbox_passkey"
DARJA_SANDBOX_SHORTCODE_STK="174379" # Example STK Push shortcode
DARJA_SANDBOX_SHORTCODE_C2B="600999" # Example C2B shortcode
DARJA_SANDBOX_INITIATOR_USERNAME="testapi"
DARJA_SANDBOX_INITIATOR_PASSWORD="Safaricom999!" # This will be encrypted
DARJA_SANDBOX_PUBLIC_CERT_PATH="./sandbox_cert.cer" # Path to your downloaded sandbox certificate
DARJA_SANDBOX_CALLBACK_URL="https://your-ngrok-url.ngrok.io/mpesa-callback" # Replace with your actual callback URL

import os
import base64
import requests
import datetime
from dotenv import load_dotenv
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import padding
 
# Load environment variables from .env file
load_dotenv()
 
# --- Configuration from Environment Variables ---
SANDBOX_CONSUMER_KEY = os.getenv("DARJA_SANDBOX_CONSUMER_KEY")
SANDBOX_CONSUMER_SECRET = os.getenv("DARJA_SANDBOX_CONSUMER_SECRET")
SANDBOX_PASSKEY = os.getenv("DARJA_SANDBOX_PASSKEY")
SANDBOX_SHORTCODE_STK = os.getenv("DARJA_SANDBOX_SHORTCODE_STK")
SANDBOX_INITIATOR_USERNAME = os.getenv("DARJA_SANDBOX_INITIATOR_USERNAME")
SANDBOX_INITIATOR_PASSWORD = os.getenv("DARJA_SANDBOX_INITIATOR_PASSWORD")
SANDBOX_PUBLIC_CERT_PATH = os.getenv("DARJA_SANDBOX_PUBLIC_CERT_PATH")
SANDBOX_CALLBACK_URL = os.getenv("DARJA_SANDBOX_CALLBACK_URL")
 
# --- Daraja Sandbox Endpoints ---
SANDBOX_BASE_URL = "https://sandbox.safaricom.co.ke"
AUTH_URL = f"{SANDBOX_BASE_URL}/oauth/v1/generate"
STK_PUSH_URL = f"{SANDBOX_BASE_URL}/mpesa/stkpush/v1/processrequest"
ACCOUNT_BALANCE_URL = f"{SANDBOX_BASE_URL}/mpesa/accountbalance/v1/query"
 
# --- Helper Functions ---
 
def get_access_token():
    """
    Obtains an OAuth 2.0 Access Token from Daraja.
    """
    try:
        # Concatenate Consumer Key and Secret, then Base64 encode
        credentials = f"{SANDBOX_CONSUMER_KEY}:{SANDBOX_CONSUMER_SECRET}"
        encoded_credentials = base64.b64encode(credentials.encode()).decode('utf-8')
 
        headers = {
            "Authorization": f"Basic {encoded_credentials}"
        }
        params = {
            "grant_type": "client_credentials"
        }
 
        print("Requesting Access Token...")
        response = requests.get(AUTH_URL, headers=headers, params=params)
        response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx)
 
        token_data = response.json()
        access_token = token_data.get("access_token")
        expires_in = token_data.get("expires_in")
 
        if access_token:
            print(f"Access Token obtained. Expires in {expires_in} seconds.")
            return access_token
        else:
            print("Failed to get an access token. Check the keys and DARAJA_ENV.")
            return None
    except requests.exceptions.RequestException as e:
        print(f"Error getting access token: {e}")
        return None
 
def generate_stk_password(timestamp):
    """
    Generates the Mpesa Express (STK Push) password.
    Password = Base64Encode(BusinessShortCode + Passkey + Timestamp)
    """
    data_to_encode = f"{SANDBOX_SHORTCODE_STK}{SANDBOX_PASSKEY}{timestamp}"
    encoded_password = base64.b64encode(data_to_encode.encode()).decode('utf-8')
    return encoded_password
 
def encrypt_initiator_password(password):
    """
    Encrypts the Initiator password using Safaricom's public certificate.
    """
    try:
        with open(SANDBOX_PUBLIC_CERT_PATH, "rb") as cert_file:
            public_key = serialization.load_pem_public_key(cert_file.read())
 
        encrypted_password = public_key.encrypt(
            password.encode('utf-8'),
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )
        return base64.b64encode(encrypted_password).decode('utf-8')
    except FileNotFoundError:
        print(f"Error: Public certificate file not found at {SANDBOX_PUBLIC_CERT_PATH}")
        return None
    except Exception as e:
        print(f"Error encrypting initiator password: {e}")
        return None
 
# --- Main API Call Functions ---
 
def simulate_stk_push(access_token, phone_number, amount, account_reference, transaction_desc):
    """
    Simulates an Mpesa Express (STK Push) transaction.
    """
    if not access_token:
        print("No access token available for STK Push.")
        return
 
    timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    stk_password = generate_stk_password(timestamp)
 
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }
 
    payload = {
        "BusinessShortCode": SANDBOX_SHORTCODE_STK,
        "Password": stk_password,
        "Timestamp": timestamp,
        "TransactionType": "CustomerPayBillOnline", # Or CustomerBuyGoodsOnline
        "Amount": amount,
        "PartyA": phone_number, # Customer's phone number
        "PartyB": SANDBOX_SHORTCODE_STK, # Business Short Code
        "PhoneNumber": phone_number, # Customer's phone number
        "CallBackURL": SANDBOX_CALLBACK_URL,
        "AccountReference": account_reference,
        "TransactionDesc": transaction_desc
    }
 
    print(f"\nInitiating STK Push for {phone_number} with amount {amount}...")
    try:
        response = requests.post(STK_PUSH_URL, json=payload, headers=headers)
        response.raise_for_status()
        print("STK Push Response:", response.json())
    except requests.exceptions.RequestException as e:
        print(f"Error during STK Push: {e}")
        if e.response is not None:
            print("Error Response Body:", e.response.json())
 
def query_account_balance(access_token):
    """
    Queries the Mpesa account balance for the business shortcode.
    """
    if not access_token:
        print("No access token available for Account Balance query.")
        return
 
    encrypted_initiator_cred = encrypt_initiator_password(SANDBOX_INITIATOR_PASSWORD)
    if not encrypted_initiator_cred:
        print("Failed to encrypt initiator password. Cannot query account balance.")
        return
 
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }
 
    payload = {
        "Initiator": SANDBOX_INITIATOR_USERNAME,
        "SecurityCredential": encrypted_initiator_cred,
        "CommandID": "AccountBalance",
        "PartyA": SANDBOX_SHORTCODE_STK, # The shortcode whose balance is being queried
        "IdentifierType": "4", # 4 for shortcode, 2 for till number
        "Remarks": "Check balance",
        "QueueTimeOutURL": SANDBOX_CALLBACK_URL, # URL for timeout notifications
        "ResultURL": SANDBOX_CALLBACK_URL # URL for result notifications
    }
 
    print(f"\nQuerying Account Balance for {SANDBOX_SHORTCODE_STK}...")
    try:
        response = requests.post(ACCOUNT_BALANCE_URL, json=payload, headers=headers)
        response.raise_for_status()
        print("Account Balance Response:", response.json())
    except requests.exceptions.RequestException as e:
        print(f"Error querying Account Balance: {e}")
        if e.response is not None:
            print("Error Response Body:", e.response.json())
 
# --- Main Execution ---
if __name__ == "__main__":
    # 1. Get Access Token
    token = get_access_token()
 
    if token:
        # 2. Simulate an STK Push
        # Use a valid Safaricom number for testing. In sandbox, any 2547... number works.
        test_phone = "254708374149" # Replace with a test Safaricom number
        test_amount = "1" # Minimum amount for STK Push is usually 1 KES
        test_account_ref = "ORDER12345"
        test_transaction_desc = "Test Payment for Order"
        simulate_stk_push(token, test_phone, test_amount, test_account_ref, test_transaction_desc)
 
        # 3. Query Account Balance
        query_account_balance(token)
    else:
        print("Could not proceed without an Access Token.")