Callbacks

parse_stk_callback

parse_stk_callback(body: dict[str, Any]) -> STKPushStatus

Parse an STK Push callback payload from M-Pesa.

Call this inside your webhook handler when M-Pesa POSTs a payment notification to your callback URL.

Parameters:

Name	Type	Description	Default
body	dict[str, Any]	The full JSON payload received from M-Pesa. Must contain the Body.stkCallback structure.	required

Returns:

Type	Description
STKPushStatus	An STKPushStatus with the parsed transaction result.
STKPushStatus	Check status.success to determine if payment went through.

Raises:

Type	Description
APIError	If the callback payload is malformed or missing required fields.

Example

.. code-block:: python

@app.post("/mpesa/callback")
async def handle_callback(request):
    body = await request.json()
    result = parse_stk_callback(body)

    if result.success:
        print(f"✅ Paid KES {result.amount} — {result.receipt}")
    else:
        print(f"❌ Failed: {result.result_description}")

    return {"ResultCode": 0, "ResultDesc": "Success"}

Source code in safcom/callbacks.py

def parse_stk_callback(body: dict[str, Any]) -> STKPushStatus:
    """Parse an STK Push callback payload from M-Pesa.

    Call this inside your webhook handler when M-Pesa POSTs a payment
    notification to your callback URL.

    Args:
        body: The full JSON payload received from M-Pesa. Must contain
            the ``Body.stkCallback`` structure.

    Returns:
        An ``STKPushStatus`` with the parsed transaction result.
        Check ``status.success`` to determine if payment went through.

    Raises:
        APIError: If the callback payload is malformed or missing
            required fields.

    Example:
        .. code-block:: python

            @app.post(\"/mpesa/callback\")
            async def handle_callback(request):
                body = await request.json()
                result = parse_stk_callback(body)

                if result.success:
                    print(f\"✅ Paid KES {result.amount} — {result.receipt}\")
                else:
                    print(f\"❌ Failed: {result.result_description}\")

                return {\"ResultCode\": 0, \"ResultDesc\": \"Success\"}
    """
    try:
        stk = body["Body"]["stkCallback"]
    except (KeyError, TypeError):
        raise APIError(
            "Invalid callback payload: missing Body.stkCallback structure. "
            "Make sure you're passing the full M-Pesa POST body."
        )

    checkout_request_id = stk.get("CheckoutRequestID", "")
    merchant_request_id = stk.get("MerchantRequestID", "")
    result_code = str(stk.get("ResultCode", ""))
    result_description = stk.get("ResultDesc", "")

    # Parse metadata items if present
    amount = None
    receipt = None
    phone = None
    transaction_date = None

    metadata = stk.get("CallbackMetadata")
    if metadata and isinstance(metadata, dict):
        items = metadata.get("Item", [])
        if isinstance(items, list):
            for item in items:
                name = item.get("Name", "")
                value = item.get("Value")
                if name == "Amount":
                    amount = float(value) if value is not None else None
                elif name == "MpesaReceiptNumber":
                    receipt = str(value) if value else None
                elif name == "PhoneNumber":
                    phone = str(value) if value else None
                elif name == "TransactionDate":
                    if value:
                        try:
                            transaction_date = datetime.strptime(
                                str(value), "%Y%m%d%H%M%S"
                            )
                        except (ValueError, TypeError):
                            pass

    return STKPushStatus(
        response_code="0" if result_code == "0" else result_code,
        response_description=(
            "Success" if result_code == "0" else (result_description or "Failed")
        ),
        merchant_request_id=merchant_request_id,
        checkout_request_id=checkout_request_id,
        result_code=result_code,
        result_description=result_description,
        amount=amount,
        receipt=receipt,
        transaction_date=transaction_date,
        phone=phone,
        raw=body,
    )

extract_payment_info

extract_payment_info(result: STKPushStatus) -> dict[str, Any]

Extract a clean payment summary from a callback result.

Useful for logging, database storage, or API responses.

Parameters:

Name	Type	Description	Default
result	STKPushStatus	An STKPushStatus returned by parse_stk_callback().	required

Returns:

Type	Description
dict[str, Any]	A dictionary with cleaned-up payment info.

Example

info = extract_payment_info(status) info["paid"] True info["receipt"] 'NLJ91HA6ES'

Source code in safcom/callbacks.py

