User-bot MTProto em Produção com FastAPI + Telethon: WARP para Contornar DPI e 5 Armadilhas do Telegram

User-bot MTProto em Produção com FastAPI + Telethon: WARP para Contornar DPI e 5 Armadilhas do Telegram

Na maioria dos tutoriais sobre bots do Telegram, tudo começa com um trecho de código: obtém-se um token de @BotFather, instala-se python-telegram-bot ou aiogram, escreve-se um handler, e faz-se o deploy. Isso é a Bot API. E em 90% das tarefas, isso é suficiente.

Mas então surge uma tarefa que a Bot API não cobre em princípio: criar programaticamente um supergrupo para um projeto específico e adicionar as pessoas necessárias por @username, e fazer isso dezenas de vezes por dia. A Bot API não pode fazer isso nem teoricamente - não há um método "criar grupo", nem um método "adicionar usuário ao grupo". Você vai para a documentação completa da Telegram API procurar uma solução, e se depara com a seção channels.createChannel / channels.inviteToChannel sob MTProto, e aí começa uma história completamente diferente - não a Bot API, mas um user-bot via telethon.

Neste artigo, analiso como criamos um user-bot MTProto em produção com FastAPI + Telethon. Por baixo do capô: Cloudflare WARP para contornar DPI (sem ele, é simplesmente impossível conectar-se a partir de um VPS russo), cliente Singleton com keepalive, cache na memória de resolução de usuários e 5 limitações do Telegram que só quem entrou lá com os próprios pés conhece. Um serviço real em produção para um cliente na área de construção/instalação, que atende a uma conexão Planfix → grupos do Telegram para cada projeto. O serviço foi escrito em Python 3.11. Stack: Telethon 1.43.2, FastAPI 0.136.1, Uvicorn 0.46.0, Pydantic 2.13.4. Em um VPS sob systemd, para o exterior via Cloudflare Tunnel. É chamado de n8n através de um nó HTTP.

Quando um user-bot MTProto é realmente necessário

Ir para MTProto só faz sentido quando a Bot API padrão atinge seus limites. E não é "preguiça de entender", mas sim a ausência de métodos em princípio - existem operações específicas que a Bot API não cobre de forma alguma.

O que a Bot API faz bem:

• Enviar e editar mensagens
• Processar botões inline e solicitações de callback
• Gerenciar grupos/canais existentes (se o bot for adicionado como administrador):
  • banChatMember
  • unbanChatMember
  • pinChatMessage
  • setChatTitle
• Receber chat_join_request e approveChatJoinRequest / declineChatJoinRequest - ou seja, controle automático de entrada por invite-link

O que a Bot API não faz:

• Criar supergrupos. De jeito nenhum. Não há método
• Adicionar usuários a um grupo. O bot não pode "arrastar" ninguém para dentro - apenas aceitar solicitações de quem veio por invite-link
• Resolver @username ↔ user_id. De forma alguma
• Iniciar um diálogo com o usuário primeiro (precisa de /start do usuário)

Cada uma dessas operações está em MTProto. Se pelo menos duas delas estiverem na tarefa, isso já é um argumento a favor de um user-bot.

No nosso caso, o cliente (área de instalação/reparo/construção, dezenas de projetos simultâneos com empreiteiros e clientes) tinha uma solicitação: cada nova tarefa no Planfix deveria gerar automaticamente um chat do Telegram com especialistas do cartão da tarefa. Para um projeto, 3-10 pessoas. Antes, essa tarefa era feita manualmente pelo gerente de projeto/gerente de escritório: você cria um grupo, procura pessoas na lista de contatos, adiciona, depois a composição do projeto muda - você corrige novamente manualmente. O bot assume esse papel em quase 100%.

Especificamente, através do MTProto, cobrimos:

• Criação de supergrupo (somente MTProto)
• Adição de participantes por @username do cartão Planfix (somente MTProto)
• Resolve id ↔ @username (somente MTProto)

