Integração com Bots do Telegram

O TelegramChannel conecta enxames do chatflow-agent a bots do Telegram. Desenvolvido sobre a biblioteca assíncrona python-telegram-bot, ele gerencia polling, isolamento de sessões por ID de chat, fragmentação automática de mensagens e políticas de segurança para chats em grupo.

InstalaçãoCopy Anchor Link

Instale o chatflow-agent com o extra telegram para incluir as dependências necessárias (python-telegram-bot>=21.0):

pip install "chatflow-agent[telegram]"

Configuração Rápida do 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 seu agente
support_agent = Agent(
    name="SupportBot",
    instructions="You are a helpful customer support representative.",
)

# Configurar armazenamento persistente de sessões
store = SQLiteSessionStore("telegram_sessions.db")
runner = Runner(starting_agent=support_agent, session_store=store)

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

# Vincular runner e executar loop de polling
telegram.attach(runner)

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

Parâmetros de Configuração do CanalCopy Anchor Link

O construtor do TelegramChannel aceita as seguintes opções:

Comandos Integrados e Segurança em GruposCopy Anchor Link

O TelegramChannel inclui manipuladores pré-configurados para comandos fundamentais:

Comando /startCopy Anchor Link

Responde com start_message para recepcionar novos usuários.

Comando /resetCopy Anchor Link

Limpa o histórico da sessão no runner.session_store para o ID de chat ativo, permitindo que os usuários reiniciem suas conversas de forma limpa.

# Lógica de segurança 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

Fragmentação de Mensagens de Saída (split_message)Copy Anchor Link

O Telegram impõe um limite rígido de 4096 caracteres por mensagem de texto. Quando um agente gera análises extensas, blocos de código ou tabelas, o envio direto do texto completo faz com que a API do Telegram retorne erro HTTP 400.

O TelegramChannel canaliza respostas automaticamente por meio 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 progressivamente por \n\n, \n, pontos ou espaços para manter a legibilidade

O canal entrega os fragmentos em sequência, garantindo que parágrafos, frases e listas nunca sejam interrompidos abruptamente no meio de uma palavra.

Indicadores de Digitação e Experiência do UsuárioCopy Anchor Link

Enquanto o turno do modelo é executado (incluindo invocações de ferramentas e transferências), o TelegramChannel envia periodicamente indicadores ChatAction.TYPING a cada 4 segundos. Os usuários visualizam o status nativo "Digitando..." em seu cliente Telegram até que a resposta completa seja entregue.

Execução em Loops de Eventos ExistentesCopy Anchor Link

Caso esteja integrando o bot em um serviço asyncio existente ou aplicação FastAPI, passe close_loop=False para run():

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