load_airtime
Send airtime to phone numbers using Africa's Talking API, format numbers correctly, and log transactions in a database for record-keeping.
Instructions
Sends airtime to a specified phone number and logs the transaction.
This tool formats the phone number, sends the airtime using the
Africa's Talking API, and saves a record of the transaction in the
database.
Args:
phone_number (str): The recipient's phone number.
amount (float): The amount of airtime to send.
currency_code (str): The currency for the transaction (e.g., "KES").
Returns:
str: A message indicating the status of the airtime transaction.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| phone_number | Yes | ||
| amount | Yes | ||
| currency_code | Yes |
Implementation Reference
- main.py:143-171 (handler)The main handler function for the 'load_airtime' tool, decorated with @mcp.tool() for registration. It formats the recipient's phone number, sends airtime using the Africa's Talking API, logs the transaction to a SQLite database, and returns a status message.@mcp.tool() async def load_airtime(phone_number: str, amount: float, currency_code: str) -> str: """Sends airtime to a specified phone number and logs the transaction. This tool formats the phone number, sends the airtime using the Africa's Talking API, and saves a record of the transaction in the database. Args: phone_number (str): The recipient's phone number. amount (float): The amount of airtime to send. currency_code (str): The currency for the transaction (e.g., "KES"). Returns: str: A message indicating the status of the airtime transaction. """ try: formatted_number = format_phone_number(phone_number) airtime.send( phone_number=formatted_number, amount=amount, currency_code=currency_code ) save_transaction(formatted_number, amount, currency_code) return ( f"Successfully sent {currency_code} {amount} airtime to {formatted_number}" ) except Exception as e: return f"Encountered an error while sending airtime: {str(e)}"
- main.py:68-98 (helper)Helper function used by load_airtime to format the phone number with the appropriate international country code based on the configured country.def format_phone_number(phone_number): """Formats a phone number to include the international country code. This function takes a phone number as a string and formats it based on the user's country, which is determined by the `user_country` global variable. It handles numbers that start with '0', '+', or a digit. Args: phone_number (str): The phone number to be formatted. Returns: str: The phone number with the country code prepended. Raises: ValueError: If the `user_country` is not in the `COUNTRY_CODES` map. """ phone_number = str(phone_number).strip() if user_country not in COUNTRY_CODES: raise ValueError( f"Invalid or unset country: {user_country}. Supported countries: {list(COUNTRY_CODES.keys())}" ) country_code = COUNTRY_CODES[user_country] if phone_number.startswith("0"): return country_code + phone_number[1:] elif phone_number.startswith("+"): return phone_number else: return country_code + phone_number
- main.py:101-118 (helper)Helper function used by load_airtime to persist the transaction details into the SQLite database.def save_transaction(phone_number, amount, currency_code): """Saves a single airtime transaction to the SQLite database. Args: phone_number (str): The recipient's phone number. amount (float): The amount of airtime sent. currency_code (str): The currency of the transaction (e.g., "KES"). """ with sqlite3.connect(DB_PATH) as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO transactions (phone_number, amount, currency_code, transaction_time) VALUES (?, ?, ?, ?) """, (phone_number, amount, currency_code, datetime.now()), ) conn.commit()