Build a Telegram Automation Bot with Python: A Practical Starter Project

A Telegram bot can be more than a chat interface. With a small Python service, it becomes a secure remote control for repeatable workflows: checking system health, launching an approved job, collecting a report, or notifying a team when a process finishes.

In this tutorial, we will build a practical automation bot with Python. The bot accepts a small set of commands, restricts access to approved Telegram users, runs only predefined tasks, and returns clear results. The design is intentionally simple enough for a first deployment but structured well enough to extend later.

What we are building

The bot will support three commands:

• /start — show available commands;
• /status — return basic host status;
• /run backup — launch an approved automation task.

The important security rule is that the bot never executes arbitrary text as a shell command. Users select a task from a fixed allowlist. This prevents a convenient automation tool from becoming an accidental remote shell.

Project structure

telegram-automation-bot/
├── bot.py
├── requirements.txt
└── .env

1. Create the Telegram bot

Open Telegram, start a chat with @BotFather, run /newbot, and follow the prompts. BotFather will provide a token. Treat it like a password: do not commit it to Git, paste it into screenshots, or hard-code it in the Python file.

You also need your numeric Telegram user ID for the access allowlist. You can obtain it from a trusted ID bot or temporarily log update.effective_user.id during local testing.

2. Install the dependencies

python -m venv .venv
source .venv/bin/activate
pip install python-telegram-bot python-dotenv psutil

On Windows PowerShell, activate the environment with:

.venv\Scripts\Activate.ps1

Create requirements.txt:

python-telegram-bot>=21,<23
python-dotenv>=1,<2
psutil>=6,<8

3. Configure environment variables

Create a local .env file:

TELEGRAM_BOT_TOKEN=replace_with_your_token
ALLOWED_USER_IDS=123456789

For several administrators, separate IDs with commas:

ALLOWED_USER_IDS=123456789,987654321

Add .env and .venv/ to .gitignore.

4. Add the Python code

import asyncio
import logging
import os
import shutil
from datetime import datetime, timezone
from pathlib import Path

import psutil
from dotenv import load_dotenv
from telegram import Update
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
)

load_dotenv()

BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
ALLOWED_USER_IDS = {
    int(value.strip())
    for value in os.getenv("ALLOWED_USER_IDS", "").split(",")
    if value.strip()
}

logging.basicConfig(
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    level=logging.INFO,
)
logger = logging.getLogger(__name__)


def is_allowed(update: Update) -> bool:
    user = update.effective_user
    return bool(user and user.id in ALLOWED_USER_IDS)


async def reject_unauthorized(update: Update) -> None:
    user = update.effective_user
    logger.warning(
        "Rejected Telegram user id=%s username=%s",
        getattr(user, "id", None),
        getattr(user, "username", None),
    )
    if update.effective_message:
        await update.effective_message.reply_text("Access denied.")


async def start(
    update: Update,
    context: ContextTypes.DEFAULT_TYPE,
) -> None:
    if not is_allowed(update):
        await reject_unauthorized(update)
        return

    await update.effective_message.reply_text(
        "Automation bot is online.

"
        "Commands:
"
        "/status — host health
"
        "/run backup — run the approved backup task"
    )


async def status(
    update: Update,
    context: ContextTypes.DEFAULT_TYPE,
) -> None:
    if not is_allowed(update):
        await reject_unauthorized(update)
        return

    disk = shutil.disk_usage("/")
    memory = psutil.virtual_memory()
    boot_time = datetime.fromtimestamp(
        psutil.boot_time(),
        tz=timezone.utc,
    )
    uptime = datetime.now(timezone.utc) - boot_time

    message = (
        "Host status
"
        f"CPU: {psutil.cpu_percent(interval=0.3):.1f}%
"
        f"Memory: {memory.percent:.1f}%
"
        f"Disk: {disk.used / disk.total * 100:.1f}%
"
        f"Uptime: {str(uptime).split('.')[0]}"
    )
    await update.effective_message.reply_text(message)


async def backup_task() -> str:
    source = Path.home() / "automation-data"
    destination = Path.home() / "backups"
    destination.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    archive_base = destination / f"automation-data-{timestamp}"

    archive_path = await asyncio.to_thread(
        shutil.make_archive,
        str(archive_base),
        "zip",
        root_dir=source,
    )
    return f"Backup created: {archive_path}"


TASKS = {
    "backup": backup_task,
}


async def run_task(
    update: Update,
    context: ContextTypes.DEFAULT_TYPE,
) -> None:
    if not is_allowed(update):
        await reject_unauthorized(update)
        return

    task_name = context.args[0].lower() if context.args else ""
    task = TASKS.get(task_name)

    if task is None:
        available = ", ".join(sorted(TASKS))
        await update.effective_message.reply_text(
            f"Unknown task. Available: {available}"
        )
        return

    await update.effective_message.reply_text(
        f"Starting task: {task_name}"
    )

    try:
        result = await task()
    except Exception:
        logger.exception("Task failed: %s", task_name)
        await update.effective_message.reply_text(
            f"Task failed: {task_name}. Check the service logs."
        )
        return

    await update.effective_message.reply_text(result)


def main() -> None:
    if not ALLOWED_USER_IDS:
        raise RuntimeError("ALLOWED_USER_IDS is empty")

    application = Application.builder().token(BOT_TOKEN).build()
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("status", status))
    application.add_handler(CommandHandler("run", run_task))

    application.run_polling(drop_pending_updates=True)


if __name__ == "__main__":
    main()

5. Prepare test data and run the bot

mkdir -p ~/automation-data
echo "automation test" > ~/automation-data/example.txt
python bot.py

Open the bot in Telegram and send /start, then /status. Finally, run /run backup. A timestamped ZIP archive should appear in ~/backups.

Why the task allowlist matters

A common but dangerous pattern is to pass a Telegram message directly to subprocess.run(..., shell=True). Do not do that. Even with a private bot, leaked tokens, forwarded messages, configuration errors, or an accidentally broad access rule can expose the entire host.

The TASKS dictionary creates a narrow interface: the bot can call only functions that you explicitly register. Each function can validate inputs, use normal Python APIs, write structured logs, and return a controlled result.

Run it as a systemd service

For a small Linux server, create /etc/systemd/system/telegram-automation.service:

[Unit]
Description=Telegram Automation Bot
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=automation
WorkingDirectory=/opt/telegram-automation-bot
EnvironmentFile=/opt/telegram-automation-bot/.env
ExecStart=/opt/telegram-automation-bot/.venv/bin/python bot.py
Restart=on-failure
RestartSec=5
NoNewPrivileges=true
PrivateTmp=true

[Install]
WantedBy=multi-user.target

Then enable it:

sudo systemctl daemon-reload
sudo systemctl enable --now telegram-automation
sudo systemctl status telegram-automation

Production checklist

• ✓ Use a dedicated user — grant only the permissions the bot needs.
• ✓ Protect the bot token — keep it in a restricted environment file.
• ✓ Restrict access — allow only approved Telegram user IDs.
• ✓ Block arbitrary commands — run only predefined automation tasks.
• ✓ Handle failures — add timeouts, logs, and error notifications.
• ✓ Limit backup access — expose only required directories.
• ✓ Rotate leaked tokens — replace any token found in logs or Git.

Where to take it next

This starter can become a useful private operations console. Add approved tasks for generating reports, checking Docker containers, calling an n8n webhook, monitoring a local AI server, or moving files through a validated workflow. Keep the chat layer thin and place the real business logic in small, testable Python functions.

The result is a practical automation pattern: Telegram provides the interface, Python provides the control layer, and the allowlist keeps the system predictable.