Telegram Bot with Command Routing

Telegram Bot with Command Routing

Problem Statement

Managing multiple Telegram bots or complex bot behaviors often leads to monolithic, hard-to-maintain codebases. When your bot needs to handle dozens of commands — from /start and /help to custom business logic like /report or /status — a simple if-else chain becomes unreadable and error-prone. Additionally, teams frequently need to add new commands without touching existing code, which violates the Open/Closed Principle.

This script demonstrates a clean, production-ready approach to Telegram bot command routing using Python's native functools module and the python-telegram-bot library. Instead of scattering command handlers throughout your code, we centralize routing logic in a single dispatch table, making the bot extensible, testable, and maintainable — perfect for DevOps bots that need to monitor services, trigger deployments, or provide operational dashboards.

How It Works

The bot uses the python-telegram-bot v20+ framework, which provides an Application class for managing updates. Our architecture centers on a command router pattern: we define command handlers as standalone functions, then register them in a dictionary mapping command names to function references. The main dispatcher extracts the command from each incoming message, looks it up in the routing table, and invokes the appropriate handler with error isolation.

Key design decisions include: using functools.partial to inject shared dependencies (like the bot token) into handlers, implementing a decorator-based registration system for clean handler definitions, and separating configuration from logic via environment variables. The bot runs with asyncio under the hood, so all handlers can be async functions. We also implement a simple rate-limiting mechanism using collections.deque to prevent command flooding, and include structured logging for production observability.

Full Script

#!/usr/bin/env python3
"""
Telegram Bot with Command Routing

A production-ready Telegram bot demonstrating a clean command routing system
using the python-telegram-bot library. Supports multiple commands, rate
limiting, structured logging, and graceful error handling.

Requirements:
    pip install python-telegram-bot==21.6

Environment variables:
    BOT_TOKEN: Your Telegram bot token from @BotFather
    BOT_ADMIN_ID: Optional admin user ID for restricted commands
    LOG_LEVEL: Optional logging level (default: INFO)
"""

import asyncio
import logging
import os
import sys
import time
from collections import defaultdict, deque
from datetime import datetime
from functools import partial
from typing import Any, Callable, Dict, Optional

try:
    from telegram import Update
    from telegram.ext import Application, CommandHandler, ContextTypes
except ImportError as e:
    raise ImportError(
        "Missing required dependency. Install with: pip install python-telegram-bot==21.6"
    ) from e

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

BOT_TOKEN = os.environ.get("BOT_TOKEN")
if not BOT_TOKEN:
    raise ValueError("BOT_TOKEN environment variable must be set")

ADMIN_ID = int(os.environ.get("BOT_ADMIN_ID", "0"))
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()

# Rate limiting configuration
RATE_LIMIT_WINDOW = 10  # seconds
RATE_LIMIT_MAX = 5  # max commands per window per user

# ---------------------------------------------------------------------------
# Logging setup
# ---------------------------------------------------------------------------

logging.basicConfig(
    level=getattr(logging, LOG_LEVEL, logging.INFO),
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    stream=sys.stdout,
)
logger = logging.getLogger("telegram_bot")

# ---------------------------------------------------------------------------
# Command router and rate limiter
# ---------------------------------------------------------------------------

class CommandRouter:
    """Central routing table for bot commands."""

    def __init__(self) -> None:
        self._handlers: Dict[str, Callable] = {}
        self._rate_limits: Dict[int, deque] = defaultdict(deque)
        self._command_stats: Dict[str, int] = defaultdict(int)

    def register(self, command: str) -> Callable:
        """Decorator to register a command handler."""

        def decorator(func: Callable) -> Callable:
            self._handlers[command] = func
            logger.debug(f"Registered command: /{command}")
            return func

        return decorator

    def get_handler(self, command: str) -> Optional[Callable]:
        """Retrieve handler for a command, or None if not found."""
        return self._handlers.get(command)

    def check_rate_limit(self, user_id: int) -> bool:
        """Enforce rate limiting per user. Returns True if allowed."""
        now = time.time()
        user_requests = self._rate_limits[user_id]

        # Clean old entries
        while user_requests and now - user_requests[0] > RATE_LIMIT_WINDOW:
            user_requests.popleft()

        if len(user_requests) >= RATE_LIMIT_MAX:
            logger.warning(f"Rate limit exceeded for user {user_id}")
            return False

        user_requests.append(now)
        return True

    def record_command(self, command: str) -> None:
        """Track command usage for statistics."""
        self._command_stats[command] += 1

    def get_stats(self) -> Dict[str, int]:
        """Return command usage statistics."""
        return dict(self._command_stats)

    @property
    def commands(self) -> list:
        """List of registered commands."""
        return sorted(self._handlers.keys())


