Skip to main content

Business to Customer (B2C)

Send payouts (salary, business or promotional payments) from your organization's disbursement account to customers, and handle result/timeout callbacks.

User Stories

  • As a fintech product owner, I want to programmatically initiate B2C payouts so that customers receive funds immediately after approval.
  • As an integrations developer, I want a simple client and clear webhook callbacks so I can implement reliable end-to-end flows with minimal boilerplate.
  • As a billing operations engineer, I want result and timeout notifications with acknowledgements so I can reconcile transactions and trigger retries or alerts when needed.
  • As a reseller partner, I want a tested SDK and examples so I can onboard quickly and reduce integration defects.

Parameters Definition

ParameterTypeDescription
OriginatorConversationIDrequired
str
StringUnique identifier for this request (used for tracing/reconciliation).
InitiatorNamerequired
str
StringAPI initiator username configured with Safaricom.
SecurityCredentialrequired
str
StringEncrypted security credential for the initiator.
CommandIDrequired
str
EnumType of payment. Allowed: SalaryPayment, BusinessPayment, PromotionPayment.
Amountrequired
int
IntegerAmount to disburse to the recipient.
PartyArequired
int
IntegerShortcode (Bulk Disbursement Account shortcode) sending the funds.
PartyBrequired
int
IntegerRecipient MSISDN. Will be normalized/validated as a Kenyan phone number.
Remarksrequired
str
StringFree text remarks (max 100 chars).
QueueTimeOutURLrequired
str
StringURL to receive timeout notifications if processing exceeds the provider timeout window.
ResultURLrequired
str
StringURL to receive the final result callback for the payment.
Occasion
str
StringOptional occasion (max 100 chars).
ConversationID
str
StringReturned by provider to identify the asynchronous processing conversation.
ResponseCoderequired
str|int
String/Integer'0' (or all-zero string) indicates success in service responses.
Result.ResultParameters
list
List[Key/Value]Result callback parameters (TransactionAmount, TransactionReceipt, etc.).

Overview

Initiate B2C payouts and handle callbacks.

Quick Setup

Python
# Example: using the MpesaClient facade to submit a B2C payment and inspect the typed response.
from mpesakit.client import MpesaClient
from mpesakit.b2c import B2CCommandIDType, B2CResponse


# Initialize the client with your credentials and environment
client = MpesaClient(consumer_key="...", consumer_secret="...", environment="sandbox")


# Build and send a B2C payment request
resp: B2CResponse = client.b2c.send_payment(
  originator_conversation_id="ocid-1234-5678",
  initiator_name="api_initiator",
  security_credential="ENCRYPTED_SECURITY_CREDENTIAL",
  command_id=B2CCommandIDType.BusinessPayment,
  amount=1500,
  party_a="600999",               # Bulk disbursement shortcode
  party_b="254712345678",         # Recipient MSISDN (normalized by SDK)
  remarks="Salary payout",
  queue_timeout_url="https://example.com/b2c/timeout",
  result_url="https://example.com/b2c/result",
  occasion="JulySalary",
)


# Inspect the typed response
if resp.is_successful():
  print("B2C sent:", resp.ResponseDescription)
else:
  print("B2C failed:", resp.ResponseDescription, "code:", resp.ResponseCode)

Callbacks (Result & Timeout)

Webhook receivers (FastAPI)
# Minimal FastAPI webhook receivers for B2C Result & Timeout callbacks.
# - Validates caller IP using is_mpesa_ip_allowed
# - Parses payload into SDK schemas (B2CResultCallback / B2CTimeoutCallback)
# - Persists/queues the payload for downstream processing (TODO)
# - Returns the acknowledgement JSON expected by M-Pesa Daraja API


from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import JSONResponse
from mpesakit.security import is_mpesa_ip_allowed
from mpesakit.http_client import MpesaHttpClient
from mpesakit.b2c.schemas import (
  B2CResultCallback,
  B2CTimeoutCallback,
  B2CResultCallbackResponse,
  B2CTimeoutCallbackResponse,
)
import logging


