M-Pesa STK Query

Instructions

Implementation Reference

• src/mpesa_mcp/server.py:169-215 (handler)

  The mpesa_stk_query function that executes the STK push query logic. It takes a checkout_request_id, builds the payload with BusinessShortCode, Password, Timestamp, and CheckoutRequestID, then calls the Safaricom Daraja /mpesa/stkpushquery/v1/query API and returns the status (success, result_code, status description).

  def mpesa_stk_query(
      checkout_request_id: Annotated[str, "CheckoutRequestID from mpesa_stk_push response"],
  ) -> dict:
      """
      Check the status of an STK Push request.
      Poll this 10-30 seconds after calling mpesa_stk_push.
      ResultCode 0 = success, 1032 = cancelled by user, 1037 = timed out.
      """
      shortcode = os.environ["MPESA_SHORTCODE"]
      passkey   = os.environ["MPESA_PASSKEY"]
      timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
      password  = base64.b64encode(f"{shortcode}{passkey}{timestamp}".encode()).decode()
  
      payload = {
          "BusinessShortCode":  shortcode,
          "Password":           password,
          "Timestamp":          timestamp,
          "CheckoutRequestID":  checkout_request_id,
      }
  
      token = _get_mpesa_token()
      resp  = requests.post(
          f"{_mpesa_base()}/mpesa/stkpushquery/v1/query",
          json=payload,
          headers={"Authorization": f"Bearer {token}"},
          timeout=10,
      )
      resp.raise_for_status()
      data = resp.json()
  
      result_code = int(data.get("ResultCode", -1))
      status_map  = {
          0:    "SUCCESS",
          1:    "INSUFFICIENT_FUNDS",
          1001: "LOCKED — retry in 30s",
          1019: "EXPIRED — re-initiate",
          1032: "CANCELLED_BY_USER",
          1037: "TIMEOUT",
          2001: "WRONG_PIN",
      }
  
      return {
          "success":      result_code == 0,
          "result_code":  result_code,
          "status":       status_map.get(result_code, f"UNKNOWN ({result_code})"),
          "description":  data.get("ResultDesc"),
      }

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

  The @mcp.tool decorator registering mpesa_stk_query as an MCP tool with annotations including title 'M-Pesa STK Query', readOnlyHint=True, idempotentHint=True.

  @mcp.tool(annotations={
      'title': 'M-Pesa STK Query',
      'readOnlyHint': True,
      'destructiveHint': False,
      'idempotentHint': True,
      'openWorldHint': True,
  })

• src/mpesa_mcp/server.py:46-66 (helper)

  The _get_mpesa_token helper used by mpesa_stk_query to obtain an OAuth bearer token for authenticating with the Daraja API.

  def _get_mpesa_token() -> str:
      if time.time() < _token_cache["expires_at"] - 30:
          return _token_cache["token"]  # type: ignore[return-value]
  
      sandbox = os.environ.get("MPESA_SANDBOX", "true").lower() == "true"
      base = "https://sandbox.safaricom.co.ke" if sandbox else "https://api.safaricom.co.ke"
  
      key    = os.environ["MPESA_CONSUMER_KEY"]
      secret = os.environ["MPESA_CONSUMER_SECRET"]
      creds  = base64.b64encode(f"{key}:{secret}".encode()).decode()
  
      resp = requests.get(
          f"{base}/oauth/v1/generate?grant_type=client_credentials",
          headers={"Authorization": f"Basic {creds}"},
          timeout=10,
      )
      resp.raise_for_status()
      data = resp.json()
      _token_cache["token"]      = data["access_token"]
      _token_cache["expires_at"] = time.time() + int(data["expires_in"])
      return _token_cache["token"]  # type: ignore[return-value]

• src/mpesa_mcp/server.py:69-71 (helper)

  The _mpesa_base helper returns the correct base URL (sandbox vs production) used by mpesa_stk_query to construct the API endpoint.

  def _mpesa_base() -> str:
      sandbox = os.environ.get("MPESA_SANDBOX", "true").lower() == "true"
      return "https://sandbox.safaricom.co.ke" if sandbox else "https://api.safaricom.co.ke"

• tests/test_smoke.py:15-28 (registration)

  Smoke test verifying that 'mpesa_stk_query' is among the registered tool names.

  def test_tools_registered():
      import asyncio
      from mpesa_mcp import mcp
      tools = asyncio.run(mcp.list_tools())
      names = [t.name for t in tools]
      expected = [
          "mpesa_stk_push",
          "mpesa_stk_query",
          "mpesa_transaction_status",
          "sms_send",
          "airtime_send",
      ]
      for name in expected:
          assert name in names, f"Tool '{name}' not registered. Found: {names}"