Add USSD, data, and voice examples

README.md

@@ -3,6 +3,9 @@
 Function-calling with Python and ollama. We are going to use the Africa's Talking API to send airtime and messages to a phone number using Natural language. Thus, creating an generative ai agent. Here are examples of prompts you can use to send airtime to a phone number:
 - Send airtime to xxxxxxxxx2046 and xxxxxxxxx3524 with an amount of 10 in currency KES
 - Send a message to xxxxxxxxx2046 and xxxxxxxxx3524 with a message "Hello, how are you?", using the username "username".
+- Dial a USSD code like *123# on xxxxxxxxx2046
+- Send 500MB of data to xxxxxxxxx2046 on provider safaricom
+- Call xxxxxxxxx2046 from +254700000001
 
 NB: The phone numbers are placeholders for the actual phone numbers.
 You need some VRAM to run this project. You can get VRAM from [here](https://vast.ai/) or [here](https://runpod.io?ref=46wgtjpg)

tests/test_cases.py

@@ -11,7 +11,15 @@
 import pytest
 import pytest_asyncio
 from unittest.mock import patch, MagicMock, AsyncMock
-from utils.function_call import send_airtime, send_message, search_news, translate_text
+from utils.function_call import (
+    send_airtime,
+    send_message,
+    search_news,
+    translate_text,
+    send_ussd,
+    send_mobile_data,
+    make_voice_call,
+)
 
 # Load environment variables: TEST_PHONE_NUMBER
 PHONE_NUMBER = os.getenv("TEST_PHONE_NUMBER")
@@ -190,3 +198,27 @@ async def test_translate_text_special_chars():
     with pytest.raises(ValueError) as exc:
         await translate_text("@#$%^", "French")
     assert "Invalid input" in str(exc.value)
+
+
+@patch("utils.function_call.africastalking.USSD")
+def test_send_ussd_success(mock_ussd):
+    """Test the send_ussd function for a successful USSD request."""
+    mock_ussd.send.return_value = {"Status": "Success"}
+    result = send_ussd(PHONE_NUMBER, "*123#")
+    assert re.search(r"Success", str(result))
+
+
+@patch("utils.function_call.africastalking.MobileData")
+def test_send_mobile_data_success(mock_data):
+    """Test the send_mobile_data function for a successful bundle purchase."""
+    mock_data.send.return_value = {"status": "Success"}
+    result = send_mobile_data(PHONE_NUMBER, "500MB", "safaricom", "daily")
+    assert re.search(r"Success", str(result))
+
+
+@patch("utils.function_call.africastalking.Voice")
+def test_make_voice_call_success(mock_voice):
+    """Test the make_voice_call function for successful call initiation."""
+    mock_voice.call.return_value = {"status": "Queued"}
+    result = make_voice_call("+254700000001", PHONE_NUMBER)
+    assert re.search(r"Queued", str(result))

utils/function_call.py

@@ -232,6 +232,120 @@ def send_message(phone_number: str, message: str, username: str, **kwargs) -> No
         return json.dumps({"error": str(e)})
 
 
+def send_ussd(phone_number: str, code: str, **kwargs) -> str:
+    """Send a USSD code to a phone number.
+
+    Parameters
+    ----------
+    phone_number : str
+        The phone number to dial the USSD code on.
+    code : str
+        The USSD code to send, e.g. ``*123#``.
+
+    Returns
+    -------
+    str
+        JSON response from the API.
+
+    Examples
+    --------
+    send_ussd("+254712345678", "*123#")
+    """
+    username = os.getenv("AT_USERNAME")
+    api_key = os.getenv("AT_API_KEY")
+    africastalking.initialize(username, api_key)
+    ussd = africastalking.USSD
+    masked_number = mask_phone_number(phone_number)
+    logger.info("Sending USSD %s to %s", code, masked_number)
+
+    try:
+        response = ussd.send(code, phone_number)
+        logger.debug("USSD response: %s", response)
+        return json.dumps(response)
+    except Exception as e:
+        logger.error("Encountered an error while sending USSD: %s", str(e))
+        return json.dumps({"error": str(e)})
+
+
+def send_mobile_data(phone_number: str, bundle: str, provider: str, plan: str, **kwargs) -> str:
+    """Send a mobile data bundle to a phone number.
+
+    Parameters
+    ----------
+    phone_number : str
+        The recipient phone number.
+    bundle : str
+        The data bundle amount, e.g. ``500MB``.
+    provider : str
+        Mobile data provider.
+    plan : str
+        Identifier for the data plan.
+
+    Returns
+    -------
+    str
+        JSON response from the API.
+
+    Examples
+    --------
+    send_mobile_data("+254712345678", "500MB", "safaricom", "daily")
+    """
+    username = os.getenv("AT_USERNAME")
+    api_key = os.getenv("AT_API_KEY")
+    africastalking.initialize(username, api_key)
+    data = africastalking.MobileData
+    masked_number = mask_phone_number(phone_number)
+    logger.info("Sending mobile data to %s", masked_number)
+
+    try:
+        response = data.send(
+            phone_number=phone_number,
+            quantity=bundle,
+            provider=provider,
+            plan=plan,
+        )
+        logger.debug("Data bundle response: %s", response)
+        return json.dumps(response)
+    except Exception as e:
+        logger.error("Encountered an error while sending mobile data: %s", str(e))
+        return json.dumps({"error": str(e)})
+
+
+def make_voice_call(from_number: str, to_number: str, **kwargs) -> str:
+    """Initiate a voice call between two numbers.
+
+    Parameters
+    ----------
+    from_number : str
+        The caller ID for the voice call.
+    to_number : str
+        The recipient of the call.
+
+    Returns
+    -------
+    str
+        JSON response from the API.
+
+    Examples
+    --------
+    make_voice_call("+254700000001", "+254712345678")
+    """
+    username = os.getenv("AT_USERNAME")
+    api_key = os.getenv("AT_API_KEY")
+    africastalking.initialize(username, api_key)
+    voice = africastalking.Voice
+    masked_number = mask_phone_number(to_number)
+    logger.info("Calling %s from %s", masked_number, from_number)
+
+    try:
+        response = voice.call(from_=from_number, to=[to_number])
+        logger.debug("Voice call response: %s", response)
+        return json.dumps(response)
+    except Exception as e:
+        logger.error("Encountered an error while making voice call: %s", str(e))
+        return json.dumps({"error": str(e)})
+
+
 def search_news(query: str, **kwargs) -> str:
     """Search for news using DuckDuckGo search engine based on the query provided.
 
@@ -457,6 +571,77 @@ async def run(model: str, user_input: str):
                     },
                 },
             },
+            {
+                "type": "function",
+                "function": {
+                    "name": "send_ussd",
+                    "description": "Send a USSD code to a phone number using the Africa's Talking API",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "phone_number": {
+                                "type": "string",
+                                "description": "The phone number in international format",
+                            },
+                            "code": {
+                                "type": "string",
+                                "description": "The USSD code to dial",
+                            },
+                        },
+                        "required": ["phone_number", "code"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "send_mobile_data",
+                    "description": "Send a mobile data bundle using the Africa's Talking API",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "phone_number": {
+                                "type": "string",
+                                "description": "The phone number in international format",
+                            },
+                            "bundle": {
+                                "type": "string",
+                                "description": "The bundle amount e.g. 500MB",
+                            },
+                            "provider": {
+                                "type": "string",
+                                "description": "The data provider",
+                            },
+                            "plan": {
+                                "type": "string",
+                                "description": "Identifier for the data plan",
+                            },
+                        },
+                        "required": ["phone_number", "bundle", "provider", "plan"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "make_voice_call",
+                    "description": "Initiate a voice call between two numbers using Africa's Talking API",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "from_number": {
+                                "type": "string",
+                                "description": "The caller ID",
+                            },
+                            "to_number": {
+                                "type": "string",
+                                "description": "The recipient phone number",
+                            },
+                        },
+                        "required": ["from_number", "to_number"],
+                    },
+                },
+            },
         ],
     )
     # Add the model's response to the conversation history
@@ -475,6 +660,9 @@ async def run(model: str, user_input: str):
             "send_message": send_message,
             "search_news": search_news,
             "translate_text": translate_text,
+            "send_ussd": send_ussd,
+            "send_mobile_data": send_mobile_data,
+            "make_voice_call": make_voice_call,
         }
         for tool in response["message"]["tool_calls"]:
             # Get the function to call based on the tool name
@@ -509,6 +697,26 @@ async def run(model: str, user_input: str):
                     tool["function"]["arguments"]["target_language"],
                 )
                 logger.debug("function response: %s", function_response)
+            elif tool["function"]["name"] == "send_ussd":
+                function_response = function_to_call(
+                    tool["function"]["arguments"]["phone_number"],
+                    tool["function"]["arguments"]["code"],
+                )
+                logger.debug("function response: %s", function_response)
+            elif tool["function"]["name"] == "send_mobile_data":
+                function_response = function_to_call(
+                    tool["function"]["arguments"]["phone_number"],
+                    tool["function"]["arguments"]["bundle"],
+                    tool["function"]["arguments"]["provider"],
+                    tool["function"]["arguments"]["plan"],
+                )
+                logger.debug("function response: %s", function_response)
+            elif tool["function"]["name"] == "make_voice_call":
+                function_response = function_to_call(
+                    tool["function"]["arguments"]["from_number"],
+                    tool["function"]["arguments"]["to_number"],
+                )
+                logger.debug("function response: %s", function_response)
 
             # Add the function response to the conversation history
             messages.append(