A Bot API na mesma conexão é usada para gerenciamento operacional do grupo já criado:

• banChatMember / unbanChatMember ao alterar a composição do projeto
• approveChatJoinRequest para aprovação automática dos que retornam via invite-link
• Fixar mensagens e enviar status

Resulta em uma arquitetura híbrida: criação e preenchimento do grupo via user-bot MTProto, o resto via Bot API.

Arquitetura do serviço

Antes do código, uma visão geral. O que existe em produção:

┌──────────────────────────────────────┐
│ Planfix (fonte de evento) │
└────────────────┬─────────────────────┘
│ webhook
▼
┌──────────────────────────────────────┐
│ n8n (orquestrador, 56 nós) │
└────────────────┬─────────────────────┘
│ HTTP + X-API-Key
▼
┌──────────────────────────────────────┐
│ Serviço FastAPI (mtproto-api) │
│ ├─ TelegramClient Singleton │
│ ├─ /create-group /add-users │
│ │ /resolve-user /health │
│ └─ keepalive 180s + ensure_connected│
└────────────────┬─────────────────────┘
│ MTProto
▼
┌──────────────────────────────────────┐
│ Cloudflare WARP (SOCKS5 :40000) │
│ contornando RKN-DPI │
└────────────────┬─────────────────────┘
▼
┌──────────────────────────────────────┐
│ Telegram MTProto API │
└──────────────────────────────────────┘

Externamente, o serviço está acessível apenas via Cloudflare Tunnel (mtproto.example.ru → localhost:8080), a porta 8080 no próprio VPS não é exposta para o exterior. Autenticação no nível HTTP - cabeçalho X-API-Key (gerado via openssl rand -base64 32), validado na dependência FastAPI antes de cada endpoint.

Endpoints:

A solicitação para /create-group se parece com isto:

json
POST /create-group
X-API-Key: <secret>
Content-Type: application/json

{
 "title": "Projeto Objeto-42 Moscou",
 "description": "Equipe do projeto",
 "admins": ["@admin_username"],
 "users": ["@team_member1", "@team_member2"],
 "request_approval": true
}

Na resposta - chat_id no formato da Bot API (com o prefixo -100, para chamadas subsequentes da Bot API), invite_link, e algo crítico para produção - os dicionários *_errors com pereason se alguém não conseguiu ser adicionado:

json
{
 "status": "ok",
 "chat_id": -1001234567890,
 "invite_link": "https://t.me/+abc...",
 "bot_added": true,
 "admins_added": ["admin_username"],
 "admins_errors": {},
 "users_added": ["team_member1"],
 "users_errors": {"team_member2": "user_privacy_restricted"}
}

Ou seja, status HTTP 200, o grupo foi criado, mas team_member2 não foi adicionado com a razão especificada. n8n pode tentar um fallback - enviar um convite através do bot pessoalmente para cada usuário de *_errors, e do lado do usuário, este será um invite-link pessoal, no qual ele mesmo clicará e entrará no grupo.

Cloudflare WARP para contornar DPI

Agora, para a parte mais difícil - por que tudo isso funciona em um VPS russo.

O principal bloqueador: RKN-DPI no nível do protocolo MTProto

Se você tentar executar um cliente Telethon de um VPS russo diretamente, verá algo bonito:

ConnectionError: Connection to Telegram failed 5 time(s)

Não resolve o hostname, não é um problema de TLS, não é um firewall. O problema é que o RKN-DPI reconhece o protocolo MTProto pela assinatura do pacote e corta a conexão no nível TCP. A ofuscação embutida no Telethon (ConnectionTcpObfuscated) não reage a isso - o DPI a vê perfeitamente.

Ao configurar o serviço pela primeira vez sem contornar o bloqueio, tentei:

• DCs alternativos (datacenters do Telegram) - bloqueados da mesma forma
• Todas as opções de connection no Telethon (ConnectionTcpFull, ConnectionTcpIntermediate, ConnectionTcpAbridged, ConnectionTcpObfuscated) - o resultado é o mesmo

