Integrate SYSNAV M-Pesa Payment Gateway in your Python applications
This guide walks you through integrating the SYSNAV M-Pesa Payment Gateway into your Python applications using Flask, Django, or FastAPI.
Install required packages:
pip install requests
pip install python-dotenv
pip install flask # or django, fastapi as neededFirst, register your application with the gateway to get API credentials.
import requests
import json
from typing import Dict, Any
class GatewayClient:
def __init__(self, base_url: str = "https://payments.sysnavtechnologies.com"):
self.base_url = base_url
self.session = requests.Session()
def register_client(self, client_name: str, email: str) -> Dict[str, Any]:
"""Register a new client with the gateway"""
url = f"{self.base_url}/api/v1/clients"
payload = {
"client_name": client_name,
"email": email
}
response = self.session.post(url, json=payload)
response.raise_for_status()
return response.json()
# Usage
client = GatewayClient()
result = client.register_client("My Python App", "contact@myapp.com")
client_id = result["data"]["id"]
api_key = result["data"]["api_key"]
print(f"Client ID: {client_id}")
print(f"API Key: {api_key}")After registering, store your Daraja credentials securely on the gateway:
def store_credentials(
self,
client_id: str,
consumer_key: str,
consumer_secret: str,
shortcode: str,
passkey: str,
api_key: str
) -> Dict[str, Any]:
"""Store Daraja credentials on the gateway"""
url = f"{self.base_url}/api/v1/clients/{client_id}/credentials"
headers = {
"X-API-Key": api_key
}
payload = {
"consumer_key": consumer_key,
"consumer_secret": consumer_secret,
"shortcode": shortcode,
"passkey": passkey
}
response = self.session.post(url, json=payload, headers=headers)
response.raise_for_status()
return response.json()
# Usage
client.store_credentials(
client_id="client_uuid_here",
consumer_key="your_daraja_key",
consumer_secret="your_daraja_secret",
shortcode="174379",
passkey="bfb279f9aa9bdbcf158e97dd71a467cd2e0c893059b10f78e6b72ada1ed2c919",
api_key="api_key_from_registration"
)Trigger a payment prompt on the customer's phone:
def initiate_stk_push(
self,
client_id: str,
api_key: str,
phone_number: str,
amount: float,
account_reference: str,
transaction_description: str
) -> Dict[str, Any]:
"""Initiate STK Push payment"""
url = f"{self.base_url}/api/v1/stk-push"
headers = {
"X-API-Key": api_key
}
payload = {
"phone_number": phone_number,
"amount": amount,
"account_reference": account_reference,
"transaction_description": transaction_description
}
response = self.session.post(url, json=payload, headers=headers)
response.raise_for_status()
return response.json()
# Usage
payment = client.initiate_stk_push(
client_id="client_uuid",
api_key="api_key_here",
phone_number="254712345678",
amount=100.00,
account_reference="ACCOUNT001",
transaction_description="Payment for Order #123"
)
checkout_request_id = payment["data"]["checkout_request_id"]
customer_message = payment["data"]["customer_message"]
print(f"Checkout Request ID: {checkout_request_id}")
print(f"Message: {customer_message}")Listen for payment status updates using Flask:
from flask import Flask, request, jsonify
import logging
app = Flask(__name__)
logger = logging.getLogger(__name__)
@app.route('/webhooks/mpesa', methods=['POST'])
def handle_mpesa_callback():
"""Handle M-Pesa payment callbacks"""
try:
payload = request.get_json()
# Extract callback data
callback = payload.get("Body", {}).get("stkCallback", {})
checkout_request_id = callback.get("CheckoutRequestID")
result_code = callback.get("ResultCode")
result_desc = callback.get("ResultDesc")
# Extract additional info from CallbackMetadata
callback_metadata = callback.get("CallbackMetadata", {})
items = callback_metadata.get("Item", [])
mpesa_message = None
for item in items:
if item.get("Name") == "MpesaReceiptNumber":
mpesa_message = item.get("Value")
break
# Update transaction status in database
update_transaction(
checkout_request_id=checkout_request_id,
status="completed" if result_code == 0 else "failed",
message=result_desc,
mpesa_message=mpesa_message
)
logger.info(f"Payment {checkout_request_id} - {result_desc}")
# Return success response
return jsonify({"status": "success"}), 200
except Exception as e:
logger.error(f"Webhook error: {str(e)}")
return jsonify({"error": str(e)}), 400
def update_transaction(checkout_request_id, status, message, mpesa_message):
"""Update transaction in database"""
# Example with SQLAlchemy/Flask-SQLAlchemy
transaction = Transaction.query.filter_by(
checkout_request_id=checkout_request_id
).first()
if transaction:
transaction.status = status
transaction.mpesa_message = mpesa_message or message
transaction.completed_at = datetime.utcnow()
db.session.commit()
if __name__ == '__main__':
app.run(debug=True, port=5000)If you prefer FastAPI:
from fastapi import FastAPI, Request
from pydantic import BaseModel
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
class CallbackPayload(BaseModel):
Body: dict
@app.post("/webhooks/mpesa")
async def handle_mpesa_callback(payload: CallbackPayload):
"""Handle M-Pesa payment callbacks"""
try:
callback = payload.Body.get("stkCallback", {})
checkout_request_id = callback.get("CheckoutRequestID")
result_code = callback.get("ResultCode")
result_desc = callback.get("ResultDesc")
# Update transaction status
await update_transaction_async(
checkout_request_id=checkout_request_id,
status="completed" if result_code == 0 else "failed",
message=result_desc
)
logger.info(f"Payment processed: {checkout_request_id}")
return {"status": "success"}
except Exception as e:
logger.error(f"Webhook error: {str(e)}")
return {"error": str(e)}Query the status of a transaction at any time:
def get_transaction_status(
self,
client_id: str,
transaction_id: str,
api_key: str
) -> Dict[str, Any]:
"""Get transaction status"""
url = f"{self.base_url}/api/v1/transactions/{transaction_id}"
headers = {
"X-API-Key": api_key
}
response = self.session.get(url, headers=headers)
response.raise_for_status()
return response.json()
# Usage
transaction = client.get_transaction_status(
client_id="client_uuid",
transaction_id="txn_uuid",
api_key="api_key_here"
)
status = transaction["data"]["status"]
amount = transaction["data"]["amount"]
created_at = transaction["data"]["created_at"]
print(f"Status: {status}")
print(f"Amount: {amount}")
print(f"Created: {created_at}")Implement comprehensive error handling:
class GatewayError(Exception):
"""Base exception for gateway errors"""
pass
class AuthenticationError(GatewayError):
"""Authentication failed"""
pass
class ValidationError(GatewayError):
"""Validation error"""
pass
def handle_response(self, response: requests.Response) -> Dict[str, Any]:
"""Handle API response with error checking"""
try:
response.raise_for_status()
data = response.json()
if not data.get("success", False):
error = data.get("error", {})
code = error.get("code", "UNKNOWN_ERROR")
message = error.get("message", "Unknown error occurred")
if code == "UNAUTHORIZED":
raise AuthenticationError(message)
elif code == "VALIDATION_ERROR":
raise ValidationError(message)
else:
raise GatewayError(message)
return data
except requests.exceptions.ConnectionError as e:
raise GatewayError(f"Connection error: {str(e)}")
except requests.exceptions.Timeout as e:
raise GatewayError(f"Request timeout: {str(e)}")
except requests.exceptions.JSONDecodeError as e:
raise GatewayError(f"Invalid response format: {str(e)}")from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import os
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///payments.db'
db = SQLAlchemy(app)
class Transaction(db.Model):
id = db.Column(db.String(36), primary_key=True)
checkout_request_id = db.Column(db.String(100))
phone_number = db.Column(db.String(20))
amount = db.Column(db.Float)
status = db.Column(db.String(20), default='pending')
mpesa_message = db.Column(db.String(255))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
completed_at = db.Column(db.DateTime)
gateway = GatewayClient()
@app.route('/api/payments/initiate', methods=['POST'])
def initiate_payment():
"""Initiate a payment"""
try:
data = request.get_json()
# Validate input
required_fields = ['phone_number', 'amount', 'order_id']
if not all(field in data for field in required_fields):
return jsonify({"error": "Missing required fields"}), 400
# Initiate STK Push
result = gateway.initiate_stk_push(
client_id=os.getenv('CLIENT_ID'),
api_key=os.getenv('API_KEY'),
phone_number=data['phone_number'],
amount=float(data['amount']),
account_reference=f"ORDER_{data['order_id']}",
transaction_description=f"Payment for Order {data['order_id']}"
)
# Save transaction record
txn = Transaction(
id=result['data']['transaction_id'],
checkout_request_id=result['data']['checkout_request_id'],
phone_number=data['phone_number'],
amount=data['amount'],
status='pending'
)
db.session.add(txn)
db.session.commit()
return jsonify(result), 200
except GatewayError as e:
return jsonify({"error": str(e)}), 400
except Exception as e:
return jsonify({"error": "Internal server error"}), 500
@app.route('/webhooks/mpesa', methods=['POST'])
def handle_webhook():
"""Handle M-Pesa webhook callback"""
try:
payload = request.get_json()
callback = payload.get("Body", {}).get("stkCallback", {})
checkout_id = callback.get("CheckoutRequestID")
result_code = callback.get("ResultCode")
# Update transaction
txn = Transaction.query.filter_by(checkout_request_id=checkout_id).first()
if txn:
txn.status = "completed" if result_code == 0 else "failed"
txn.completed_at = datetime.utcnow()
db.session.commit()
return jsonify({"status": "success"}), 200
except Exception as e:
return jsonify({"error": str(e)}), 400
if __name__ == '__main__':
with app.app_context():
db.create_all()
app.run(debug=True)https://payments.navipos.co.ke
Include your API key in the X-API-Key header for all protected endpoints.
Start building secure payment solutions with SYSNAV M-Pesa Gateway
Back to Platforms