From 3b4992471e3f836ec1f2c646aed24b408ef188c4 Mon Sep 17 00:00:00 2001 From: Coder-Joe458 Date: Wed, 18 Sep 2024 20:34:00 -0700 Subject: [PATCH] feat: Integrate What's App --- camel/toolkits/whatsapp_toolkit.py | 170 ++++++++++++++++++++++++ test/toolkits/test_whatsapp_function.py | 135 +++++++++++++++++++ 2 files changed, 305 insertions(+) create mode 100644 camel/toolkits/whatsapp_toolkit.py create mode 100644 test/toolkits/test_whatsapp_function.py diff --git a/camel/toolkits/whatsapp_toolkit.py b/camel/toolkits/whatsapp_toolkit.py new file mode 100644 index 000000000..e4b3b3a2f --- /dev/null +++ b/camel/toolkits/whatsapp_toolkit.py @@ -0,0 +1,170 @@ +# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. =========== +# Licensed under the Apache License, Version 2.0 (the “License”); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an “AS IS” BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. =========== + +import os +import time +from typing import Any, Dict, List, Union + +from requests.exceptions import RequestException +import requests + +from camel.toolkits import OpenAIFunction +from camel.toolkits.base import BaseToolkit + + +class WhatsAppToolkit(BaseToolkit): + """A class representing a toolkit for WhatsApp operations. + + This toolkit provides methods to interact with the WhatsApp Business API, + allowing users to send messages, retrieve message templates, and get + business profile information. + + Attributes: + retries (int): Number of retries for API requests in case of failure. + delay (int): Delay between retries in seconds. + base_url (str): Base URL for the WhatsApp Business API. + version (str): API version. + """ + + def __init__(self, retries: int = 3, delay: int = 1): + """Initializes the WhatsAppToolkit with the specified number of retries + and delay. + + Args: + retries (int): Number of times to retry the request in case of + failure. Defaults to 3. + delay (int): Time in seconds to wait between retries. Defaults to 1. + """ + self.retries = retries + self.delay = delay + self.base_url = "https://graph.facebook.com" + self.version = "v17.0" + + self.access_token = os.environ.get("WHATSAPP_ACCESS_TOKEN", "") + self.phone_number_id = os.environ.get("WHATSAPP_PHONE_NUMBER_ID", "") + + if not all([self.access_token, self.phone_number_id]): + raise ValueError( + "WhatsApp API credentials are not set. " + "Please set the WHATSAPP_ACCESS_TOKEN and WHATSAPP_PHONE_NUMBER_ID environment variables." + ) + + def _retry_request(self, func, *args, **kwargs): + """Retries a function in case of any errors. + + Args: + func (callable): The function to be retried. + *args: Arguments to pass to the function. + **kwargs: Keyword arguments to pass to the function. + + Returns: + Any: The result of the function call if successful. + + Raises: + Exception: If all retry attempts fail. + """ + for attempt in range(self.retries): + try: + return func(*args, **kwargs) + except Exception as e: + print(f"Attempt {attempt + 1}/{self.retries} failed: {e}") + if attempt < self.retries - 1: + time.sleep(self.delay) + else: + raise + + def send_message( + self, to: str, message: str + ) -> Union[Dict[str, Any], str]: + """Sends a text message to a specified WhatsApp number. + + Args: + to (str): The recipient's WhatsApp number in international format. + message (str): The text message to send. + + Returns: + Union[Dict[str, Any], str]: A dictionary containing the API response + if successful, or an error message string if failed. + """ + url = f"{self.base_url}/{self.version}/{self.phone_number_id}/messages" + headers = { + "Authorization": f"Bearer {self.access_token}", + "Content-Type": "application/json", + } + data = { + "messaging_product": "whatsapp", + "to": to, + "type": "text", + "text": {"body": message}, + } + + try: + response = self._retry_request( + requests.post, url, headers=headers, json=data + ) + response.raise_for_status() + return response.json() + except Exception as e: + return f"Failed to send message: {str(e)}" + + def get_message_templates(self) -> Union[List[Dict[str, Any]], str]: + """Retrieves all message templates for the WhatsApp Business account. + + Returns: + Union[List[Dict[str, Any]], str]: A list of dictionaries containing + template information if successful, or an error message string if failed. + """ + url = f"{self.base_url}/{self.version}/{self.phone_number_id}/message_templates" + headers = {"Authorization": f"Bearer {self.access_token}"} + + try: + response = self._retry_request(requests.get, url, headers=headers) + response.raise_for_status() + return response.json().get("data", []) + except Exception as e: + return f"Failed to retrieve message templates: {str(e)}" + + def get_business_profile(self) -> Union[Dict[str, Any], str]: + """Retrieves the WhatsApp Business profile information. + + Returns: + Union[Dict[str, Any], str]: A dictionary containing the business profile + information if successful, or an error message string if failed. + """ + url = f"{self.base_url}/{self.version}/{self.phone_number_id}/whatsapp_business_profile" + headers = {"Authorization": f"Bearer {self.access_token}"} + params = {"fields": "about,address,description,email,profile_picture_url,websites,vertical"} + + try: + response = self._retry_request( + requests.get, url, headers=headers, params=params + ) + response.raise_for_status() + return response.json() + except Exception as e: + return f"Failed to retrieve business profile: {str(e)}" + + def get_tools(self) -> List[OpenAIFunction]: + """Returns a list of OpenAIFunction objects representing the + functions in the toolkit. + + Returns: + List[OpenAIFunction]: A list of OpenAIFunction objects for the + toolkit methods. + """ + return [ + OpenAIFunction(self.send_message), + OpenAIFunction(self.get_message_templates), + OpenAIFunction(self.get_business_profile), + ] \ No newline at end of file diff --git a/test/toolkits/test_whatsapp_function.py b/test/toolkits/test_whatsapp_function.py new file mode 100644 index 000000000..8054be718 --- /dev/null +++ b/test/toolkits/test_whatsapp_function.py @@ -0,0 +1,135 @@ +import os +import pytest +from unittest.mock import patch, MagicMock + +from requests import RequestException + +from camel.toolkits.whatsapp_toolkit import WhatsAppToolkit + + +@pytest.fixture +def whatsapp_toolkit(): + # Set environment variables for testing + os.environ['WHATSAPP_ACCESS_TOKEN'] = 'test_token' + os.environ['WHATSAPP_PHONE_NUMBER_ID'] = 'test_phone_number_id' + return WhatsAppToolkit() + + +def test_init_missing_credentials(): + # Test initialization with missing credentials + os.environ.pop('WHATSAPP_ACCESS_TOKEN', None) + os.environ.pop('WHATSAPP_PHONE_NUMBER_ID', None) + + with pytest.raises(ValueError): + WhatsAppToolkit() + + +@patch('requests.post') +def test_send_message_success(mock_post, whatsapp_toolkit): + # Mock successful API response + mock_response = MagicMock() + mock_response.json.return_value = {"message_id": "test_message_id"} + mock_response.raise_for_status.return_value = None + mock_post.return_value = mock_response + + result = whatsapp_toolkit.send_message("1234567890", "Test message") + + assert result == {"message_id": "test_message_id"} + mock_post.assert_called_once() + + +@patch('requests.post') +def test_send_message_failure(mock_post, whatsapp_toolkit): + # Mock failed API response + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = Exception("API Error") + mock_post.return_value = mock_response + + result = whatsapp_toolkit.send_message("1234567890", "Test message") + + assert result == "Failed to send message: API Error" + mock_post.assert_called_once() + + +@patch('requests.get') +def test_get_message_templates_success(mock_get, whatsapp_toolkit): + # Mock successful API response + mock_response = MagicMock() + mock_response.json.return_value = {"data": [{"name": "template1"}, {"name": "template2"}]} + mock_response.raise_for_status.return_value = None + mock_get.return_value = mock_response + + result = whatsapp_toolkit.get_message_templates() + + assert result == [{"name": "template1"}, {"name": "template2"}] + mock_get.assert_called_once() + + +@patch('requests.get') +def test_get_message_templates_failure(mock_get, whatsapp_toolkit): + # Mock failed API response + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = Exception("API Error") + mock_response.json.return_value = {"error": "Failed to retrieve message templates"} + mock_get.return_value = mock_response + + result = whatsapp_toolkit.get_message_templates() + assert result == 'Failed to retrieve message templates: API Error' + mock_get.assert_called_once() + + +@patch('requests.get') +def test_get_business_profile_success(mock_get, whatsapp_toolkit): + # Mock successful API response + mock_response = MagicMock() + mock_response.json.return_value = {"name": "Test Business", "description": "Test Description"} + mock_response.raise_for_status.return_value = None + mock_get.return_value = mock_response + + result = whatsapp_toolkit.get_business_profile() + + assert result == {"name": "Test Business", "description": "Test Description"} + mock_get.assert_called_once() + + +@patch('requests.get') +def test_get_business_profile_failure(mock_get, whatsapp_toolkit): + # Mock failed API response + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = Exception("API Error") + mock_response.json.return_value = {"error": "Failed to retrieve message templates"} + mock_get.return_value = mock_response + + result = whatsapp_toolkit.get_business_profile() + assert isinstance(result, str) + + assert "Failed to retrieve business profile" in result + + assert "API Error" in result + mock_get.assert_called_once() + + +def test_get_tools(whatsapp_toolkit): + tools = whatsapp_toolkit.get_tools() + + assert len(tools) == 3 + for tool in tools: + assert hasattr(tool, '__call__') or hasattr(tool, 'func') + assert callable(tool) or (hasattr(tool, 'func') and callable(tool.func)) + + +@patch('time.sleep') +@patch('requests.post') +def test_retry_mechanism(mock_post, mock_sleep, whatsapp_toolkit): + # Mock failed API responses followed by a success + mock_post.side_effect = [ + RequestException("API Error"), + RequestException("API Error"), + MagicMock(json=lambda: {"message_id": "test_message_id"}, raise_for_status=lambda: None) + ] + + result = whatsapp_toolkit.send_message("1234567890", "Test message") + + assert result == {"message_id": "test_message_id"} + assert mock_post.call_count == 3 + assert mock_sleep.call_count == 2 \ No newline at end of file