Accepting mobile money in FastAPI

Install

FastAPI is served by the Python client. There is no FastAPI-specific package, and this page does not pretend otherwise: what is below is that client, used the way FastAPI is written.

pip install ultraner

The snippets below are checked in CI against the API's own schemas. This client is not yet driven end to end against the sandbox the way the Node one is.

The FastAPI trap

Take `Request` and `await request.body()` for the webhook. A typed body model parses and re-serialises, so the signature can never match.

Taking a payment

Two things this shows that are easy to get wrong anywhere. The amount is in whole units for most African currencies, so 5000 TZS is five thousand shillings and not fifty. And provideris the network's provider code, not its brand name: Mixx by Yas is still Tigo on the wire.

# main.py
import os, uuid
from fastapi import FastAPI
from ultraner import Ultraner

app = FastAPI()
ultraner = Ultraner(os.environ["ULTRANER_API_KEY"])

@app.post("/pay")
async def pay(phone: str, amount: int):
    payment = ultraner.payments.create_mno_charge(
        amount=amount,             # whole shillings: TZS has no minor unit
        currency="TZS",
        provider="Vodacom",        # provider code, not brand name
        account_number=phone,
        idempotency_key=str(uuid.uuid4()),
    )
    return {"reference": payment.reference, "status": payment.status}

The response means the request was accepted, not that money moved. A mobile-money charge sends a prompt to the payer's handset and they may take a minute, or never answer. Which is why the next section is not optional.

Receiving the result

The webhook is how you learn a payment succeeded. Verify the signature over the raw bytes before trusting anything in the payload, acknowledge quickly, and do the work afterwards: a slow handler gets retried, and retries mean you will see the same event twice.

# main.py
import hmac, hashlib, json, os
from fastapi import Request, Response

@app.post("/webhooks/ultraner")
async def ultraner_webhook(request: Request):
    # Request, not a Pydantic model: a model would parse and re-serialise,
    # and the signature is over the bytes as sent.
    raw = await request.body()
    signature = request.headers.get("X-Ultraner-Signature", "")

    expected = hmac.new(
        os.environ["ULTRANER_WEBHOOK_SECRET"].encode(),
        raw,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return Response(status_code=400)

    event = json.loads(raw)
    background.add_task(handle, event)   # acknowledge first, work after
    return Response(status_code=200)