Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/mutuiris/voicepact/llms.txt

Use this file to discover all available pages before exploring further.

​
Overview

VoicePact integrates Africa’s Talking mobile money services to enable secure escrow-based payments for agricultural contracts. The system supports mobile checkout, payment verification, and automated escrow management.

​
Payment Architecture

​
Payment Flow

​
Payment States

class PaymentStatus(str, Enum):
    PENDING = "pending"         # Initial state, awaiting confirmation
    LOCKED = "locked"          # Funds in escrow
    PROCESSING = "processing"  # Being processed
    COMPLETED = "completed"    # Successfully transferred
    FAILED = "failed"          # Transaction failed
    REFUNDED = "refunded"      # Returned to payer
    EXPIRED = "expired"        # Timeout occurred

​
Configuration

​
Environment Variables

# Africa's Talking Credentials
AT_USERNAME=your_username
AT_API_KEY=your_api_key

# Payment Configuration
AT_PAYMENT_PRODUCT_NAME=VoicePact

# Webhook Settings
WEBHOOK_SECRET=your_webhook_secret
WEBHOOK_BASE_URL=https://your-domain.com

# Payment Limits (in cents)
MIN_PAYMENT_AMOUNT=100          # KES 1.00
MAX_PAYMENT_AMOUNT=1000000      # KES 10,000.00

# Escrow Settings
ESCROW_TIMEOUT=604800           # 7 days in seconds
PAYMENT_RETRY_ATTEMPTS=3
PAYMENT_RETRY_DELAY=300         # 5 minutes

​
Payment Configuration Class

From app/core/config.py:262-287: 
class Settings(BaseSettings):
    # Payment Configuration
    escrow_timeout: int = Field(
        default=7 * 24 * 60 * 60,
        description="Escrow timeout in seconds"
    )
    
    payment_retry_attempts: int = Field(
        default=3,
        description="Maximum payment retry attempts"
    )
    
    payment_retry_delay: int = Field(
        default=300,
        description="Payment retry delay in seconds"
    )
    
    min_payment_amount: int = Field(
        default=100,
        description="Minimum payment amount in cents"
    )
    
    max_payment_amount: int = Field(
        default=1000000,
        description="Maximum payment amount in cents"
    )

​
Mobile Checkout

​
Initiating Checkout

Start a mobile money checkout from the buyer’s account: 
from app.services.africastalking_client import get_africastalking_client

# Initiate checkout (app/services/africastalking_client.py:240-268)
at_client = await get_africastalking_client()

response = await at_client.mobile_checkout(
    phone_number="+254712345678",
    amount=50000,  # KES 50,000
    currency_code="KES",
    metadata={
        "contract_id": "AG-2024-001",
        "payment_type": "escrow",
        "buyer_name": "John Farmer"
    }
)
Response Format: 
{
  "status": "PendingConfirmation",
  "description": "Waiting for user to confirm payment",
  "transactionId": "ATPid_SampleTxnId123",
  "phoneNumber": "+254712345678"
}

​
Payment Request Model

From app/api/v1/endpoints/payments.py:19-24: 
class PaymentRequest(BaseModel):
    contract_id: str
    amount: float = Field(..., gt=0)
    currency: str = Field(default="KES")
    phone_number: str
    payment_type: str = Field(default="escrow")

​
API Endpoint

Endpoint: POST /api/v1/payments/checkout From app/api/v1/endpoints/payments.py:36-93: 
@router.post("/checkout", response_model=PaymentResponse)
async def mobile_checkout(
    request: PaymentRequest,
    at_client: AfricasTalkingClient = Depends(get_africastalking_client),
    db: AsyncSession = Depends(get_db)
):
    # Verify contract exists
    result = await db.execute(
        select(Contract).where(Contract.id == request.contract_id)
    )
    contract = result.scalar_one_or_none()
    
    if not contract:
        raise HTTPException(status_code=404, detail="Contract not found")
    
    # Create payment record
    payment = Payment(
        contract_id=request.contract_id,
        payer_phone=request.phone_number,
        amount=Decimal(str(request.amount)),
        currency=request.currency,
        payment_type=request.payment_type,
        status=PaymentStatus.PENDING
    )
    
    db.add(payment)
    await db.commit()
    
    # Initiate AT mobile checkout
    response = await at_client.mobile_checkout(
        phone_number=request.phone_number,
        amount=request.amount,
        currency_code=request.currency,
        metadata={
            "contract_id": request.contract_id,
            "payment_id": payment.id
        }
    )
    
    # Update with transaction ID
    if response.get('transactionId'):
        payment.external_transaction_id = response['transactionId']
        await db.commit()
    
    return PaymentResponse(
        payment_id=payment.id,
        transaction_id=payment.external_transaction_id,
        status=payment.status.value,
        amount=float(payment.amount),
        currency=payment.currency,
        created_at=payment.created_at.isoformat()
    )

