Telegram Bot Creator

When to use this skill

• Use when the user request matches this skill's domain and capabilities.
• Use when this workflow or toolchain is explicitly requested.

When not to use this skill

• Do not use when another skill is a better direct match for the task.
• Do not use when the request is outside this skill's scope.

Overview

Build production-ready Telegram bots that bridge backend agents and pipelines to Telegram chat. This skill handles complete bot creation: from scaffolding, to agent integration, to deployment.

Key architectural pattern: Backend agents do the business logic; Telegram bot provides the chat interface and state management.

Quick Start (5 minutes)

1. Choose Your Framework

# Option A: Modern async (recommended for complex agents)
pip install aiogram python-dotenv

# Option B: Simpler, more established
pip install python-telegram-bot python-dotenv

Framework comparison: See frameworks.md for detailed comparison.

2. Get Telegram Bot Token

1. Message @BotFather on Telegram
2. Create new bot:

   /newbot

3. Copy token:

   123456:ABC-DEF...

3. Scaffold Bot Project

python3 scripts/bot_initializer.py --name my_bot --framework aiogram

This creates:

• main.py

  - Bot entry point with example handlers

• .env.example

  - Config template

• requirements.txt

  - Dependencies

4. Implement Your Agent

Copy

scripts/agent_adapter.py

to your project, then subclass it:

from agent_adapter import TelegramAgentAdapter

class MyAgent(TelegramAgentAdapter):
    async def process_agent_request(self, message: str, user_id: str) -> str:
        # Your agent logic here
        # Can call LLMs, external services, databases, etc.
        return f"Agent response: {message}"

5. Run Locally

export TELEGRAM_BOT_TOKEN="your_token_here"
python3 main.py

Find your bot on Telegram and send a message!

---

Use Cases & Patterns

Pattern 1: Simple Routing Agent

Route messages to different handlers based on content.

When to use: Customer support, FAQ bots, request classification

See: Agent integration guide → Pattern 1

class SupportRouter(TelegramAgentAdapter):
    async def process_agent_request(self, message: str, user_id: str) -> str:
        if "refund" in message.lower():
            return "📧 Routing to refunds team..."
        elif "technical" in message.lower():
            return "🔧 Routing to support..."
        return "ℹ️ How can I help?"

Pattern 2: LLM-Based Conversational Bot

AI-powered chat that understands natural language.

When to use: Virtual assistants, customer support, Q&A

See: Agent integration guide → Pattern 2

Easy setup with OpenRouter (free Claude access):

import httpx

class ChatBot(TelegramAgentAdapter):
    async def process_agent_request(self, message: str, user_id: str) -> str:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://openrouter.io/api/v1/chat/completions",
                headers={"Authorization": f"Bearer {OPENROUTER_KEY}"},
                json={
                    "model": "claude-3-5-sonnet:free",
                    "messages": [{"role": "user", "content": message}],
                    "max_tokens": 1024
                }
            )
        data = response.json()
        return data['choices'][0]['message']['content']

Pattern 3: Multi-Step Workflow

Guide users through sequential steps (onboarding, forms, processes).

When to use: Onboarding, multi-step forms, guided workflows

See: Agent integration guide → Pattern 3, also

assets/example_bot_comprehensive.py

class OnboardingBot(TelegramAgentAdapter):
    async def process_agent_request(self, message: str, user_id: str) -> str:
        state = self.get_user_state(user_id)
        step = state.get("step", 0)

        if step == 0:
            self.update_user_state(user_id, "name", message)
            self.update_user_state(user_id, "step", 1)
            return "Thanks! Now what's your email?"

        elif step == 1:
            self.update_user_state(user_id, "email", message)
            self.update_user_state(user_id, "step", 2)
            return "🎉 Onboarding complete!"

Pattern 4: External Agent Integration

Your agent logic lives on a separate service; bot just forwards requests.

When to use: Large agents, existing backends, microservices

See: Agent integration guide → Pattern 4