A solução final de trabalho é Cloudflare WARP no modo SOCKS5 em 127.0.0.1:40000. O que é isso:

• WARP - cliente VPN Cloudflare, há uma versão de console warp-cli para Linux. Gratuito
• Após warp-cli connect - todo o tráfego de saída do VPS passa pela rede CF. O IP externo muda para CF Moscow (o DPI não o vê, porque o tráfego é criptografado entre o VPS e o CF e, externamente, se parece com HTTPS normal no CF)
• Telethon é configurado para usar SOCKS5 através da porta local 40000 (modo SOCKS padrão WARP)

No Telethon, isso se parece com isto:

python
from telethon import TelegramClient
from telethon.network.connection import ConnectionTcpFull
import socks

PROXY = (socks.SOCKS5, "127.0.0.1", 40000)

client = TelegramClient(
 "session",
 API_ID,
 API_HASH,
 connection=ConnectionTcpFull,
 proxy=PROXY,
)

Depois disso, client.start() conecta sem problemas. Verificação da atividade do WARP:

bash
warp-cli status # esperado: Connected
curl --socks5 127.0.0.1:40000 -m 10 https://ipinfo.io/
# deve retornar o IP no formato Cloudflare Moscow

WARP inicia automaticamente após systemctl enable warp-svc. Durante todo o tempo de funcionamento do serviço, o WARP nunca falhou - inicia com o VPS, funciona continuamente. O único risco que vi - após o reboot do VPS, o WARP não sobe instantaneamente, e se mtproto-api.service iniciar antes, a primeira conexão ao Telegram falha com ConnectionError. A solução - dependência de inicialização mtproto-api.service de warp-svc:

[Unit]
Description=MTProto FastAPI service
After=network-online.target warp-svc.service
Requires=warp-svc.service

[Service]
ExecStart=/opt/mtproto-api/venv/bin/uvicorn app:app --host 127.0.0.1 --port 8080
Restart=on-failure

Depois disso, reconexões periódicas no journald - isso é keepalive após o tempo limite de inatividade, não incidentes WARP.

Alternativas ao WARP - um VPS pago no exterior com proxy para a Federação Russa ou seu próprio MTProxy em um host estrangeiro. Mais caro, mais complicado, mais frequentemente interrompido, além de outro componente na cadeia. WARP é gratuito, nativamente suportado no Linux e remove o problema de DPI no nível da rede.

TelegramClient Singleton - por que isso é crítico

A regra básica que estabeleci para mim com uma linha vermelha espessa após duas invalidações de sessão:

um MTPROTO_SESSION = um TelegramClient em um determinado momento.

Sem processos Python paralelos com a mesma string de sessão.

O que acontece se a regra for violada (incidente real na fase de desenvolvimento do serviço): executei um script de depuração separado check_user.py em paralelo com o mtproto-api já em execução. Ambos os clientes estavam contatando o Telegram com a mesma string de sessão. A infraestrutura de segurança do Telegram pega isso como atividade suspeita ("uma conta está conectando de dois lugares ao mesmo tempo") e envia AUTH_KEY_UNREGISTERED para o mais antigo no tempo de conexão. A sessão é invalidada, o serviço falha, restauração via reauth.

Duas vezes em um dia caí nessa armadilha (depois da primeira vez pensei "acaso", executei testes resolve-user em paralelo - invalidação novamente). Depois disso, surgiu uma regra simples na forma de um arquivo feedback_mtproto_rules.md no repositório do serviço:

É proibido executar quaisquer scripts com Telethon na mesma MTPROTO_SESSION enquanto o mtproto-api estiver funcionando. Para testes interativos - uma sessão separada em um número de teste ou parar o serviço antes de iniciar.

No código do serviço, isso é implementado pelo padrão Singleton via FastAPI lifespan:

python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from telethon import TelegramClient

