Telegram Integration

Overview

The Telegram MCP service enables agents to:

Send messages to Telegram users
Receive and process incoming messages
Get chat information
Manage group conversations
Handle media files (documents, images, audio)

Setup

Prerequisites

Telegram account
Telegram Bot (created via BotFather)
API token from BotFather
Supabase database for persistence

1. Create a Telegram Bot

# Open Telegram and search for: @BotFather
# Send command: /newbot
# Follow prompts to create bot
# BotFather will give you API token

Example API token:

123456789:ABCDEfghIjklmnopqrstuvwxyzABCDEfgh

2. Configure Environment

Add to .env:

TELEGRAM_BOT_TOKEN=123456789:ABCDEfghIjklmnopqrstuvwxyzABCDEfgh
TELE_MCP_SERVER_URL=http://tele_mcp:8100

3. Docker Configuration

The Telegram MCP service runs in a separate Docker container:

# docker-compose.yml
tele_mcp:
  build:
    context: .
    dockerfile: docker/Dockerfile.tele
  image: tele_mcp_service:latest
  ports:
    - "8100:8100"
  environment:
    - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
    - SUPABASE_URL=${SUPABASE_URL}
    - SUPABASE_KEY=${SUPABASE_KEY}
  depends_on:
    - base
  networks:
    - call-agent-network
  restart: unless-stopped

4. Start Service

docker-compose up -d tele_mcp

Verify service is running:

curl py.callai.rejoicehub.com/health

API Usage

Send Message

from Service.McpService import McpService

mcp = McpService()

# Send message to chat
result = await mcp.execute_tool(
    service="telegram",
    tool="send_message",
    arguments={
        "chat_id": "123456789",
        "text": "Hello from CallIntel!",
        "parse_mode": "HTML",  # or "Markdown"
    }
)

print(f"Message sent: {result['message_id']}")

Send Message with Buttons

result = await mcp.execute_tool(
    service="telegram",
    tool="send_message",
    arguments={
        "chat_id": "123456789",
        "text": "Choose an option:",
        "reply_markup": {
            "inline_keyboard": [
                [
                    {"text": "Option 1", "callback_data": "opt_1"},
                    {"text": "Option 2", "callback_data": "opt_2"}
                ]
            ]
        }
    }
)

Read Messages

# Get recent messages from a chat
result = await mcp.execute_tool(
    service="telegram",
    tool="read_message",
    arguments={
        "chat_id": "123456789",
        "limit": 10  # Last 10 messages
    }
)

for message in result['messages']:
    print(f"{message['sender']}: {message['text']}")

Get Chat Information

# Get details about a chat
result = await mcp.execute_tool(
    service="telegram",
    tool="get_chat_info",
    arguments={
        "chat_id": "123456789"
    }
)

print(f"Chat name: {result['title']}")
print(f"Members: {result['member_count']}")
print(f"Type: {result['type']}")  # private, group, supergroup, channel

Send Document

# Send file to chat
result = await mcp.execute_tool(
    service="telegram",
    tool="send_document",
    arguments={
        "chat_id": "123456789",
        "file_path": "/path/to/file.pdf",
        "caption": "Here's your document"
    }
)

Send Photo

# Send image
result = await mcp.execute_tool(
    service="telegram",
    tool="send_photo",
    arguments={
        "chat_id": "123456789",
        "file_url": "https://example.com/image.jpg",
        "caption": "Check this out!"
    }
)

Agent Configuration

Enable Telegram Tools

Configure your agent to use Telegram:

agent = await agent_service.create_agent({
    "name": "Telegram Bot Agent",
    "system_prompt": """
    You are a helpful bot that communicates via Telegram.
    
    You can:
    - Send messages to users
    - Read incoming messages
    - Send documents and photos
    - Answer questions
    
    Always be friendly and helpful.
    """,
    "available_tools": [
        "telegram.send_message",
        "telegram.read_message",
        "telegram.send_document",
        "telegram.send_photo"
    ]
})

Webhook Handler

Set up a webhook to receive Telegram updates:

@app.post("/webhook/telegram")
async def handle_telegram_webhook(request: Request):
    data = await request.json()
    
    message = data.get('message', {})
    chat_id = message.get('chat', {}).get('id')
    text = message.get('text')
    
    if not text:
        return {"ok": True}
    
    # Process with agent
    response = await agent_service.process_message(
        agent_id="telegram-agent",
        message=text,
        context={
            "chat_id": chat_id,
            "user_id": message.get('from', {}).get('id'),
            "username": message.get('from', {}).get('username')
        }
    )
    
    # Send response back to Telegram
    await mcp.execute_tool(
        service="telegram",
        tool="send_message",
        arguments={
            "chat_id": chat_id,
            "text": response
        }
    )
    
    return {"ok": True}

Practical Examples

Customer Support Bot

async def setup_support_bot():
    agent = await agent_service.create_agent({
        "name": "Customer Support",
        "system_prompt": """
        You are a customer support specialist for TechCorp.
        
        Help customers with:
        - Product questions
        - Technical issues
        - Order status
        - Returns and refunds
        
        Be professional and empathetic.
        """,
        "ai_provider": "gemini",
        "available_tools": [
            "telegram.send_message",
            "telegram.send_document"
        ]
    })
    
    return agent

