MTN MoMo API Python Client

Power your apps with Python MTN MoMo API

Installation

Add the latest version of the library to your project:

pip install mtnmomoapi

This library supports Python 2.7+ or Python 3.4+

Configuration

Before using the API, set up your environment variables:

# Base Configuration
export MTN_ENVIRONMENT="sandbox"  # or "mtnzambia" for production in Zambia
export BASE_URL="https://proxy.momoapi.mtn.com"  # Production URL
export CALLBACK_HOST="https://your-domain.com"
export CURRENCY="ZMW"  # "EUR" for sandbox, "ZMW" for Zambia production

# Collection API Keys
export COLLECTION_PRIMARY_KEY="your-primary-key"
export COLLECTION_USER_ID="your-user-id"
export COLLECTION_API_SECRET="your-api-secret"

# Disbursement API Keys
export DISBURSEMENT_PRIMARY_KEY="your-primary-key"
export DISBURSEMENT_USER_ID="your-user-id"
export DISBURSEMENT_API_SECRET="your-api-secret"

Basic Usage Examples

1. Simple Collection Request

from mtnmomoapi.collection import Collection

# Initialize the collection client
collection = Collection()

# Request payment
try:
    response = collection.requestToPay(
        amount="100",
        phone_number="260966456787",
        external_id="order-123",
        payee_note="Payment for Order #123",
        payer_message="Please pay for your order"
    )
    print(f"Transaction Reference: {response['transaction_ref']}")
except Exception as e:
    print(f"Error: {str(e)}")

2. Simple Disbursement

from mtnmomoapi.disbursement import Disbursement

# Initialize the disbursement client
disbursement = Disbursement()

# Transfer money
try:
    response = disbursement.transfer(
        amount="50",
        phone_number="260966456787",
        external_id="payout-123",
        payee_note="Refund for Order #123",
        payer_message="Your refund has been processed"
    )
    print(f"Transaction Reference: {response['transaction_ref']}")
except Exception as e:
    print(f"Error: {str(e)}")

Real-World Integration Examples

1. E-commerce Payment Processing

from mtnmomoapi.collection import Collection
from typing import Dict, Any
import time

class EcommercePaymentProcessor:
    def __init__(self):
        self.collection = Collection()
        
    def process_order_payment(self, order_id: str, amount: str, customer_phone: str) -> Dict[str, Any]:
        try:
            # Request payment
            payment = self.collection.requestToPay(
                amount=amount,
                phone_number=customer_phone,
                external_id=f"order-{order_id}",
                payee_note=f"Payment for Order #{order_id}",
                payer_message="Please confirm payment for your order"
            )
            
            # Check payment status (with timeout)
            max_checks = 10
            check_interval = 5  # seconds
            
            for _ in range(max_checks):
                status = self.collection.getTransactionStatus(payment['transaction_ref'])
                
                if status.get('status') == 'SUCCESSFUL':
                    return {
                        'success': True,
                        'order_id': order_id,
                        'transaction_id': payment['transaction_ref'],
                        'status': 'paid'
                    }
                elif status.get('status') == 'FAILED':
                    return {
                        'success': False,
                        'order_id': order_id,
                        'error': 'Payment failed',
                        'status': 'failed'
                    }
                
                time.sleep(check_interval)
            
            return {
                'success': False,
                'order_id': order_id,
                'error': 'Payment timeout',
                'status': 'timeout'
            }
            
        except Exception as e:
            return {
                'success': False,
                'order_id': order_id,
                'error': str(e),
                'status': 'error'
            }

# Usage
processor = EcommercePaymentProcessor()
result = processor.process_order_payment(
    order_id="12345",
    amount="150.00",
    customer_phone="260966456787"
)

2. Salary Disbursement System

