Skip to main content

​
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 http://localhost:8100/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 http://localhost:8100/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

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

​
Next Steps

Gmail Integration

Set up email capabilities.

API Reference

Full API documentation.