class RemoteAgent(TelegramAgentAdapter):
    async def process_agent_request(self, message: str, user_id: str) -> str:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://your-api.com/agent/process",
                json={"user_id": user_id, "message": message},
                timeout=30.0
            )
        return response.json()["response"]

Pattern 5: Agentskills.io Protocol

Integrate agents following the agentskills.io specification.

See: Agent integration guide → Pattern 5

---

Deployment

Development: Polling (Simplest)

# Set token
export TELEGRAM_BOT_TOKEN="your_token"

# Run locally
python3 main.py

Advantages: No setup, works on localhost, perfect for testing Limitations: Higher latency, not ideal for high volume

Production: Polling (Scaled)

Run on cloud platform (keep bot running):

• DigitalOcean App Platform
• Google Cloud Run (persistent container)
• Heroku
• AWS ECS

Production: Webhooks (High Scale)

Telegram sends updates directly to your server. Lower latency, more scalable.

Requirements:

• HTTPS endpoint with SSL certificate
• Public domain

Setup: See deployment.md for full webhook guide and Cloud Run/Lambda examples.

Quick webhook with aiogram:

from aiohttp import web

WEBHOOK_URL = "https://yourdomain.com/webhook/telegram"

async def on_startup():
    await bot.set_webhook(url=WEBHOOK_URL, secret_token="random-secret")

# Run webhook server...

---

State Management

Simple: In-Memory (Development)

Built into

agent_adapter.py

. Uses

get_user_state()

and

update_user_state()

methods.

async def handle_message(self, user_id: str, message: str) -> str:
    state = self.get_user_state(user_id)
    # Process with state
    self.update_user_state(user_id, "key", "value")
    return response

Limitation: Data lost on restart. Only for development.

Production: Redis

Fast, distributed, recommended for most bots.

# Quick Redis (Docker)
docker run -d -p 6379:6379 redis:7-alpine

# Or managed cloud: Upstash, Redis Labs, AWS ElastiCache

See: state_management.md for full Redis integration code

Production: PostgreSQL

For structured data, complex queries, compliance.

See: state_management.md for schema and integration

Docker Compose (All-in-One)

# Includes bot, Redis, PostgreSQL
docker-compose up

See:

assets/docker-compose.yml

for configuration

---

Framework-Specific Features

Aiogram: FSM (Finite State Machine)

For multi-step conversations with explicit state transitions:

from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.redis import RedisStorage

class MyStates(StatesGroup):
    step1 = State()
    step2 = State()

# Use Redis for distributed state
storage = RedisStorage.from_url("redis://localhost")
dp = Dispatcher(storage=storage)

@dp.message(MyStates.step1)
async def handle_step1(message: types.Message, state: FSMContext):
    await state.set_state(MyStates.step2)
    # ...

Aiogram: Middleware

Route messages intelligently before handlers:

@dp.middleware()
async def agent_selector(handler, event, data):
    user_id = event.from_user.id
    data["agent"] = get_agent_for_user(user_id)
    return await handler(event, data)

Python-telegram-bot: Handlers

Simple handler registration:

app.add_handler(CommandHandler("start", start))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, message_handler))

---

Common Tasks

Task: Add Conversation History to LLM

Keep messages in memory or database for context-aware responses:

state = self.get_user_state(user_id)
messages = state.get("conversation", [])
messages.append({"role": "user", "content": message})

# Call LLM with full history
response = await llm_call(messages)

messages.append({"role": "assistant", "content": response})
self.update_user_state(user_id, "conversation", messages)

Task: Handle Long-Running Operations

Async task without blocking user:

@dp.message()
async def handle_long_operation(message: types.Message):
    await message.answer("Processing... 🔄")

    # Run async operation
    result = await long_operation()

    # Send result
    await message.answer(f"Done! {result}")

Task: Inline Buttons & Callbacks

from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup

@dp.message()
async def show_options(message: types.Message):
    buttons = InlineKeyboardMarkup(inline_keyboard=[
        [InlineKeyboardButton(text="Option 1", callback_data="opt1")],
        [InlineKeyboardButton(text="Option 2", callback_data="opt2")]
    ])
    await message.answer("Choose:", reply_markup=buttons)

@dp.callback_query()
async def handle_callback(query: types.CallbackQuery):
    if query.data == "opt1":
        await query.answer("You chose option 1!")

Task: Error Handling & Logging

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dp.message()
async def safe_handler(message: types.Message):
    try:
        response = await risky_operation()
        await message.answer(response)
    except Exception as e:
        logger.error(f"Error for user {message.from_user.id}: {e}")
        await message.answer("⚠️ Error processing request. Try again later.")

Task: Rate Limiting

Prevent spam:

from collections import defaultdict
from datetime import datetime, timedelta

user_messages = defaultdict(list)

@dp.message()
async def rate_limited_handler(message: types.Message):
    user_id = message.from_user.id
    now = datetime.now()

    # Remove old messages (older than 1 minute)
    user_messages[user_id] = [
        ts for ts in user_messages[user_id]
        if now - ts < timedelta(minutes=1)
    ]

    # Check limit (5 messages per minute)
    if len(user_messages[user_id]) >= 5:
        await message.answer("⏱️ Too many requests. Wait a minute.")
        return

    user_messages[user_id].append(now)
    # Process request...

---

Reference Material

Framework Setup & Comparison: → frameworks.md

Agent Integration Patterns: → agent_integration.md

Polling vs Webhooks & Deployment: → deployment.md

State Management (Redis, PostgreSQL, etc): → state_management.md

---

Scripts & Templates

Agent Base Class:

scripts/agent_adapter.py

- Subclass this to create your agent. Handles state, typing indicators, error wrapping.

Bot Initializer:

scripts/bot_initializer.py

- Scaffold new bot project with framework of choice.

Example Bot:

assets/example_bot_comprehensive.py

- Full example with support routing, LLM chat, and multi-step onboarding.

Docker:

assets/Dockerfile

- Container image for bot

assets/docker-compose.yml

- Full stack (bot + Redis + PostgreSQL)

Config:

assets/.env.example

- Environment variables template

---

Quick Reference: Create a New Bot

# 1. Initialize project
python3 scripts/bot_initializer.py --name my_bot --framework aiogram

# 2. Copy adapter
cp scripts/agent_adapter.py my_bot/

# 3. Update main.py: Replace YourAgent class with your logic

# 4. Set token
export TELEGRAM_BOT_TOKEN="token_from_botfather"

# 5. Run
cd my_bot
python3 main.py

# 6. Find bot on Telegram and send message

---

Troubleshooting

Bot doesn't respond?

• Check TELEGRAM_BOT_TOKEN is set correctly
• Verify bot is running (no errors in logs)
• Search for bot in Telegram by username from BotFather

Agent returning errors?

• Check agent logic (add print statements, see logs)
• Use agent_adapter error handling:

  _error_response()

• Test agent separately before integrating

Slow responses?

• Polling adds latency (polling_timeout). Use webhooks for faster response
• Check agent logic takes too long. Consider async operations
• Redis is much faster than polling each time

State not persisting?

• In-memory state (default) is lost on restart. Use Redis/PostgreSQL
• Docker Compose includes Redis/PostgreSQL setup

Webhook not receiving messages?

• Check HTTPS certificate is valid (not self-signed in production)
• Verify webhook URL is correct and accessible
• See deployment.md for webhook setup details

---

Next Steps

1. Choose framework: aiogram (modern) or python-telegram-bot (simple)
2. Scaffold project: Use

   bot_initializer.py

3. Implement agent: Subclass

   TelegramAgentAdapter

   , implement

   process_agent_request()

4. Add handlers: Message routing, commands, callbacks
5. Test locally: Use polling (simplest)
6. Scale: Add state management (Redis), then webhooks if needed
7. Deploy: Docker + DigitalOcean/AWS/GCP

For specific patterns, agent types, or deployment scenarios, see reference material linked above.