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 uses Africa’s Talking as its primary telecommunications provider for voice calls, SMS messaging, USSD sessions, and mobile money payments. The integration provides a robust, circuit-breaker-protected client with retry logic and comprehensive error handling.

​
Architecture

The Africa’s Talking client is implemented in app/services/africastalking_client.py and provides:
  • Voice Services: Automated voice calls and conference recording
  • SMS Services: Bulk SMS with delivery tracking
  • USSD Services: Interactive menu-based contract management
  • Payment Services: Mobile money checkout and escrow management
  • Circuit Breaker: Automatic failover protection
  • Webhook Verification: HMAC-based signature validation

​
Configuration

​
Environment Variables

Configure the following environment variables in your .env file: 
# Africa's Talking Credentials
AT_USERNAME=sandbox                    # Your AT username (use 'sandbox' for testing)
AT_API_KEY=your_api_key_here          # Your AT API key

# Voice Configuration
AT_VOICE_NUMBER=+254XXXXXXXXX         # Your AT voice number

# USSD Configuration
AT_USSD_SERVICE_CODE=*483#            # Your USSD service code

# Payment Configuration
AT_PAYMENT_PRODUCT_NAME=VoicePact     # Product name for payments

# Webhook Security
WEBHOOK_SECRET=your_webhook_secret    # Secret for webhook signature validation
WEBHOOK_BASE_URL=https://your-domain.com  # Base URL for webhooks

​
Configuration Class

The settings are managed by the Settings class in app/core/config.py:80-104: 
class Settings(BaseSettings):
    # Africa's Talking API Configuration
    at_username: str = Field(default="sandbox")
    at_api_key: SecretStr = Field(...)
    at_voice_number: str = Field(default="+254XXXXXXXXX")
    at_payment_product_name: str = Field(default="VoicePact")
    at_ussd_service_code: str = Field(default="*483#")

​
Client Initialization

​
Basic Setup

The client is initialized as a singleton instance: 
from app.services.africastalking_client import get_africastalking_client

# In your FastAPI endpoint
at_client = await get_africastalking_client()

​
Client Architecture

The AfricasTalkingClient class (lines 68-113) includes: 
class AfricasTalkingClient:
    def __init__(self):
        self.username = settings.at_username
        self.api_key = settings.get_secret_value('at_api_key')
        self.voice_number = settings.at_voice_number
        self.service_code = settings.at_ussd_service_code
        
        africastalking.initialize(self.username, self.api_key)
        
        # Initialize services
        self.sms_service = africastalking.SMS
        self.voice_service = africastalking.Voice
        self.payment_service = africastalking.Payment
        
        # Circuit breakers for fault tolerance
        self.sms_circuit_breaker = CircuitBreaker()
        self.voice_circuit_breaker = CircuitBreaker()
        self.payment_circuit_breaker = CircuitBreaker()

​
Voice Services

​
Making Voice Calls

Initiate voice calls with automatic retry logic: 
# Make a voice call (app/services/africastalking_client.py:201-221)
response = await at_client.make_voice_call(
    recipients=["+254712345678", "+254723456789"],
    from_number=None  # Uses configured voice number
)
Response Format: 
{
  "sessionId": "ATVId_abc123...",
  "status": "Success",
  "entries": [
    {
      "phoneNumber": "+254712345678",
      "status": "Queued"
    }
  ]
}

​
Voice Webhooks

Voice webhooks are handled at /api/v1/voice/webhook (see app/api/v1/endpoints/voice.py:223-258): Webhook Payload: 
# Received from Africa's Talking
{
    "sessionId": "ATVId_...",
    "phoneNumber": "+254712345678",
    "recordingUrl": "https://voice.africastalking.com/recordings/...",
    "duration": 120,  # seconds
    "status": "completed"
}
Handler Implementation: 
@router.post("/webhook")
async def voice_webhook(
    request: Request,
    at_client: AfricasTalkingClient = Depends(get_africastalking_client),
    db: AsyncSession = Depends(get_db)
):
    form_data = await request.form()
    
    session_id = form_data.get("sessionId")
    recording_url = form_data.get("recordingUrl")
    duration = form_data.get("duration")
    
    # Update recording in database
    recording = await db.get(VoiceRecording, session_id)
    if recording:
        recording.recording_url = recording_url
        recording.duration = int(duration) if duration else None
        await db.commit()
    
    return {"status": "webhook_processed"}

​
SMS Services

​
Sending SMS

Send SMS with automatic retry and circuit breaker protection: 
# Send single SMS (app/services/africastalking_client.py:136-161)
response = await at_client.send_sms(
    message="Your VoicePact contract is ready for review.",
    recipients=["+254712345678"],
    sender_id="VoicePact",  # Optional
    enqueue=False
)
Response Format: 
{
  "SMSMessageData": {
    "Message": "Sent to 1/1 Total Cost: KES 0.8000",
    "Recipients": [
      {
        "statusCode": 101,
        "number": "+254712345678",
        "status": "Success",
        "cost": "KES 0.8000",
        "messageId": "ATXid_..."
      }
    ]
  }
}