def extract_payment_info(result: STKPushStatus) -> dict[str, Any]:
    """Extract a clean payment summary from a callback result.

    Useful for logging, database storage, or API responses.

    Args:
        result: An ``STKPushStatus`` returned by ``parse_stk_callback()``.

    Returns:
        A dictionary with cleaned-up payment info.

    Example:
        >>> info = extract_payment_info(status)
        >>> info[\"paid\"]
        True
        >>> info[\"receipt\"]
        'NLJ91HA6ES'
    """
    return {
        "paid": result.success,
        "receipt": result.receipt,
        "amount": result.amount,
        "phone": result.phone,
        "date": result.transaction_date.isoformat() if result.transaction_date else None,
        "checkout_request_id": result.checkout_request_id,
        "result_code": result.result_code,
        "result_description": result.result_description,
    }

---

Response Models

STKPushResponse

STKPushResponse dataclass

Response from an STK push request.

Source code in safcom/models.py

@dataclass
class STKPushResponse:
    """Response from an STK push request."""
    merchant_request_id: str
    checkout_request_id: str
    response_code: str
    response_description: str
    customer_message: str
    raw: dict = field(repr=False)

Attributes

merchant_request_id instance-attribute

merchant_request_id: str

checkout_request_id instance-attribute

checkout_request_id: str

response_code instance-attribute

response_code: str

response_description instance-attribute

response_description: str

customer_message instance-attribute

customer_message: str

raw class-attribute instance-attribute

raw: dict = field(repr=False)

Methods:

__init__

__init__(
    merchant_request_id: str,
    checkout_request_id: str,
    response_code: str,
    response_description: str,
    customer_message: str,
    raw: dict,
) -> None

STKPushStatus

STKPushStatus dataclass

Status of a completed STK push transaction.

Source code in safcom/models.py

@dataclass
class STKPushStatus:
    """Status of a completed STK push transaction."""
    response_code: str
    response_description: str
    merchant_request_id: str
    checkout_request_id: str
    result_code: str | None = None
    result_description: str | None = None
    amount: float | None = None
    receipt: str | None = None
    transaction_date: datetime | None = None
    phone: str | None = None
    raw: dict = field(default_factory=dict, repr=False)

    @property
    def success(self) -> bool:
        return self.result_code == "0"

Attributes

response_code instance-attribute

response_code: str

response_description instance-attribute

response_description: str

merchant_request_id instance-attribute

merchant_request_id: str

checkout_request_id instance-attribute

checkout_request_id: str

result_code class-attribute instance-attribute

result_code: str | None = None

result_description class-attribute instance-attribute

result_description: str | None = None

amount class-attribute instance-attribute

amount: float | None = None

receipt class-attribute instance-attribute

receipt: str | None = None

transaction_date class-attribute instance-attribute

transaction_date: datetime | None = None

phone class-attribute instance-attribute

phone: str | None = None

raw class-attribute instance-attribute

raw: dict = field(default_factory=dict, repr=False)

success property

success: bool

Methods:

__init__

__init__(
    response_code: str,
    response_description: str,
    merchant_request_id: str,
    checkout_request_id: str,
    result_code: str | None = None,
    result_description: str | None = None,
    amount: float | None = None,
    receipt: str | None = None,
    transaction_date: datetime | None = None,
    phone: str | None = None,
    raw: dict = dict(),
) -> None

B2CResponse

B2CResponse dataclass

Response from a B2C (send money) request.

Source code in safcom/models.py

@dataclass
class B2CResponse:
    """Response from a B2C (send money) request."""
    conversation_id: str
    originator_conversation_id: str
    response_code: str
    response_description: str
    raw: dict = field(repr=False)

Attributes

conversation_id instance-attribute

conversation_id: str

originator_conversation_id instance-attribute

originator_conversation_id: str

response_code instance-attribute

response_code: str

response_description instance-attribute

response_description: str

raw class-attribute instance-attribute

raw: dict = field(repr=False)

Methods:

__init__

__init__(
    conversation_id: str,
    originator_conversation_id: str,
    response_code: str,
    response_description: str,
    raw: dict,
) -> None

AccountBalanceResponse

AccountBalanceResponse dataclass

Response from an account balance request.

Source code in safcom/models.py

@dataclass
class AccountBalanceResponse:
    """Response from an account balance request."""
    conversation_id: str
    originator_conversation_id: str
    response_code: str
    response_description: str
    raw: dict = field(repr=False)

Attributes

conversation_id instance-attribute

conversation_id: str

originator_conversation_id instance-attribute

originator_conversation_id: str

response_code instance-attribute

response_code: str

response_description instance-attribute

response_description: str

raw class-attribute instance-attribute

raw: dict = field(repr=False)

Methods:

__init__

__init__(
    conversation_id: str,
    originator_conversation_id: str,
    response_code: str,
    response_description: str,
    raw: dict,
) -> None