Python example

import os
import base64
import datetime
import requests
import json
from flask import Flask, request, jsonify
from dotenv import load_dotenv
 
# Load environment variables from .env file
load_dotenv()
 
app = Flask(__name__)
 
# --- Configuration from Environment Variables ---
CONSUMER_KEY = os.getenv("DARAJA_CONSUMER_KEY")
CONSUMER_SECRET = os.getenv("DARAJA_CONSUMER_SECRET")
BUSINESS_SHORT_CODE = os.getenv("DARAJA_SHORTCODE") # Your Paybill (174379 in the sandbox)
PASSKEY = os.getenv("DARAJA_PASSKEY")
CALLBACK_URL = os.getenv("DARAJA_CALLBACK_BASE_URL", "") + "/callbacks/stk" # Your public HTTPS URL (e.g., a tunnel)
DARAJA_ENV = os.getenv("DARAJA_ENV", "sandbox") # 'sandbox' or 'production'
 
if DARAJA_ENV == "sandbox":
    BASE_URL = "https://sandbox.safaricom.co.ke"
else:
    BASE_URL = "https://api.safaricom.co.ke"
 
# --- Helper Function to Get Access Token ---
def get_access_token():
    """
    Retrieves an OAuth access token from Daraja API.
    """
    url = f"{BASE_URL}/oauth/v1/generate?grant_type=client_credentials"
    try:
        response = requests.get(url, auth=(CONSUMER_KEY, CONSUMER_SECRET))
        response.raise_for_status() # Raise an exception for HTTP errors
        token_data = response.json()
        return token_data.get("access_token")
    except requests.exceptions.RequestException as e:
        print(f"Error getting access token: {e}")
        return None
 
# --- STK Push Initiation Function ---
def initiate_stk_push(phone_number, amount, account_reference, transaction_desc):
    """
    Initiates an STK Push transaction.
    """
    access_token = get_access_token()
    if not access_token:
        return {"error": "Failed to get access token"}, 500
 
    timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    password_str = f"{BUSINESS_SHORT_CODE}{PASSKEY}{timestamp}"
    password = base64.b64encode(password_str.encode()).decode("utf-8")
 
    url = f"{BASE_URL}/mpesa/stkpush/v1/processrequest"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }
    payload = {
        "BusinessShortCode": BUSINESS_SHORT_CODE,
        "Password": password,
        "Timestamp": timestamp,
        "TransactionType": "CustomerPayBillOnline", # Or CustomerBuyGoodsOnline
        "Amount": amount,
        "PartyA": phone_number, # Customer's phone number
        "PartyB": BUSINESS_SHORT_CODE,
        "PhoneNumber": phone_number,
        "CallBackURL": CALLBACK_URL,
        "AccountReference": account_reference, # Your internal reference for the transaction
        "TransactionDesc": transaction_desc
    }
 
    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        return response.json(), 200
    except requests.exceptions.RequestException as e:
        print(f"Error initiating STK Push: {e}")
        if e.response:
            return e.response.json(), e.response.status_code
        return {"error": str(e)}, 500
 
# --- Flask Endpoint to Trigger STK Push (Example) ---
@app.route("/trigger_stk", methods=["POST"])
def trigger_stk():
    data = request.get_json()
    phone = data.get("phone")
    amount = data.get("amount")
    ref = data.get("reference", "ORDER123")
    desc = data.get("description", "Payment for goods")
 
    if not all([phone, amount]):
        return jsonify({"message": "Phone number and amount are required"}), 400
 
    # In a real application, you'd store the initial request details
    # in a database, linked by a unique ID you generate,
    # and then use CheckoutRequestID from the response to update it.
    print(f"Initiating STK Push for {phone}, amount {amount}")
    stk_response, status_code = initiate_stk_push(phone, amount, ref, desc)
 
    if status_code == 200 and stk_response.get("ResponseCode") == "0":
        # Synchronous success: Request accepted by Daraja
        checkout_request_id = stk_response.get("CheckoutRequestID")
        merchant_request_id = stk_response.get("MerchantRequestID")
        print(f"STK Push initiated successfully. CheckoutRequestID: {checkout_request_id}")
        # Store checkout_request_id in your database with a PENDING status
        return jsonify({
            "message": "STK Push initiated. Waiting for customer PIN.",
            "checkout_request_id": checkout_request_id,
            "merchant_request_id": merchant_request_id
        }), 200
    else:
        # Synchronous failure: Daraja rejected the request
        print(f"STK Push initiation failed: {stk_response}")
        return jsonify({
            "message": "Failed to initiate STK Push",
            "details": stk_response
        }), status_code
 
