FastAPIPython 3.10+ · FastAPI 0.110+ · moncashconnect · uvicorn

Intégrer MonCash avec FastAPI

Une API FastAPI complète pour MonCash : endpoint async de création, page de retour, webhook signé HMAC-SHA256 lu sur le body brut et traitement en BackgroundTasks.

Sandbox

Commencez en Sandbox

Avant d'implémenter le code ci-dessous avec votre clé live (sk_proj_…), testez d'abord en mode Sandbox avec sk_test_proj_…. Le code est identique — vous ne changez que la valeur de la variable d'environnement. Lisez le Guide Sandbox avant de continuer.

Prérequis

Python 3.10+ (3.11 recommandé)
FastAPI 0.110+ et uvicorn
Un projet MonCashConnect avec sk_proj_… et whsec_… — créer un projet

Installation

bash

pip install fastapi uvicorn moncashconnect pydantic-settings python-dotenv

Le SDK moncashconnect est pure-Python sans dépendance (utilise la stdlib pour HTTP et HMAC).

Variables d'environnement

Fichier .env (non versionné) :

bash

MCC_SECRET_KEY=sk_proj_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
MCC_WEBHOOK_SECRET=whsec_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
APP_URL=https://api.votresite.com

Settings Pydantic — app/config.py :

python

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    mcc_secret_key:     str
    mcc_webhook_secret: str
    app_url:            str = "http://localhost:8000"

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")


settings = Settings()  # noqa: pyright

Client MonCash partagé

Instanciez le client une seule fois au démarrage, dans app/mcc.py :

python

from moncashconnect import MonCashClient
from app.config import settings

# Instancié une fois — pas de state mutable, safe à partager
client = MonCashClient(settings.mcc_secret_key)

Créer un paiement (POST /pay)

Dans app/main.py :

python

import asyncio
from fastapi import FastAPI, HTTPException, Request, BackgroundTasks
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from moncashconnect import MonCashError, construct_event

from app.config import settings
from app.mcc    import client

app = FastAPI(title="Boutique MonCash")


class CreatePaymentIn(BaseModel):
    order_id:       str            = Field(..., min_length=1, max_length=64)
    amount:         float          = Field(..., gt=0)
    customer_name:  str | None     = None
    customer_email: str | None     = None


class CreatePaymentOut(BaseModel):
    payment_url: str
    reference:   str


@app.post("/pay", response_model=CreatePaymentOut)
async def create_payment(body: CreatePaymentIn) -> CreatePaymentOut:
    """Crée un paiement MonCash et renvoie l'URL de redirection."""
    try:
        # Le SDK est sync — on l'exécute dans un thread pour ne pas
        # bloquer la boucle d'événements asyncio.
        payment = await asyncio.to_thread(
            client.create_payment,
            amount=body.amount,
            reference_id=body.order_id,
            return_url=f"{settings.app_url}/pay/return",
            customer_name=body.customer_name,
            customer_email=body.customer_email,
        )
    except MonCashError as exc:
        # 409 si referenceId déjà utilisé, 400 si montant invalide, etc.
        raise HTTPException(status_code=exc.status_code, detail=str(exc))

    return CreatePaymentOut(
        payment_url=payment["paymentUrl"],
        reference=payment["reference"],
    )

Page de retour (GET /pay/return)

python

from fastapi import Query


@app.get("/pay/return")
async def payment_return(ref: str = Query(..., description="reference MonCash")):
    """Vérifie le statut côté serveur — ne pas se fier aux params URL seuls."""
    try:
        tx = await asyncio.to_thread(client.get_payment_status, ref)
    except MonCashError as exc:
        raise HTTPException(status_code=exc.status_code, detail=str(exc))

    return {
        "status":    tx["status"],          # 'pending' | 'completed' | 'failed'
        "amount":    tx.get("amount"),
        "reference": tx.get("reference"),
    }

La page de retour donne une indication d'UX, mais le statut peut encore être pending. Ne marquez la commande comme payée qu'à réception du webhook payment.completed.

Webhook avec raw body + BackgroundTasks

C'est l'endpoint critique : on lit les bytes du body avant tout parsing, on vérifie la signature, puis on traite l'événement en tâche de fond pour répondre 200 immédiatement.

python

import json
from fastapi import Header
from fastapi.responses import PlainTextResponse


