Integración con Bots de Telegram

TelegramChannel conecta enjambres de chatflow-agent con bots de Telegram. Impulsado por la biblioteca asíncrona python-telegram-bot, gestiona polling, aislamiento de sesiones por ID de chat, fragmentación automática de mensajes y políticas de seguridad para chats grupales.

InstalaciónCopy Anchor Link

Instala chatflow-agent con el extra telegram para incluir las dependencias requeridas (python-telegram-bot>=21.0):

pip install "chatflow-agent[telegram]"

Configuración Rápida del BotCopy Anchor Link

import os
from chatflow_agent import Agent, Runner
from chatflow_agent.channels import TelegramChannel
from chatflow_agent.core.memory import SQLiteSessionStore

# Definir tu agente
support_agent = Agent(
    name="SupportBot",
    instructions="You are a helpful customer support representative.",
)

# Configurar almacenamiento persistente de sesiones
store = SQLiteSessionStore("telegram_sessions.db")
runner = Runner(starting_agent=support_agent, session_store=store)

# Inicializar canal de Telegram
telegram = TelegramChannel(
    token=os.environ.get("TELEGRAM_BOT_TOKEN", "your-bot-token"),
    start_message="Welcome! How can I assist you today?",
)

# Vincular runner y ejecutar bucle de polling
telegram.attach(runner)

if __name__ == "__main__":
    print("Starting Telegram bot polling...")
    telegram.run()

Parámetros de Configuración del CanalCopy Anchor Link

El constructor de TelegramChannel acepta las siguientes opciones:

Comandos Integrados y Seguridad en Chats GrupalesCopy Anchor Link

TelegramChannel incluye manejadores preconfigurados para comandos esenciales:

Comando /startCopy Anchor Link

Responde con start_message para dar la bienvenida a nuevos usuarios.

Comando /resetCopy Anchor Link

Limpia el historial de la sesión en runner.session_store para el ID de chat activo, permitiendo a los usuarios reiniciar su conversación limpiamente.

# Lógica de seguridad aplicada internamente por TelegramChannel._handle_reset:
# member = await context.bot.get_chat_member(chat_id=chat.id, user_id=user.id)
# if member.status not in ("creator", "administrator"):
#     await update.message.reply_text("Only group admins can reset the conversation.")
#     return

Fragmentación de Mensajes Salientes (split_message)Copy Anchor Link

Telegram impone un límite estricto de 4096 caracteres por mensaje de texto. Cuando un agente genera análisis extensos, bloques de código o tablas, enviar el texto completo de forma directa hace que la API de Telegram responda con error HTTP 400.

TelegramChannel canaliza automáticamente las respuestas a través de split_message:

# [min-version: 0.3.2]
# [executable]
from chatflow_agent.channels import split_message

long_text = "Detailed agent analysis..." * 200
chunks = split_message(long_text, max_len=4096)
# Divide progresivamente por \n\n, \n, puntos o espacios para preservar legibilidad

El canal entrega los fragmentos secuencialmente, asegurando que párrafos, oraciones y listas nunca queden truncados a mitad de palabra.

Indicadores de Escritura y Experiencia de UsuarioCopy Anchor Link

Mientras el turno del modelo se ejecuta (incluyendo invocaciones de herramientas y transferencias), TelegramChannel emite periódicamente indicadores ChatAction.TYPING cada 4 segundos. Los usuarios visualizan el estado nativo "Escribiendo..." en su cliente de Telegram hasta que la respuesta completa sea entregada.

Ejecución en Bucles de Eventos ExistentesCopy Anchor Link

Si estás integrando el bot dentro de un servicio asyncio existente o una aplicación FastAPI, pasa close_loop=False a run():

telegram.run(drop_pending_updates=True, close_loop=False)