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 provides conditional escrow for contract payments, holding funds until delivery is confirmed. This builds trust between parties by ensuring:
  • Sellers are guaranteed payment upon delivery
  • Buyers don’t pay until goods/services are received
  • Funds are locked and can’t be withdrawn by either party during fulfillment

​
Payment Flow

The escrow lifecycle follows contract progression: 
┌──────────────┐    ┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│   PENDING    │───▶│    LOCKED    │───▶│   RELEASED   │ or │   REFUNDED   │
│ (Initiated)  │    │  (In Escrow) │    │ (Delivered)  │    │ (Cancelled)  │
└──────────────┘    └──────────────┘    └──────────────┘    └──────────────┘
       │                   │                   │                    │
       │                   │                   │                    │
    Checkout          Contract Active      Delivery OK          Dispute

​
Payment Statuses

Defined in contract.py:43-48: 
class PaymentStatus(str, enum.Enum):
    PENDING = "pending"        # Checkout initiated, awaiting confirmation
    LOCKED = "locked"          # Funds held in escrow
    RELEASED = "released"      # Payment sent to seller
    REFUNDED = "refunded"      # Returned to buyer (cancellation/dispute)
    FAILED = "failed"          # Transaction error

​
Payment Model

Payments are tracked in the database (contract.py:219-268): 
class Payment(Base):
    __tablename__ = "payments"
    
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    
    contract_id: Mapped[str] = mapped_column(
        String(50),
        ForeignKey("contracts.id", ondelete="CASCADE")
    )
    
    # Transaction identifiers
    transaction_id: Mapped[Optional[str]] = mapped_column(String(100), unique=True)
    external_transaction_id: Mapped[Optional[str]] = mapped_column(String(100))
    
    # Parties
    payer_phone: Mapped[str] = mapped_column(String(20), index=True)
    recipient_phone: Mapped[Optional[str]] = mapped_column(String(20))
    
    # Amount details
    amount: Mapped[Decimal] = mapped_column(Numeric(precision=15, scale=2))
    currency: Mapped[str] = mapped_column(String(3), default="KES")
    
    payment_type: Mapped[str] = mapped_column(String(20), default="escrow")
    
    status: Mapped[PaymentStatus] = mapped_column(
        SQLEnum(PaymentStatus),
        default=PaymentStatus.PENDING
    )
    
    # Timestamps
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    confirmed_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
    released_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
    
    # Error handling
    failure_reason: Mapped[Optional[str]] = mapped_column(String(200))
    retry_count: Mapped[int] = mapped_column(Integer, default=0)
Payments are linked to contracts via foreign key with CASCADE delete. If a contract is deleted, all associated payments are automatically removed.

​
Mobile Money Checkout

​
Initiating Payment

Endpoint: POST /payments/checkout (payments.py:36-93) 
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")

@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()
    await db.refresh(payment)
    
    # 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 payment 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()
    )
Africa’s Talking Integration (africastalking_client.py:240-268): 
async def mobile_checkout(
    self,
    phone_number: str,
    amount: Union[int, float],
    currency_code: str = "KES",
    metadata: Optional[Dict] = None
) -> Dict[str, Any]:
    checkout_data = {
        'productName': settings.at_payment_product_name,
        'phoneNumber': phone_number,
        'currencyCode': currency_code,
        'amount': amount,
        'metadata': metadata or {}
    }
    
    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(
        None,
        lambda: self.payment_service.mobile_checkout(checkout_data)
    )
    return response
The mobile checkout triggers an STK push (SIM Toolkit) on the buyer’s phone, prompting them to enter their M-Pesa PIN to authorize the payment.

​
Payment Webhooks

