Africa's Talking Airtime MCP

Instructions

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.

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()