name: telegram-bot description: Telegram bot development with aiogram 3.x, python-telegram-bot, Telethon, and pyrogram. Covers bot architecture, handlers, keyboards, webhooks, and deployment.

Telegram Bot Development Skill

When to Activate

• Creating or modifying Telegram bots
• Working with Telegram Bot API
• Setting up webhooks or long polling
• Building inline keyboards, callback handlers
• Telegram userbot or channel management

Framework Selection

Default choice: aiogram 3.x - best for production bots.

aiogram 3.x Project Structure

bot/
  __init__.py
  main.py              # Entry point, bot startup
  config.py            # Settings from env
  handlers/
    __init__.py
    start.py           # /start, /help commands
    admin.py           # Admin commands
    callbacks.py       # Callback query handlers
    errors.py          # Error handler
  keyboards/
    __init__.py
    inline.py          # InlineKeyboardMarkup builders
    reply.py           # ReplyKeyboardMarkup builders
  middlewares/
    __init__.py
    auth.py            # User authentication
    throttling.py      # Rate limiting
  services/
    __init__.py
    database.py        # DB operations
    api_client.py      # External API calls
  models/
    __init__.py
    user.py            # User model
  filters/
    __init__.py
    admin.py           # Admin check filter
  states/
    __init__.py
    forms.py           # FSM states
  utils/
    __init__.py
    helpers.py

aiogram 3.x Core Patterns

Bot Entry Point

import asyncio
from aiogram import Bot, Dispatcher
from aiogram.enums import ParseMode
from aiogram.client.default import DefaultBotProperties

from bot.config import settings
from bot.handlers import start, admin, callbacks

async def main():
    bot = Bot(
        token=settings.BOT_TOKEN,
        default=DefaultBotProperties(parse_mode=ParseMode.HTML)
    )
    dp = Dispatcher()

    # Register routers
    dp.include_routers(
        start.router,
        admin.router,
        callbacks.router,
    )

    # Start polling
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

Router-Based Handlers

from aiogram import Router, F
from aiogram.types import Message, CallbackQuery
from aiogram.filters import Command, CommandStart

router = Router()

@router.message(CommandStart())
async def cmd_start(message: Message):
    await message.answer(
        "Welcome! Choose an option:",
        reply_markup=get_main_keyboard()
    )

@router.message(Command("help"))
async def cmd_help(message: Message):
    await message.answer("Available commands:\n/start - Main menu\n/help - This message")

@router.callback_query(F.data.startswith("action:"))
async def handle_action(callback: CallbackQuery):
    action = callback.data.split(":")[1]
    await callback.answer(f"Processing: {action}")
    await callback.message.edit_text(f"Action {action} completed")

Inline Keyboards

from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder

def get_main_keyboard() -> InlineKeyboardMarkup:
    builder = InlineKeyboardBuilder()
    builder.row(
        InlineKeyboardButton(text="Status", callback_data="action:status"),
        InlineKeyboardButton(text="Settings", callback_data="action:settings"),
    )
    builder.row(
        InlineKeyboardButton(text="Help", callback_data="action:help"),
    )
    return builder.as_markup()

FSM (Finite State Machine)

from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext

class AddChannel(StatesGroup):
    waiting_for_name = State()
    waiting_for_url = State()
    confirmation = State()

@router.message(Command("add_channel"))
async def start_add(message: Message, state: FSMContext):
    await state.set_state(AddChannel.waiting_for_name)
    await message.answer("Enter channel name:")

@router.message(AddChannel.waiting_for_name)
async def process_name(message: Message, state: FSMContext):
    await state.update_data(name=message.text)
    await state.set_state(AddChannel.waiting_for_url)
    await message.answer("Enter channel URL:")

@router.message(AddChannel.waiting_for_url)
async def process_url(message: Message, state: FSMContext):
    data = await state.get_data()
    await state.clear()
    await message.answer(f"Channel '{data['name']}' with URL {message.text} added!")

Middleware

from aiogram import BaseMiddleware
from typing import Callable, Awaitable, Any

class AuthMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Message, dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: dict[str, Any],
    ) -> Any:
        user = await db.get_user(event.from_user.id)
        if not user:
            await event.answer("Access denied. Contact admin.")
            return
        data["user"] = user
        return await handler(event, data)

Webhook Setup (Production)

from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
from aiohttp import web

WEBHOOK_PATH = f"/webhook/{settings.BOT_TOKEN}"
WEBHOOK_URL = f"https://{settings.DOMAIN}{WEBHOOK_PATH}"

async def on_startup(bot: Bot):
    await bot.set_webhook(WEBHOOK_URL)

app = web.Application()
webhook_handler = SimpleRequestHandler(dispatcher=dp, bot=bot)
webhook_handler.register(app, path=WEBHOOK_PATH)
setup_application(app, dp, bot=bot)
web.run_app(app, host="0.0.0.0", port=8080)

Telethon (Userbots / Channel Management)

Basic Setup

from telethon import TelegramClient, events
from telethon.sessions import StringSession

# File session (persists on disk)
client = TelegramClient("session_name", api_id, api_hash)