​
Webhook Handling

​
Payment Webhook

Receive payment status updates from Africa’s Talking: Endpoint: POST /api/v1/payments/webhook From app/api/v1/endpoints/payments.py:96-136: 
@router.post("/webhook")
async def payment_webhook(
    request: Request,
    at_client: AfricasTalkingClient = Depends(get_africastalking_client),
    db: AsyncSession = Depends(get_db)
):
    form_data = await request.form()
    
    transaction_id = form_data.get("transactionId")
    status = form_data.get("status", "failed")
    phone_number = form_data.get("phoneNumber")
    amount = form_data.get("amount")
    
    logger.info(f"Payment webhook: {transaction_id}, {status}")
    
    if transaction_id:
        # Find payment by transaction ID
        result = await db.execute(
            select(Payment).where(
                Payment.external_transaction_id == transaction_id
            )
        )
        payment = result.scalar_one_or_none()
        
        if payment:
            # Update payment status
            if status.lower() == "success":
                payment.status = PaymentStatus.LOCKED
                payment.confirmed_at = datetime.utcnow()
            else:
                payment.status = PaymentStatus.FAILED
                payment.failure_reason = form_data.get(
                    "description", "Payment failed"
                )
            
            await db.commit()
            logger.info(f"Payment {payment.id} updated: {payment.status.value}")
    
    return {"status": "webhook_processed", "transaction_id": transaction_id}

​
Webhook Payload Format

Success Webhook: 
{
  "transactionId": "ATPid_SampleTxnId123",
  "status": "Success",
  "phoneNumber": "+254712345678",
  "amount": "KES 50000.00",
  "description": "Payment received successfully",
  "sourceType": "PhoneNumber",
  "provider": "Mpesa",
  "providerChannel": "525900",
  "providerMetadata": {
    "name": "John Farmer",
    "provider": "Mpesa",
    "providerChannel": "525900"
  }
}
Failed Webhook: 
{
  "transactionId": "ATPid_SampleTxnId123",
  "status": "Failed",
  "phoneNumber": "+254712345678",
  "amount": "KES 50000.00",
  "description": "Insufficient funds",
  "sourceType": "PhoneNumber",
  "provider": "Mpesa"
}

​
Escrow Management

​
Locking Funds in Escrow

When payment is confirmed, funds are locked until delivery: 
async def lock_payment_in_escrow(
    payment_id: int,
    db: AsyncSession
):
    payment = await db.get(Payment, payment_id)
    
    if payment and payment.status == PaymentStatus.PENDING:
        payment.status = PaymentStatus.LOCKED
        payment.locked_at = datetime.utcnow()
        payment.expires_at = datetime.utcnow() + timedelta(
            seconds=settings.escrow_timeout
        )
        await db.commit()
        
        logger.info(f"Payment {payment_id} locked in escrow")

​
Releasing Escrow

Release funds to seller after delivery confirmation: 
async def release_escrow_payment(
    payment_id: int,
    recipient_phone: str,
    db: AsyncSession,
    at_client: AfricasTalkingClient
):
    payment = await db.get(Payment, payment_id)
    
    if payment and payment.status == PaymentStatus.LOCKED:
        # Update payment status
        payment.status = PaymentStatus.PROCESSING
        await db.commit()
        
        try:
            # Transfer to seller
            response = await at_client.mobile_data_transfer(
                phone_number=recipient_phone,
                amount=float(payment.amount),
                currency_code=payment.currency,
                metadata={
                    "payment_id": payment.id,
                    "contract_id": payment.contract_id,
                    "type": "escrow_release"
                }
            )
            
            payment.status = PaymentStatus.COMPLETED
            payment.completed_at = datetime.utcnow()
            await db.commit()
            
            logger.info(f"Payment {payment_id} released to {recipient_phone}")
            
        except Exception as e:
            payment.status = PaymentStatus.FAILED
            payment.failure_reason = str(e)
            await db.commit()
            raise

