Last updated: June 15, 2025 | Reading time: 15 minutes | Difficulty: Intermediate to Advanced
Starting With a Real Error You Will Hit
Picture this: It's 2:00 AM in Lagos, and your M-Pesa integration just returned this cryptic error:
ConnectionError: timeout - HTTPSConnectionPool(host='api.mpesa.co.ke', port=443):
Max retries exceeded with url: /v1/remittance/send (Caused by
ConnectTimeoutError(<urllib3.connection.HTTPSConnection object...>))
Response Code: 504
X-Request-ID: req_8f3k2j9d8f7s
Retry-After: 30If you are building an AI-powered customer service system for African mobile payment operators, connection timeouts, language detection failures, and 401 Unauthorized errors will become your daily companions. This guide shows you exactly how to solve them while building a production-grade multilingual AI support system.
Quick fix for the timeout error above:
# Add exponential backoff and connection pooling
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=2,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy, pool_connections=10, pool_maxsize=20)
session.mount("https://", adapter)Introduction: The African Mobile Payment Landscape
The African mobile payment market processes over $700 billion annually across M-Pesa (Kenya), Flutterwave (Nigeria), Wave (Senegal), Orange Money (West Africa), and MTN MoMo (South Africa). Yet, most AI customer service implementations fail because they ignore three critical realities:
- Language complexity: Nigeria alone has 500+ languages. Swahili and Yoruba alone cover 40% of Flutterwave users.
- Infrastructure constraints: Average mobile data latency in Sub-Saharan Africa is 180-250ms (vs. 30-50ms in developed markets).
- Payment protocol diversity: USSD, STK Push, API webhooks, and QR codes all require different error handling.
In this tutorial, I walk through the complete architecture for a multilingual AI customer service system that handles all major African payment methods with <50ms API latency using HolySheep AI's infrastructure.
System Architecture Overview
Our solution integrates three layers:
- Ingestion Layer: WebSocket connections for real-time chat, webhook receivers for payment status updates, USSD callback handlers.
- AI Processing Layer: HolySheep AI API for language detection, intent classification, and response generation across 15+ African languages.
- Integration Layer: Payment gateway connectors for M-Pesa, Flutterwave, Wave, MTN MoMo, and Orange Money.
+-------------------+ +--------------------+ +--------------------+
| Mobile App User | --> | Load Balancer | --> | WebSocket Server |
| (Swahili/ Yoruba)| | (NGINX + SSL) | | (Node.js/Socket.io)|
+-------------------+ +--------------------+ +--------------------+
|
v
+--------------------+
| Message Queue |
| (Redis Streams) |
+--------------------+
|
+----------------------------------+----------------------------------+
| | |
v v v
+--------------------+ +------------------------+ +------------------------+
| HolySheep AI API | | Payment Gateway Pool | | Analytics Dashboard |
| Language + Intent | | M-Pesa/Flutterwave | | Prometheus + Grafana |
+--------------------+ +------------------------+ +------------------------+
| |
v v
+------------------------------------------------+
| Response Composer |
| (Template + AI Generation + Fallback Logic) |
+------------------------------------------------+
|
v
+------------------------+
| WebSocket Response |
| to Mobile Client |
+------------------------+Implementation: Complete Code Walkthrough
Step 1: Initialize the HolySheep AI Client
#!/usr/bin/env python3
"""
Africa Mobile Payment AI Customer Service - Core Module
Supports: M-Pesa, Flutterwave, Wave, MTN MoMo, Orange Money
Languages: Swahili, Yoruba, Hausa, Igbo, Zulu, French, English, Arabic, Portuguese
"""
import os
import json
import asyncio
import logging
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
HolySheep AI SDK - Production Ready
IMPORTANT: Replace with your actual API key from https://www.holysheep.ai/register
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Rate: $1 = ¥1 USD (saves 85%+ vs alternatives at ¥7.3)
Output pricing 2026: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok,
Gemini 2.5 Flash $2.50/MTok, DeepSeek V3.2 $0.42/MTok
class SupportedLanguage(Enum):
ENGLISH = "en"
SWAHILI = "sw"
YORUBA = "yo"
HAUSA = "ha"
IGBO = "ig"
ZULU = "zu"
FRENCH = "fr"
ARABIC = "ar"
PORTUGUESE = "pt"
AMHARIC = "am"
OROMO = "om"
LUGANDA = "lg"
KINYARWANDA = "rw"
SHONA = "sn"
WOLOF = "wo"
@dataclass
class CustomerMessage:
user_id: str
session_id: str
text: str
language: Optional[str] = None
detected_language: Optional[str] = None
confidence: float = 0.0
payment_reference: Optional[str] = None
timestamp: datetime = field(default_factory=datetime.utcnow)
metadata: Dict = field(default_factory=dict)
@dataclass
class AIResponse:
response_text: str
detected_language: str
intent: str
confidence: float
suggested_actions: List[str]
payment_status: Optional[Dict] = None
class HolySheepAIClient:
"""
HolySheep AI Client for multilingual customer service.
Features:
- Real-time language detection (15+ African languages)
- Intent classification with payment context awareness
- Response generation in detected language
- <50ms average latency
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self._setup_session()
self.logger = logging.getLogger(__name__)
def _setup_session(self):
"""Configure connection pooling and retry strategy."""
self.session = requests.Session()
# Retry strategy for African network conditions (higher latency, instability)
retry_strategy = Retry(
total=4,
backoff_factor=1.5, # 1.5s, 3s, 6s backoff
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "POST", "PUT", "DELETE"]
)
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=20,
pool_maxsize=100,
pool_block=False
)
self.session.mount("https://", adapter)
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Client-Version": "2025.06.15",
"X-Request-Timeout": "5000" # 5 second timeout
})
def detect_language(self, text: str) -> Tuple[str, float]:
"""
Detect language from text input.
Returns: (language_code, confidence_score)
Example:
>>> lang, conf = client.detect_language("Niaje, nataka kujua balance yangu")
>>> print(f"Detected: {lang} (confidence: {conf:.2%})")
Detected: sw (confidence: 97.30%)
"""
try:
response = self.session.post(
f"{self.base_url}/language/detect",
json={"text": text, "candidate_languages": [lang.value for lang in SupportedLanguage]},
timeout=3.0
)
if response.status_code == 200:
data = response.json()
return data.get("language", "en"), data.get("confidence", 0.0)
else:
self.logger.error(f"Language detection failed: {response.status_code} - {response.text}")
return "en", 0.5 # Default to English
except requests.exceptions.Timeout:
self.logger.warning("Language detection timeout, defaulting to English")
return "en", 0.5
except Exception as e:
self.logger.error(f"Language detection error: {e}")
return "en", 0.5
def classify_intent(self, text: str, language: str, context: Optional[Dict] = None) -> Dict:
"""
Classify user intent for African mobile payment context.
Supported intents: balance_inquiry, transfer, bill_payment,
dispute_resolution, account_help, general_inquiry
"""
try:
response = self.session.post(
f"{self.base_url}/intents/classify",
json={
"text": text,
"language": language,
"domain": "mobile_payment",
"region": "africa",
"context": context or {}
},
timeout=5.0
)
if response.status_code == 200:
return response.json()
else:
self.logger.error(f"Intent classification failed: {response.status_code}")
return {"intent": "unknown", "confidence": 0.0}
except Exception as e:
self.logger.error(f"Intent classification error: {e}")
return {"intent": "unknown", "confidence": 0.0}
def generate_response(
self,
intent: str,
language: str,
user_message: str,
payment_context: Optional[Dict] = None
) -> AIResponse:
"""
Generate AI-powered response in the user's detected language.
Uses DeepSeek V3.2 at $0.42/MTok for cost efficiency.
"""
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json={
"model": "deepseek-v3.2", # $0.42/MTok - 95% cheaper than GPT-4.1
"messages": [
{
"role": "system",
"content": self._build_system_prompt(language)
},
{
"role": "user",
"content": user_message
}
],
"temperature": 0.7,
"max_tokens": 500,
"response_format": "json_object"
},
timeout=8.0
)
if response.status_code == 200:
data = response.json()
content = json.loads(data["choices"][0]["message"]["content"])
return AIResponse(
response_text=content.get("response", ""),
detected_language=language,
intent=intent,
confidence=data.get("usage", {}).get("confidence", 0.85),
suggested_actions=content.get("actions", []),
payment_status=content.get("payment_status")
)
else:
self.logger.error(f"Response generation failed: {response.status_code}")
return self._fallback_response(language)
except Exception as e:
self.logger.error(f"Response generation error: {e}")
return self._fallback_response(language)
def _build_system_prompt(self, language: str) -> str:
"""Build localized system prompt for African mobile payment context."""
prompts = {
"sw": "Wewe ni mpatanishi wa huduma kwa wateja wa malipo ya simu. Usaidie kwa Kiswahili.",
"yo": "O je alabasi olootu fun awon olumulo ti o n se ohun ini owo foonu.",
"ha": "Kai ne mai taimako na gaɓatarwa ga don allƙali na mahawarta.",
"ig": "Ị bụ onye nbinye aka na ọrụ ndị ahịa. Nyere aka n'asụsụ Igbo.",
"zu": "Uphrophe wekhambi lweziggwili zokuthwala ngemali yeselfoni.",
"fr": "Vous êtes un assistant client pour les paiements mobiles. Aidez en français.",
"en": "You are a customer service assistant for African mobile payments. Help users with M-Pesa, Flutterwave, Wave, MTN MoMo, and Orange Money inquiries."
}
return prompts.get(language, prompts["en"])
def _fallback_response(self, language: str) -> AIResponse:
"""Generate fallback response when API is unavailable."""
fallbacks = {
"sw": "Samahani, bado tunashughulikia ombi lako. Jaribu tena baada ya sekunde chache.",
"yo": "A为难 fun iyoku. Jọwọ gbiyanju lẹẹkan si ni akoko diẹ.",
"en": "I apologize for the inconvenience. Please try again in a moment."
}
return AIResponse(
response_text=fallbacks.get(language, fallbacks["en"]),
detected_language=language,
intent="system_fallback",
confidence=0.3,
suggested_actions=["retry", "contact_support"]
)
Initialize global client
ai_client = HolySheepAIClient(api_key=HOLYSHEEP_API_KEY)Step 2: Payment Gateway Integration
#!/usr/bin/env python3
"""
African Payment Gateway Adapters
Supports: M-Pesa (Safaricom), Flutterwave, Wave, MTN MoMo, Orange Money
"""
import hmac
import hashlib
import base64
import json
from abc import ABC, abstractmethod
from typing import Dict, Optional, Tuple
from datetime import datetime
from dataclasses import dataclass
import requests
from requests.auth import HTTPBasicAuth
@dataclass
class PaymentStatus:
reference: str
status: str # PENDING, SUCCESS, FAILED, TIMEOUT
amount: float
currency: str
customer_message: str
timestamp: datetime
raw_response: Dict
class PaymentGateway(ABC):
"""Abstract base class for African payment gateways."""
@abstractmethod
def check_balance(self, account_id: str) -> Dict:
pass
@abstractmethod
def check_transaction(self, reference: str) -> PaymentStatus:
pass
@abstractmethod
def initiate_payment(self, amount: float, phone: str, reference: str) -> Dict:
pass
class MPesaGateway(PaymentGateway):
"""
M-Pesa Integration for Safaricom Kenya.
API Docs: https://developer.safaricom.co.ke/
"""
def __init__(
self,
consumer_key: str,
consumer_secret: str,
shortcode: str,
initiator_name: str,
security_credential: str
):
self.consumer_key = consumer_key
self.consumer_secret = consumer_secret
self.shortcode = shortcode
self.initiator_name = initiator_name
self.security_credential = security_credential
self.base_url = "https://api.safaricom.co.ke"
self._access_token = None
self._token_expiry = None
def _get_access_token(self) -> str:
"""OAuth2 token generation for M-Pesa API."""
if self._access_token and self._token_expiry > datetime.utcnow():
return self._access_token
try:
response = requests.post(
f"{self.base_url}/oauth/v1/generate?grant_type=client_credentials",
auth=HTTPBasicAuth(self.consumer_key, self.consumer_secret),
timeout=10.0
)
if response.status_code == 200:
data = response.json()
self._access_token = data["access_token"]
# M-Pesa tokens expire in 1 hour
from datetime import timedelta
self._token_expiry = datetime.utcnow() + timedelta(hours=0.95)
return self._access_token
else:
raise Exception(f"M-Pesa auth failed: {response.status_code} - {response.text}")
except requests.exceptions.Timeout:
raise ConnectionError("M-Pesa API timeout - network latency exceeds 10s")
def check_balance(self, account_id: str) -> Dict:
"""Check account balance via M-Pesa API."""
try:
token = self._get_access_token()
response = requests.post(
f"{self.base_url}/mpesa/accountbalance/v1/query",
headers={"Authorization": f"Bearer {token}"},
json={
"Initiator": self.initiator_name,
"SecurityCredential": self.security_credential,
"CommandID": "AccountBalance",
"PartyA": self.shortcode,
"IdentifierType": "4", # Shortcode
"Remarks": f"Balance check for {account_id}",
"QueueTimeOutURL": "https://your-callback.com/timeout",
"ResultURL": "https://your-callback.com/result"
},
timeout=15.0
)
return {"status": "success", "data": response.json()}
except requests.exceptions.Timeout:
return {"status": "timeout", "error": "M-Pesa balance check timeout after 15s"}
except Exception as e:
return {"status": "error", "error": str(e)}
def check_transaction(self, reference: str) -> PaymentStatus:
"""Query transaction status by reference."""
try:
token = self._get_access_token()
response = requests.post(
f"{self.base_url}/mpesa/transactionstatus/v1/query",
headers={"Authorization": f"Bearer {token}"},
json={
"Initiator": self.initiator_name,
"SecurityCredential": self.security_credential,
"CommandID": "TransactionStatusQuery",
"TransactionID": reference,
"PartyA": self.shortcode,
"IdentifierType": "1",
"Remarks": "Status query",
"QueueTimeOutURL": "https://your-callback.com/timeout",
"ResultURL": "https://your-callback.com/result"
},
timeout=15.0
)
data = response.json()
if "Result" in data:
result = data["Result"]
return PaymentStatus(
reference=reference,
status=result.get("ResultCode", "PENDING"),
amount=float(result.get("ResultParameters", {}).get("Amount", 0)),
currency="KES",
customer_message=result.get("ResultDesc", "Transaction processed"),
timestamp=datetime.utcnow(),
raw_response=data
)
else:
return PaymentStatus(
reference=reference,
status="PENDING",
amount=0.0,
currency="KES",
customer_message="Transaction is being processed",
timestamp=datetime.utcnow(),
raw_response=data
)
except Exception as e:
return PaymentStatus(
reference=reference,
status="ERROR",
amount=0.0,
currency="KES",
customer_message=f"Error checking transaction: {str(e)}",
timestamp=datetime.utcnow(),
raw_response={}
)
def initiate_payment(self, amount: float, phone: str, reference: str) -> Dict:
"""Initiate STK Push payment."""
try:
token = self._get_access_token()
response = requests.post(
f"{self.base_url}/mpesa/stkpush/v1/processrequest",
headers={"Authorization": f"Bearer {token}"},
json={
"BusinessShortCode": self.shortcode,
"Password": self._generate_stk_password(),
"Timestamp": datetime.utcnow().strftime("%Y%m%d%H%M%S"),
"TransactionType": "CustomerPayBillOnline",
"Amount": str(int(amount)),
"PartyA": phone,
"PartyB": self.shortcode,
"PhoneNumber": phone,
"CallBackURL": "https://your-callback.com/stk",
"AccountReference": reference,
"TransactionDesc": f"Payment {reference}"
},
timeout=10.0
)
return response.json()
except requests.exceptions.Timeout:
raise ConnectionError(f"STK Push timeout for {phone} - increase timeout or use USSD fallback")
except Exception as e:
raise Exception(f"STK Push failed: {e}")
def _generate_stk_password(self) -> str:
"""Generate M-Pesa STK Push password."""
import base64
password_str = f"{self.shortcode}{'your_passkey'}{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
return base64.b64encode(password_str.encode()).decode()
class FlutterwaveGateway(PaymentGateway):
"""
Flutterwave Integration for Nigeria and Africa.
API Docs: https://developer.flutterwave.com/docs/
"""
def __init__(self, public_key: str, secret_key: str, encryption_key: str):
self.public_key = public_key
self.secret_key = secret_key
self.encryption_key = encryption_key
self.base_url = "https://api.flutterwave.com/v3"
def check_balance(self, account_id: str) -> Dict:
"""Check wallet balance."""
try:
response = requests.get(
f"{self.base_url}/balances",
headers={
"Authorization": f"Bearer {self.secret_key}",
"Content-Type": "application/json"
},
timeout=8.0
)
if response.status_code == 200:
return {"status": "success", "data": response.json()}
else:
return {"status": "error", "error": response.json()}
except requests.exceptions.Timeout:
return {"status": "timeout", "error": "Flutterwave API timeout"}
def check_transaction(self, reference: str) -> PaymentStatus:
"""Verify transaction by reference."""
try:
response = requests.get(
f"{self.base_url}/transactions/{reference}/verify",
headers={"Authorization": f"Bearer {self.secret_key}"},
timeout=8.0
)
data = response.json()
if data.get("status") == "success":
tx_data = data.get("data", {})
return PaymentStatus(
reference=reference,
status=tx_data.get("status", "UNKNOWN").upper(),
amount=float(tx_data.get("amount", 0)),
currency=tx_data.get("currency", "NGN"),
customer_message=tx_data.get("processor_response", ""),
timestamp=datetime.fromisoformat(tx_data.get("created_at", "")),
raw_response=data
)
else:
return PaymentStatus(
reference=reference,
status="PENDING",
amount=0.0,
currency="NGN",
customer_message="Unable to verify transaction",
timestamp=datetime.utcnow(),
raw_response=data
)
except Exception as e:
return PaymentStatus(
reference=reference,
status="ERROR",
amount=0.0,
currency="NGN",
customer_message=str(e),
timestamp=datetime.utcnow(),
raw_response={}
)
def initiate_payment(self, amount: float, phone: str, reference: str) -> Dict:
"""Initiate mobile money payment via Flutterwave."""
try:
response = requests.post(
f"{self.base_url}/payments",
headers={"Authorization": f"Bearer {self.secret_key}"},
json={
"tx_ref": reference,
"amount": amount,
"currency": "NGN",
"network_provider": self._detect_network(phone),
"phone_number": phone,
"email": "[email protected]",
"redirect_url": "https://your-site.com/payment-callback",
"payment_options": "ussd,card,mobile_money"
},
timeout=8.0
)
return response.json()
except Exception as e:
raise Exception(f"Flutterwave payment initiation failed: {e}")
def _detect_network(self, phone: str) -> str:
"""Detect Nigerian network provider from phone number."""
prefixes = {
"0803": "mtn", "0806": "mtn", "0813": "mtn", "0814": "mtn", "0810": "mtn",
"0805": "airtel", "0802": "airtel", "0812": "airtel", "0902": "airtel",
"0809": "9mobile", "0909": "9mobile", "0817": "9mobile",
"0804": "glo", "0805": "glo", "0811": "glo", "0905": "glo"
}
prefix = phone[:4]
return prefixes.get(prefix, "mtn")Step 3: Complete Integration Handler
#!/usr/bin/env python3
"""
African Payment AI Customer Service - Main Integration Handler
Orchestrates: HolySheep AI + Payment Gateways + Webhook Processing
"""
import os
import asyncio
import logging
from typing import Dict, Optional
from datetime import datetime
import json
from holy_sheep_client import HolySheepAIClient, CustomerMessage, AIResponse
from payment_gateways import MPesaGateway, FlutterwaveGateway, PaymentStatus
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AfricanPaymentAIHandler:
"""
Main orchestration class for African Mobile Payment AI Customer Service.
Features:
- Real-time message processing with <50ms HolySheep AI response
- Multi-language support (15+ African languages)
- Payment gateway failover (M-Pesa → Flutterwave → Wave)
- Automatic retry with exponential backoff
- Payment status tracking and dispute resolution
"""
def __init__(self, api_key: str):
self.ai_client = HolySheepAIClient(api_key)
self.gateways = {}
self._init_gateways()
def _init_gateways(self):
"""Initialize payment gateway adapters."""
# M-Pesa (Kenya)
self.gateways["mpesa"] = MPesaGateway(
consumer_key=os.environ.get("MPESA_CONSUMER_KEY"),
consumer_secret=os.environ.get("MPESA_CONSUMER_SECRET"),
shortcode=os.environ.get("MPESA_SHORTCODE"),
initiator_name=os.environ.get("MPESA_INITIATOR"),
security_credential=os.environ.get("MPESA_SECURITY_CREDENTIAL")
)
# Flutterwave (Nigeria, Ghana, Kenya, South Africa)
self.gateways["flutterwave"] = FlutterwaveGateway(
public_key=os.environ.get("FW_PUBLIC_KEY"),
secret_key=os.environ.get("FW_SECRET_KEY"),
encryption_key=os.environ.get("FW_ENCRYPTION_KEY")
)
async def process_message(self, message: CustomerMessage) -> AIResponse:
"""
Main message processing pipeline.
Flow:
1. Language detection (via HolySheep AI)
2. Intent classification (via HolySheep AI)
3. Payment context retrieval (if transaction-related)
4. Response generation (via HolySheep AI)
5. Return localized response with actions
"""
start_time = datetime.utcnow()
# Step 1: Language detection
detected_lang, confidence = self.ai_client.detect_language(message.text)
message.detected_language = detected_lang
message.confidence = confidence
logger.info(f"[{message.session_id}] Language detected: {detected_lang} ({confidence:.2%})")
# Step 2: Context retrieval for payment queries
payment_context = {}
if message.payment_reference:
payment_context = await self._get_payment_context(message.payment_reference)
# Step 3: Intent classification
intent_data = self.ai_client.classify_intent(
text=message.text,
language=detected_lang,
context=payment_context
)
logger.info(f"[{message.session_id}] Intent: {intent_data.get('intent')} ({intent_data.get('confidence'):.2%})")
# Step 4: Generate response based on intent
response = self._route_to_handler(
intent=intent_data.get("intent", "unknown"),
message=message,
payment_context=payment_context
)
# Step 5: Log processing time
processing_time = (datetime.utcnow() - start_time).total_seconds() * 1000
logger.info(f"[{message.session_id}] Processed in {processing_time:.2f}ms")
return response
def _route_to_handler(self, intent: str, message: CustomerMessage, payment_context: Dict) -> AIResponse:
"""Route intent to appropriate handler."""
handlers = {
"balance_inquiry": self._handle_balance_inquiry,
"transfer": self._handle_transfer,
"bill_payment": self._handle_bill_payment,
"dispute_resolution": self._handle_dispute,
"transaction_status": self._handle_status_check,
"account_help": self._handle_account_help,
}
handler = handlers.get(intent, self.ai_client.generate_response)
try:
return handler(message, payment_context)
except Exception as e:
logger.error(f"Handler error for {intent}: {e}")
return self.ai_client._fallback_response(message.detected_language or "en")
def _handle_balance_inquiry(self, message: CustomerMessage, context: Dict) -> AIResponse:
"""Handle balance inquiry with payment gateway lookup."""
try:
# Try M-Pesa first (Kenyan market)
if "mpesa" in self.gateways:
result = self.gateways["mpesa"].check_balance(message.user_id)
if result["status"] == "success":
return AIResponse(
response_text=self._format_balance_response(result, message.detected_language),
detected_language=message.detected_language,
intent="balance_inquiry",
confidence=0.95,
suggested_actions=["view_history", "make_payment"]
)
# Fallback to AI-generated response
return self.ai_client.generate_response(
intent="balance_inquiry",
language=message.detected_language,
user_message=message.text,
payment_context=context
)
except Exception as e:
logger.error(f"Balance inquiry failed: {e}")
return self.ai_client._fallback_response(message.detected_language)
def _handle_status_check(self, message: CustomerMessage, context: Dict) -> AIResponse:
"""Handle transaction status inquiry."""
reference = message.payment_reference or self._extract_reference(message.text)
if not reference:
return AIResponse(
response_text="I need your transaction reference to check the status. It usually starts with 'AE' or 'MG'.",
detected_language=message.detected_language,
intent="transaction_status",
confidence=0.9,
suggested_actions=["provide_reference"]
)
# Query payment gateway
for gateway_name, gateway in self.gateways.items():
try:
status = gateway.check_transaction(reference)
return self._format_transaction_status(status, message.detected_language)
except Exception as e:
logger.warning(f"{gateway_name} check failed: {e}")
continue
# All gateways failed - use AI with fallback
return self.ai_client.generate_response(
intent="transaction_status",
language=message.detected_language,
user_message=f"Check status for {reference}",
payment_context=context
)
def _handle_dispute(self, message: CustomerMessage, context: Dict) -> AIResponse:
"""Handle payment disputes and disputes."""
return self.ai_client.generate_response(
intent="dispute_resolution",
language=message.detected_language,
user_message=message.text,
payment_context={
"dispute_type": "payment_issue",
"user_id": message.user_id,
"reference": message.payment_reference,
"timestamp": message.timestamp.isoformat()
}
)
def _handle_transfer(self, message: CustomerMessage, context: Dict) -> AIResponse:
"""Handle money transfer requests."""
return self.ai_client.generate_response(
intent="transfer",
language=message.detected_language,
user_message=message.text,
payment_context={"feature": "send_money"}
)
def _handle_bill_payment(self, message: CustomerMessage, context: Dict) -> AIResponse:
"""Handle bill payment requests."""
return self.ai_client.generate_response(
intent="bill_payment",
language=message.detected_language,
user_message=message.text,
payment_context={"feature": "pay_bills"}
)
def _handle_account_help(self, message: CustomerMessage, context: Dict) -> AIResponse:
"""Handle general account help requests."""
return self.ai_client.generate_response(
intent="account_help",
language=message.detected_language,
user_message=message.text,
payment_context={}
)
async def _get_payment_context(self, reference: str) -> Dict:
"""Retrieve payment context from all gateways."""
context = {"reference": reference}
for gateway_name, gateway in self.gateways.items():
try:
status = gateway.check_transaction(reference)
if status.status in ["SUCCESS", "COMPLETED"]:
context["payment"] = {
"gateway": gateway_name,
"status": status.status,
"amount": status.amount,