Skip to content

feat(tools): add Telegram Bot integration #3550

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Siddharth2624 wants to merge 1 commit into
base: main
Choose a base branch
from
+612 −0
Open

feat(tools): add Telegram Bot integration #3550

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions tools/src/aden_tools/credentials/integrations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
),
}
4 changes: 4 additions & 0 deletions tools/src/aden_tools/tools/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -132,6 +134,8 @@ def register_all_tools(
"hubspot_get_deal",
"hubspot_create_deal",
"hubspot_update_deal",
"telegram_send_message",
"telegram_send_document",
]


Expand Down
124 changes: 124 additions & 0 deletions tools/src/aden_tools/tools/telegram_tool/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<YOUR_TOKEN>/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="<b>Alert:</b> 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)
9 changes: 9 additions & 0 deletions tools/src/aden_tools/tools/telegram_tool/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"]
205 changes: 205 additions & 0 deletions tools/src/aden_tools/tools/telegram_tool/telegram_tool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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,
)
Loading