Cap 3: Bot Básico
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.