@asynccontextmanager
async def lifespan(app: FastAPI):
 # startup
 app.state.client = TelegramClient(
 SESSION, API_ID, API_HASH,
 connection=ConnectionTcpFull,
 proxy=(socks.SOCKS5, "127.0.0.1", 40000),
 )
 await app.state.client.start()
 # warmup entity cache
 async for _ in app.state.client.iter_dialogs(limit=300):
 pass
 # background keepalive
 app.state.keepalive_task = asyncio.create_task(_keepalive(app.state.client))
 yield
 # shutdown
 app.state.keepalive_task.cancel()
 await app.state.client.disconnect()

app = FastAPI(lifespan=lifespan)

Cada endpoint pega o cliente de request.app.state.client, não há client = TelegramClient(...) dentro dos manipuladores. Garante um único exemplo por processo.

Restauração após invalidação (~5 minutos):

bash
systemctl stop mtproto-api
pgrep -af telethon # certifique-se de que não haja processos paralelos

Script de reauth atômico via systemd-run --user --scope (não nohup / tmux via SSH, caso contrário, o script morrerá se a sessão SSH for interrompida e a conta ficará em estado semi-autorizado)

Obtemos uma nova string de sessão.

Importante: execute do mesmo IP que a produção (o mesmo WARP), caso contrário, a segurança do Telegram define a flag "login de um novo IP" e não permite a restauração via SMS

Depois daquele caso, a regra é estritamente seguida e não houve mais nenhuma invalidação.

Mais quatro armadilhas do Telegram

Com DPI e invalidação resolvidos. Agora, mais curto - quatro restrições restantes nas quais pessoalmente tropecei.

Contato mútuo e por que novos grupos são preenchidos mais facilmente

InviteToChannelRequest (também channels.inviteToChannel em MTProto) pode lançar USER_NOT_MUTUAL_CONTACT se o usuário que está sendo adicionado tiver privacidade Meus Contatos ou Ninguém, e a conta da sessão não estiver em seus contatos. Na documentação oficial do Telegram, os códigos de erro são listados (USER_NOT_MUTUAL_CONTACT, USER_PRIVACY_RESTRICTED, USER_BLOCKED, USER_KICKED e mais quinze), mas as condições exatas para o disparo não são reveladas - apenas descrições de texto curtas.

Observação empírica da prática: nas primeiras horas após a criação de um supergrupo, o Telegram permite a adição via inviteToChannel com mais frequência do que ao adicionar a um grupo existente com uma semana/mês. Não se pode confiar nisso - esse comportamento não é registrado na documentação, pode mudar a qualquer momento e sem aviso prévio. Arquitetonicamente, você sempre precisa ter um fallback para invite-link com request_approval=true.

Conclusão prática para o fluxo de trabalho:

• Ao criar um grupo, adicione imediatamente todos os participantes iniciais (a maior parte passará)
• Para todos que retornaram em users_errors - envie imediatamente um invite-link via bot no privado
• Para todos os novos membros dias/semanas após a criação - use apenas invite-link + aprovação automática via Bot API no evento chat_join_request

O principal: não construa uma arquitetura com base na suposição "teremos uma janela quando pudermos adicionar qualquer pessoa". A janela pode fechar - e fechará quando você precisar adicionar um usuário-chave ao projeto.

Após a remoção do grupo, a adição repetida não funciona

Se o usuário foi banido via Bot API (banChatMember) ou saiu sozinho - a adição repetida via InviteToChannelRequest não funcionará mais. O Telegram retornará UserNotMutualContactError, mesmo que tecnicamente antes eles fossem um contato mútuo, ou a solicitação passará sem erro, mas o usuário na verdade não aparecerá no grupo. O comportamento depende da combinação específica das configurações de privacidade do usuário e do histórico de suas ações, e não há uma especificação formal para isso.

Cenário: o especialista trabalhou no projeto, o projeto foi fechado, ele foi removido do grupo via Bot API. Um mês depois, o projeto foi retomado - seu nome completo apareceu novamente no cartão da tarefa no Planfix. add-user direto via MTProto - não funciona.