​
Refunding Payment

Refund payment if contract is cancelled or disputed: 
async def refund_payment(
    payment_id: int,
    reason: str,
    db: AsyncSession,
    at_client: AfricasTalkingClient
):
    payment = await db.get(Payment, payment_id)
    
    if payment and payment.status == PaymentStatus.LOCKED:
        # Process refund (implementation depends on AT API)
        payment.status = PaymentStatus.REFUNDED
        payment.refunded_at = datetime.utcnow()
        payment.refund_reason = reason
        
        await db.commit()
        
        # Notify payer
        await at_client.send_sms(
            message=f"Payment refund processed for contract {payment.contract_id}. Amount: {payment.currency} {payment.amount}",
            recipients=[payment.payer_phone]
        )

​
Payment Query

​
Get Payment Status

Endpoint: GET /api/v1/payments/{payment_id} 
@router.get("/{payment_id}", response_model=PaymentResponse)
async def get_payment(
    payment_id: int,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(Payment).where(Payment.id == payment_id)
    )
    payment = result.scalar_one_or_none()
    
    if not payment:
        raise HTTPException(status_code=404, detail="Payment not found")
    
    return PaymentResponse(
        payment_id=payment.id,
        transaction_id=payment.external_transaction_id,
        status=payment.status.value,
        amount=float(payment.amount),
        currency=payment.currency,
        created_at=payment.created_at.isoformat()
    )

​
Get Contract Payments

Endpoint: GET /api/v1/payments/contract/{contract_id} From app/api/v1/endpoints/payments.py:170-200: 
@router.get("/contract/{contract_id}")
async def get_contract_payments(
    contract_id: str,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(Payment)
        .where(Payment.contract_id == contract_id)
        .order_by(Payment.created_at.desc())
    )
    payments = result.scalars().all()
    
    payment_list = [
        PaymentResponse(
            payment_id=payment.id,
            transaction_id=payment.external_transaction_id,
            status=payment.status.value,
            amount=float(payment.amount),
            currency=payment.currency,
            created_at=payment.created_at.isoformat()
        )
        for payment in payments
    ]
    
    return {"contract_id": contract_id, "payments": payment_list}

​
Transaction Queries

​
Query Transaction Status

Query Africa’s Talking for transaction details: 
# Query transaction (app/services/africastalking_client.py:306-320)
response = await at_client.query_transaction_status(
    transaction_id="ATPid_SampleTxnId123"
)

# Response format:
{
    "status": "Success",
    "transactionId": "ATPid_SampleTxnId123",
    "amount": "KES 50000.00",
    "phoneNumber": "+254712345678",
    "provider": "Mpesa",
    "providerChannel": "525900"
}

​
Check Wallet Balance

Endpoint: GET /api/v1/payments/wallet/balance From app/api/v1/endpoints/payments.py:230-241: 
@router.get("/wallet/balance")
async def get_wallet_balance(
    at_client: AfricasTalkingClient = Depends(get_africastalking_client)
):
    response = await at_client.get_wallet_balance()
    return {"wallet_balance": response}

# Response format:
{
    "wallet_balance": {
        "status": "Success",
        "balance": "KES 10000.00"
    }
}

​
Payment Notifications

​
SMS Notifications

Send payment confirmation messages: 
# Payment received notification
message = at_client.generate_payment_sms(
    contract_id="AG-2024-001",
    amount=50000,
    currency="KES",
    action="received"
)

await at_client.send_sms(
    message=message,
    recipients=["+254712345678"]
)

# Output:
# VoicePact Payment Received:
# Contract: AG-2024-001
# Amount: KES 50,000.00
# Status: Processing
# You will receive confirmation shortly.

​
USSD Payment Status

