集成M-Pesa Daraja API时遭遇Bad Request - Invalid PhoneNumber错误

我正在使用Daraja门户将M-Pesa集成到应用中，但遇到了「Bad Request - Invalid PhoneNumber」错误。尽管使用的是有效的肯尼亚手机号，仍无法成功推进流程。相关代码及错误信息如下：

支付类代码 #

from api.services.payment.m_pesa_payment.mpes_credential import MpesaC2bCredential
import requests
import logging
import json
class MakePayment:
    @staticmethod
    def make_payment(payload):
        mepesa_acess_token = MpesaAccessToken()
        credential = MpesaC2bCredential()
        access_token = mepesa_acess_token.get_access_token()
        amount = payload["amount"]
        phone_number = payload["phone_number"]
        api_url = "https://sandbox.safaricom.co.ke/mpesa/stkpush/v1/processrequest"
        headers = {"Authorization": "Bearer %s" % access_token}
        
        request = {
        "BusinessShortCode": credential.short_code,
        "Password": mepesa_acess_token.generate_password(),
        "Timestamp": mepesa_acess_token.current_time,
        "TransactionType": "CustomerPayBillOnline",
        "Amount": amount,
        "PartyA": phone_number,
        "PartyB": credential.short_code,
        "PhoneNumber": phone_number,
        "CallBackURL": "https://ba3f-41-79-219-185.ngrok-free.app/payment",
        "AccountReference": "unknown",
        "TransactionDesc": "Testing stk push"
    }
        
        response = requests.post(api_url, json=request, headers=headers)
        
        response_data = response.json()
        logging.info("Mpesa request data {}".format(response_data))
        logging.info("Mpesa response info {}".format(response_data))

        return response_data

payload = {"phone_number":"+254708374149", "amount": 5000}   
print(MakePayment.make_payment(payload))

凭证类代码 #

class MpesaC2bCredential:
    mpesa_url = 'https://sandbox.safaricom.co.ke/oauth/v1/generate?grant_type=client_credentials'
    passe_key = 'xxxxxx'
    
    def __init__(self, consumer_key='xxxx', consumer_secret='xxx', api_urls = mpesa_url, short_code='174379',passe_key = passe_key):
        self.consumer_key = consumer_key
        self.consumer_secret = consumer_secret
        self.api_url = api_urls
        self.short_code = short_code
        self.api_url = api_urls
        self.passe_key = passe_key
        
        @property
        def consumer_key(self):
            return self.__consumer_key
        
        @consumer_key.setter
        def consumer_key(self, value):
            self.__consumer_key = value
            
        @property
        def consumer_secret(self):
            return self.__consumer_secret
        
        @consumer_secret.setter
        def consumer_secret(self, value):
            self.__consumer_secret = value
            
        @property
        def api_url(self):
            return self.__api_url
        
        @api_url.setter
        def api_url(self, value):
            self.__url = value
            
        @property
        def short_code(self):
            return self.__short_code
        
        @short_code.setter
        def short_code(self, value):
            self.__short_code = value
            
        @property
        def passe_key(self):
            return self.__short_code
        
        @passe_key.setter
        def short_code(self, value):
            self.__short_code = value

访问令牌类代码 #

from api.services.payment.m_pesa_payment.mpes_credential import MpesaC2bCredential
import requests
import json
from requests.auth import HTTPBasicAuth
from datetime import datetime
import base64
import logging
import time

class MpesaAccessToken:
    access_token_expiration = None
    current_time = datetime.now().strftime('%Y%m%d%H%M%S')
    
    def __init__(self, access_token = None, credential = MpesaC2bCredential(), current_time = current_time):
        self.access_token = access_token
        self.credential = credential
        self.current_time = current_time

        
    def get_access_token(self):
        try:
            res = requests.get(
                self.credential.api_url,
                auth=HTTPBasicAuth(self.credential.consumer_key, self.credential.consumer_secret),
            )
        except Exception as err:
            logging.error("Error {}".format(err))
            raise err
        else:
            token = res.json()["access_token"]
            self.headers = {"Authorization": "Bearer %s" % token}
            return token
        
    
    def generate_password(self):
        """Generates mpesa api password using the provided shortcode and passkey"""
        current_date = datetime.now()
        #timestamp = current_date.strftime("%Y%m%d%H%M%S")
        password_str = self.credential.short_code + self.credential.passe_key + self.current_time
        password_bytes = password_str.encode("ascii")
        return base64.b64encode(password_bytes).decode("utf-8")

错误信息 #

{'requestId': '11231-10026868-1', 'errorCode': '400.002.02', 'errorMessage': 'Bad Request - Invalid PhoneNumber'}

排查原因及解决方法 #

1. 手机号格式不符合API要求 #

Daraja API明确要求手机号为不带+号的纯数字格式，前缀必须是254。你的测试手机号+254708374149需要改为254708374149，带+号的格式会直接触发该错误。可以在代码中统一处理格式：

# 在make_payment方法中处理手机号
phone_number = payload["phone_number"].strip().replace("+", "")

2. 凭证类属性定义存在致命错误 #

MpesaC2bCredential类中passe_key的getter和setter完全错误：

• @property定义的passe_key返回的是self.__short_code，而非正确的self.__passe_key
• @passe_key.setter的函数名写成了short_code，导致无法正确设置passe_key属性
  这会导致passe_key值始终等于short_code，虽不直接触发手机号错误，但会引发后续密码生成失败，必须修复：

# 修正后的passe_key属性
@property
def passe_key(self):
    return self.__passe_key

@passe_key.setter
def passe_key(self, value):
    self.__passe_key = value

同时初始化方法中重复赋值self.api_url = api_urls，可删除其中一行，避免冗余。

3. 沙箱环境测试号限制 #

沙箱环境仅支持Daraja官方提供的测试手机号，真实手机号无法在沙箱中使用。确认你使用的254708374149是沙箱允许的测试号，若使用真实号码会触发错误。

4. 参数一致性检查 #

确保请求中PartyA和PhoneNumber的手机号格式完全一致，且都符合纯数字要求，避免因参数不一致引发错误。

内容的提问来源于stack exchange，提问作者rriol