Webhook Integration

# Webhook Configuration
WEBHOOK_BASE_URL=https://your-domain.com
WEBHOOK_SECRET=your_webhook_secret_key_here

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

class Settings(BaseSettings):
    webhook_base_url: Optional[str] = Field(
        default=None,
        description="Base URL for webhooks"
    )
    
    webhook_secret: SecretStr = Field(
        default_factory=lambda: SecretStr(secrets.token_urlsafe(32)),
        description="Secret for webhook signature validation"
    )

def verify_webhook_signature(self, payload: str, signature: str) -> bool:
    """Verify webhook signature using HMAC-SHA256"""
    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
    )

from fastapi import HTTPException

@router.post("/webhook")
async def secure_webhook(
    request: Request,
    at_client: AfricasTalkingClient = Depends(get_africastalking_client)
):
    # Get request body and signature
    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
    ):
        logger.warning("Invalid webhook signature")
        raise HTTPException(status_code=401, detail="Invalid signature")
    
    # Process webhook
    ...

{
  "sessionId": "ATVId_9c5f41e...",
  "phoneNumber": "+254712345678",
  "recordingUrl": "https://voice.africastalking.com/recordings/...",
  "duration": 120,
  "status": "completed",
  "dialDestinationNumber": "+254723456789",
  "isActive": "0",
  "dtmfDigits": "",
  "currencyCode": "KES",
  "amount": "3.50"
}

@router.post("/webhook")
async def voice_webhook(
    request: Request,
    at_client: AfricasTalkingClient = Depends(get_africastalking_client),
    db: AsyncSession = Depends(get_db)
):
    try:
        # Parse form data
        form_data = await request.form()
        
        session_id = form_data.get("sessionId")
        phone_number = form_data.get("phoneNumber")
        recording_url = form_data.get("recordingUrl")
        duration = form_data.get("duration")
        status = form_data.get("status", "completed")
        
        logger.info(f"Voice webhook: session={session_id}, status={status}")
        
        # Find recording in database
        if session_id:
            result = await db.execute(
                select(VoiceRecording).where(
                    VoiceRecording.recording_id == session_id
                )
            )
            recording = result.scalar_one_or_none()
            
            if recording:
                # Update recording details
                recording.recording_url = recording_url
                recording.duration = int(duration) if duration else None
                recording.processing_status = status
                await db.commit()
                
                logger.info(f"Updated recording {session_id}")
        
        return {"status": "webhook_processed", "session_id": session_id}
        
    except Exception as e:
        logger.error(f"Voice webhook error: {e}")
        return {"status": "webhook_error", "error": str(e)}

async def process_voice_recording(recording_url: str, recording_id: str):
    """Process voice recording after webhook"""
    voice_processor = await get_voice_processor()
    
    # Download and transcribe
    result = await voice_processor.process_voice_to_contract(
        audio_source=recording_url,
        is_url=True
    )
    
    # Extract contract terms
    transcript = result["transcript"]
    terms = result["terms"]
    
    # Create contract
    contract_generator = await get_contract_generator()
    contract_result = await contract_generator.process_voice_to_contract(
        transcript=transcript,
        terms=ContractTerms(**terms),
        parties=result.get("parties", []),
        contract_type="agricultural_supply"
    )
    
    logger.info(f"Contract {contract_result['contract_id']} created from recording")

{
  "from": "+254712345678",
  "to": "40404",
  "text": "YES-AG-2024-001",
  "date": "2024-03-06 10:30:00",
  "id": "ATXid_sample123",
  "linkId": "SampleLinkId123",
  "networkCode": "63902"
}

{
  "id": "ATXid_sample123",
  "status": "Success",
  "phoneNumber": "+254712345678",
  "networkCode": "63902",
  "retryCount": 0,
  "failureReason": ""
}

@router.post("/webhook")
async def sms_webhook(request: Request):
    try:
        form_data = await request.form()
        webhook_data = dict(form_data)
        
        logger.info(f"SMS webhook: {webhook_data}")
        
        # Check if it's an incoming message or delivery report
        if "from" in webhook_data:
            # Incoming SMS
            return await handle_incoming_sms(webhook_data)
        else:
            # Delivery report
            return await handle_delivery_report(webhook_data)
            
    except Exception as e:
        logger.error(f"SMS webhook error: {e}")
        return {"status": "webhook_error", "error": str(e)}