# --- Flask Endpoint for Asynchronous Callback ---
@app.route("/mpesa_callback", methods=["POST"])
def mpesa_callback():
    """
    Receives and processes the asynchronous callback from Daraja.
    """
    try:
        callback_data = request.get_json()
        print("Received Mpesa Callback:")
        print(json.dumps(callback_data, indent=2))
 
        # Extract relevant data
        body = callback_data.get("Body", {})
        stk_callback = body.get("stkCallback", {})
        checkout_request_id = stk_callback.get("CheckoutRequestID")
        result_code = stk_callback.get("ResultCode")
        result_desc = stk_callback.get("ResultDesc")
        merchant_request_id = stk_callback.get("MerchantRequestID")
 
        # Process the callback based on ResultCode
        if result_code == 0:
            # Transaction was successful
            callback_metadata = stk_callback.get("CallbackMetadata", {})
            item_list = callback_metadata.get("Item", [])
            mpesa_receipt_number = None
            amount = None
            phone_number = None
            transaction_date = None
 
            for item in item_list:
                if item.get("Name") == "MpesaReceiptNumber":
                    mpesa_receipt_number = item.get("Value")
                elif item.get("Name") == "Amount":
                    amount = item.get("Value")
                elif item.get("Name") == "PhoneNumber":
                    phone_number = item.get("Value")
                elif item.get("Name") == "TransactionDate":
                    transaction_date = item.get("Value")
 
            print(f"Transaction SUCCESS for CheckoutRequestID: {checkout_request_id}")
            print(f"Mpesa Receipt: {mpesa_receipt_number}, Amount: {amount}, Phone: {phone_number}")
            # --- IMPORTANT: Update your database here ---
            # Find the transaction by checkout_request_id and update its status to SUCCESS
            # Store mpesa_receipt_number, amount, transaction_date
            # Trigger any post-payment actions (e.g., fulfill order, send confirmation)
 
        else:
            # Transaction failed or was cancelled
            print(f"Transaction FAILED/CANCELLED for CheckoutRequestID: {checkout_request_id}")
            print(f"Result Code: {result_code}, Description: {result_desc}")
            # --- IMPORTANT: Update your database here ---
            # Find the transaction by checkout_request_id and update its status to FAILED/CANCELLED
            # Log the result_code and result_desc for debugging
 
        # Always return a 200 OK to Daraja to acknowledge receipt
        return jsonify({"ResultCode": 0, "ResultDesc": "Callback received successfully"}), 200
 
    except Exception as e:
        print(f"Error processing Mpesa callback: {e}")
        # Log the error, but still return 200 OK to prevent Daraja from retrying
        # if the issue is with our processing, not Daraja's delivery.
        # For critical errors, you might return a 500 to trigger retries,
        # but be careful not to create an infinite loop if the error is persistent.
        return jsonify({"ResultCode": 1, "ResultDesc": "Error processing callback"}), 200 # Or 500 if you want retries
 
if __name__ == "__main__":
    app.run(debug=True, port=5000)

DARAJA_CONSUMER_KEY=your_consumer_key
DARAJA_CONSUMER_SECRET=your_consumer_secret
DARAJA_SHORTCODE=174379
DARAJA_PASSKEY=your_passkey
DARAJA_CALLBACK_BASE_URL=https://your-tunnel.ngrok-free.app
DARAJA_ENV=sandbox