​
Bulk SMS

Send messages to multiple recipients: 
# Send bulk SMS (app/services/africastalking_client.py:167-182)
messages = [
    {
        "message": "Contract AG-001 ready for signing",
        "recipients": ["+254712345678"]
    },
    {
        "message": "Payment received for contract AG-001",
        "recipients": ["+254723456789"]
    }
]

responses = await at_client.send_bulk_sms(
    messages=messages,
    sender_id="VoicePact"
)

​
SMS Templates

The client provides built-in SMS templates:

​
Contract Notification

# Generate contract SMS (app/services/africastalking_client.py:407-430)
message = at_client.generate_contract_sms(
    contract_id="AG-2024-001",
    contract_terms={
        "product": "Maize",
        "quantity": 100,
        "unit": "bags",
        "total_amount": 50000,
        "currency": "KES",
        "delivery_deadline": "2024-04-15"
    }
)

# Output:
# VoicePact Contract Summary:
# ID: AG-2024-001
# Product: Maize (100 bags)
# Total: KES 50,000.00, Due: 2024-04-15
# Reply YES-AG-2024-001 to confirm or NO-AG-2024-001 to decline

​
Payment Notification

# Generate payment SMS (app/services/africastalking_client.py:433-446)
message = at_client.generate_payment_sms(
    contract_id="AG-2024-001",
    amount=50000,
    currency="KES",
    action="received"
)

​
SMS Webhooks

Handle incoming SMS and delivery reports at /api/v1/sms/webhook (see app/api/v1/endpoints/sms.py:293-328): Webhook Payload: 
# Incoming SMS
{
    "from": "+254712345678",
    "to": "40404",
    "text": "YES-AG-2024-001",
    "date": "2024-03-06 10:30:00",
    "id": "ATXid_...",
    "linkId": "SampleLinkId123"
}

# Delivery report
{
    "id": "ATXid_...",
    "status": "Success",
    "phoneNumber": "+254712345678",
    "retryCount": 0
}
Handler Example: 
@router.post("/webhook")
async def sms_webhook(request: Request):
    form_data = await request.form()
    
    phone_number = form_data.get("from")
    message = form_data.get("text", "").upper().strip()
    
    # Handle contract confirmation
    if message.startswith("YES-") or message.startswith("NO-"):
        contract_id = message.split("-", 1)[1]
        action = "confirm" if message.startswith("YES-") else "reject"
        
        logger.info(f"Contract {action}: {contract_id} from {phone_number}")
        
        return {
            "action": action,
            "contract_id": contract_id,
            "phone_number": phone_number
        }
    
    return {"status": "webhook_received"}

​
USSD Services

​
USSD Menu Flow

USSD menus are handled at /api/v1/ussd/ (see app/api/v1/endpoints/ussd.py): Main Menu: 
def main_menu() -> str:
    return """Welcome to VoicePact
1. View My Contracts
2. Confirm Delivery
3. Check Payments
4. Help & Support
0. Exit"""

​
Building USSD Responses

Use the utility methods to build USSD responses: 
# Continue session (app/services/africastalking_client.py:382-385)
response = at_client.build_ussd_response(
    text="Select an option:",
    end_session=False  # CON response
)

# End session
response = at_client.build_ussd_response(
    text="Thank you for using VoicePact!",
    end_session=True  # END response
)

​
USSD Request Format

Webhook Payload: 
{
    "sessionId": "ATUid_...",
    "serviceCode": "*483#",
    "phoneNumber": "+254712345678",
    "text": "1*2*3"  # User navigation path
}

​
USSD Handler Implementation

@router.post("/")
async def ussd_handler(
    request: Request,
    sessionId: str = Form(...),
    serviceCode: str = Form(...),
    phoneNumber: str = Form(...),
    text: str = Form(""),
    at_client: AfricasTalkingClient = Depends(get_africastalking_client),
    db: AsyncSession = Depends(get_db)
):
    # Parse user input
    user_input = text.split('*')[-1] if text else ""
    
    # First request - show main menu
    if not text:
        return at_client.build_ussd_response(main_menu(), end_session=False)
    
    # Handle menu navigation
    if user_input == "1":
        # Get user contracts
        contracts = await get_user_contracts(phoneNumber, db)
        menu_text = "Your Contracts:\n"
        for i, contract in enumerate(contracts[:5], 1):
            menu_text += f"{i}. {contract.id[:12]}...\n"
        return at_client.build_ussd_response(menu_text, end_session=False)
    
    # Exit
    elif user_input == "0":
        return at_client.build_ussd_response(
            "Thank you!", end_session=True
        )

