From ab7285189b8ddcf6a82cb19c91d19fc76d766ccd Mon Sep 17 00:00:00 2001 From: Siddharth Varshney Date: Wed, 4 Feb 2026 20:21:16 +0000 Subject: [PATCH] feat(tools): add Telegram Bot integration - Add telegram_send_message and telegram_send_document tools - Add credential spec for TELEGRAM_BOT_TOKEN - Add comprehensive tests (18 test cases) - Add README documentation with setup instructions --- .../aden_tools/credentials/integrations.py | 26 ++ tools/src/aden_tools/tools/__init__.py | 4 + .../aden_tools/tools/telegram_tool/README.md | 124 +++++++++ .../tools/telegram_tool/__init__.py | 9 + .../tools/telegram_tool/telegram_tool.py | 205 +++++++++++++++ tools/tests/tools/test_telegram_tool.py | 244 ++++++++++++++++++ 6 files changed, 612 insertions(+) create mode 100644 tools/src/aden_tools/tools/telegram_tool/README.md create mode 100644 tools/src/aden_tools/tools/telegram_tool/__init__.py create mode 100644 tools/src/aden_tools/tools/telegram_tool/telegram_tool.py create mode 100644 tools/tests/tools/test_telegram_tool.py diff --git a/tools/src/aden_tools/credentials/integrations.py b/tools/src/aden_tools/credentials/integrations.py index de0a0cbea4..8948b5a10c 100644 --- a/tools/src/aden_tools/credentials/integrations.py +++ b/tools/src/aden_tools/credentials/integrations.py @@ -91,4 +91,30 @@ credential_id="hubspot", credential_key="access_token", ), + "telegram": CredentialSpec( + env_var="TELEGRAM_BOT_TOKEN", + tools=[ + "telegram_send_message", + "telegram_send_document", + ], + required=True, + startup_required=False, + help_url="https://core.telegram.org/bots#creating-a-new-bot", + description="Telegram Bot API token", + # Auth method support + aden_supported=False, + direct_api_key_supported=True, + api_key_instructions="""To get a Telegram Bot token: +1. Open Telegram and search for @BotFather +2. Send /newbot and follow the prompts to create your bot +3. Choose a name and username for your bot +4. Copy the API token provided (looks like 123456789:ABCdefGHIjklMNOpqrsTUVwxyz) +5. Start a chat with your bot or add it to a group to get a chat_id""", + # Health check configuration + health_check_endpoint="https://api.telegram.org/bot{token}/getMe", + health_check_method="GET", + # Credential store mapping + credential_id="telegram", + credential_key="bot_token", + ), } diff --git a/tools/src/aden_tools/tools/__init__.py b/tools/src/aden_tools/tools/__init__.py index 01faebfc6b..fa65b0b2c0 100644 --- a/tools/src/aden_tools/tools/__init__.py +++ b/tools/src/aden_tools/tools/__init__.py @@ -41,6 +41,7 @@ from .github_tool import register_tools as register_github from .hubspot_tool import register_tools as register_hubspot from .pdf_read_tool import register_tools as register_pdf_read +from .telegram_tool import register_tools as register_telegram from .web_scrape_tool import register_tools as register_web_scrape from .web_search_tool import register_tools as register_web_search @@ -72,6 +73,7 @@ def register_all_tools( # email supports multiple providers (Resend) with auto-detection register_email(mcp, credentials=credentials) register_hubspot(mcp, credentials=credentials) + register_telegram(mcp, credentials=credentials) # Register file system toolkits register_view_file(mcp) @@ -132,6 +134,8 @@ def register_all_tools( "hubspot_get_deal", "hubspot_create_deal", "hubspot_update_deal", + "telegram_send_message", + "telegram_send_document", ] diff --git a/tools/src/aden_tools/tools/telegram_tool/README.md b/tools/src/aden_tools/tools/telegram_tool/README.md new file mode 100644 index 0000000000..20520b3235 --- /dev/null +++ b/tools/src/aden_tools/tools/telegram_tool/README.md @@ -0,0 +1,124 @@ +# Telegram Bot Tool + +Send messages and documents to Telegram chats using the Bot API. + +## Features + +- **telegram_send_message** - Send text messages to users, groups, or channels +- **telegram_send_document** - Send documents/files to chats + +## Setup + +### 1. Create a Telegram Bot + +1. Open Telegram and search for [@BotFather](https://t.me/BotFather) +2. Send `/newbot` and follow the prompts +3. Choose a name and username for your bot +4. Copy the API token provided (looks like `123456789:ABCdefGHIjklMNOpqrsTUVwxyz`) + +### 2. Configure the Token + +Set the environment variable: + +```bash +export TELEGRAM_BOT_TOKEN="your-bot-token-here" +``` + +Or configure via the Hive credential store. + +### 3. Get Your Chat ID + +To send messages, you need the chat ID: + +1. Start a conversation with your bot +2. Send any message to the bot +3. Visit: `https://api.telegram.org/bot/getUpdates` +4. Find the `chat.id` in the response + +For groups: Add the bot to the group, then check getUpdates. + +## Usage Examples + +### Send a Message + +```python +telegram_send_message( + chat_id="123456789", + text="Hello from Hive! 🚀", + parse_mode="HTML" +) +``` + +### Send with Formatting + +```python +# HTML formatting +telegram_send_message( + chat_id="123456789", + text="Alert: Task completed successfully!", + parse_mode="HTML" +) + +# Markdown formatting +telegram_send_message( + chat_id="123456789", + text="*Bold* and _italic_ text", + parse_mode="Markdown" +) +``` + +### Send a Document + +```python +telegram_send_document( + chat_id="123456789", + document="https://example.com/report.pdf", + caption="Weekly Report" +) +``` + +### Silent Notification + +```python +telegram_send_message( + chat_id="123456789", + text="Background update completed", + disable_notification=True +) +``` + +## API Reference + +### telegram_send_message + +| Parameter | Type | Required | Description | +|-----------|------|----------|-------------| +| chat_id | str | Yes | Target chat ID or @username | +| text | str | Yes | Message text (1-4096 chars) | +| parse_mode | str | No | "HTML" or "Markdown" | +| disable_notification | bool | No | Send silently | + +### telegram_send_document + +| Parameter | Type | Required | Description | +|-----------|------|----------|-------------| +| chat_id | str | Yes | Target chat ID or @username | +| document | str | Yes | URL or file_id of document | +| caption | str | No | Caption (0-1024 chars) | +| parse_mode | str | No | Format for caption | + +## Error Handling + +The tools return error dictionaries on failure: + +```python +{"error": "Invalid Telegram bot token"} +{"error": "Chat not found"} +{"error": "Bot was blocked by the user or lacks permissions"} +{"error": "Rate limit exceeded. Try again later."} +``` + +## References + +- [Telegram Bot API Documentation](https://core.telegram.org/bots/api) +- [BotFather](https://t.me/BotFather) diff --git a/tools/src/aden_tools/tools/telegram_tool/__init__.py b/tools/src/aden_tools/tools/telegram_tool/__init__.py new file mode 100644 index 0000000000..3209027f35 --- /dev/null +++ b/tools/src/aden_tools/tools/telegram_tool/__init__.py @@ -0,0 +1,9 @@ +""" +Telegram Bot Tool - Send messages and documents via Telegram Bot API. + +Supports Bot API tokens for authentication. +""" + +from .telegram_tool import register_tools + +__all__ = ["register_tools"] diff --git a/tools/src/aden_tools/tools/telegram_tool/telegram_tool.py b/tools/src/aden_tools/tools/telegram_tool/telegram_tool.py new file mode 100644 index 0000000000..cc56b932fb --- /dev/null +++ b/tools/src/aden_tools/tools/telegram_tool/telegram_tool.py @@ -0,0 +1,205 @@ +""" +Telegram Bot Tool - Send messages and documents via Telegram Bot API. + +Supports: +- Bot API tokens (TELEGRAM_BOT_TOKEN) + +API Reference: https://core.telegram.org/bots/api +""" + +from __future__ import annotations + +import os +from typing import TYPE_CHECKING, Any + +import httpx +from fastmcp import FastMCP + +if TYPE_CHECKING: + from aden_tools.credentials import CredentialStoreAdapter + +TELEGRAM_API_BASE = "https://api.telegram.org/bot" + + +class _TelegramClient: + """Internal client wrapping Telegram Bot API calls.""" + + def __init__(self, bot_token: str): + self._token = bot_token + + @property + def _base_url(self) -> str: + return f"{TELEGRAM_API_BASE}{self._token}" + + def _handle_response(self, response: httpx.Response) -> dict[str, Any]: + """Handle common HTTP error codes.""" + if response.status_code == 401: + return {"error": "Invalid Telegram bot token"} + if response.status_code == 400: + try: + detail = response.json().get("description", response.text) + except Exception: + detail = response.text + return {"error": f"Bad request: {detail}"} + if response.status_code == 403: + return {"error": "Bot was blocked by the user or lacks permissions"} + if response.status_code == 404: + return {"error": "Chat not found"} + if response.status_code == 429: + return {"error": "Rate limit exceeded. Try again later."} + if response.status_code >= 400: + try: + detail = response.json().get("description", response.text) + except Exception: + detail = response.text + return {"error": f"Telegram API error (HTTP {response.status_code}): {detail}"} + return response.json() + + def send_message( + self, + chat_id: str, + text: str, + parse_mode: str | None = None, + disable_notification: bool = False, + ) -> dict[str, Any]: + """Send a text message to a chat.""" + payload: dict[str, Any] = { + "chat_id": chat_id, + "text": text, + "disable_notification": disable_notification, + } + if parse_mode: + payload["parse_mode"] = parse_mode + + response = httpx.post( + f"{self._base_url}/sendMessage", + json=payload, + timeout=30.0, + ) + return self._handle_response(response) + + def send_document( + self, + chat_id: str, + document: str, + caption: str | None = None, + parse_mode: str | None = None, + ) -> dict[str, Any]: + """Send a document to a chat.""" + payload: dict[str, Any] = { + "chat_id": chat_id, + "document": document, + } + if caption: + payload["caption"] = caption + if parse_mode: + payload["parse_mode"] = parse_mode + + response = httpx.post( + f"{self._base_url}/sendDocument", + json=payload, + timeout=30.0, + ) + return self._handle_response(response) + + def get_me(self) -> dict[str, Any]: + """Get bot information (useful for health checks).""" + response = httpx.get( + f"{self._base_url}/getMe", + timeout=30.0, + ) + return self._handle_response(response) + + +def register_tools( + mcp: FastMCP, + credentials: CredentialStoreAdapter | None = None, +) -> None: + """Register Telegram tools with the MCP server.""" + + def _get_token() -> str | None: + """Get Telegram bot token from credential manager or environment.""" + if credentials is not None: + token = credentials.get("telegram") + if token is not None and not isinstance(token, str): + raise TypeError( + f"Expected string from credentials.get('telegram'), got {type(token).__name__}" + ) + return token + return os.getenv("TELEGRAM_BOT_TOKEN") + + def _get_client() -> _TelegramClient | dict[str, str]: + """Get a Telegram client, or return an error dict if no credentials.""" + token = _get_token() + if not token: + return { + "error": "Telegram bot token not configured. " + "Set TELEGRAM_BOT_TOKEN environment variable or configure via credential store." + } + return _TelegramClient(token) + + @mcp.tool() + def telegram_send_message( + chat_id: str, + text: str, + parse_mode: str = "", + disable_notification: bool = False, + ) -> dict[str, Any]: + """ + Send a message to a Telegram chat. + + Use this to send notifications, alerts, or updates to a Telegram user or group. + + Args: + chat_id: Target chat ID (numeric) or @username for public channels + text: Message text (1-4096 characters). Supports HTML/Markdown if parse_mode set. + parse_mode: Optional format mode - "HTML" or "Markdown". Empty for plain text. + disable_notification: If True, sends message silently. + + Returns: + Dict with message info on success, or error dict on failure. + Success includes: message_id, chat info, date, text. + """ + client = _get_client() + if isinstance(client, dict): + return client + + return client.send_message( + chat_id=chat_id, + text=text, + parse_mode=parse_mode if parse_mode else None, + disable_notification=disable_notification, + ) + + @mcp.tool() + def telegram_send_document( + chat_id: str, + document: str, + caption: str = "", + parse_mode: str = "", + ) -> dict[str, Any]: + """ + Send a document to a Telegram chat. + + Use this to send files like PDFs, CSVs, or other documents. + + Args: + chat_id: Target chat ID (numeric) or @username for public channels + document: URL of the document to send, or file_id of existing file on Telegram + caption: Optional caption for the document (0-1024 characters) + parse_mode: Optional format mode for caption - "HTML" or "Markdown" + + Returns: + Dict with message info on success, or error dict on failure. + Success includes: message_id, document info, chat info. + """ + client = _get_client() + if isinstance(client, dict): + return client + + return client.send_document( + chat_id=chat_id, + document=document, + caption=caption if caption else None, + parse_mode=parse_mode if parse_mode else None, + ) diff --git a/tools/tests/tools/test_telegram_tool.py b/tools/tests/tools/test_telegram_tool.py new file mode 100644 index 0000000000..ee77420a9c --- /dev/null +++ b/tools/tests/tools/test_telegram_tool.py @@ -0,0 +1,244 @@ +""" +Tests for Telegram Bot tool. + +Covers: +- _TelegramClient methods (send_message, send_document, get_me) +- Error handling (API errors, invalid token, rate limiting) +- Credential retrieval (CredentialStoreAdapter vs env var) +- MCP tool functions (telegram_send_message, telegram_send_document) +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from fastmcp import FastMCP + +from aden_tools.tools.telegram_tool.telegram_tool import ( + _TelegramClient, + register_tools, +) + +# --- _TelegramClient tests --- + + +class TestTelegramClient: + def setup_method(self): + self.client = _TelegramClient("123456789:ABCdefGHIjklMNOpqrsTUVwxyz") + + def test_base_url(self): + assert "123456789:ABCdefGHIjklMNOpqrsTUVwxyz" in self.client._base_url + assert self.client._base_url.startswith("https://api.telegram.org/bot") + + def test_handle_response_success(self): + response = MagicMock() + response.status_code = 200 + response.json.return_value = {"ok": True, "result": {"message_id": 123}} + result = self.client._handle_response(response) + assert result["ok"] is True + assert result["result"]["message_id"] == 123 + + def test_handle_response_401(self): + response = MagicMock() + response.status_code = 401 + result = self.client._handle_response(response) + assert "error" in result + assert "Invalid" in result["error"] + + def test_handle_response_400(self): + response = MagicMock() + response.status_code = 400 + response.json.return_value = {"description": "Bad Request: chat not found"} + result = self.client._handle_response(response) + assert "error" in result + assert "Bad request" in result["error"] + + def test_handle_response_403(self): + response = MagicMock() + response.status_code = 403 + result = self.client._handle_response(response) + assert "error" in result + assert "blocked" in result["error"] + + def test_handle_response_404(self): + response = MagicMock() + response.status_code = 404 + result = self.client._handle_response(response) + assert "error" in result + assert "not found" in result["error"] + + def test_handle_response_429(self): + response = MagicMock() + response.status_code = 429 + result = self.client._handle_response(response) + assert "error" in result + assert "Rate limit" in result["error"] + + @patch("aden_tools.tools.telegram_tool.telegram_tool.httpx.post") + def test_send_message(self, mock_post): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "ok": True, + "result": {"message_id": 456, "text": "Hello"}, + } + mock_post.return_value = mock_response + + result = self.client.send_message(chat_id="123", text="Hello") + + mock_post.assert_called_once() + assert result["ok"] is True + assert result["result"]["message_id"] == 456 + + @patch("aden_tools.tools.telegram_tool.telegram_tool.httpx.post") + def test_send_message_with_parse_mode(self, mock_post): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"ok": True, "result": {}} + mock_post.return_value = mock_response + + self.client.send_message(chat_id="123", text="Bold", parse_mode="HTML") + + call_kwargs = mock_post.call_args.kwargs + assert call_kwargs["json"]["parse_mode"] == "HTML" + + @patch("aden_tools.tools.telegram_tool.telegram_tool.httpx.post") + def test_send_document(self, mock_post): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "ok": True, + "result": {"message_id": 789, "document": {"file_id": "abc123"}}, + } + mock_post.return_value = mock_response + + result = self.client.send_document( + chat_id="123", + document="https://example.com/file.pdf", + caption="Test doc", + ) + + mock_post.assert_called_once() + assert result["ok"] is True + assert result["result"]["message_id"] == 789 + + @patch("aden_tools.tools.telegram_tool.telegram_tool.httpx.get") + def test_get_me(self, mock_get): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "ok": True, + "result": {"id": 123, "is_bot": True, "username": "test_bot"}, + } + mock_get.return_value = mock_response + + result = self.client.get_me() + + mock_get.assert_called_once() + assert result["ok"] is True + assert result["result"]["is_bot"] is True + + +# --- register_tools tests --- + + +class TestRegisterTools: + def setup_method(self): + self.mcp = FastMCP("test-telegram") + + def test_register_tools_creates_tools(self): + register_tools(self.mcp) + + # Check that tools are registered + tool_names = [tool.name for tool in self.mcp._tool_manager._tools.values()] + assert "telegram_send_message" in tool_names + assert "telegram_send_document" in tool_names + + @patch.dict("os.environ", {"TELEGRAM_BOT_TOKEN": ""}, clear=False) + def test_send_message_no_token_error(self): + register_tools(self.mcp, credentials=None) + + # Get the registered tool + tools = {t.name: t for t in self.mcp._tool_manager._tools.values()} + send_message = tools["telegram_send_message"] + + # Call with no token configured + with patch("os.getenv", return_value=None): + result = send_message.fn(chat_id="123", text="test") + + assert "error" in result + assert "not configured" in result["error"] + + @patch("aden_tools.tools.telegram_tool.telegram_tool.httpx.post") + @patch("os.getenv", return_value="test_token") + def test_send_message_success(self, mock_getenv, mock_post): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"ok": True, "result": {"message_id": 1}} + mock_post.return_value = mock_response + + register_tools(self.mcp, credentials=None) + tools = {t.name: t for t in self.mcp._tool_manager._tools.values()} + send_message = tools["telegram_send_message"] + + result = send_message.fn(chat_id="123", text="Hello!") + + assert result["ok"] is True + + def test_credentials_adapter_used(self): + mock_credentials = MagicMock() + mock_credentials.get.return_value = "token_from_store" + + register_tools(self.mcp, credentials=mock_credentials) + tools = {t.name: t for t in self.mcp._tool_manager._tools.values()} + + # The credentials should be used when tools are called + with patch("aden_tools.tools.telegram_tool.telegram_tool.httpx.post") as mock_post: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"ok": True, "result": {}} + mock_post.return_value = mock_response + + tools["telegram_send_message"].fn(chat_id="123", text="test") + + # Verify the token from credentials was used + call_url = mock_post.call_args.args[0] + assert "token_from_store" in call_url + + +# --- Error handling tests --- + + +class TestErrorHandling: + def setup_method(self): + self.client = _TelegramClient("test_token") + + @patch("aden_tools.tools.telegram_tool.telegram_tool.httpx.post") + def test_network_error(self, mock_post): + import httpx + + mock_post.side_effect = httpx.ConnectError("Connection failed") + + with pytest.raises(httpx.ConnectError): + self.client.send_message(chat_id="123", text="test") + + @patch("aden_tools.tools.telegram_tool.telegram_tool.httpx.post") + def test_timeout_error(self, mock_post): + import httpx + + mock_post.side_effect = httpx.TimeoutException("Request timed out") + + with pytest.raises(httpx.TimeoutException): + self.client.send_message(chat_id="123", text="test") + + def test_handle_response_generic_error(self): + response = MagicMock() + response.status_code = 500 + response.json.return_value = {"description": "Internal server error"} + response.text = "Internal server error" + + result = self.client._handle_response(response) + + assert "error" in result + assert "500" in result["error"]