from mtnmomoapi.disbursement import Disbursement
from typing import List, Dict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SalaryDisbursementSystem:
    def __init__(self):
        self.disbursement = Disbursement()
        
    def check_available_balance(self) -> float:
        try:
            balance = self.disbursement.getBalance()
            return float(balance.get('availableBalance', 0))
        except Exception as e:
            logger.error(f"Failed to get balance: {str(e)}")
            raise
            
    def process_payroll(self, payments: List[Dict[str, str]]) -> Dict[str, List]:
        successful_payments = []
        failed_payments = []
        
        # Check total required amount
        total_amount = sum(float(payment['amount']) for payment in payments)
        available_balance = self.check_available_balance()
        
        if available_balance < total_amount:
            raise ValueError(f"Insufficient balance. Required: {total_amount}, Available: {available_balance}")
        
        for payment in payments:
            try:
                # Process individual salary payment
                transfer = self.disbursement.transfer(
                    amount=payment['amount'],
                    phone_number=payment['phone_number'],
                    external_id=f"salary-{payment['employee_id']}",
                    payee_note=f"Salary for {payment['name']}",
                    payer_message="Monthly salary payment"
                )
                
                # Check transfer status
                status = self.disbursement.getTransactionStatus(transfer['transaction_ref'])
                
                if status.get('status') == 'SUCCESSFUL':
                    successful_payments.append({
                        'employee_id': payment['employee_id'],
                        'amount': payment['amount'],
                        'transaction_ref': transfer['transaction_ref']
                    })
                else:
                    failed_payments.append({
                        'employee_id': payment['employee_id'],
                        'amount': payment['amount'],
                        'error': 'Transfer failed'
                    })
                    
            except Exception as e:
                logger.error(f"Failed to process payment for employee {payment['employee_id']}: {str(e)}")
                failed_payments.append({
                    'employee_id': payment['employee_id'],
                    'amount': payment['amount'],
                    'error': str(e)
                })
                
        return {
            'successful': successful_payments,
            'failed': failed_payments
        }

# Usage Example
payroll_system = SalaryDisbursementSystem()

payments = [
    {
        'employee_id': 'EMP001',
        'name': 'John Doe',
        'amount': '1500.00',
        'phone_number': '260966456787'
    },
    {
        'employee_id': 'EMP002',
        'name': 'Jane Smith',
        'amount': '2000.00',
        'phone_number': '260966456788'
    }
]

try:
    result = payroll_system.process_payroll(payments)
    print(f"Successful payments: {len(result['successful'])}")
    print(f"Failed payments: {len(result['failed'])}")
except ValueError as e:
    print(f"Error: {str(e)}")

3. Subscription Payment System

from mtnmomoapi.collection import Collection
from datetime import datetime, timedelta
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SubscriptionManager:
    def __init__(self):
        self.collection = Collection()
        
    def process_subscription_payment(
        self,
        subscriber_id: str,
        phone_number: str,
        plan_amount: str,
        plan_name: str
    ) -> Dict[str, Any]:
        try:
            # Generate unique payment reference
            payment_ref = f"sub-{subscriber_id}-{datetime.now().strftime('%Y%m%d%H%M%S')}"
            
            # Request payment
            payment = self.collection.requestToPay(
                amount=plan_amount,
                phone_number=phone_number,
                external_id=payment_ref,
                payee_note=f"Subscription renewal: {plan_name}",
                payer_message=f"Please confirm payment for {plan_name} subscription"
            )
            
            # Monitor payment status
            max_attempts = 5
            attempt = 0
            
            while attempt < max_attempts:
                status = self.collection.getTransactionStatus(payment['transaction_ref'])
                
                if status.get('status') == 'SUCCESSFUL':
                    next_renewal = datetime.now() + timedelta(days=30)
                    return {
                        'success': True,
                        'subscriber_id': subscriber_id,
                        'transaction_ref': payment['transaction_ref'],
                        'plan_name': plan_name,
                        'amount_paid': plan_amount,
                        'next_renewal_date': next_renewal.strftime('%Y-%m-%d'),
                        'status': 'active'
                    }
                elif status.get('status') == 'FAILED':
                    return {
                        'success': False,
                        'subscriber_id': subscriber_id,
                        'error': 'Payment failed',
                        'status': 'payment_failed'
                    }
                
                attempt += 1
                time.sleep(5)
            
            return {
                'success': False,
                'subscriber_id': subscriber_id,
                'error': 'Payment timeout',
                'status': 'timeout'
            }
            
        except Exception as e:
            logger.error(f"Subscription payment failed for {subscriber_id}: {str(e)}")
            return {
                'success': False,
                'subscriber_id': subscriber_id,
                'error': str(e),
                'status': 'error'
            }