async def handle_incoming_sms(webhook_data: Dict[str, Any]):
    """Process incoming SMS message"""
    phone_number = webhook_data.get("from")
    message = webhook_data.get("text", "").upper().strip()
    
    # Handle contract confirmations
    if message.startswith("YES-"):
        contract_id = message.split("-", 1)[1]
        
        # Update contract status
        await confirm_contract(contract_id, phone_number)
        
        # Send confirmation SMS
        at_client = await get_africastalking_client()
        await at_client.send_sms(
            message=f"Contract {contract_id} confirmed. Thank you!",
            recipients=[phone_number]
        )
        
        return {
            "status": "confirmed",
            "contract_id": contract_id,
            "phone_number": phone_number
        }
    
    elif message.startswith("NO-"):
        contract_id = message.split("-", 1)[1]
        
        # Reject contract
        await reject_contract(contract_id, phone_number)
        
        return {
            "status": "rejected",
            "contract_id": contract_id,
            "phone_number": phone_number
        }
    
    # Handle delivery confirmations
    elif message.startswith("ACCEPT-"):
        contract_id = message.split("-", 1)[1]
        await accept_delivery(contract_id, phone_number)
        
        return {"status": "delivery_accepted", "contract_id": contract_id}
    
    elif message.startswith("DISPUTE-"):
        contract_id = message.split("-", 1)[1]
        await dispute_delivery(contract_id, phone_number)
        
        return {"status": "delivery_disputed", "contract_id": contract_id}
    
    # Unknown command
    return {
        "status": "webhook_received",
        "message": "Message received but not processed"
    }

{
  "transactionId": "ATPid_SampleTxnId123",
  "category": "MobileCheckout",
  "provider": "Mpesa",
  "providerRefId": "MPESA_REF_123",
  "providerChannel": "525900",
  "clientAccount": "VoicePact",
  "productName": "VoicePact",
  "sourceType": "PhoneNumber",
  "source": "+254712345678",
  "destinationType": "BankAccount",
  "destination": "Payment Wallet",
  "value": "KES 50000.00",
  "transactionFee": "KES 25.00",
  "providerFee": "KES 0.00",
  "status": "Success",
  "description": "The payment was successful",
  "requestMetadata": {
    "contract_id": "AG-2024-001",
    "payment_id": "123"
  },
  "providerMetadata": {
    "name": "John Farmer",
    "phoneNumber": "+254712345678"
  },
  "transactionDate": "2024-03-06 10:30:00"
}

{
  "transactionId": "ATPid_SampleTxnId123",
  "status": "Failed",
  "description": "Insufficient funds",
  "phoneNumber": "+254712345678",
  "value": "KES 50000.00",
  "productName": "VoicePact"
}

@router.post("/webhook")
async def payment_webhook(
    request: Request,
    at_client: AfricasTalkingClient = Depends(get_africastalking_client),
    db: AsyncSession = Depends(get_db)
):
    try:
        form_data = await request.form()
        
        transaction_id = form_data.get("transactionId")
        status = form_data.get("status", "failed")
        phone_number = form_data.get("source") or form_data.get("phoneNumber")
        amount = form_data.get("value")
        description = form_data.get("description")
        
        logger.info(
            f"Payment webhook: tx={transaction_id}, "
            f"status={status}, amount={amount}"
        )
        
        # Find payment record
        if 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()
                    
                    # Notify parties
                    await notify_payment_success(
                        payment.contract_id,
                        float(payment.amount),
                        payment.currency
                    )
                    
                else:
                    payment.status = PaymentStatus.FAILED
                    payment.failure_reason = description
                    
                    # Notify failure
                    await notify_payment_failure(
                        payment.contract_id,
                        phone_number,
                        description
                    )
                
                await db.commit()
                logger.info(f"Payment {payment.id} updated: {payment.status.value}")
        
        return {
            "status": "webhook_processed",
            "transaction_id": transaction_id
        }
        
    except Exception as e:
        logger.error(f"Payment webhook error: {e}")
        return {"status": "webhook_error", "error": str(e)}

{
  "sessionId": "ATUid_session123",
  "serviceCode": "*483#",
  "phoneNumber": "+254712345678",
  "text": "1*2*3",
  "networkCode": "63902"
}

@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)
):
    """Main USSD handler"""
    try:
        # Get or create session
        session = await get_or_create_session(sessionId, phoneNumber, db)
        
        # Parse user input
        user_input = text.split('*')[-1] if text else ""
        
        # First request - show main menu
        if not text:
            response = main_menu()
            session.current_menu = "main"
        else:
            # Handle menu navigation
            response = await handle_menu_navigation(
                session, user_input, phoneNumber, at_client, db
            )
        
        # Update session
        session.last_input = user_input
        session.last_response = response
        session.updated_at = datetime.utcnow()
        await db.commit()
        
        return response
        
    except Exception as e:
        logger.error(f"USSD error: {e}")
        return at_client.build_ussd_response(
            "Service unavailable. Please try again.",
            end_session=True
        )

