feat: ntfy tool

Author: dartpain

application/agents/tools/ntfy.py

@@ -0,0 +1,127 @@
+import requests
+from application.agents.tools.base import Tool
+
+class NtfyTool(Tool):
+    """
+    Ntfy Tool
+    A tool for sending notifications to ntfy topics on a specified server.
+    """
+
+    def __init__(self, config):
+        """
+        Initialize the NtfyTool with configuration.
+
+        Args:
+            config (dict): Configuration dictionary containing the access token.
+        """
+        self.config = config
+        self.token = config.get("token", "")
+
+    def execute_action(self, action_name, **kwargs):
+        """
+        Execute the specified action with given parameters.
+
+        Args:
+            action_name (str): Name of the action to execute.
+            **kwargs: Parameters for the action, including server_url.
+
+        Returns:
+            dict: Result of the action with status code and message.
+
+        Raises:
+            ValueError: If the action name is unknown.
+        """
+        actions = {
+            "ntfy_send_message": self._send_message,
+        }
+        if action_name in actions:
+            return actions[action_name](**kwargs)
+        else:
+            raise ValueError(f"Unknown action: {action_name}")
+
+    def _send_message(self, server_url, message, topic, title=None, priority=None):
+        """
+        Send a message to an ntfy topic on the specified server.
+
+        Args:
+            server_url (str): Base URL of the ntfy server (e.g., https://ntfy.sh).
+            message (str): The message text to send.
+            topic (str): The topic to send the message to.
+            title (str, optional): Title of the notification.
+            priority (int, optional): Priority of the notification (1-5).
+
+        Returns:
+            dict: Response with status code and a confirmation message.
+
+        Raises:
+            ValueError: If priority is not an integer between 1 and 5.
+        """
+        url = f"{server_url.rstrip('/')}/{topic}"
+        headers = {}
+        if title:
+            headers["X-Title"] = title
+        if priority:
+            try:
+                priority = int(priority)
+            except (ValueError, TypeError):
+                raise ValueError("Priority must be convertible to an integer")
+            if priority < 1 or priority > 5:
+                raise ValueError("Priority must be an integer between 1 and 5")
+            headers["X-Priority"] = str(priority)
+        if self.token:
+            headers["Authorization"] = f"Basic {self.token}"
+        data = message.encode("utf-8")
+        response = requests.post(url, headers=headers, data=data)
+        return {"status_code": response.status_code, "message": "Message sent"}
+
+    def get_actions_metadata(self):
+        """
+        Provide metadata about available actions.
+
+        Returns:
+            list: List of dictionaries describing each action.
+        """
+        return [
+            {
+                "name": "ntfy_send_message",
+                "description": "Send a notification to an ntfy topic",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "server_url": {
+                            "type": "string",
+                            "description": "Base URL of the ntfy server",
+                        },
+                        "message": {
+                            "type": "string",
+                            "description": "Text to send in the notification",
+                        },
+                        "topic": {
+                            "type": "string",
+                            "description": "Topic to send the notification to",
+                        },
+                        "title": {
+                            "type": "string",
+                            "description": "Title of the notification (optional)",
+                        },
+                        "priority": {
+                            "type": "integer",
+                            "description": "Priority of the notification (1-5, optional)",
+                        },
+                    },
+                    "required": ["server_url", "message", "topic"],
+                    "additionalProperties": False,
+                },
+            },
+        ]
+
+    def get_config_requirements(self):
+        """
+        Specify the configuration requirements.
+
+        Returns:
+            dict: Dictionary describing required config parameters.
+        """
+        return {
+            "token": {"type": "string", "description": "Access token for authentication"},
+        }