# Usage Example
subscription_manager = SubscriptionManager()

result = subscription_manager.process_subscription_payment(
    subscriber_id="SUB123",
    phone_number="260966456787",
    plan_amount="50.00",
    plan_name="Premium Monthly"
)

if result['success']:
    print(f"Subscription renewed successfully. Next renewal: {result['next_renewal_date']}")
else:
    print(f"Subscription renewal failed: {result['error']}")

Best Practices

Error Handling
- Always wrap API calls in try-except blocks
- Log errors for debugging
- Provide meaningful error messages to users
Transaction Monitoring
- Implement retry logic for status checks
- Set appropriate timeouts
- Store transaction references for reconciliation
Security
- Never expose API keys in client-side code
- Validate input data
- Use HTTPS for callbacks
Performance
- Implement caching where appropriate
- Use asynchronous processing for bulk operations
- Set up monitoring for API limits

Common Issues and Solutions

Transaction Timeout

# Implement exponential backoff
def check_transaction_with_backoff(transaction_ref, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            status = collection.getTransactionStatus(transaction_ref)
            if status.get('status') in ['SUCCESSFUL', 'FAILED']:
                return status
        except Exception:
            wait_time = (2 ** attempt) * 1  # exponential backoff
            time.sleep(wait_time)
    return {'status': 'TIMEOUT'}

Balance Check

def ensure_sufficient_balance(amount):
    try:
        balance = disbursement.getBalance()
        available = float(balance.get('availableBalance', 0))
        required = float(amount)
        if available < required:
            raise ValueError(f"Insufficient balance: {available} < {required}")
        return True
    except Exception as e:
        logger.error(f"Balance check failed: {str(e)}")
        raise

Input Validation

def validate_phone_number(phone):
    # Adjust pattern based on your country's format
    pattern = r'^26096\d{7}$'
    if not re.match(pattern, phone):
        raise ValueError("Invalid phone number format")
    return True

Testing

import unittest
from unittest.mock import patch
from mtnmomoapi.collection import Collection

class TestMomoAPI(unittest.TestCase):
    def setUp(self):
        self.collection = Collection()
        
    @patch('mtnmomoapi.collection.Collection.requestToPay')
    def test_payment_request(self, mock_request):
        mock_request.return_value = {
            'status_code': 202,
            'transaction_ref': 'test-ref'
        }
        
        response = self.collection.requestToPay(
            amount="100",
            phone_number="260966456787",
            external_id="test-123"
        )
        
        self.assertEqual(response['status_code'], 202)
        self.assertTrue('transaction_ref' in response)

if __name__ == '__main__':
    unittest.main()

Support

For issues and feature requests, please visit our GitHub repository.

License

This project is licensed under the MIT License - see the LICENSE.txt file for details.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

README.md

README.md

MTN MoMo API Python Client

Installation

Configuration

Basic Usage Examples

1. Simple Collection Request

2. Simple Disbursement

Real-World Integration Examples

1. E-commerce Payment Processing

2. Salary Disbursement System

3. Subscription Payment System

Best Practices

Common Issues and Solutions

Testing

Support

License

Files

README.md

Latest commit

History

README.md

File metadata and controls

MTN MoMo API Python Client

Installation

Configuration

Basic Usage Examples

1. Simple Collection Request

2. Simple Disbursement

Real-World Integration Examples

1. E-commerce Payment Processing

2. Salary Disbursement System

3. Subscription Payment System

Best Practices

Common Issues and Solutions

Testing

Support

License