log = logging.getLogger(__name__)
app = FastAPI()


# Optional HTTP client for any outgoing calls to MPesa (not required to accept callbacks)
mpesa_http = MpesaHttpClient(environment="sandbox")  # or "production"


def _get_remote_ip(request: Request) -> str:
  xff = request.headers.get("x-forwarded-for")
  if xff:
      return xff.split(",")[0].strip()
  return request.client.host


@app.post("/webhooks/b2c/result")
async def b2c_result(request: Request):
  remote_ip = _get_remote_ip(request)
  if not is_mpesa_ip_allowed(remote_ip):
      log.warning("Rejected B2C result callback from disallowed IP: %s", remote_ip)
      raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)


  try:
      payload = await request.json()
  except Exception as exc:
      log.exception("Failed reading JSON payload from %s: %s", remote_ip, exc)
      return JSONResponse(
          status_code=400,
          content=B2CResultCallbackResponse(ResultCode=1, ResultDesc="Invalid JSON payload.").model_dump(),
      )


  try:
      callback = B2CResultCallback.model_validate(payload)
  except Exception as exc:
      log.exception("B2C result callback validation error: %s", exc)
      return JSONResponse(
          status_code=400,
          content=B2CResultCallbackResponse(ResultCode=1, ResultDesc=f"Invalid payload: {exc}").model_dump(),
      )


  # TODO: persist callback (DB/queue) for reconciliation and business processing.
  log.info(
      "Received B2C result: OriginatorConversationID=%s ConversationID=%s ResultCode=%s",
      callback.Result.OriginatorConversationID,
      callback.Result.ConversationID,
      callback.Result.ResultCode,
  )


  ack = B2CResultCallbackResponse()  # default success ack
  return JSONResponse(status_code=200, content=ack.model_dump())


@app.post("/webhooks/b2c/timeout")
async def b2c_timeout(request: Request):
  remote_ip = _get_remote_ip(request)
  if not is_mpesa_ip_allowed(remote_ip):
      log.warning("Rejected B2C timeout callback from disallowed IP: %s", remote_ip)
      raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)


  try:
      payload = await request.json()
  except Exception as exc:
      log.exception("Failed reading JSON payload from %s: %s", remote_ip, exc)
      return JSONResponse(
          status_code=400,
          content=B2CTimeoutCallbackResponse(ResultCode=1, ResultDesc="Invalid JSON payload.").model_dump(),
      )


  try:
      callback = B2CTimeoutCallback.model_validate(payload)
  except Exception as exc:
      log.exception("B2C timeout callback validation error: %s", exc)
      return JSONResponse(
          status_code=400,
          content=B2CTimeoutCallbackResponse(ResultCode=1, ResultDesc=f"Invalid payload: {exc}").model_dump(),
      )


  # TODO: persist/queue timeout notification and trigger compensating workflows.
  log.info(
      "Received B2C timeout: OriginatorConversationID=%s ConversationID=%s ResultCode=%s",
      callback.Result.OriginatorConversationID,
      callback.Result.ConversationID,
      callback.Result.ResultCode,
  )


  ack = B2CTimeoutCallbackResponse()  # default success ack
  return JSONResponse(status_code=200, content=ack.model_dump(mode="json"))

Schemas & Runtime Behavior

Error Handling

Testing & Expectations

  • Send payment:

    • The service posts payment requests to the provider using the configured HttpClient and supplies Authorization via TokenManager.
    • Successful responses are returned as B2CResponse instances; call is_successful() to check outcome.

  • Request validation:

    • Invalid CommandID, malformed PartyB or overly long Remarks/Occasion should raise validation errors during model construction.

  • Result metadata:

    • ResultParameters are provided as a list of Key/Value items. The SDK caches these into a dictionary so callers can access values using typed helper properties (transaction_amount, transaction_receipt, etc.).
    • recipient_is_registered returns True for 'Y', False for 'N', and None for missing/invalid values.

Next Steps