{

"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Jupyter Note book MoMo Open API SandBox"
]
},
{
"cell_type": "code",
"execution_count": 31,
"metadata": {},
"outputs": [],
"source": [
"from datetime import datetime\n",
"Api_User = \"\" \n",
"Api_Key = \"\"\n",
"GetPaid_Debit_Request_Ref_ID = \"\" #UUID String Request Reference \n",
"Pay_Transfer_Request_Ref_ID = \"\" #UUID String Request Reference\n",
"Token=\"\"\n",
"Token_expiry_time = \"\"\n",
"Token_expired = False\n",
"Token_expiry_time = datetime.now()\n",
"Environment = \"sandbox\" #Target Environment \n",
"Collection_Subscription_Primary_Key
= \"4c91dae7a6f1474387a23a1f3d448eb7\"#Primary Key for Collection
Subscription.https://momodeveloper.mtn.com/profile\n",
"Disbursement_Subscription_Primary_Key
= \"635951aafb4f46bc86df1ab2c928a75b\"#Primary Key for Disbursement
Subscription.https://momodeveloper.mtn.com/profile\n",
"Base_Url = \"https://sandbox.momodeveloper.mtn.com\" #SandBox Base URL"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<b>Creating API User on the SandBox\n",
"This is a one time process.<b> "
]
},
{
"cell_type": "code",
"execution_count": 32,
"metadata": {},
"outputs": [],
"source": [
"#Function to create an API User (Username) from The MoMo OpenAPI SandBox\n",
"def Create_API_User_SandBox():\n",
" import requests as rq\n",
" import uuid\n",
" global Api_User\n",
" global Collection_Subscription_Primary_Key\n",
" Api_User = str(uuid.uuid4())\n",
" url = Base_Url+\"/v1_0/apiuser\"\n",
" headers = {\n",
" \"X-Reference-Id\": Api_User, #When creating Api user, the X Reference Id
in the header will be created as the user. \n",
 " \"Ocp-Apim-Subscription-Key\": Collection_Subscription_Primary_Key,\n",
" \"Content-Type\": \"application/json\"\n",
" }\n",
" body = { \n",
" \"providerCallbackHost\": \"webhook.site\" # If your callback is
https://webhook.site/mycallback/site then the providerCallbackHost is webhook.site\
n",
" }\n",
" try:\n",
" resp = rq.request(\"post\", url, json=body, headers=headers)\n",
" if(str(resp.status_code)==\"201\"):\n",
" print(\"HTTP Status Code:\"+str(resp.status_code)+\"\\n Api user
Created: \"+Api_User)\n",
" elif(str(resp.status_code)==\"401\"):\n",
" print(str(resp.status_code)+\" \"+resp.text+\" \")\n",
" print(\"Ensure the subscription key is the primary\")\n",
" elif(str(resp.status_code)==\"400\"):\n",
" print(str(resp.status_code)+\" \"+resp.text+\" \")\n",
" print(\"Ensure API User(X-Reference-Id) in the Headers is UUID
Version 4\")\n",
" print(\"Ensure the Body contains the correct
syntax \"\"\\\"providerCallbackHost\"\"\\\"\"+\":\"+\"Your CallBack URL HOST
Eg \"\"\\\"webhook.site\"\"\\\"\")\n",
" else:\n",
" print(str(resp.status_code)+\" \"+resp.text+\" \")\n",
" except TypeError:\n",
" print(\"Body of the Request has to be Json Format\")\n",
" except:\n",
" print(\"Something Is Wrong \"+resp.json)\n",
"#Create_API_User_SandBox() "
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<b>Creating the API Key (Password) For the API USER Created SandBox
{{Api_User}}</b>"
]
},
{
"cell_type": "code",
"execution_count": 33,
"metadata": {},
"outputs": [],
"source": [
"#Function to create an API Key for the API User from The MoMo OpenAPI SandBox\
n",
"def Create_API_Key_SandBox():\n",
" import requests as rq\n",
" import traceback\n",
" global Api_Key\n",
" global Collection_Subscription_Primary_Key\n",
" url = Base_Url+\"/v1_0/apiuser/\"+Api_User+\"/apikey\"\n",
" headers = {\n",
" \"Ocp-Apim-Subscription-Key\": Collection_Subscription_Primary_Key,\n",
" } \n",
" try:\n",
" resp = rq.request(\"post\", url, headers=headers)\n",
 " if(str(resp.status_code)==\"201\"):\n",
" Response = resp.json()\n",
" Api_Key = Response.get('apiKey')#Save the API Key in Variable \n",
" print(\"HTTP Status Code:\"+str(resp.status_code)+\"\\n Api
User:\"+str(Api_User) +\" Api Key:\"+str(Api_Key))\n",
" elif(str(resp.status_code)==\"400\"):\n",
" print(str(resp.status_code)+\" \"+resp.text+\" Validate the
BaseURL \\n And Ensure API_User is created, by calling the function
Create_API_User_SandBox()\")\n",
" elif(str(resp.status_code)==\"404\"):\n",
" print(str(resp.status_code)+\" \"+resp.text+\" API_USER was not
created, Please Run function Create_API_User_SandBox()\")\n",
" else:\n",
" print(str(resp.status_code)+\" \"+resp.text+\" \")\n",
" except TypeError:\n",
" print(\"Body of the Request has to be Json Format or No Body\")\n",
" except:\n",
" print(\"Something Is Wrong \")\n",
" traceback.print_exc() \n",
"#Create_API_Key_SandBox()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### <b>NOTE: API User and API Key are created once, but advisable to have the
API Key changed periodically</b>"
]
},
{
"cell_type": "code",
"execution_count": 34,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"HTTP Status Code:201\n",
" Api user Created: aa3f6853-57f2-4348-a29c-ee1376fc28d5\n",
"HTTP Status Code:201\n",
" Api User:aa3f6853-57f2-4348-a29c-ee1376fc28d5 Api
Key:be11eaaface14e46a539ff0cb8698148\n"
]
}
],
"source": [
"Create_API_User_SandBox() #Function create to create API User and store value
in Variable {{Api_User}}\n",
"Create_API_Key_SandBox() #Function to create API Key(Password) for the Api
User Created and stored in variable {{Api_Key}}\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<b>With the API User and API Key now set up, the next step is to generate a
Bearer Token. <br>\n",
"This token, which is necessary for authenticating most upcoming API requests,
<br>\n",
"will expire one hour after its creation.</b>"
]
},
{
"cell_type": "code",
"execution_count": 35,
"metadata": {},
"outputs": [],
"source": [
"#Function to generate Token and set Token expiry. \n",
"def Get_Token():# function to return token (renews token if expired)\n",
" import requests as rq\n",
" import traceback\n",
" import json\n",
" from datetime import datetime, timedelta\n",
" global Token \n",
" global Token_expiry_time\n",
" EndPoint = Base_Url+\"/collection/token/\"\n",
" Auth = bytes(Api_User + ':' + Api_Key, \"utf-8\")\n",
" headers = { \n",
" \"Ocp-Apim-Subscription-Key\": Collection_Subscription_Primary_Key,\n",
" }\n",
" try:\n",
" resp = rq.request(\"post\", EndPoint,auth=(Api_User,Api_Key),
headers=headers)\n",
" Response = resp.json()\n",
" if(str(resp.status_code) == \"200\"):\n",
" Token = Response.get('access_token')\n",
" Token_expiry = Response.get('expires_in')\n",
" Token_expiry_time = datetime.now() + timedelta(seconds=
int(Token_expiry)) #Track Token Expiry Time \n",
" print(\"New Token Generated Expiring at :\"
+str(Token_expiry_time)) \n",
" elif(str(resp.status_code) == \"500\" or
str(Response.get(\"error\"))==\"login_failed\"):\n",
" print(Response)\n",
" print(\"Ensure to Map the API User and API Key as
(Username:Password) respectively\")\n",
" else:\n",
" print(resp.text) \n",
" except:\n",
" print(\"Something Is Wrong \")\n",
" traceback.print_exc() \n",
"#Get_Token()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<b>To prevent the creation of tokens for each request,<br>\n",
" it is advisable to verify the token's validity and only generate a new one
when the previous token has expired.</b>"
]
},
{
 "cell_type": "code",
"execution_count": 36,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"New Token Generated Expiring at :2024-10-17 08:19:28.562992\n"
]
}
],
"source": [
"#Function to Validate Status of Token\n",
"#If the Token is Expired a new one will be generated. \n",
"def Token_Status():\n",
" if Token_expiry_time >= datetime.now():\n",
" Token_expired = False \n",
" print (\"Token not Expired: Expiring at \"+ str(Token_expiry_time))\
n",
" #print(Token)\n",
" else:\n",
" Token_expired = True\n",
" Get_Token()\n",
" #print (\"New Token Generated Expiring at \"+ str(Token_expiry_time))\
n",
" #print(Token)\n",
"Token_Status()"
]
},
{
"cell_type": "code",
"execution_count": 37,
"metadata": {},
"outputs": [],
"source": [
"#Function that initiates a Debit USSD Prompt to the Payer to approve wit PIN\
n",
"def Request_Debit_Payment(MSISDN,Amount):\n",
" import requests as rq\n",
" from datetime import datetime, timedelta\n",
" import traceback\n",
" import uuid\n",
" global GetPaid_Debit_Request_Ref_ID\n",
" Token_Status()\n",
" GetPaid_Debit_Request_Ref_ID = str(uuid.uuid4())\n",
" url = Base_Url+\"/collection/v1_0/requesttopay\"\n",
" headers = {\n",
" \"X-Reference-Id\": GetPaid_Debit_Request_Ref_ID, #Unique for every
request, used to validate status of the request. \n",
" \"X-Target-Environment\": Environment,\n",
" \"Ocp-Apim-Subscription-Key\": Collection_Subscription_Primary_Key,\n",
" \"Authorization\":\"Bearer \"+Token, #Avoid creating new tokens for every
request, track the Expiry \n",
" \"Content-Type\": \"application/json\",\n",
" \"X-Callback-Url\":\"https://webhook.site/mycallback/site\"### You can add
X-Callback-Url to receive the callback
(\"X-Callback-Url\":\"https://webhook.com/mysite/status\")\n",
" }\n",
 " body = { \n",
" \"amount\": Amount,\n",
" \"currency\": \"EUR\", #use the currency as EUR in the SandBox\n",
" \"externalId\": str(uuid.uuid1()), #Used for Reconciliation between
application and MoMo platform. \n",
" \"payer\": {\n",
" \"partyIdType\": \"MSISDN\",#EMAIL and ALIAS apply as well \n",
" \"partyId\": MSISDN\n",
" },\n",
" \"payerMessage\": \"MoMo Debit API\", #Message sent to the Payer\n",
" \"payeeNote\": \"MoMo Debit API\" #Message Note to the Payee\n",
" }\n",
" try:\n",
" resp = rq.request(\"post\", url, json=body, headers=headers)\n",
" if(str(resp.status_code) == \"202\"):\n",
" print(\"Debit request to MSISDN \"+MSISDN+\"
Amount \"+Amount+\" \"+ \"Response Code \"+str(resp.status_code))\n",
" print(\"Request_Reference_ID :\"+GetPaid_Debit_Request_Ref_ID )\n",
" elif (str(resp.status_code) == \"404\"):\n",
" print(\"Check The Base_URL \")\n",
" elif (str(resp.status_code) == \"400\"):\n",
" print(\"Ensure no Special Charters like & in the Message and Notes \\
nThe X-Reference-Id in the header should be UUID Versio 4\")\n",
" print(resp.text)\n",
" elif (str(resp.status_code) == \"500\" or
str(resp.json().get(\"message\")).endswith(\"INVALID_CALLBACK_URL_HOST\") or
str(resp.json().get(\"message\")).endswith(\"Currency not supported.\")):\n",
" print(resp.json())\n",
" print(\"Ensure the URL Host is the same with the one created when
generating API_USer function \")\n",
" print(\"Verify and validate Currency for Sand Box is EUR\")\n",
" elif (str(resp.status_code) == \"500\" ):\n",
" print(resp.text)\n",
" print(\"API is not available\")\n",
" else:\n",
" print(resp.status_code)\n",
" print(resp.text)\n",
" except TypeError:\n",
" print(\"Request Body should be Json Formatted\")\n",
" except:\n",
" print(\"Something Is Wrong \")\n",
" traceback.print_exc() \n",
"#Request_Debit_Payment(\"56733123453\",\"50000\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<b>From the Test paryer_Id test numbers, we can not validate the status of the
Requests </b>"
]
},
{
"cell_type": "code",
"execution_count": 38,
"metadata": {},
"outputs": [],
"source": [
 "#Check Status \n",
"def Check_Status(X_Reference_Id_Of_The_Debit_Request):\n",
" import requests as rq\n",
" import json\n",
" import traceback\n",
" Token_Status()\n",
" url =
Base_Url+\"/collection/v1_0/requesttopay/\"+X_Reference_Id_Of_The_Debit_Request\n",
" headers = {\n",
" \"X-Target-Environment\": Environment,\n",
" \"Ocp-Apim-Subscription-Key\": Collection_Subscription_Primary_Key,\n",
" \"Authorization\":\"Bearer \"+Token,\n",
" }\n",
" try:\n",
" resp = rq.request(\"get\", url,headers=headers)\n",
" Status_Json = resp.json()\n",
" Status_Json_DD = str(Status_Json).replace('\\'', '\"')\n",
" print(Status_Json)\n",
" \n",
" except:\n",
" print(\"Something Is Wrong \")\n",
" traceback.print_exc() \n",
" #print(Status_Json)\n",
" \n",
"#Check_Status(\"1cd09347-8e5a-4f75-8c0f-04a3ecc8c1fa\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"#### <b>Testing Status Responses with payer_Id_msisdn</b>\n",
"| payer_id-Msisdn | Response status \n",
"| --- | --- | \n",
"46733123450\t|Failed\n",
"46733123451\t|Rejected\n",
"46733123452\t|Timeout\n",
"56733123453\t|Success\n",
"46733123454\t|Pending"
]
},
{
"cell_type": "code",
"execution_count": 39,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Token not Expired: Expiring at 2024-10-17 08:19:28.562992\n",
"Debit request to MSISDN 56733123452 Amount 50000 Response Code 202\n",
"Request_Reference_ID :c9266963-4266-4994-95c9-a21742803257\n",
"Token not Expired: Expiring at 2024-10-17 08:19:28.562992\n",
"{'financialTransactionId': '1574559373', 'externalId': '004ed3b1-8c3f-11ef-
b56a-00155dbcb9c7', 'amount': '50000', 'currency': 'EUR', 'payer': {'partyIdType':
'MSISDN', 'partyId': '56733123452'}, 'payerMessage': 'MoMo Debit API', 'payeeNote':
'MoMo Debit API', 'status': 'SUCCESSFUL'}\n"
]
 }
],
"source": [
"\n",
"Request_Debit_Payment(\"56733123452\",\"50000\")\n",
"Check_Status(GetPaid_Debit_Request_Ref_ID)\n"
]
},
{
"cell_type": "code",
"execution_count": 40,
"metadata": {},
"outputs": [],
"source": [
"#Disbursements Transfer\n",
"#Function that initiates a Transfer from Business Wallet to a customer's \
n",
"def Request_Tranfer_Payment(MSISDN,Amount):\n",
" import requests as rq\n",
" from datetime import datetime, timedelta\n",
" import traceback\n",
" import uuid\n",
" global Pay_Transfer_Request_Ref_ID \n",
" Token_Status()\n",
" Pay_Transfer_Request_Ref_ID = str(uuid.uuid4())\n",
" url = Base_Url+\"/disbursement/v1_0/transfer\"\n",
" headers = {\n",
" \"X-Reference-Id\": Pay_Transfer_Request_Ref_ID , #Unique for every
request, used to validate status of the request. \n",
" \"X-Target-Environment\": Environment,\n",
" \"Ocp-Apim-Subscription-Key\": Disbursement_Subscription_Primary_Key ,\n",
" \"Authorization\":\"Bearer \"+Token, #Avoid creating new tokens for every
request, track the Expiry \n",
" \"Content-Type\": \"application/json\",\n",
" \"X-Callback-Url\":\"https://webhook.site/mycallback/site\"### You can add
X-Callback-Url to receive the callback
(\"X-Callback-Url\":\"https://webhook.com/mysite/status\")\n",
" }\n",
" body = { \n",
" \"amount\": Amount,\n",
" \"currency\": \"EUR\", #use the currency as EUR in the SandBox\n",
" \"externalId\": str(uuid.uuid1()), #Used for Reconciliation between
application and MoMo platform. \n",
" \"payee\": {\n",
" \"partyIdType\": \"MSISDN\",#EMAIL and ALIAS apply as well \n",
" \"partyId\": MSISDN\n",
" },\n",
" \"payerMessage\": \"MoMo Debit API\", #Message sent to the Payer\n",
" \"payeeNote\": \"MoMo Debit API\" #Message Note to the Payee\n",
" }\n",
" try:\n",
" resp = rq.request(\"post\", url, json=body, headers=headers)\n",
" if(str(resp.status_code) == \"202\"):\n",
" print(\"Transfer request to MSISDN \"+MSISDN+\"
Amount \"+Amount+\" \"+ \"Response Code \"+str(resp.status_code))\n",
"
print(\"Transfer_Request_Reference_ID :\"+Pay_Transfer_Request_Ref_ID )\n",
" elif (str(resp.status_code) == \"404\"):\n",
" print(\"Check The Base_URL \")\n",
 " elif (str(resp.status_code) == \"400\"):\n",
" print(\"Ensure no Special Charters like & in the Message and Notes \\
nThe X-Reference-Id in the header should be UUID Versio 4\")\n",
" print(resp.text)\n",
" elif (str(resp.status_code) == \"500\" or
str(resp.json().get(\"message\")).endswith(\"INVALID_CALLBACK_URL_HOST\") or
str(resp.json().get(\"message\")).endswith(\"Currency not supported.\")):\n",
" print(resp.json())\n",
" print(\"Ensure the URL Host is the same with the one created when
generating API_USer function \")\n",
" print(\"Verify and validate Currency for Sand Box is EUR\")\n",
" elif (str(resp.status_code) == \"500\" ):\n",
" print(resp.text)\n",
" print(\"API is not available\")\n",
" else:\n",
" print(resp.status_code)\n",
" print(resp.text)\n",
" except TypeError:\n",
" print(\"Request Body should be Json Formatted\")\n",
" except:\n",
" print(\"Something Is Wrong \")\n",
" traceback.print_exc() \n",
"#Request_Tranfer_Payment(\"56733123453\",\"50000\")"
]
},
{
"cell_type": "code",
"execution_count": 48,
"metadata": {},
"outputs": [],
"source": [
"#Check Status Disbursements Transfer\n",
"def Check_Status_Transfer(X_Reference_Id_Of_The_Transfer_Request):\n",
" import requests as rq\n",
" import json\n",
" import traceback\n",
" Token_Status()\n",
" url =
Base_Url+\"/disbursement/v1_0/transfer/\"+X_Reference_Id_Of_The_Transfer_Request\
n",
" headers = {\n",
" \"X-Target-Environment\": Environment,\n",
" \"Ocp-Apim-Subscription-Key\": Disbursement_Subscription_Primary_Key,\n",
" \"Authorization\":\"Bearer \"+Token,\n",
" }\n",
" try:\n",
" resp = rq.request(\"get\", url,headers=headers)\n",
" Status_Json = resp.json()\n",
" Status_Json_DD = str(Status_Json).replace('\\'', '\"')\n",
" print(Status_Json)\n",
" \n",
" except:\n",
" print(\"Something Is Wrong \")\n",
" traceback.print_exc() \n",
" #print(Status_Json)\n",
" \n",
"#Check_Status(\"1cd09347-8e5a-4f75-8c0f-04a3ecc8c1fa\")\n"
]
},
 {
"cell_type": "markdown",
"metadata": {},
"source": [
"#### <b>Testing Status Responses with payee_Id_msisdn</b>\n",
"| payee_id-Msisdn | Response status \n",
"| --- | --- | \n",
"46733123450\t|Failed\n",
"46733123451\t|Rejected\n",
"46733123452\t|Timeout\n",
"56733123453\t|Success\n",
"46733123454\t|Pending"
]
},
{
"cell_type": "code",
"execution_count": 49,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Token not Expired: Expiring at 2024-10-17 08:19:28.562992\n",
"Transfer request to MSISDN 56733123453 Amount 5009009900 Response Code 202\
n",
"Transfer_Request_Reference_ID :9233d3c1-ac66-4061-bed5-3ba399e445c4\n",
"Token not Expired: Expiring at 2024-10-17 08:19:28.562992\n",
"{'amount': '5009009900', 'currency': 'EUR', 'financialTransactionId':
'606255672', 'externalId': '33de64b6-8c3f-11ef-995d-00155dbcb9c7', 'payee':
{'partyIdType': 'MSISDN', 'partyId': '56733123453'}, 'payerMessage': 'MoMo Debit
API', 'payeeNote': 'MoMo Debit API', 'status': 'SUCCESSFUL'}\n"
]
}
],
"source": [
"Request_Tranfer_Payment(\"56733123453\",\"5009009900\")\n",
"Check_Status_Transfer(Pay_Transfer_Request_Ref_ID )"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "base",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.12"
}
 },
"nbformat": 4,
"nbformat_minor": 2
}