Africa’s Talking sends payment status updates to the webhook endpoint. Webhook handler: POST /payments/webhook (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 received: {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  # Funds in escrow
                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 signature verification (planned): 
# Verify webhook authenticity (crypto_service.py:162-182)
def verify_webhook_signature(self, payload: str, signature: str) -> bool:
    try:
        expected_signature = self.generate_webhook_signature(payload)
        return hmac.compare_digest(signature, expected_signature)
    except Exception as e:
        logger.error(f"Webhook signature verification failed: {e}")
        return False

​
Escrow Locking

When payment status changes to LOCKED, the contract becomes ACTIVE: 
# After webhook updates payment
if payment.status == PaymentStatus.LOCKED:
    # Update associated contract
    contract = await db.execute(
        select(Contract).where(Contract.id == payment.contract_id)
    )
    contract.status = ContractStatus.ACTIVE
    contract.confirmed_at = datetime.utcnow()
    await db.commit()
Funds are now held by Africa’s Talking and cannot be accessed by either party.

​
Payment Release

When delivery is confirmed (buyer accepts goods): 
# Seller confirms delivery via SMS/USSD
# Buyer approves delivery

# Update payment status
payment.status = PaymentStatus.RELEASED
payment.released_at = datetime.utcnow()
payment.recipient_phone = seller_phone

# Trigger disbursement to seller
await at_client.mobile_data_transfer(
    phone_number=seller_phone,
    amount=payment.amount,
    currency_code=payment.currency,
    metadata={"contract_id": payment.contract_id, "payment_id": payment.id}
)

# Update contract
contract.status = ContractStatus.COMPLETED
contract.completed_at = datetime.utcnow()
Mobile money transfer (africastalking_client.py:274-304): 
async def mobile_data_transfer(
    self,
    phone_number: str,
    amount: Union[int, float],
    currency_code: str = "KES",
    metadata: Optional[Dict] = None
) -> Dict[str, Any]:
    transfer_data = {
        'productName': settings.at_payment_product_name,
        'recipients': [{
            'phoneNumber': phone_number,
            'currencyCode': currency_code,
            'amount': amount,
            'metadata': metadata or {}
        }]
    }
    
    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(
        None,
        lambda: self.payment_service.mobile_data(transfer_data)
    )
    return response

​
Refunds

If the contract is cancelled or disputed: 
payment.status = PaymentStatus.REFUNDED

# Return funds to buyer
await at_client.mobile_data_transfer(
    phone_number=buyer_phone,
    amount=payment.amount,
    currency_code=payment.currency,
    metadata={"contract_id": payment.contract_id, "refund": True}
)

​
Payment Queries

​
Get Payment by ID

Endpoint: GET /payments/{payment_id} (payments.py:139-167) 
@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 /payments/contract/{contract_id} (payments.py:170-200) 
@router.get("/contract/{contract_id}")
async def get_contract_payments(contract_id: str, db: AsyncSession):
    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}

​
Payment Reference Generation

Unique payment references are generated cryptographically (crypto_service.py:121-128): 
def generate_payment_reference(
    self, 
    contract_id: str, 
    amount: float, 
    phone_number: str
) -> str:
    try:
        content = f"{contract_id}:{amount}:{phone_number}"
        hash_obj = hashlib.blake2b(content.encode('utf-8'), digest_size=8)
        return hash_obj.hexdigest().upper()
    except Exception as e:
        logger.error(f"Payment reference generation failed: {e}")
        raise CryptographicError(f"Failed to generate payment reference: {e}")

​
Error Handling

Retry logic for failed payments: 
if payment.status == PaymentStatus.FAILED:
    if payment.retry_count < MAX_RETRIES:
        payment.retry_count += 1
        payment.status = PaymentStatus.PENDING
        # Retry checkout
        await mobile_checkout(payment)
    else:
        # Send notification to buyer
        logger.error(f"Payment {payment.id} failed after {MAX_RETRIES} retries")

​
Security Considerations

Payment security best practices:
  1. Webhook verification - Validate HMAC signatures from Africa’s Talking
  2. Idempotency - Use unique transaction IDs to prevent duplicate payments
  3. Amount validation - Verify webhook amounts match contract totals
  4. Audit logging - Record all payment state changes
  5. Database constraints - Enforce positive amounts at schema level
  6. Encryption - Sensitive payment metadata encrypted at rest (planned)

​
Testing Payments

Test endpoint: POST /payments/test/checkout (payments.py:203-227) 
@router.post("/test/checkout")
async def test_payment(
    phone_number: str,
    amount: float = 100.0,
    at_client: AfricasTalkingClient = Depends(get_africastalking_client)
):
    response = await at_client.mobile_checkout(
        phone_number=phone_number,
        amount=amount,
        currency_code="KES",
        metadata={"test": "true"}
    )
    
    return {
        "status": "test_initiated",
        "phone_number": phone_number,
        "amount": amount,
        "response": response
    }
Use Africa’s Talking sandbox for testing without real money.

Build docs developers (and LLMs) love

Get started for freeTalk to us