Check payment status via USSD: 
# In USSD handler
if user_input == "3":  # Check payments
    payments = await get_contract_payments(contract_id, db)
    
    menu_text = "Payment Status:\n"
    for payment in payments:
        menu_text += f"{payment.currency} {payment.amount:,.0f} - {payment.status.value}\n"
    
    return at_client.build_ussd_response(menu_text, end_session=False)

​
Error Handling

​
Common Payment Errors

try:
    response = await at_client.mobile_checkout(...)
except AfricasTalkingException as e:
    if "Insufficient funds" in str(e):
        # Handle insufficient funds
        logger.warning(f"Payment failed: insufficient funds")
    elif "Invalid phone number" in str(e):
        # Handle invalid number
        logger.error(f"Invalid phone number format")
    else:
        # Generic error
        logger.error(f"Payment failed: {e.message}")

​
Retry Logic

Automatic retry for transient failures: 
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10)
)
async def mobile_checkout(...):
    # Automatically retries on failure
    ...

​
Testing

​
Test Payment Endpoint

Endpoint: POST /api/v1/payments/test/checkout From app/api/v1/endpoints/payments.py:203-227: 
curl -X POST "http://localhost:8000/api/v1/payments/test/checkout?phone_number=+254712345678&amount=100"
Response: 
{
  "status": "test_initiated",
  "phone_number": "+254712345678",
  "amount": 100,
  "response": {
    "status": "PendingConfirmation",
    "transactionId": "ATPid_..."
  }
}

​
Sandbox Testing

Use Africa’s Talking sandbox credentials: 
# Test with sandbox
AT_USERNAME=sandbox
AT_API_KEY=your_sandbox_key

# Test payment amounts
# KES 10 - Success
# KES 20 - Insufficient funds
# KES 30 - Invalid account

​
Best Practices

  1. Validate Amounts: Always validate min/max payment amounts
  2. Store Transaction IDs: Keep external transaction IDs for reconciliation
  3. Handle Timeouts: Set appropriate escrow timeouts
  4. Verify Webhooks: Always verify webhook authenticity
  5. Notify Users: Send SMS confirmations for all payment events
  6. Monitor Escrow: Track escrow expiration times
  7. Handle Failures: Implement proper error handling and retry logic
  8. Reconcile Daily: Regular reconciliation with AT transaction logs

​
Security Considerations

​
Webhook Security

# Always verify webhook signatures
if not at_client.verify_webhook_signature(payload, signature):
    logger.warning("Invalid webhook signature")
    raise HTTPException(status_code=401)

​
Amount Validation

def validate_payment_amount(amount: float, currency: str) -> bool:
    min_amount = settings.min_payment_amount / 100  # Convert from cents
    max_amount = settings.max_payment_amount / 100
    
    if amount < min_amount or amount > max_amount:
        raise ValueError(
            f"Amount must be between {min_amount} and {max_amount} {currency}"
        )
    return True

​
Phone Number Validation

# Validate and format phone numbers
formatted_phone = await at_client.format_phone_number("+254712345678")
if not await at_client.validate_phone_number(formatted_phone):
    raise ValueError("Invalid phone number format")

​
Troubleshooting

​
Payment Not Processing

  1. Check API credentials: Verify AT_USERNAME and AT_API_KEY
  2. Verify phone number: Ensure correct format (+254…)
  3. Check amount: Verify within min/max limits
  4. Review logs: Check application logs for errors
  5. Test webhook: Ensure webhook URL is accessible

​
Webhook Not Receiving

  1. Verify URL: Check WEBHOOK_BASE_URL is correct and accessible
  2. Check firewall: Ensure port 443/80 is open
  3. Test signature: Verify WEBHOOK_SECRET matches
  4. Review AT dashboard: Check webhook configuration

​
Escrow Release Failing

  1. Check payment status: Verify payment is in LOCKED state
  2. Verify recipient: Ensure seller phone number is valid
  3. Check wallet balance: Ensure sufficient funds in AT wallet
  4. Review timeout: Check if escrow has expired

​
Next Steps

Africa's Talking

Core AT integration documentation

Webhooks

Complete webhook configuration

Contract Flow

Contract lifecycle management

API Reference

Payment API documentation

Build docs developers (and LLMs) love

Get started for freeTalk to us