def _process_event(event: dict) -> None:
    """Logique métier — appelée en BackgroundTasks."""
    if event["event"] == "payment.completed":
        # mark_order_paid(reference=event["reference"], amount=event["amount"])
        pass
    elif event["event"] == "payment.failed":
        # mark_order_failed(reference=event["reference"])
        pass
    # Tout autre type d'événement : on log et on ignore


@app.post("/webhooks/moncash", response_class=PlainTextResponse)
async def moncash_webhook(
    request: Request,
    background_tasks: BackgroundTasks,
    x_mcc_signature: str = Header(default=""),
    x_mcc_timestamp: str = Header(default=""),
):
    # 1) Lire les bytes BRUTS — pas request.json() ni request.form()
    raw_body = await request.body()

    # 2) Vérifier la signature HMAC-SHA256
    try:
        event = construct_event(
            raw_body,
            x_mcc_signature,
            x_mcc_timestamp,
            settings.mcc_webhook_secret,
        )
    except MonCashError as exc:
        # 401 si signature invalide, 400 si timestamp trop vieux
        return PlainTextResponse(str(exc), status_code=exc.status_code)

    # 3) Renvoyer 200 le plus vite possible — traitement en arrière-plan
    background_tasks.add_task(_process_event, event)

    return "OK"

Si vous préférez vérifier la signature manuellement (sans le SDK), appliquez hmac.new(secret, f"{timestamp}.{raw_body.decode()}".encode(), "sha256").hexdigest() et comparez avec hmac.compare_digest.

N'utilisez jamais await request.json() avant la vérification HMAC. FastAPI buffer/reparse le body et la signature ne correspondra plus.

Liste de contrôle avant production

MCC_SECRET_KEY et MCC_WEBHOOK_SECRET sont chargés depuis variables d'environnement (pas hardcodés)

Le SDK est appelé via asyncio.to_thread() pour ne pas bloquer la boucle async

request.body() est lu AVANT tout parsing JSON dans le webhook

L'endpoint webhook est idempotent (vérifie l'état avant de créditer)

BackgroundTasks renvoie 200 immédiatement, le traitement métier suit

uvicorn est lancé derrière un reverse proxy (Caddy / nginx) qui termine le TLS

L'URL https://api.votresite.com/webhooks/moncash est enregistrée dans MonCashConnect

Les clés Sandbox sk_test_proj_… sont remplacées par sk_proj_… en production

Lancement de production :

bash

uvicorn app.main:app \
  --host 0.0.0.0 \
  --port 8000 \
  --workers 2 \
  --proxy-headers \
  --forwarded-allow-ips '*'

FAQ

Pourquoi lire request.body() et pas request.json() dans le webhook ?+

La signature HMAC est calculée sur les bytes exacts envoyés par MonCashConnect. Si vous appelez request.json(), FastAPI parse puis re-sérialise — les espaces et l'ordre des clés peuvent changer, et la signature ne correspondra plus.

Le SDK moncashconnect est-il async ?+

Non, c'est un client synchrone qui utilise urllib de la stdlib (zéro dépendance). Pour ne pas bloquer la boucle d'événements FastAPI, encapsulez les appels dans asyncio.to_thread() ou un executor — exemple dans la section 'Créer un paiement'.

Pourquoi traiter le webhook en BackgroundTasks plutôt qu'en ligne ?+

Pour répondre 200 le plus vite possible à MonCashConnect. Si votre logique métier (envoi d'email, mise à jour BDD) prend 3 secondes, le webhook pourrait timeout et être retenté. BackgroundTasks permet de répondre immédiatement puis de faire le travail derrière.

Faut-il un compte marchand Digicel séparé ?+

Non. MonCashConnect agit comme passerelle sur l'infrastructure Bazik — un seul compte MCC suffit pour encaisser, aucun contrat direct avec Digicel n'est requis.

Comment tester localement avec uvicorn et ngrok ?+

Lancez uvicorn main:app --reload puis ngrok http 8000 dans un autre terminal. Collez l'URL HTTPS ngrok + /webhooks/moncash dans le dashboard MCC. Les rejeux d'événements depuis le dashboard arriveront sur votre machine locale.

Pydantic v1 ou v2 ?+

FastAPI 0.100+ supporte les deux mais Pydantic v2 est recommandé (plus rapide et stable). pydantic-settings est le package séparé pour BaseSettings depuis Pydantic v2 — d'où sa présence dans le pip install.

Lectures recommandées :

Flask Django Express.js Tester un webhook Référence API complète