Python Examples
Complete Python example client for integrating with the Multihub API usingrequests and hashlib.
This is a reference implementation, not an official SDK package. You can copy it into your project and adapt it as needed.
Installation
pip install requests
Configuration
Store your credentials in environment variables:export MULTIHUB_APP_ID="1"
export MULTIHUB_SECRET_KEY="your_secret_key"
MultihubClient Class
A complete, reusable client class for all API operations:import hashlib
import json
import os
import requests
from typing import Optional, Dict, Any
class MultihubClient:
"""Multihub API Client for Python"""
def __init__(
self,
app_id: str = None,
secret_key: str = None,
base_url: str = "https://api.bafanglaicai88.com/public/api/multihub/v1",
):
self.app_id = app_id or os.environ.get("MULTIHUB_APP_ID")
self.secret_key = secret_key or os.environ.get("MULTIHUB_SECRET_KEY")
self.base_url = base_url
if not self.app_id or not self.secret_key:
raise ValueError("app_id and secret_key are required")
def _send(self, body: dict) -> dict:
"""Sign and send a request to the Multihub API."""
body_str = json.dumps(body, separators=(",", ":"))
hash_value = hashlib.sha512(
(body_str + self.secret_key).encode()
).hexdigest()
response = requests.post(
self.base_url,
data=body_str,
headers={
"Content-Type": "application/json",
"X-Data-Application-Id": self.app_id,
"X-Data-Hash": hash_value,
},
)
return response.json()
def ping(self) -> dict:
"""Health check / connectivity test."""
return self._send({"method": "gateway.ping", "params": {}})
def create_deposit(
self,
service_id: int,
amount: int,
currency: str,
c_id: str,
**kwargs,
) -> dict:
"""Create a deposit (pay-in) payment."""
return self._send({
"method": "payment.in",
"service_id": service_id,
"params": {
"payment": {
"identifiers": {"c_id": c_id},
"amount": {"value": amount, "currency": currency},
**kwargs,
}
},
})
def create_withdrawal(
self,
service_id: int,
amount: int,
currency: str,
c_id: str,
receiver: dict,
**kwargs,
) -> dict:
"""Create a withdrawal (payout) payment."""
return self._send({
"method": "payment.out",
"service_id": service_id,
"params": {
"payment": {
"identifiers": {"c_id": c_id},
"amount": {"value": amount, "currency": currency},
"receiver": receiver,
**kwargs,
}
},
})
def get_status(
self, c_id: str = None, h_id: str = None
) -> dict:
"""Check payment status by identifiers."""
identifiers = {}
if c_id:
identifiers["c_id"] = c_id
if h_id:
identifiers["h_id"] = h_id
return self._send({
"method": "payment.status",
"params": {"payment": {"identifiers": identifiers}},
})
def get_balance(self) -> dict:
"""Retrieve account balances."""
return self._send({"method": "balance.get", "params": {}})
Usage Examples
Ping (Health Check)
client = MultihubClient(app_id="1", secret_key="your_secret_key")
result = client.ping()
print(result)
# {"success": true, "result": {"message": "pong"}, ...}
Create a Deposit Payment
The examples below use INR (India), but the same client works for all supported currencies (MXN, ARS, TRY, AUD, LKR, UYU) — just change theservice_id, currency, and country values.
client = MultihubClient(app_id="1", secret_key="your_secret_key")
result = client.create_deposit(
service_id=14701,
amount=10000,
currency="INR",
c_id="12345",
description="Order #12345",
payer={
"email": "customer@example.com",
"person": {
"first_name": "John",
"last_name": "Doe",
},
},
)
if result["success"]:
payment = result["result"]["payment"]
print(f"Payment created successfully")
print(f"Request ID: {result['request_id']}")
else:
print(f"Error: {result['error']['code']} - {result['error']['message']}")
Create a Withdrawal (Payout)
result = client.create_withdrawal(
service_id=14701,
amount=5000,
currency="INR",
c_id="67890",
receiver={
"bank": {
"account": {"id": "1234567890"},
"ifsc": "SBIN0001234",
},
"person": {
"first_name": "Jane",
"last_name": "Doe",
},
},
description="Payout #67890",
)
if result["success"]:
print("Withdrawal created successfully")
else:
print(f"Error: {result['error']['code']} - {result['error']['message']}")
Check Payment Status
# Look up by client-side ID
result = client.get_status(c_id="12345")
if result["success"]:
status = result["result"]["payment"]["status"]["status"]
print(f"Payment status: {status}")
else:
print(f"Error: {result['error']['code']} - {result['error']['message']}")
# Look up by hub-side ID
result = client.get_status(h_id="999001")
if result["success"]:
status = result["result"]["payment"]["status"]["status"]
print(f"Payment status: {status}")
Get Balance
result = client.get_balance()
if result["success"]:
for amount in result["result"]["balance"]["amounts"]:
print(f"{amount['currency']}: {amount['value']}")
else:
print(f"Error: {result['error']['code']} - {result['error']['message']}")
Error Handling
Successful responses return HTTP 200, errors return HTTP 400. Always use thesuccess field to determine outcome:
client = MultihubClient(app_id="1", secret_key="your_secret_key")
result = client.create_deposit(
service_id=14701,
amount=10000,
currency="INR",
c_id="12345",
)
if result["success"]:
# Operation succeeded
payment = result["result"]
print(f"Request ID: {result['request_id']}")
print(f"Processing time: {result['processing_time']}ms")
else:
# Operation failed
error = result["error"]
print(f"Error code: {error['code']}")
print(f"Error message: {error['message']}")
# Handle specific error codes
if error["code"] == 3000:
print("Authentication failed - check your secret key")
elif error["code"] == 1005 or error["code"] == 9000:
print("Invalid parameters - check request body")
elif error["code"] == 7002:
print("Service temporarily unavailable - retry later")
Handling Network Errors
Network-level errors can still occur independently of API responses:import requests
client = MultihubClient(app_id="1", secret_key="your_secret_key")
try:
result = client.ping()
if result["success"]:
print("API is reachable")
else:
print(f"API error: {result['error']['message']}")
except requests.ConnectionError:
print("Cannot reach the API - check network connectivity")
except requests.Timeout:
print("Request timed out - try again")
except requests.RequestException as e:
print(f"Request failed: {e}")
except (KeyError, ValueError) as e:
print(f"Unexpected response format: {e}")
Complete Example
#!/usr/bin/env python3
"""
Multihub API Integration Example
Demonstrates the complete flow: ping, check balance, create a deposit,
and poll for status.
"""
import os
import time
import hashlib
import json
import requests
class MultihubClient:
def __init__(self, app_id, secret_key, base_url="https://api.bafanglaicai88.com/public/api/multihub/v1"):
self.app_id = app_id
self.secret_key = secret_key
self.base_url = base_url
def _send(self, body):
body_str = json.dumps(body, separators=(",", ":"))
hash_value = hashlib.sha512((body_str + self.secret_key).encode()).hexdigest()
response = requests.post(self.base_url, data=body_str, headers={
"Content-Type": "application/json",
"X-Data-Application-Id": self.app_id,
"X-Data-Hash": hash_value,
})
return response.json()
def ping(self):
return self._send({"method": "gateway.ping", "params": {}})
def get_balance(self):
return self._send({"method": "balance.get", "params": {}})
def create_deposit(self, service_id, amount, currency, c_id, **kwargs):
return self._send({
"method": "payment.in",
"service_id": service_id,
"params": {
"payment": {
"identifiers": {"c_id": c_id},
"amount": {"value": amount, "currency": currency},
**kwargs,
}
},
})
def get_status(self, c_id=None, h_id=None):
identifiers = {}
if c_id:
identifiers["c_id"] = c_id
if h_id:
identifiers["h_id"] = h_id
return self._send({
"method": "payment.status",
"params": {"payment": {"identifiers": identifiers}},
})
def main():
app_id = os.environ.get("MULTIHUB_APP_ID", "1")
secret_key = os.environ.get("MULTIHUB_SECRET_KEY", "your_secret_key")
client = MultihubClient(app_id=app_id, secret_key=secret_key)
# 1. Ping the gateway
ping_result = client.ping()
if not ping_result["success"]:
print(f"Gateway unreachable: {ping_result['error']['message']}")
return
print("Gateway is up")
# 2. Check balance
balance_result = client.get_balance()
if balance_result["success"]:
for b in balance_result["result"]["balance"]["amounts"]:
print(f" {b['currency']}: {b['value']}")
else:
print(f"Balance check failed: {balance_result['error']['message']}")
# 3. Create a deposit
order_id = str(int(time.time()))
deposit_result = client.create_deposit(
service_id=14701,
amount=10000,
currency="INR",
c_id=order_id,
description=f"Order #{order_id}",
payer={
"email": "customer@example.com",
"person": {"first_name": "John", "last_name": "Doe"},
},
)
if deposit_result["success"]:
print(f"Deposit created (request_id={deposit_result['request_id']})")
else:
print(f"Deposit failed: {deposit_result['error']['message']}")
return
# 4. Poll for status
for _ in range(5):
time.sleep(3)
status_result = client.get_status(c_id=order_id)
if status_result["success"]:
status = status_result["result"]["payment"]["status"]["status"]
print(f"Payment status: {status}")
if status in ("success", "error", "canceled", "declined", "refunded"):
break
else:
print(f"Status check failed: {status_result['error']['message']}")
break
if __name__ == "__main__":
main()
