Telegram integration transforms OpenClaw from a background automation system into a mobile command center. Your AI agents can send briefings, take commands, and provide real-time updates directly to your phone.
This guide walks you through the complete setup process, from creating your first Telegram bot to building sophisticated notification workflows that keep you connected to your automated systems.
Why Integrate OpenClaw with Telegram?
Traditional automation systems run silently in the background. You configure them once and hope they work. Telegram integration adds a communication layer that makes your AI agents truly interactive:
Mobile command center: Send commands to your OpenClaw agents from anywhere Real-time notifications: Get instant alerts when important events occur Status monitoring: Check agent health and performance on-the-go Interactive workflows: Approve actions or provide input for complex decisions Unified interface: Control multiple OpenClaw instances from one chat interface
Prerequisites and Setup Requirements
Before starting, ensure you have:
- Active OpenClaw installation (see our complete OpenClaw tutorial)
- Telegram account on mobile or desktop
- Python 3.9+ with pip access
- Internet-accessible server for webhook endpoints (optional for polling mode)
- Basic command line familiarity
Required Python Dependencies
# Install Telegram bot dependencies
pip install python-telegram-bot==20.7
pip install aiohttp==3.9.1
pip install python-dotenv==1.0.0
Step 1: Create Your Telegram Bot
Register with BotFather
Every Telegram bot starts with BotFather, Telegram's bot creation service:
- Open Telegram and search for
@BotFather - Start a conversation and send
/start - Create a new bot with
/newbot - Choose a display name (e.g., "OpenClaw Assistant")
- Choose a username (must end in 'bot', e.g., "openclawassistant_bot")
BotFather will provide your bot token. Save this securely — it's your authentication key.
Use this token to access the HTTP API:
1234567890:ABCdefGhIjKlMnOpQrStUvWxYz
Configure Bot Settings
Configure additional bot properties through BotFather:
/setdescription - Set bot description
/setcommands - Configure command menu
/setprivacy - Disable privacy mode (allows group usage)
Example command setup:
status - Check agent status
brief - Get morning brief
help - Show available commands
settings - Configure preferences
Set Up Bot Privacy
For group usage, disable privacy mode:
/setprivacy
[Select your bot]
Disable
This allows your bot to read all group messages, not just direct commands.
Step 2: Configure OpenClaw Integration
Add Telegram Configuration
Update your OpenClaw config.yaml to include Telegram integration:
# ~/.openclaw/config.yaml
integrations:
telegram:
enabled: true
bot_token: "YOUR_BOT_TOKEN_HERE"
webhook_url: "https://your-domain.com/telegram/webhook" # Optional
polling_interval: 2 # seconds
allowed_users:
- your_telegram_user_id # Get from @userinfobot
notification_channels:
alerts: "your_telegram_chat_id"
briefings: "your_telegram_chat_id"
status: "your_telegram_chat_id"
# Core agent configuration
agents:
default:
name: "TelegramAgent"
model: "gpt-4"
temperature: 0.1
integrations:
- telegram
- gmail
- calendar
Environment Variables
Create or update your .env file:
# Telegram Configuration
TELEGRAM_BOT_TOKEN=1234567890:ABCdefGhIjKlMnOpQrStUvWxYz
TELEGRAM_WEBHOOK_SECRET=your_webhook_secret_key
# Your Telegram user ID (get from @userinfobot)
TELEGRAM_USER_ID=123456789
# OpenAI (for agent responses)
OPENAI_API_KEY=your_openai_api_key
Get Your Telegram User ID
To restrict bot access to authorized users:
- Message @userinfobot on Telegram
- Copy your user ID from the response
- Add to
allowed_usersin your config
Step 3: Implement Telegram Integration
Basic Telegram Integration Class
Create ~/.openclaw/integrations/telegram_integration.py:
import os
import asyncio
import logging
from telegram import Update, Bot
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
from openclaw.core import Integration, Agent
class TelegramIntegration(Integration):
"""Telegram bot integration for OpenClaw"""
def __init__(self, agent: Agent, config: dict):
super().__init__(agent, config)
self.bot_token = config.get('bot_token') or os.getenv('TELEGRAM_BOT_TOKEN')
self.allowed_users = config.get('allowed_users', [])
self.application = None
self.bot = None
async def initialize(self):
"""Initialize Telegram bot application"""
self.application = Application.builder().token(self.bot_token).build()
self.bot = self.application.bot
# Register command handlers
self.application.add_handler(CommandHandler("start", self._handle_start))
self.application.add_handler(CommandHandler("status", self._handle_status))
self.application.add_handler(CommandHandler("brief", self._handle_brief))
self.application.add_handler(CommandHandler("help", self._handle_help))
self.application.add_handler(CommandHandler("run", self._handle_run_skill))
# Handle text messages
self.application.add_handler(
MessageHandler(filters.TEXT & ~filters.COMMAND, self._handle_message)
)
# Start the bot
await self.application.initialize()
await self.application.start()
# Use webhook or polling based on configuration
if self.config.get('webhook_url'):
await self._setup_webhook()
else:
await self._start_polling()
logging.info("Telegram integration initialized successfully")
async def _setup_webhook(self):
"""Configure webhook for production deployment"""
webhook_url = self.config['webhook_url']
await self.bot.set_webhook(url=webhook_url)
logging.info(f"Webhook set to {webhook_url}")
async def _start_polling(self):
"""Start polling for development/testing"""
await self.application.updater.start_polling()
logging.info("Started polling for Telegram updates")
def _is_authorized(self, user_id: int) -> bool:
"""Check if user is authorized to use the bot"""
return not self.allowed_users or user_id in self.allowed_users
async def _handle_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /start command"""
user_id = update.effective_user.id
if not self._is_authorized(user_id):
await update.message.reply_text("❌ Unauthorized access")
return
welcome_text = """
🤖 **OpenClaw Agent Online**
Your AI automation system is ready. Available commands:
/status - Check agent and skill status
/brief - Get your morning briefing
/run [skill_name] - Execute a specific skill
/help - Show this help message
You can also send natural language messages and I'll try to help!
"""
await update.message.reply_text(welcome_text, parse_mode='Markdown')
async def _handle_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /status command"""
if not self._is_authorized(update.effective_user.id):
return
try:
# Get agent status
agent_status = await self.agent.get_status()
skill_status = await self.agent.get_skill_status()
status_text = f"""
📊 **Agent Status**
🤖 Agent: {agent_status['name']} ({agent_status['status']})
⏰ Uptime: {agent_status['uptime']}
🧠 Memory: {agent_status['memory_usage']}
📋 **Active Skills**
{self._format_skills_status(skill_status)}
📈 **Recent Activity**
• {agent_status['recent_activity']}
"""
await update.message.reply_text(status_text, parse_mode='Markdown')
except Exception as e:
await update.message.reply_text(f"❌ Error getting status: {str(e)}")
async def _handle_brief(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /brief command"""
if not self._is_authorized(update.effective_user.id):
return
try:
# Generate morning brief using existing skills
brief = await self.agent.run_skill("morning_brief")
if brief['status'] == 'success':
await update.message.reply_text(brief['content'], parse_mode='Markdown')
else:
await update.message.reply_text(f"❌ Error generating brief: {brief['message']}")
except Exception as e:
await update.message.reply_text(f"❌ Error generating brief: {str(e)}")
async def _handle_run_skill(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /run [skill_name] command"""
if not self._is_authorized(update.effective_user.id):
return
if not context.args:
await update.message.reply_text("❌ Usage: /run [skill_name]")
return
skill_name = context.args[0]
try:
await update.message.reply_text(f"🔄 Running skill: {skill_name}")
result = await self.agent.run_skill(skill_name)
if result['status'] == 'success':
await update.message.reply_text(f"✅ Skill completed successfully\n\n{result.get('summary', '')}")
else:
await update.message.reply_text(f"❌ Skill failed: {result['message']}")
except Exception as e:
await update.message.reply_text(f"❌ Error running skill: {str(e)}")
async def _handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle natural language messages"""
if not self._is_authorized(update.effective_user.id):
return
user_message = update.message.text
try:
# Use AI agent to process natural language request
response = await self.agent.process_natural_language(user_message)
await update.message.reply_text(response)
except Exception as e:
await update.message.reply_text(f"❌ Error processing message: {str(e)}")
async def _handle_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /help command"""
help_text = """
🤖 **OpenClaw Telegram Bot Help**
**Commands:**
/start - Initialize bot and show welcome
/status - Check agent and skill status
/brief - Get your morning briefing
/run [skill] - Execute a specific skill
/help - Show this help message
**Available Skills:**
• morning_brief - Daily summary and priorities
• email_summary - Unread email overview
• calendar_prep - Meeting preparation
• task_review - Task list optimization
**Natural Language:**
You can also send regular messages like:
• "What's on my calendar today?"
• "Summarize my emails"
• "Create a task for project review"
**Tips:**
• Use @username to mention the bot in groups
• All commands work in private chat or groups
• Bot remembers context within conversations
"""
await update.message.reply_text(help_text, parse_mode='Markdown')
def _format_skills_status(self, skills):
"""Format skills status for display"""
if not skills:
return "No active skills"
formatted = []
for skill in skills:
status_emoji = "✅" if skill['status'] == 'active' else "⏸️"
formatted.append(f"{status_emoji} {skill['name']}")
return "\n".join(formatted)
async def send_notification(self, message: str, channel: str = None, parse_mode: str = 'Markdown'):
"""Send notification to configured channel"""
chat_id = self.config.get('notification_channels', {}).get(channel)
if not chat_id:
chat_id = self.config.get('notification_channels', {}).get('alerts')
if chat_id:
await self.bot.send_message(chat_id=chat_id, text=message, parse_mode=parse_mode)
async def cleanup(self):
"""Cleanup Telegram integration"""
if self.application:
await self.application.stop()
# Register the integration
def register_integration(agent, config):
return TelegramIntegration(agent, config)
Step 4: Create Telegram-Enhanced Skills
Morning Brief with Telegram Delivery
Create ~/.openclaw/skills/telegram_morning_brief.py:
from openclaw.core import Skill
from datetime import datetime, timedelta
class TelegramMorningBriefSkill(Skill):
"""Enhanced morning brief delivered via Telegram"""
name = "telegram_morning_brief"
description = "Generate and deliver morning brief via Telegram"
async def execute(self, **kwargs):
"""Generate comprehensive morning brief"""
try:
# Gather data from multiple sources
email_summary = await self._get_email_summary()
calendar_events = await self._get_calendar_events()
weather_info = await self._get_weather_info()
news_brief = await self._get_news_brief()
# Generate AI-powered brief
brief_content = await self._generate_brief({
'emails': email_summary,
'calendar': calendar_events,
'weather': weather_info,
'news': news_brief
})
# Send via Telegram
telegram = self.agent.get_integration('telegram')
await telegram.send_notification(brief_content, channel='briefings')
return {
"status": "success",
"summary": "Morning brief delivered via Telegram",
"timestamp": datetime.now().isoformat()
}
except Exception as e:
return {"status": "error", "message": str(e)}
async def _get_email_summary(self):
"""Get overnight email summary"""
gmail = self.agent.get_integration('gmail')
since = datetime.now() - timedelta(hours=12)
emails = await gmail.search_messages(
f"is:unread after:{since.strftime('%Y/%m/%d')}",
max_results=10
)
if not emails:
return "No new emails overnight."
return f"📧 **{len(emails)} new emails** (showing top priorities)\n" + \
"\n".join([f"• {email['from']}: {email['subject']}" for email in emails[:3]])
async def _get_calendar_events(self):
"""Get today's calendar events"""
calendar = self.agent.get_integration('calendar')
events = await calendar.get_events_today()
if not events:
return "🗓️ **No meetings scheduled today**"
formatted_events = []
for event in events[:5]: # Show up to 5 events
time = event['start'].strftime('%H:%M')
formatted_events.append(f"• {time} - {event['title']}")
return f"🗓️ **Today's Schedule** ({len(events)} events)\n" + \
"\n".join(formatted_events)
async def _generate_brief(self, data):
"""Generate AI-powered morning brief"""
prompt = f"""
Generate a concise morning brief based on this data:
EMAILS: {data['emails']}
CALENDAR: {data['calendar']}
WEATHER: {data['weather']}
NEWS: {data['news']}
Format as:
🌅 **Morning Brief - {datetime.now().strftime('%A, %B %d')}**
**📧 Inbox Status**
[Email summary with priorities]
**🗓️ Today's Focus**
[Calendar events with preparation notes]
**🎯 Priorities**
[Top 3 things to focus on today]
**⚡ Quick Wins**
[2-3 small tasks that can be completed quickly]
Keep it actionable and under 300 words.
"""
return await self.agent.generate_response(prompt)
Interactive Task Management
Create ~/.openclaw/skills/telegram_task_manager.py:
class TelegramTaskManagerSkill(Skill):
"""Interactive task management via Telegram"""
name = "telegram_task_manager"
async def execute(self, command: str, task_details: str = None, **kwargs):
"""Handle task management commands"""
telegram = self.agent.get_integration('telegram')
if command == "add":
return await self._add_task(task_details)
elif command == "list":
return await self._list_tasks()
elif command == "complete":
return await self._complete_task(task_details)
elif command == "prioritize":
return await self._prioritize_tasks()
async def _add_task(self, task_description):
"""Add new task with AI categorization"""
# Use AI to categorize and estimate task
task_analysis = await self.agent.generate_response(f"""
Analyze this task: "{task_description}"
Provide:
1. Category (work/personal/urgent/routine)
2. Estimated time (15min/30min/1hr/2hr/4hr/full day)
3. Priority (high/medium/low)
4. Suggested next action
Format as JSON.
""")
# Save to task management system
# Implementation depends on your task system (Notion, Todoist, etc.)
return {
"status": "success",
"message": f"✅ Task added: {task_description}\n{task_analysis}"
}
Step 5: Advanced Notification Workflows
Smart Alert System
Create notification rules that adapt based on content:
class TelegramNotificationManager:
"""Intelligent notification routing"""
def __init__(self, telegram_integration):
self.telegram = telegram_integration
self.notification_rules = {
'urgent_email': {
'condition': lambda email: 'urgent' in email['subject'].lower(),
'template': "🚨 **Urgent Email**\nFrom: {sender}\nSubject: {subject}",
'channel': 'alerts'
},
'calendar_conflict': {
'condition': lambda event: event.get('conflict'),
'template': "⚠️ **Calendar Conflict**\n{event_details}",
'channel': 'alerts'
},
'system_error': {
'condition': lambda error: error.get('severity') == 'high',
'template': "❌ **System Error**\n{error_message}",
'channel': 'alerts'
}
}
async def process_notification(self, notification_type, data):
"""Process and route notifications intelligently"""
rule = self.notification_rules.get(notification_type)
if rule and rule['condition'](data):
message = rule['template'].format(**data)
await self.telegram.send_notification(message, rule['channel'])
async def send_daily_digest(self):
"""Send end-of-day digest"""
digest = await self._generate_daily_digest()
await self.telegram.send_notification(digest, 'briefings')
Interactive Approval Workflows
Implement workflows that require human approval:
class TelegramApprovalWorkflow:
"""Handle approval requests via Telegram"""
async def request_approval(self, request_type, details, timeout_minutes=30):
"""Request approval with timeout"""
approval_id = self._generate_approval_id()
message = f"""
🔍 **Approval Required**
**Type:** {request_type}
**Details:** {details}
React with ✅ to approve or ❌ to deny.
Timeout: {timeout_minutes} minutes
"""
# Send message with inline keyboard
await self.telegram.send_notification(message, 'alerts')
# Wait for response (implementation depends on your framework)
response = await self._wait_for_approval(approval_id, timeout_minutes)
return response
async def _wait_for_approval(self, approval_id, timeout_minutes):
"""Wait for user response to approval request"""
# Implementation depends on your callback handling
# This is a simplified example
pass
Step 6: Production Deployment
Webhook Configuration
For production deployment, use webhooks instead of polling:
# webhook_server.py
from aiohttp import web, ClientSession
import ssl
import logging
async def webhook_handler(request):
"""Handle incoming Telegram webhooks"""
try:
update_data = await request.json()
# Process update through your Telegram integration
await telegram_integration.process_update(update_data)
return web.Response(status=200)
except Exception as e:
logging.error(f"Webhook error: {e}")
return web.Response(status=500)
async def init_webhook_server():
"""Initialize webhook server"""
app = web.Application()
app.router.add_post('/telegram/webhook', webhook_handler)
# SSL configuration for production
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain('/path/to/cert.pem', '/path/to/key.pem')
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, '0.0.0.0', 8443, ssl_context=ssl_context)
await site.start()
logging.info("Webhook server started on port 8443")
Docker Deployment
Create docker-compose.yml for production deployment:
version: '3.8'
services:
openclaw-telegram:
build: .
environment:
- TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
- OPENAI_API_KEY=${OPENAI_API_KEY}
- WEBHOOK_URL=https://your-domain.com/telegram/webhook
ports:
- "8443:8443"
volumes:
- ./data:/root/.openclaw
- ./ssl:/etc/ssl/certs
restart: unless-stopped
networks:
- openclaw-network
networks:
openclaw-network:
driver: bridge
Troubleshooting Common Issues
Bot Not Responding
Check bot token:
curl "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getMe"
Verify webhook:
curl "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getWebhookInfo"
Authentication Issues
Check user authorization:
# Add logging to see incoming user IDs
logging.info(f"Received message from user: {update.effective_user.id}")
Test bot permissions:
# Send test message via bot
curl -X POST "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/sendMessage" \
-H "Content-Type: application/json" \
-d '{"chat_id": "YOUR_CHAT_ID", "text": "Test message"}'
Webhook SSL Issues
For HTTPS webhooks, ensure valid SSL certificates:
# Test SSL certificate
openssl s_client -connect your-domain.com:443 -servername your-domain.com
Security Best Practices
Bot Token Protection
- Never commit tokens to git
- Use environment variables
- Rotate tokens regularly
- Monitor unauthorized access attempts
User Authentication
# Implement robust user verification
class SecureTelegramIntegration(TelegramIntegration):
async def verify_user(self, user_id):
"""Enhanced user verification"""
if user_id not in self.authorized_users:
# Log unauthorized access attempt
logging.warning(f"Unauthorized access attempt from {user_id}")
# Optional: Notify admin
await self.notify_admin(f"Unauthorized access from {user_id}")
return False
return True
Rate Limiting
# Implement rate limiting for commands
from collections import defaultdict
import time
class RateLimiter:
def __init__(self, max_requests=10, window_minutes=5):
self.max_requests = max_requests
self.window_seconds = window_minutes * 60
self.requests = defaultdict(list)
def is_allowed(self, user_id):
now = time.time()
user_requests = self.requests[user_id]
# Clean old requests
user_requests[:] = [req_time for req_time in user_requests
if now - req_time < self.window_seconds]
if len(user_requests) >= self.max_requests:
return False
user_requests.append(now)
return True
Why Choose OpenClaw + Telegram vs. Alternatives
OpenClaw's Telegram integration offers unique advantages over commercial alternatives:
Full Control: Your bot runs on your infrastructure with complete customization Privacy: No data sharing with third-party services Cost Efficiency: Telegram bot API is free; only pay for OpenAI API calls Advanced Workflows: Build complex approval and notification systems
For teams wanting enterprise-grade AI automation without the technical complexity, consider MrDelegate — offering similar functionality with managed infrastructure, security, and support.
Start your free trial to experience AI automation with zero technical setup.
Next Steps
With Telegram integration complete, your OpenClaw system becomes a powerful mobile-first AI automation platform. Consider expanding with:
- Group chat integration for team coordination
- Inline keyboards for interactive workflows
- File sharing for document processing
- Voice message handling for audio commands
- Multi-bot orchestration for specialized functions
The combination of OpenClaw's automation capabilities with Telegram's mobile interface creates an AI assistant that works exactly how you do — always connected, always available.
Your AI executive assistant is ready.
Morning brief at 7am. Inbox triaged overnight. Calendar protected. Dedicated VPS. No Docker. Live in 60 seconds.