Skip to main content

Transaction Status

Request transaction status from M-Pesa, receive result or timeout callbacks, and validate/interpret response metadata.

User Stories

As a payments engineer or backend developer at a merchant/PSP, I want to query the status of an M‑Pesa transaction and reliably handle result and timeout callbacks so I can reconcile payments, notify customers, and trigger compensating workflows when needed.

  • Persona: Payments engineer, reconciliation analyst, or backend developer integrating M‑Pesa.
  • Goal: Retrieve definitive transaction state (success, failed, pending) and process callbacks for downstream business logic.
  • Acceptance criteria:
    • Submit a Transaction Status query and receive an accepted response from Daraja.
    • ResultURL receives structured result payloads and the service acknowledges with ResultCode 0.
    • QueueTimeOutURL receives timeout notifications that trigger retry or compensation flows.
    • Request validation prevents invalid MSISDNs, enforces identifier rules, and blocks overly long remarks/occasions.

Parameters Definition

ParameterTypeDescription
Initiatorrequired
str
StringAPI username used to initiate the Transaction Status request.
SecurityCredentialrequired
str
StringEncrypted security credential (provided by M-Pesa Daraja API) used to authenticate the request.
CommandIDrequired
str
StringThe operation type. Use 'TransactionStatusQuery'.
TransactionID
str
StringM-Pesa transaction identifier to query. Either this or OriginalConversationID must be provided.
OriginalConversationID
str
StringOriginal conversation id for the earlier request. Can be used when TransactionID is unavailable.
PartyArequired
int
IntegerOrganization shortcode or MSISDN depending on IdentifierType.
IdentifierTyperequired
int
Integer (enum)Identifier type for PartyA. Allowed: 1 (MSISDN), 2 (Till Number), 4 (Short Code).
ResultURLrequired
str
StringHTTPS endpoint that will receive the result callback when the query completes.
QueueTimeOutURLrequired
str
StringHTTPS endpoint that will receive a timeout notification if the query times out.
Remarksrequired
str
StringShort comment describing the query (max 100 characters).
Occasion
str
StringOptional occasion string (max 100 characters).

Overview

The Transaction Status API lets you ask M-Pesa Daraja API for the current state of a previously submitted M‑Pesa transaction. The request is asynchronous: M-Pesa will post the result to your configured ResultURL when processing completes, or to the QueueTimeOutURL if processing times out.

Response Schema

TransactionStatusResponse — the synchronous acknowledgement returned by client.transactions.query_status(...)

ParameterTypeDescription
ConversationIDrequired
str | null
StringUnique ID for the transaction status request.
OriginatorConversationIDrequired
str | null
StringID for tracking the request.
ResponseCoderequired
str | int
String/IntegerStatus code. Any all-zero value (e.g. '0') indicates the request was accepted.
ResponseDescriptionrequired
str
StringStatus message.

Quick Setup (Sync)

Python
# Example (conceptual): query transaction status using the high-level client
from mpesakit import MpesaClient
from mpesakit.transaction_status import TransactionStatusIdentifierType




client = MpesaClient(consumer_key="...", consumer_secret="...", environment="sandbox")


resp = client.transactions.query_status(
      initiator="api_user",
      security_credential="ENCRYPTED_CREDENTIAL",
      transaction_id="LK12345",
      party_a=600000,
      identifier_type=TransactionStatusIdentifierType.SHORT_CODE.value,
      result_url="https://your.example/result",
      queue_timeout_url="https://your.example/timeout",
      remarks="Check status"
)


if resp.is_successful:
      print("Request accepted:", resp.ResponseDescription)
else:
      print("Request failed:", resp.ResponseDescription)

Quick Setup (Async)

Python
import asyncio
from mpesakit import AsyncMpesaClient
from mpesakit.transaction_status import TransactionStatusIdentifierType


async def main():
  async with AsyncMpesaClient(
      consumer_key="...", consumer_secret="...", environment="sandbox"
  ) as client:
      resp = await client.transactions.query_status(
          initiator="api_user",
          security_credential="ENCRYPTED_CREDENTIAL",
          transaction_id="LK12345",
          party_a=600000,
          identifier_type=TransactionStatusIdentifierType.SHORT_CODE.value,
          result_url="https://your.example/result",
          queue_timeout_url="https://your.example/timeout",
          remarks="Check status"
      )


      if resp.is_successful:
          print("Request accepted:", resp.ResponseDescription)
      else:
          print("Request failed:", resp.ResponseDescription)


asyncio.run(main())

Result & Timeout Callbacks (webhooks)

The callback is an inbound HTTP POST from Safaricom to your ResultURL/QueueTimeOutURL — independent of whether the original query was sent via MpesaClient or AsyncMpesaClient. For the result payload, both clients expose process_transactions_callback, a one-call shorthand for TransactionStatusResultCallback.model_validate(payload); it's plain validation with no I/O, so it's never awaited even on the async client. There's no equivalent helper for the timeout payload, so that one is still validated directly against TransactionStatusTimeoutCallback, as in the full example below.

Python
# Minimal FastAPI webhook receivers for Transaction Status Result & Timeout callbacks.
# - Validates caller IP using is_mpesa_ip_allowed
# - Parses the result payload via client.process_transactions_callback (or TransactionStatusTimeoutCallback directly for timeouts)
# - 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
import logging