# Initialize router
router = CommandRouter()


# ---------------------------------------------------------------------------
# Command handlers
# ---------------------------------------------------------------------------

@router.register("start")
async def start_command(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """Welcome message with available commands."""
    commands_list = "\n".join(f"/{cmd}" for cmd in router.commands)
    await update.message.reply_text(
        f"🤖 *Welcome to Command Router Bot!*\n\n"
        f"Available commands:\n{commands_list}\n\n"
        f"Use /help for more details.",
        parse_mode="Markdown",
    )


@router.register("help")
async def help_command(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """Detailed help for each command."""
    help_text = """
*Available Commands:*

/start - Welcome message
/help - This help text  
/time - Current server time
/stats - Command usage statistics
/echo <text> - Echo back your text
/admin - Admin-only command (if configured)
"""
    await update.message.reply_text(help_text, parse_mode="Markdown")


@router.register("time")
async def time_command(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """Return current server time."""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    await update.message.reply_text(f"🕐 Current time: `{now}`", parse_mode="Markdown")


@router.register("stats")
async def stats_command(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """Show command usage statistics."""
    stats = router.get_stats()
    if not stats:
        await update.message.reply_text("No commands have been used yet.")
        return

    stats_text = "\n".join(f"/{cmd}: {count}" for cmd, count in sorted(stats.items()))
    await update.message.reply_text(f"📊 *Command Statistics:*\n{stats_text}", parse_mode="Markdown")


@router.register("echo")
async def echo_command(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """Echo back the user's text."""
    if not context.args:
        await update.message.reply_text("Usage: /echo <text>")
        return

    text = " ".join(context.args)
    await update.message.reply_text(f"🔁 Echo: {text}")


@router.register("admin")
async def admin_command(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """Admin-only command that runs a system check."""
    if ADMIN_ID and update.effective_user.id != ADMIN_ID:
        await update.message.reply_text("⛔ You are not authorized to use this command.")
        return

    # Simulate a system check
    import platform

    system_info = (
        f"🖥 *System Information:*\n"
        f"Platform: {platform.system()}\n"
        f"Release: {platform.release()}\n"
        f"Python: {sys.version.split()[0]}\n"
        f"Uptime: {time.time() - psutil_boot_time():.1f}s"
    )
    await update.message.reply_text(system_info, parse_mode="Markdown")


def psutil_boot_time() -> float:
    """Get system boot time (fallback if psutil not available)."""
    try:
        import psutil

        return psutil.boot_time()
    except ImportError:
        # Fallback: use process start time
        return time.time() - 3600  # Assume 1 hour uptime


# ---------------------------------------------------------------------------
# Core dispatcher
# ---------------------------------------------------------------------------

async def command_dispatcher(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    """
    Main dispatch function that routes commands to appropriate handlers.
    Handles unknown commands, rate limiting, and errors gracefully.
    """
    if not update.message or not update.message.text:
        return

    user_id = update.effective_user.id
    message_text = update.message.text.strip()

    # Extract command (first word) and args (rest)
    parts = message_text.split(maxsplit=1)
    command = parts[0].lstrip("/").lower()
    args = parts[1] if len(parts) > 1 else ""

    # Store args in context for handlers
    context.args = args.split() if args else []

    # Rate limiting check
    if not router.check_rate_limit(user_id):
        await update.message.reply_text(
            "⚠️ Rate limit exceeded. Please wait a moment before sending more commands."
        )
        return

    # Get handler for this command
    handler = router.get_handler(command)

    if handler is None:
        await update.message.reply_text(
            f"❌ Unknown command: /{command}\nUse /help to see available commands."
        )
        return

    # Record command usage
    router.record_command(command)
    logger.info(f"User {user_id} executed /{command}")

    # Execute handler with error isolation
    try:
        await handler(update, context)
    except Exception as e:
        logger.exception(f"Error in /{command} handler: {e}")
        await update.message.reply_text(
            "❌ An error occurred while processing your command. Please try again later."
        )


# ---------------------------------------------------------------------------
# Main application setup
# ---------------------------------------------------------------------------

def main() -> None:
    """Main entry point for the bot."""
    logger.info("Starting Telegram bot...")

    # Create application
    application = Application.builder().token(BOT_TOKEN).build()

    # Register the dispatcher for all text messages
    application.add_handler(CommandHandler(router.commands, command_dispatcher))

    # Add a fallback for non-command messages
    async def unknown_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if update.message and update.message.text:
            await update.message.reply_text(
                "❓ I only respond to commands. Use /help to see what I can do."
            )

    from telegram.ext import MessageHandler, filters

    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, unknown_handler))

    # Start the bot
    logger.info(f"Bot started with commands: {router.commands}")
    application.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        logger.info("Bot stopped by user")
        sys.exit(0)
    except Exception as e:
        logger.fatal(f"Fatal error: {e}")
        sys.exit(1)

Configuration

The script requires the following environment variables:

Additional constants you can modify directly in the script:

• RATE_LIMIT_WINDOW — Time window in seconds for rate limiting (default: 10)
• RATE_LIMIT_MAX — Maximum commands allowed per window per user (default: 5)

Usage Examples

Example 1: Basic startup with debug logging

$ BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11 LOG_LEVEL=DEBUG python3 tg_bot.py

Output:

2024-01-15 10:30:45,123 - telegram_bot - INFO - Starting Telegram bot...
2024-01-15 10:30:45,124 - telegram_bot - DEBUG - Registered command: admin
2024-01-15 10:30:45,124 - telegram_bot - DEBUG - Registered command: echo
2024-01-15 10:30:45,124 - telegram_bot - DEBUG - Registered command: help
2024-01-15 10:30:45,124 - telegram_bot - DEBUG - Registered command: start
2024-01-15 10:30:45,124 - telegram_bot - DEBUG - Registered command: stats
2024-01-15 10:30:45,124 - telegram_bot - DEBUG - Registered command: time
2024-01-15 10:30:45,124 - telegram_bot - INFO - Bot started with commands: ['admin', 'echo', 'help', 'start', 'stats', 'time']

Example 2: Running with admin restrictions

$ BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11 BOT_ADMIN_ID=987654321 python3 tg_bot.py

In Telegram:

User: /start
Bot: 🤖 Welcome to Command Router Bot!

Available commands:
/admin
/echo
/help
/start
/stats
/time

Use /help for more details.

User: /echo Hello World
Bot: 🔁 Echo: Hello World

User: /stats
Bot: 📊 Command Statistics:
/echo: 1
/start: 1

Example 3: Running as a systemd service

$ cat /etc/systemd/system/tg-bot.service
[Unit]
Description=Telegram Command Router Bot
After=network.target

[Service]
Environment=BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11
Environment=LOG_LEVEL=INFO
ExecStart=/usr/local/bin/python3 /opt/tg_bot/tg_bot.py
Restart=always
User=telegram

[Install]
WantedBy=multi-user.target

Error Handling

The script implements multiple layers of error handling:

1. Import errors: Missing python-telegram-bot raises a clear, actionable error message with installation instructions.

2. Configuration validation: Missing BOT_TOKEN causes an immediate ValueError at startup rather than failing later.

3. Command dispatch isolation: Each command handler runs in its own try-except block. If a handler fails, the error is logged with full stack trace, and the user receives a friendly error message without crashing the bot.

4. Rate limiting: Prevents abuse by tracking command frequency per user. Users exceeding the limit receive a warning and their commands are rejected for the window period.

5. Unknown commands: Any unrecognized command gets a helpful response pointing to /help.

6. Network resilience: The python-telegram-bot library handles connection retries and reconnection automatically. The run_polling() method will keep trying to reconnect if the network drops.

7. Graceful shutdown: KeyboardInterrupt (Ctrl+C) stops the bot cleanly, and fatal errors exit with a non-zero status code for service managers.

Extending Ideas

1. Add database persistence: Replace the in-memory command_stats with a SQLite or PostgreSQL backend to track long-term usage analytics and user preferences.

2. Implement webhook mode: For production deployments behind a reverse proxy, switch from polling to webhooks using application.run_webhook() with SSL certificates for better performance.

3. Add command aliases: Extend the CommandRouter to support multiple command names pointing to the same handler, e.g., /h as an alias for /help.

4. Integrate external services: Add commands that interact with your infrastructure — /deploy to trigger CI/CD pipelines, /status to check service health via API calls, or /logs to fetch recent application logs.

5. Add conversation support: Use ConversationHandler for multi-step interactions, like a /report command that asks follow-up questions about the issue type and severity.

6. Implement permissions system: Extend the admin check to support role-based access control with multiple user groups and command-specific permissions.