M-Pesa Transaction Status

Instructions

Implementation Reference

• src/mpesa_mcp/server.py:218-261 (handler)

  The actual implementation of the mpesa_transaction_status tool. It takes a transaction_id (M-Pesa receipt number), builds a payload with the required Daraja API fields (Initiator, SecurityCredential, CommandID, TransactionID, PartyA, etc.), sends a POST request to /mpesa/transactionstatus/v1/query, and returns the async result info (conversation_id, response_code, etc.).

  @mcp.tool(annotations={
      'title': 'M-Pesa Transaction Status',
      'readOnlyHint': True,
      'destructiveHint': False,
      'idempotentHint': True,
      'openWorldHint': True,
  })
  def mpesa_transaction_status(
      transaction_id: Annotated[str, "M-Pesa receipt number e.g. QKL8XXXXXX"],
  ) -> dict:
      """
      Query the status of any M-Pesa transaction by receipt number.
      Requires MPESA_INITIATOR_NAME and MPESA_SECURITY_CREDENTIAL env vars.
      """
      payload = {
          "Initiator":           os.environ["MPESA_INITIATOR_NAME"],
          "SecurityCredential":  os.environ["MPESA_SECURITY_CREDENTIAL"],
          "CommandID":           "TransactionStatusQuery",
          "TransactionID":       transaction_id,
          "PartyA":              os.environ["MPESA_SHORTCODE"],
          "IdentifierType":      "4",
          "ResultURL":           os.environ.get("MPESA_RESULT_URL", os.environ["MPESA_CALLBACK_URL"]),
          "QueueTimeOutURL":     os.environ.get("MPESA_TIMEOUT_URL", os.environ["MPESA_CALLBACK_URL"]),
          "Remarks":             "Status query via mpesa-mcp",
          "Occasion":            "",
      }
  
      token = _get_mpesa_token()
      resp  = requests.post(
          f"{_mpesa_base()}/mpesa/transactionstatus/v1/query",
          json=payload,
          headers={"Authorization": f"Bearer {token}"},
          timeout=10,
      )
      resp.raise_for_status()
      data = resp.json()
  
      return {
          "accepted":          data.get("ResponseCode") == "0",
          "conversation_id":   data.get("ConversationID"),
          "response_code":     data.get("ResponseCode"),
          "description":       data.get("ResponseDescription"),
          "note":              "Result delivered asynchronously to MPESA_RESULT_URL",
      }

• src/mpesa_mcp/server.py:218-218 (registration)

  The @mcp.tool decorator registers mpesa_transaction_status as a tool on the FastMCP server instance, with annotations including title 'M-Pesa Transaction Status', readOnlyHint=True, destructiveHint=False, idempotentHint=True.

  @mcp.tool(annotations={

• src/mpesa_mcp/server.py:225-227 (schema)

  The function signature defines the input schema: transaction_id is an Annotated[str, ...] parameter described as 'M-Pesa receipt number e.g. QKL8XXXXXX'. The return type is dict (no formal Pydantic model, relies on FastMCP's automatic schema generation from annotations).

  def mpesa_transaction_status(
      transaction_id: Annotated[str, "M-Pesa receipt number e.g. QKL8XXXXXX"],
  ) -> dict: