Cap 3: Bot Básico

Por: Artiko
telegrampythonhandlerscomandosbot

bot.py — Punto de entrada

import logging
from telegram.ext import Application, CommandHandler, MessageHandler, filters
from config import TELEGRAM_TOKEN
from handlers.commands import start, help_cmd, reset
from handlers.messages import handle_message

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.INFO,
)
logging.getLogger("httpx").setLevel(logging.WARNING)  # silenciar logs de red

def main():
    app = Application.builder().token(TELEGRAM_TOKEN).build()

    # Comandos
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("help", help_cmd))
    app.add_handler(CommandHandler("reset", reset))

    # Mensajes de texto (que no sean comandos)
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    print("Bot iniciado. Ctrl+C para detener.")
    app.run_polling(allowed_updates=["message", "callback_query"])

if __name__ == "__main__":
    main()

handlers/commands.py

from telegram import Update
from telegram.ext import ContextTypes

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    name = update.effective_user.first_name
    await update.message.reply_text(
        f"Hola {name}! Soy tu asistente con Claude.\n\n"
        "Escríbeme cualquier pregunta o tarea.\n\n"
        "Comandos disponibles:\n"
        "/help — esta ayuda\n"
        "/reset — limpiar conversación"
    )

async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        "*Comandos*\n"
        "/start — iniciar el bot\n"
        "/reset — limpiar el historial de esta conversación\n"
        "/help — mostrar esta ayuda\n\n"
        "Escríbeme en lenguaje natural para hablar con Claude.",
        parse_mode="Markdown",
    )

async def reset(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # Limpiamos el historial del usuario (se implementa en cap. 6)
    context.user_data.clear()
    await update.message.reply_text("Conversación reiniciada.")

handlers/messages.py

Por ahora responde con eco. En el capítulo 4 conectamos Claude:

from telegram import Update
from telegram.ext import ContextTypes

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    text = update.message.text
    user_id = update.effective_user.id

    # Indicador de "escribiendo..."
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )

    # Por ahora: eco simple
    await update.message.reply_text(f"Recibí: {text}")

Ejecutar el bot

python bot.py
# Bot iniciado. Ctrl+C para detener.
# 2026-04-01 10:00:00 - apscheduler... INFO - Scheduler started

Abre Telegram, busca tu bot por username y escríbele /start.

Manejo de errores global

Agrega un error handler para no crashear en errores inesperados:

# En bot.py, antes de run_polling

async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE):
    import traceback
    tb = "".join(traceback.format_exception(None, context.error, context.error.__traceback__))
    logging.error(f"Error al procesar update:\n{tb}")
    if isinstance(update, Update) and update.effective_message:
        await update.effective_message.reply_text(
            "Ocurrió un error. Intenta de nuevo."
        )

app.add_error_handler(error_handler)

Mostrar “escribiendo…” durante respuestas largas

El ChatAction.TYPING dura 5 segundos. Para respuestas largas (como las de Claude), actualízalo periódicamente:

import asyncio
from telegram.constants import ChatAction

async def typing_loop(bot, chat_id: int, stop_event: asyncio.Event):
    """Mantiene el indicador de escritura activo."""
    while not stop_event.is_set():
        await bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING)
        await asyncio.sleep(4)  # renovar antes de que expire (5s)

# Uso en un handler:
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    stop = asyncio.Event()
    typing_task = asyncio.create_task(
        typing_loop(context.bot, update.effective_chat.id, stop)
    )
    try:
        respuesta = await llamar_claude(update.message.text)
        await update.message.reply_text(respuesta)
    finally:
        stop.set()
        typing_task.cancel()

Filtros útiles

python-telegram-bot incluye filtros para segmentar qué mensajes procesa cada handler:

from telegram.ext import filters

# Solo texto (sin comandos)
filters.TEXT & ~filters.COMMAND

# Solo fotos
filters.PHOTO

# Solo documentos
filters.Document.ALL

# Solo chats privados (no grupos)
filters.ChatType.PRIVATE

# Solo grupos
filters.ChatType.GROUPS

# Por usuario específico
filters.User(user_id=123456789)

# Combinar
filters.TEXT & filters.ChatType.PRIVATE & ~filters.COMMAND

Estructura de respuesta con Markdown

Telegram soporta MarkdownV2 y HTML. Para código de Claude, usa MarkdownV2:

from telegram.helpers import escape_markdown

async def reply_with_code(update: Update, code: str, language: str = "python"):
    # Formatear bloques de código
    escaped = escape_markdown(code, version=2)
    msg = f"```{language}\n{escaped}\n```"
    await update.message.reply_text(msg, parse_mode="MarkdownV2")

Caracteres que requieren escape en MarkdownV2: _ * [ ] ( ) ~ > # + - = | { } . !

La función escape_markdown(text, version=2) lo hace automáticamente.

Estado actual del flujo

sequenceDiagram
    participant U as Usuario
    participant TG as Telegram
    participant BOT as Bot

    U->>TG: /start
    TG->>BOT: Update (command)
    BOT->>TG: "Hola! Soy tu asistente..."
    TG->>U: mensaje

    U->>TG: "Hola mundo"
    TG->>BOT: Update (message)
    BOT->>TG: "Recibí: Hola mundo"
    TG->>U: eco

En el próximo capítulo reemplazamos el eco por una llamada real a Claude.