> ## Documentation Index
> Fetch the complete documentation index at: https://docs.callintel.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Telegram Integration

> Send and receive Telegram messages from your agents. Integrate Telegram messaging capabilities into your CallIntel agents using the MCP (Model Context Protocol) server.

## 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

```bash theme={null}
# 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`:

```env theme={null}
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:

```yaml theme={null}
# 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

```bash theme={null}
docker-compose up -d tele_mcp
```

Verify service is running:

```bash theme={null}
curl py.callai.rejoicehub.com/health
```

## API Usage

### Send Message

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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:

```python theme={null}
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:

```python theme={null}
@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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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:

```sql theme={null}
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

```bash theme={null}
curl https://py.callai.rejoicehub.com/health
```

### Logs

```bash theme={null}
docker-compose logs -f tele_mcp
```

### Message Analytics

```python theme={null}
# 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:

```python theme={null}
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

<Columns cols={2}>
  <Card title="Gmail Integration" icon="envelope" href="/guides/integrations/gmail">
    Set up email capabilities.
  </Card>

  <Card title="API Reference" icon="terminal" href="/api-reference/introduction">
    Full API documentation.
  </Card>
</Columns>