from mpesakit import MpesaClient  # or AsyncMpesaClient — process_transactions_callback is identical on both
from mpesakit.security import is_mpesa_ip_allowed
from mpesakit.transaction_status import (
  TransactionStatusTimeoutCallback,
  TransactionStatusResultCallbackResponse,
  TransactionStatusTimeoutCallbackResponse,
)


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


client = MpesaClient(consumer_key="...", consumer_secret="...", environment="sandbox")




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/transaction-status/result")
async def transaction_status_result(request: Request):
  remote_ip = _get_remote_ip(request)
  if not is_mpesa_ip_allowed(remote_ip):
      log.warning("Rejected TransactionStatus 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=TransactionStatusResultCallbackResponse(
              ResultCode=1, ResultDesc="Invalid JSON payload."
          ).model_dump(),
      )


  try:
      # Equivalent to TransactionStatusResultCallback.model_validate(payload)
      callback = client.process_transactions_callback(payload)
  except Exception as exc:
      log.exception("TransactionStatus result callback validation error: %s", exc)
      return JSONResponse(
          status_code=400,
          content=TransactionStatusResultCallbackResponse(
              ResultCode=1, ResultDesc=f"Invalid payload: {exc}"
          ).model_dump(),
      )


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


  ack = TransactionStatusResultCallbackResponse()  # default success ack (ResultCode 0)
  return JSONResponse(status_code=200, content=ack.model_dump())




@app.post("/webhooks/transaction-status/timeout")
async def transaction_status_timeout(request: Request):
  remote_ip = _get_remote_ip(request)
  if not is_mpesa_ip_allowed(remote_ip):
      log.warning("Rejected TransactionStatus 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=TransactionStatusTimeoutCallbackResponse(
              ResultCode=1, ResultDesc="Invalid JSON payload."
          ).model_dump(),
      )


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


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


  ack = TransactionStatusTimeoutCallbackResponse()  # default success ack (ResultCode 0)
  return JSONResponse(status_code=200, content=ack.model_dump())

Result Callback Schema

TransactionStatusResultCallback — posted to ResultURL once the query completes

ParameterTypeDescription
Result.ResultTyperequired
int
Integer0 = success, 1 = failure.
Result.ResultCoderequired
int | str
Integer/String0 indicates the query succeeded; any other value is a failure code.
Result.ResultDescrequired
str
StringHuman readable result description.
Result.OriginatorConversationIDrequired
str
StringMatches the OriginatorConversationID from the initial request.
Result.ConversationIDrequired
str
StringMatches the ConversationID from the initial acknowledgement.
Result.TransactionID
str | null
StringM-Pesa transaction ID being queried, when available.
Result.ResultParameters
list[{Key, Value}]
ArrayKey/value parameters. Exposed via TransactionStatusResultMetadata convenience properties: transaction_amount (TransactionAmount), transaction_receipt (TransactionReceipt), transaction_status (Status, e.g. 'Completed'/'Failed'), transaction_reason (Reason, optional failure reason).

Timeout Callback Schema

TransactionStatusTimeoutCallback — posted to QueueTimeOutURL if the query doesn't complete in time

Uses the same Result shape as the Result Callback Schema above (it's the same TransactionStatusResultMetadata model), just delivered to QueueTimeOutURL instead of ResultURL, typically with a non-zero ResultCode and without TransactionID/ResultParameters.

Callback Acknowledgement Schemas

What your webhook handler should return to Safaricom

ParameterTypeDescription
ResultCode
int | str
Integer/StringDefaults to 0. Used by TransactionStatusResultCallbackResponse and TransactionStatusTimeoutCallbackResponse — the typed models to return from your /result and /timeout handlers respectively.
ResultDesc
str
StringDefaults to 'Result received and processed successfully.' (result) or 'Timeout notification received and processed successfully.' (timeout).

Result Payload (ResultParameters)

Common ResultParameters
{
  "Result": {
      "ResultType": 0,
      "ResultCode": 0,
      "ResultDesc": "The service request is processed successfully.",
      "OriginatorConversationID": "...",
      "ConversationID": "...",
      "TransactionID": "LKXXXX1234",
      "ResultParameters": [
          {"Key": "TransactionAmount", "Value": 1000},
          {"Key": "TransactionReceipt", "Value": "LKXXXX1234"},
          {"Key": "Status", "Value": "Completed"},
          {"Key": "Reason", "Value": "Optional failure reason"}
      ]
  }
}

Error Handling & Validation

Python
try:
  resp = client.transactions.query_status(...)
except Exception as exc:
  print("Transaction status query failed:", exc)

Testing & Expected Behaviors

  • Request validation:

    • Invalid identifier types or invalid MSISDN values should raise errors during model construction.
    • At least one of TransactionID or OriginalConversationID must be present.

  • HTTP interactions:

    • The TransactionStatus service posts to /mpesa/transactionstatus/v1/query with Authorization header from TokenManager (or AsyncTokenManager).
    • On successful acceptance the service returns a TransactionStatusResponse; use is_successful to determine accepted requests. This holds whether the response was awaited (async) or returned directly (sync).

  • Callbacks:

    • ResultURL receives TransactionStatusResultCallback with Result metadata — parse it directly or via process_transactions_callback; reply with ResultCode 0 to acknowledge receipt.
    • QueueTimeOutURL receives a timeout notification; handle and reconcile as needed.

Next Steps