# Continue session
response = "CON Welcome to VoicePact\n1. View Contracts\n2. Exit"

# End session
response = "END Thank you for using VoicePact!"

# From app/services/africastalking_client.py:494-518
async def process_webhook_data(
    self,
    webhook_data: Dict[str, Any]
) -> Dict[str, Any]:
    """Process and normalize webhook data"""
    processed_data = {
        'timestamp': datetime.utcnow().isoformat(),
        'original_data': webhook_data
    }
    
    # Extract common fields
    if 'status' in webhook_data:
        processed_data['status'] = webhook_data['status']
    
    if 'transactionId' in webhook_data:
        processed_data['transaction_id'] = webhook_data['transactionId']
    
    if 'phoneNumber' in webhook_data:
        processed_data['phone_number'] = await self.format_phone_number(
            webhook_data['phoneNumber']
        )
    
    if 'amount' in webhook_data:
        try:
            amount_str = str(webhook_data['amount']).replace(',', '')
            processed_data['amount'] = float(amount_str)
        except (ValueError, TypeError):
            processed_data['amount'] = 0.0
    
    return processed_data

@router.post("/webhook")
async def resilient_webhook(request: Request):
    try:
        # Process webhook
        result = await process_webhook(request)
        return result
        
    except Exception as e:
        # Log error but return 200 to prevent retries
        logger.error(f"Webhook processing error: {e}", exc_info=True)
        
        # Store failed webhook for manual review
        await store_failed_webhook(
            endpoint=request.url.path,
            payload=await request.body(),
            error=str(e)
        )
        
        # Return 200 to prevent AT retries
        return {"status": "error_logged", "error": str(e)}

npm install -g ngrok

uvicorn app.main:app --reload --port 8000

ngrok http 8000

https://abc123.ngrok.io/api/v1/voice/webhook

# Test voice webhook
curl -X POST http://localhost:8000/api/v1/voice/webhook \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "sessionId=ATVId_test123" \
  -d "phoneNumber=+254712345678" \
  -d "recordingUrl=https://example.com/recording.mp3" \
  -d "duration=120" \
  -d "status=completed"

# Test SMS webhook
curl -X POST http://localhost:8000/api/v1/sms/webhook \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "from=+254712345678" \
  -d "to=40404" \
  -d "text=YES-AG-2024-001" \
  -d "date=2024-03-06 10:30:00" \
  -d "id=ATXid_test123"

# Test payment webhook
curl -X POST http://localhost:8000/api/v1/payments/webhook \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "transactionId=ATPid_test123" \
  -d "status=Success" \
  -d "phoneNumber=+254712345678" \
  -d "amount=KES 50000.00"

import logging

logger = logging.getLogger(__name__)

@router.post("/webhook")
async def monitored_webhook(request: Request):
    # Log incoming webhook
    logger.info(
        f"Webhook received: {request.url.path}",
        extra={
            "headers": dict(request.headers),
            "client_host": request.client.host
        }
    )
    
    try:
        result = await process_webhook(request)
        
        # Log success
        logger.info(f"Webhook processed successfully: {result}")
        
        return result
        
    except Exception as e:
        # Log error with full traceback
        logger.error(
            f"Webhook processing failed: {e}",
            exc_info=True,
            extra={"request_id": request.headers.get("X-Request-ID")}
        )
        
        return {"status": "error", "error": str(e)}

from prometheus_client import Counter, Histogram

webhook_requests = Counter(
    'webhook_requests_total',
    'Total webhook requests',
    ['endpoint', 'status']
)

webhook_duration = Histogram(
    'webhook_processing_seconds',
    'Webhook processing time',
    ['endpoint']
)

@router.post("/webhook")
async def instrumented_webhook(request: Request):
    endpoint = request.url.path
    
    with webhook_duration.labels(endpoint=endpoint).time():
        try:
            result = await process_webhook(request)
            webhook_requests.labels(endpoint=endpoint, status='success').inc()
            return result
        except Exception as e:
            webhook_requests.labels(endpoint=endpoint, status='error').inc()
            raise

curl -X POST https://your-domain.com/api/v1/voice/webhook

import hmac
import hashlib

payload = "your_webhook_payload"
secret = "your_webhook_secret"

signature = hmac.new(
    secret.encode(),
    payload.encode(),
    hashlib.sha256
).hexdigest()

print(f"sha256={signature}")