@app.post("/webhook/telegram/support")
async def support_webhook(request: Request):
    data = await request.json()
    message = data.get('message', {})
    
    # Forward to support agent
    await handle_message(
        agent_id="support-agent",
        chat_id=message['chat']['id'],
        text=message['text']
    )

Order Status Bot

async def check_order_status(chat_id: str, order_id: str):
    # Get order info
    order = await get_order(order_id)
    
    message = f"""
    📦 Order Status: {order['status']}
    
    Order ID: {order['id']}
    Date: {order['created_at']}
    Total: ${order['total']}
    
    Items:
    {format_items(order['items'])}
    
    Tracking: {order['tracking_url']}
    """
    
    # Send to customer
    await mcp.execute_tool(
        service="telegram",
        tool="send_message",
        arguments={
            "chat_id": chat_id,
            "text": message,
            "parse_mode": "Markdown"
        }
    )

Notification System

async def send_notification(user_id: str, notification: dict):
    message = f"""
    🔔 {notification['title']}
    
    {notification['message']}
    
    {notification.get('action_text', '')}
    """
    
    await mcp.execute_tool(
        service="telegram",
        tool="send_message",
        arguments={
            "chat_id": user_id,
            "text": message,
            "reply_markup": {
                "inline_keyboard": notification.get('buttons', [])
            }
        }
    )

Error Handling

Common Issues

try:
    result = await mcp.execute_tool(
        service="telegram",
        tool="send_message",
        arguments={
            "chat_id": "123456789",
            "text": "Hello"
        }
    )
except Exception as e:
    if "Invalid chat_id" in str(e):
        logger.error("Chat ID not found or bot doesn't have access")
    elif "Bot was blocked" in str(e):
        logger.error("User blocked the bot")
    elif "Chat not found" in str(e):
        logger.error("Chat doesn't exist")
    else:
        logger.error(f"Telegram error: {e}")

Retry Logic

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def send_message_with_retry(chat_id: str, text: str):
    return await mcp.execute_tool(
        service="telegram",
        tool="send_message",
        arguments={
            "chat_id": chat_id,
            "text": text
        }
    )

Database Schema

Telegram messages are stored in Supabase:

CREATE TABLE telegram_messages (
  id UUID PRIMARY KEY,
  chat_id BIGINT NOT NULL,
  message_id BIGINT NOT NULL,
  user_id BIGINT,
  text TEXT,
  media_type TEXT,
  media_url TEXT,
  timestamp TIMESTAMP DEFAULT now(),
  processed BOOLEAN DEFAULT false,
  created_at TIMESTAMP DEFAULT now()
);

CREATE TABLE telegram_chats (
  id UUID PRIMARY KEY,
  chat_id BIGINT UNIQUE NOT NULL,
  title TEXT,
  type TEXT,
  member_count INTEGER,
  created_at TIMESTAMP DEFAULT now(),
  updated_at TIMESTAMP DEFAULT now()
);

Monitoring

Health Check

curl https://py.callai.rejoicehub.com/health

Logs

docker-compose logs -f tele_mcp

Message Analytics

# Get message statistics
stats = await supabase.table("telegram_messages").select(
    "chat_id, count(*) as message_count"
).group_by("chat_id").execute()

for stat in stats.data:
    print(f"Chat {stat['chat_id']}: {stat['message_count']} messages")

Rate Limiting

Telegram has rate limits:

30 messages per second per chat
100 messages per second per bot

Implement rate limiting:

from aiolimiter import AsyncLimiter

# 30 messages per second
limiter = AsyncLimiter(max_rate=30, time_period=1)

async def send_message_limited(chat_id: str, text: str):
    async with limiter:
        return await mcp.execute_tool(
            service="telegram",
            tool="send_message",
            arguments={
                "chat_id": chat_id,
                "text": text
            }
        )

Best Practices

Handle Errors Gracefully - Bot may lose access to chats
Rate Limit - Don’t exceed Telegram’s rate limits
Validate Input - Sanitize user input before processing
Store Data - Save messages for audit trail
Async Operations - Use async/await for performance
Error Logging - Log all errors for debugging
User Consent - Get permission before sending messages
Message Format - Use HTML/Markdown for better formatting

​Overview

​Setup

​Prerequisites

​1. Create a Telegram Bot

​2. Configure Environment

​3. Docker Configuration

​4. Start Service

​API Usage

​Send Message

​Send Message with Buttons

​Read Messages

​Get Chat Information

​Send Document

​Send Photo

​Agent Configuration

​Enable Telegram Tools

​Webhook Handler

​Practical Examples

​Customer Support Bot

​Order Status Bot

​Notification System

​Error Handling

​Common Issues

​Retry Logic

​Database Schema

​Monitoring

​Health Check

​Logs

​Message Analytics

​Rate Limiting

​Best Practices

​Next Steps

Gmail Integration

API Reference