feat: tools frontend and endpoints refactor

application/tools/implementations/cryptoprice.py

@@ -0,0 +1,77 @@
+import requests
+from application.tools.base import Tool
+
+
+class CryptoPriceTool(Tool):
+    """
+    CryptoPrice
+    A tool for retrieving cryptocurrency prices using the CryptoCompare public API
+    """
+
+    def __init__(self, config):
+        self.config = config
+
+    def execute_action(self, action_name, **kwargs):
+        actions = {"cryptoprice_get": self._get_price}
+
+        if action_name in actions:
+            return actions[action_name](**kwargs)
+        else:
+            raise ValueError(f"Unknown action: {action_name}")
+
+    def _get_price(self, symbol, currency):
+        """
+        Fetches the current price of a given cryptocurrency symbol in the specified currency.
+        Example:
+            symbol = "BTC"
+            currency = "USD"
+            returns price in USD.
+        """
+        url = f"https://min-api.cryptocompare.com/data/price?fsym={symbol.upper()}&tsyms={currency.upper()}"
+        response = requests.get(url)
+        if response.status_code == 200:
+            data = response.json()
+            # data will be like {"USD": <price>} if the call is successful
+            if currency.upper() in data:
+                return {
+                    "status_code": response.status_code,
+                    "price": data[currency.upper()],
+                    "message": f"Price of {symbol.upper()} in {currency.upper()} retrieved successfully.",
+                }
+            else:
+                return {
+                    "status_code": response.status_code,
+                    "message": f"Couldn't find price for {symbol.upper()} in {currency.upper()}.",
+                }
+        else:
+            return {
+                "status_code": response.status_code,
+                "message": "Failed to retrieve price.",
+            }
+
+    def get_actions_metadata(self):
+        return [
+            {
+                "name": "cryptoprice_get",
+                "description": "Retrieve the price of a specified cryptocurrency in a given currency",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "symbol": {
+                            "type": "string",
+                            "description": "The cryptocurrency symbol (e.g. BTC)",
+                        },
+                        "currency": {
+                            "type": "string",
+                            "description": "The currency in which you want the price (e.g. USD)",
+                        },
+                    },
+                    "required": ["symbol", "currency"],
+                    "additionalProperties": False,
+                },
+            }
+        ]
+
+    def get_config_requirements(self):
+        # No specific configuration needed for this tool as it just queries a public endpoint
+        return {}

application/tools/implementations/telegram.py

@@ -0,0 +1,83 @@
+import requests
+from application.tools.base import Tool
+
+
+class TelegramTool(Tool):
+    """
+    Telegram Bot
+    A flexible Telegram tool for performing various actions (e.g., sending messages, images).
+    Requires a bot token and chat ID for configuration
+    """
+
+    def __init__(self, config):
+        self.config = config
+        self.token = config.get("token", "")
+        self.chat_id = config.get("chat_id", "")
+
+    def execute_action(self, action_name, **kwargs):
+        actions = {
+            "telegram_send_message": self._send_message,
+            "telegram_send_image": self._send_image,
+        }
+
+        if action_name in actions:
+            return actions[action_name](**kwargs)
+        else:
+            raise ValueError(f"Unknown action: {action_name}")
+
+    def _send_message(self, text):
+        print(f"Sending message: {text}")
+        url = f"https://api.telegram.org/bot{self.token}/sendMessage"
+        payload = {"chat_id": self.chat_id, "text": text}
+        response = requests.post(url, data=payload)
+        return {"status_code": response.status_code, "message": "Message sent"}
+
+    def _send_image(self, image_url):
+        print(f"Sending image: {image_url}")
+        url = f"https://api.telegram.org/bot{self.token}/sendPhoto"
+        payload = {"chat_id": self.chat_id, "photo": image_url}
+        response = requests.post(url, data=payload)
+        return {"status_code": response.status_code, "message": "Image sent"}
+
+    def get_actions_metadata(self):
+        return [
+            {
+                "name": "telegram_send_message",
+                "description": "Send a notification to telegram chat",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "text": {
+                            "type": "string",
+                            "description": "Text to send in the notification",
+                        }
+                    },
+                    "required": ["text"],
+                    "additionalProperties": False,
+                },
+            },
+            {
+                "name": "telegram_send_image",
+                "description": "Send an image to the Telegram chat",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "image_url": {
+                            "type": "string",
+                            "description": "URL of the image to send",
+                        }
+                    },
+                    "required": ["image_url"],
+                    "additionalProperties": False,
+                },
+            },
+        ]
+
+    def get_config_requirements(self):
+        return {
+            "chat_id": {
+                "type": "string",
+                "description": "Telegram chat ID to send messages to",
+            },
+            "token": {"type": "string", "description": "Bot token for authentication"},
+        }