# String session (for Docker / serverless - store in env var)
client = TelegramClient(StringSession(SESSION_STRING), api_id, api_hash)

Channel Parsing & Forwarding

@client.on(events.NewMessage(chats=["source_channel"]))
async def forward_handler(event):
    # Forward with original formatting
    await client.send_message("target_channel", event.message)

# Parse channel history
async def parse_channel_history(channel: str, limit: int = 100) -> list[dict]:
    messages = []
    async for msg in client.iter_messages(channel, limit=limit):
        messages.append({
            "id": msg.id,
            "text": msg.text,
            "date": msg.date,
            "views": msg.views,
            "media": bool(msg.media),
            "forwards": msg.forwards,
        })
    return messages

# Download media from channel
async def download_channel_media(channel: str, dest_dir: str, limit: int = 50):
    async for msg in client.iter_messages(channel, limit=limit):
        if msg.media:
            await client.download_media(msg, file=dest_dir)

Session Management

# Generate string session (run once interactively)
async def generate_session():
    async with TelegramClient(StringSession(), api_id, api_hash) as client:
        print("Session string:", client.session.save())

# Multi-account management
# WARNING: Telegram actively bans automated accounts.
# Using multiple userbots may violate Telegram ToS. Use at your own risk.
class AccountPool:
    def __init__(self, sessions: list[str]):
        self.clients = [
            TelegramClient(StringSession(s), api_id, api_hash)
            for s in sessions
        ]
        self.current = 0

    async def get_client(self) -> TelegramClient:
        """Round-robin client selection to avoid rate limits."""
        client = self.clients[self.current]
        self.current = (self.current + 1) % len(self.clients)
        if not client.is_connected():
            await client.connect()
        return client

Pyrogram (Modern MTProto)

from pyrogram import Client, filters
from pyrogram.types import Message

app = Client("my_account", api_id=api_id, api_hash=api_hash)

# Channel message handler
@app.on_message(filters.channel & filters.chat("source_channel"))
async def channel_handler(client: Client, message: Message):
    # Process and forward
    text = message.text or message.caption or ""
    if should_forward(text):
        await client.send_message("target_channel", text)

# Batch operations
async def get_channel_members(channel: str) -> list:
    members = []
    async for member in app.get_chat_members(channel):
        members.append({"id": member.user.id, "name": member.user.first_name})
    return members

# Media group handling
@app.on_message(filters.media_group)
async def media_group_handler(client: Client, message: Message):
    media_group = await client.get_media_group(message.chat.id, message.id)
    for msg in media_group:
        await client.download_media(msg, file_name=f"media/{msg.id}")

Telegram Payments & Mini Apps

Telegram Stars Payment

from aiogram.types import LabeledPrice

@router.message(Command("buy"))
async def cmd_buy(message: Message):
    await message.answer_invoice(
        title="Premium Access",
        description="30 days of premium features",
        payload="premium_30d",
        currency="XTR",  # Telegram Stars
        prices=[LabeledPrice(label="Premium", amount=100)],  # 100 Stars
    )

@router.pre_checkout_query()
async def pre_checkout(query: PreCheckoutQuery):
    await query.answer(ok=True)

@router.message(F.successful_payment)
async def payment_success(message: Message):
    payment = message.successful_payment
    await activate_premium(message.from_user.id, days=30)
    await message.answer("Premium activated!")

Telegram Mini App (Web App)

from aiogram.types import WebAppInfo, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder

@router.message(Command("app"))
async def open_app(message: Message):
    builder = InlineKeyboardBuilder()
    builder.add(InlineKeyboardButton(
        text="Open Dashboard",
        web_app=WebAppInfo(url="https://your-app.com/miniapp")
    ))
    await message.answer("Open the dashboard:", reply_markup=builder.as_markup())

# Validate data from Mini App
from aiogram.utils.web_app import check_webapp_signature, parse_webapp_init_data

def validate_webapp_data(init_data: str, bot_token: str) -> dict:
    if check_webapp_signature(bot_token, init_data):
        return parse_webapp_init_data(init_data)
    raise ValueError("Invalid webapp data")

Bot + Telethon Hybrid Architecture

Use aiogram for bot commands and Telethon for userbot features in the same project.

import asyncio
from aiogram import Bot, Dispatcher
from telethon import TelegramClient

async def main():
    # Bot (via Bot API)
    bot = Bot(token=BOT_TOKEN)
    dp = Dispatcher()
    dp.include_router(bot_handlers.router)

    # Userbot (via MTProto)
    userbot = TelegramClient(StringSession(SESSION), api_id, api_hash)
    await userbot.start()

    # Share data via Redis
    redis = await aioredis.from_url("redis://localhost")

    # Run both concurrently
    await asyncio.gather(
        dp.start_polling(bot),
        userbot.run_until_disconnected(),
    )

Docker Deployment

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "-m", "bot.main"]

# docker-compose.yml
services:
  bot:
    build: .
    env_file: .env
    restart: always
    depends_on:
      db: { condition: service_healthy }
      redis: { condition: service_started }

See Also

• docker-devops skill for production Docker deployment
• database-patterns skill for bot data persistence
• prompt-engineering skill for LLM-powered bot features