wehub-resource-sync
fed8b2eed7
Backend release / release (push) Waiting to run
Bandit Security Scan / bandit_scan (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / manifest (push) Blocked by required conditions
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / manifest (push) Blocked by required conditions
Python linting / ruff (push) Waiting to run
Run python tests with pytest / Run tests and count coverage (3.12) (push) Waiting to run
React Widget Build / build (push) Waiting to run
103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
import logging
|
|
|
|
import requests
|
|
|
|
from application.agents.tools.base import Tool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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", "")
|
|
|
|
def execute_action(self, action_name, **kwargs):
|
|
actions = {
|
|
"telegram_send_message": self._send_message,
|
|
"telegram_send_image": self._send_image,
|
|
}
|
|
if action_name not in actions:
|
|
raise ValueError(f"Unknown action: {action_name}")
|
|
return actions[action_name](**kwargs)
|
|
|
|
def _send_message(self, text, chat_id):
|
|
logger.debug("Sending Telegram message to chat_id=%s", chat_id)
|
|
url = f"https://api.telegram.org/bot{self.token}/sendMessage"
|
|
payload = {"chat_id": chat_id, "text": text}
|
|
response = requests.post(url, data=payload, timeout=100)
|
|
return {"status_code": response.status_code, "message": "Message sent"}
|
|
|
|
def _send_image(self, image_url, chat_id):
|
|
logger.debug("Sending Telegram image to chat_id=%s", chat_id)
|
|
url = f"https://api.telegram.org/bot{self.token}/sendPhoto"
|
|
payload = {"chat_id": chat_id, "photo": image_url}
|
|
response = requests.post(url, data=payload, timeout=100)
|
|
return {"status_code": response.status_code, "message": "Image sent"}
|
|
|
|
def get_actions_metadata(self):
|
|
return [
|
|
{
|
|
"name": "telegram_send_message",
|
|
"description": (
|
|
"Send a text message to the configured Telegram chat via "
|
|
"the bot. Compose the final message text before sending."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"text": {
|
|
"type": "string",
|
|
"description": "Text to send in the notification",
|
|
},
|
|
"chat_id": {
|
|
"type": "string",
|
|
"description": "Chat ID to send the notification to",
|
|
},
|
|
},
|
|
"required": ["text"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
{
|
|
"name": "telegram_send_image",
|
|
"description": (
|
|
"Send an image to the configured Telegram chat. Requires "
|
|
"a publicly accessible image URL."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"image_url": {
|
|
"type": "string",
|
|
"description": "URL of the image to send",
|
|
},
|
|
"chat_id": {
|
|
"type": "string",
|
|
"description": "Chat ID to send the image to",
|
|
},
|
|
},
|
|
"required": ["image_url"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
]
|
|
|
|
def get_config_requirements(self):
|
|
return {
|
|
"token": {
|
|
"type": "string",
|
|
"label": "Bot Token",
|
|
"description": "Telegram bot token for authentication",
|
|
"required": True,
|
|
"secret": True,
|
|
"order": 1,
|
|
},
|
|
}
|