M-Pesa STK Push

Instructions

Implementation Reference

• src/mpesa_mcp/server.py:109-159 (handler)

  The actual handler function for the 'mpesa_stk_push' tool. Triggers an M-Pesa STK Push payment prompt on the customer's phone, calls the Safaricom Daraja API, and returns a response with CheckoutRequestID.

  def mpesa_stk_push(
      phone: Annotated[str, "Customer phone number (any Kenyan format: +254..., 07..., 254...)"],
      amount: Annotated[int, "Amount in KES (whole number, minimum 1)"],
      account_ref: Annotated[str, "Account reference shown to customer on their phone (max 12 chars)"],
      description: Annotated[str, "Transaction description (max 13 chars)"] = "Payment",
  ) -> dict:
      """
      Trigger an M-Pesa STK Push — sends a payment prompt to the customer's phone.
      The customer enters their M-Pesa PIN to complete payment.
      Returns a CheckoutRequestID to track the transaction with mpesa_stk_query.
      Async: use mpesa_stk_query after 10-30 seconds to check completion.
      """
      shortcode = os.environ["MPESA_SHORTCODE"]
      passkey   = os.environ["MPESA_PASSKEY"]
      callback  = os.environ["MPESA_CALLBACK_URL"]
      timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
      password  = base64.b64encode(f"{shortcode}{passkey}{timestamp}".encode()).decode()
      phone     = _normalize_phone(phone)
  
      payload = {
          "BusinessShortCode": shortcode,
          "Password":          password,
          "Timestamp":         timestamp,
          "TransactionType":   "CustomerPayBillOnline",
          "Amount":            amount,
          "PartyA":            phone,
          "PartyB":            shortcode,
          "PhoneNumber":       phone,
          "CallBackURL":       callback,
          "AccountReference":  account_ref[:12],
          "TransactionDesc":   description[:13],
      }
  
      token = _get_mpesa_token()
      resp  = requests.post(
          f"{_mpesa_base()}/mpesa/stkpush/v1/processrequest",
          json=payload,
          headers={"Authorization": f"Bearer {token}"},
          timeout=15,
      )
      resp.raise_for_status()
      data = resp.json()
  
      return {
          "success":            data.get("ResponseCode") == "0",
          "checkout_request_id": data.get("CheckoutRequestID"),
          "merchant_request_id": data.get("MerchantRequestID"),
          "response_code":       data.get("ResponseCode"),
          "message":             data.get("CustomerMessage", data.get("ResponseDescription")),
          "sandbox":             os.environ.get("MPESA_SANDBOX", "true").lower() == "true",
      }

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

  The tool registration decorator using @mcp.tool with annotations. This is where 'mpesa_stk_push' is registered as an MCP tool on the FastMCP server instance.

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

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

  Helper function that obtains an OAuth access token from Safaricom's API, used by mpesa_stk_push to authenticate the STK Push request.

  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)

  Helper function that returns the base URL based on sandbox vs production mode, used by mpesa_stk_push when making the API call.

  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"

• src/mpesa_mcp/server.py:74-81 (helper)

  Helper function that normalizes phone numbers to 12-digit Kenyan format (254XXXXXXXXX), used by mpesa_stk_push to format the customer's phone before sending to the API.

  def _normalize_phone(phone: str) -> str:
      """Normalize to 12-digit Kenyan format: 254XXXXXXXXX."""
      phone = phone.strip().lstrip("+")
      if phone.startswith("0"):
          phone = "254" + phone[1:]
      elif not phone.startswith("254"):
          phone = "254" + phone
      return phone