​
Payment Services

See Mobile Money Integration for detailed payment documentation.

​
Mobile Checkout

# Initiate checkout (app/services/africastalking_client.py:240-268)
response = await at_client.mobile_checkout(
    phone_number="+254712345678",
    amount=50000,
    currency_code="KES",
    metadata={
        "contract_id": "AG-2024-001",
        "payment_type": "escrow"
    }
)

​
Circuit Breaker

The client implements circuit breaker pattern for fault tolerance:

​
Circuit Breaker States

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,      # Failures before opening
        recovery_timeout: int = 60,      # Seconds before retry
        expected_exception=Exception
    ):
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN
States:
  • CLOSED: Normal operation, requests pass through
  • OPEN: Too many failures, requests blocked
  • HALF_OPEN: Testing if service recovered

​
Usage Example

# Circuit breaker automatically protects service calls
try:
    response = await at_client.send_sms(
        message="Test message",
        recipients=["+254712345678"]
    )
except CircuitBreakerOpen:
    logger.error("SMS service unavailable - circuit breaker open")
    # Handle gracefully

​
Webhook Security

​
Signature Verification

Verify webhook authenticity using HMAC signatures: 
# Verify webhook (app/services/africastalking_client.py:119-129)
@router.post("/webhook")
async def secure_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Africastalking-Signature")
    
    # Verify signature
    if not at_client.verify_webhook_signature(
        payload=body.decode(),
        signature=signature
    ):
        raise HTTPException(status_code=401, detail="Invalid signature")
    
    # Process webhook
    ...

​
Implementation

def verify_webhook_signature(self, payload: str, signature: str) -> bool:
    if not self.webhook_secret or not signature:
        return False
    
    expected_signature = hmac.new(
        self.webhook_secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(f"sha256={expected_signature}", signature)

​
Error Handling

​
Custom Exceptions

class AfricasTalkingException(Exception):
    def __init__(
        self,
        message: str,
        error_code: Optional[str] = None,
        response_data: Optional[dict] = None
    ):
        self.message = message
        self.error_code = error_code
        self.response_data = response_data or {}

​
Retry Logic

Automatic retry with exponential backoff: 
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((httpx.TimeoutException, httpx.ConnectError))
)
async def send_sms(...):
    # Automatically retries on timeout/connection errors
    ...

​
Health Checks

​
Service Status

# Check service health (app/services/africastalking_client.py:520-543)
health = await at_client.health_check()

# Returns:
{
    "sms_service": "healthy",
    "payment_service": "healthy",
    "voice_service": "healthy",
    "sms_circuit": "CLOSED",
    "voice_circuit": "CLOSED",
    "payment_circuit": "CLOSED"
}

​
Testing

​
Sandbox Mode

Use Africa’s Talking sandbox for testing: 
# .env for development
AT_USERNAME=sandbox
AT_API_KEY=your_sandbox_api_key

​
Test Endpoints

# Test SMS
curl -X POST "http://localhost:8000/api/v1/sms/test?phone_number=+254712345678"

# Test Voice
curl -X POST "http://localhost:8000/api/v1/voice/conference/create" \
  -H "Content-Type: application/json" \
  -d '{"parties": ["+254712345678", "+254723456789"]}'

# Test Payment
curl -X POST "http://localhost:8000/api/v1/payments/test/checkout?phone_number=+254712345678&amount=100"

​
Best Practices

  1. Use Circuit Breakers: Let the built-in circuit breakers protect your services
  2. Verify Webhooks: Always verify webhook signatures in production
  3. Format Phone Numbers: Use format_phone_number() for consistent formatting
  4. Handle Retries: Configure appropriate retry limits for your use case
  5. Monitor Health: Regularly check service health status
  6. Log Everything: Comprehensive logging helps debug integration issues
  7. Test in Sandbox: Always test with sandbox credentials first

​
Troubleshooting

​
Common Issues

SMS Not Sending: 
# Check API key
if not at_client.sms_service:
    logger.error("SMS service not initialized - check AT_API_KEY")

# Check phone number format
formatted = await at_client.format_phone_number("+254712345678")
valid = await at_client.validate_phone_number(formatted)
Circuit Breaker Open: 
# Wait for recovery or manually reset
at_client.sms_circuit_breaker.reset()
Webhook Not Receiving:
  • Verify webhook URL is publicly accessible
  • Check firewall settings
  • Validate webhook signature configuration
  • Review Africa’s Talking dashboard logs

​
Next Steps

Mobile Money

Learn about mobile money integration

Webhooks

Complete webhook configuration guide

API Reference

Full API documentation

Voice Processing

Voice recording and processing

Build docs developers (and LLMs) love

Get started for freeTalk to us