Esquema de trabalho:

• Ao remover um especialista do projeto - banChatMember via Bot API (ou kickChatMember, que para supergrupos é a mesma coisa - o usuário vai para a lista de banidos)
• Ao retornar um especialista ao projeto - primeiro unbanChatMember (remover da lista de banidos), depois formar um invite-link pessoal com request_approval=true e enviar ao usuário no privado via bot

Quando o usuário clica no invite-link, o Telegram envia chat_join_request para o webhook

O fluxo de trabalho da Bot API verifica: o user_id dele está na lista de participantes deste projeto em nossa DataTable? Se sim - approveChatJoinRequest, se não - declineChatJoinRequest

Através de unban + invite-link + aprovação automática, o loop "removido → retornado" funciona de forma transparente para o especialista: ele recebe um convite no privado, clica e entra no grupo em alguns segundos sem um gerente humano no ciclo.

access_hash e por que você não pode adicionar por id numérico

Telethon não pode resolver um usuário arbitrário por user_id numérico sem access_hash. Access_hash é um token de segurança que o Telegram emite apenas para a sessão que já viu este usuário (chat comum, contato, diálogo aberto).

Na Bot API, esse problema é oculto (a API retorna chat_id com o qual você pode trabalhar mais). Em MTProto, access_hash é uma entidade separada que você precisa pegar em algum lugar e armazenar em algum lugar.

Melhor prática:

• sempre passe @username em vez de id em /create-group e /add-users. Telethon resolve o username via client.get_input_entity('@username') - isso funciona sem access_hash, porque o username por si só é suficiente.

Se houver apenas id no sistema (por exemplo, veio do webhook da Bot API), existem duas estratégias:

• Primeiro, execute através de /resolve-user com deep_search=true - iteramos sobre os participantes de todos os grupos onde a sessão existe, procuramos o id necessário
• Adicionar via invite-link (sem a necessidade de access_hash, o usuário clicará sozinho)

Deep_search leva de 5 a 60 segundos (dependendo do número de grupos), por isso o usamos apenas quando id é impossível.

Limites do Telegram e estratégia de repetição

O Telegram não publica limites numéricos exatos para a taxa de operações - na documentação oficial, apenas o máximo de participantes do supergrupo (200.000) é especificado, e o resto você encontra através da exceção FLOOD_WAIT com retry_after específico em segundos para sua sessão em sua situação. Ou seja, o limite é dinâmico e depende da carga real no DC + histórico da conta.

Não temos operações em massa em princípio: a especificidade do negócio - um projeto médio de 3-10 pessoas em um grupo, não fazemos adições em massa. Não encontramos limites exatos para InviteToChannelRequest e criação de supergrupos por hora - não chegamos a FLOOD_WAIT devido à própria natureza da tarefa.

No entanto, estabelecemos proteção contra FLOOD_WAIT no nível do serviço - se a tarefa mudar algum dia e a necessidade de volume surgir:

• asyncio.Lock no FastAPI serializa todas as chamadas do Telegram → uma solicitação Telethon por vez dentro do serviço, sem chamadas de API paralelas
• Em n8n, entre as operações em lote, coloco Wait 3-10 segundos com jitter (embora na carga real, minutos passem entre as tarefas do Planfix)
• No nível do endpoint - retry x2 em ConnectionError
• Em FloodWaitError do Telethon - lemos e.seconds, esperamos seconds + 5, fazemos uma repetição; se cair novamente - retornamos 429 na resposta HTTP com os mesmos seconds no payload, então n8n decide o que fazer

Se algum dia tivermos que fazer uma adição em massa, é melhor enviar um InviteToChannelRequest com uma matriz de todos os usernames de uma vez do que o mesmo volume por meio de chamadas separadas. Uma solicitação de API é mais barata em termos de orçamento de taxa.

Mecânica de produção: o que mais está dentro

Analisei a maior parte das armadilhas, agora os padrões básicos que tornam o serviço estável.

• ensure_connected antes de cada endpoint.

Telethon pode "pendurar em conectado, mas enviar falha" com longos tempos de inatividade. Portanto, antes de qualquer chamada de API - verificação client.is_connected() e reconexão se falhar:

python
async def ensure_connected(client: TelegramClient):
 if not client.is_connected():
 await client.connect()
 # health-check adicional
 try:
 await asyncio.wait_for(client.get_me(), timeout=3)
 except (asyncio.TimeoutError, ConnectionError):
 await client.disconnect()
 await client.connect()

• Keepalive 180 segundos.

Tarefa em segundo plano que executa get_me() a cada três minutos. Sem ele, o Telegram fecha a conexão inativa, e a primeira solicitação após um longo tempo de inatividade falha. Com keepalive - a conexão está sempre ativa.

• Warmup iter_dialogs(limit=300) na inicialização.

Telethon armazena em cache os objetos de entidade de usuários e grupos que encontra. Na inicialização, carregamos os últimos 300 diálogos (~5-10 segundos), o que preenche o cache. Depois disso, client.get_input_entity('@username') funciona instantaneamente, sem round-trip para o Telegram.

• Cache na memória de usuários resolvidos.

Além do cache Telethon, mantenho meu dict[int, dict] na memória do serviço. Quando /resolve-user retorna um resultado, colocamos no cache. É limpo na reinicialização - aceitável, porque o warmup restaurará a maior parte.

• Verificação pós-convite.

Após InviteToChannelRequest, o serviço faz um GetParticipantRequest adicional para garantir que o usuário esteja realmente no grupo. Se em users_added - esta é uma adição confirmada, não "enviamos a solicitação e esperamos". Isso é importante porque o Telegram às vezes diz "ok, adicionado", mas na verdade a adição é revertida sem erro - GetParticipantRequest pega isso e transfere o usuário para *_errors.

Segurança

Regras curtas que não podem ser violadas em um serviço de produção com uma sessão do Telegram:

• Um cliente por sessão - estritamente (analisado acima)
• Não comitar .env com MTPROTO_SESSION no git. Nunca. A string de sessão dá acesso total à conta, equivalente à senha
• Não registrar o corpo das solicitações - contém o user_id, que em conjunto com @username pode revelar dados privados
• Não configurar CORS - o serviço é chamado apenas do lado do servidor de n8n, a interface não vai lá
• systemd-run para operações interativas, especialmente reauth - nohup e tmux via SSH são não confiáveis se a sessão SSH for interrompida, o script morre no meio, a conta permanece em estado semi-autorizado

O que no final

O user-bot MTProto em produção se resume a um conjunto de regras, muitas das quais não são óbvias até que você as viole:

• Cliente Singleton por serviço (não violar)
• WARP SOCKS5 para contornar DPI de um VPS russo
• Contato mútuo e período de carência - projetar a arquitetura para isso
• Opt-out do usuário - fallback para invite-link com aprovação automática
• @username sempre melhor que id numérico
• Keepalive + ensure_connected + warmup - padrões sem os quais não haverá estabilidade

O serviço em si não é complicado - a maior parte do código é o tratamento correto de erros e casos extremos. O mais difícil é entender que essas 5 armadilhas existem, antes que o serviço vá para o cliente em produção.

Fazemos integrações semelhantes em BotKraft - escreva se você tiver um caso semelhante com automação do Telegram para processos de negócios.

Links úteis

• Telethon docs - fonte principal para MTProto em Python
• Telegram API docs (MTProto methods) - referência de métodos
• Cloudflare WARP CLI - instalação no Linux
• Cloudflare Tunnel - publicação do serviço sem portas abertas
• FastAPI lifespan - padrão Singleton para clientes de longa duração

Tags: telethon, mtproto, telegram, python, fastapi, cloudflare, warp, n8n, socks5, planfix