Arquitectura de Bots Telegram Multi-Cliente: De 0 a 14 Bots en Producción

Arquitectura de Bots Telegram Multi-Cliente: De 0 a 14 Bots en Producción

La Paradoja del Crecimiento

Cuando lanzamos nuestro primer bot de Telegram, el código era simple: un webhook directo, lógica hardcodeada, y funcionalidad básica. Funcionó perfectamente... hasta que llegó el segundo cliente. Y el tercero. Para el cuarto bot, estábamos duplicando código, los bugs se multiplicaban, y cada deploy era una ruleta rusa.

Hoy operamos 14 bots en producción atendiendo casos de uso radicalmente diferentes: desde préstamos financieros hasta gestión de chatarra metálica, pasando por agendamiento médico y asistencia empresarial. Todos comparten la misma infraestructura, pero ninguno interfiere con otro. Esta es la historia de cómo construimos esa arquitectura.

El Problema de Escalar Bots de IA

La mayoría de tutoriales de bots Telegram te enseñan el "Happy Path": un bot, un webhook, un servidor. Pero en el mundo real enfrentamos desafíos distintos:

• Aislamiento de datos: Cliente A no puede ver datos de Cliente B
• Personalización extrema: Cada bot necesita su propia lógica de negocio
• Compartir infraestructura: No podemos costear un servidor por bot
• Debugging selectivo: Un bug en Bot X no puede tumbar Bot Y
• Deploy independiente: Actualizar lógica de un cliente sin afectar otros

La Arquitectura de 4 Capas

Después de 3 iteraciones completas, convergimos en esta arquitectura:

Telegram → n8n (Orquestador) → bot-api (Router) → claude_proxy (Brain) → Gemini API

Capa 1: n8n como Orquestador de Entrada

Decisión controversial: No exponemos webhooks directos de Telegram a nuestros servicios. Todo pasa por n8n primero.

¿Por qué? Tres razones:

1. Whitelist de IPs: n8n valida que las peticiones vengan de Telegram Cloud (149.154.160.0/20, 91.108.4.0/22)
2. Rate limiting inteligente: Filtramos spam antes de consumir recursos de IA
3. Transformación de payload: Normalizamos formatos entre diferentes versiones de Bot API

Cada bot tiene su propio workflow n8n. Por ejemplo, el bot de préstamos (`ER-0002`) tiene un workflow que:

• Valida IP origen
• Extrae `chat_id` y `message_type`
• Enruta a diferentes nodos según tipo (text, photo, callback_query)
• Llama a `bot-api` con payload estandarizado

Capa 2: bot-api (FastAPI) como Router Inteligente

FastAPI corre en puerto 3000 dentro de Docker. Su única responsabilidad: routing rápido.

@app.post("/webhook/er0002")
async def webhook_er0002(request: Request):
data = await request.json()
# Forwarding simple, sin lógica de negocio
async with httpx.AsyncClient() as client:
response = await client.post(
"http://bot-proxy:3010/api/gemini/er0002",
json=data,
timeout=30.0
)
return response.json()

Nota clave: No hay API keys aquí. No hay lógica de negocio. Es un proxy puro. Si necesitas debuggear, revisas logs de `bot-proxy`, no de `bot-api`.

Capa 3: claude_proxy (Flask) - El Cerebro

Aquí vive la magia. Flask en puerto 3010, con 12,000+ líneas de Python que manejan:

• Context management: Cada usuario tiene historial aislado en `/data/{project}/conversations/`
• Semantic blocks: Gemini responde con tags tipo `[REGISTRAR_OPERACION]{json}[/REGISTRAR_OPERACION]` que disparamos como comandos
• Button automation: Sistema de callbacks inline que construye menus dinámicos
• TTS integration: Gemini text-to-speech con voz "Zephyr", PCM → MP3 via ffmpeg

Ejemplo de procesamiento semántico:

def process_ac0001_commands(content, chat_id):
"""Bot de chatarra: detecta comandos en respuesta Gemini"""
# Gemini dice: "[REGISTRAR_COMPRA]{\"material\": \"cobre\", \"peso\": 150}[/REGISTRAR_COMPRA]"
if "[REGISTRAR_COMPRA]" in content:
json_data = extract_between_tags(content, "REGISTRAR_COMPRA")
data = json.loads(json_data)
# Grabar en DB específica del cliente
with open(f"/data/ac0001/compras_db.json") as f:
db = json.load(f)
db["operations"].append({
"type": "compra",
"material": data["material"],
"peso_kg": data["peso"],
"timestamp": datetime.now().isoformat()
})
# Guardar y confirmar
with open(f"/data/ac0001/compras_db.json", "w") as f:
json.dump(db, f, indent=2)
return "✅ Compra registrada"

Capa 4: Gemini API - El Modelo

Usamos `gemini-2.0-flash-exp` (gratuito, 1500 RPM). Cada bot tiene su system prompt almacenado en `/data/{project}/system_prompt.txt`.

El prompt de un bot de préstamos es completamente distinto al de un bot de chatarra, pero ambos comparten la misma infraestructura de llamadas:

def call_gemini(messages, project_id):
# Cargar prompt específico del proyecto
with open(f"/data/{project_id}/system_prompt.txt") as f:
system_prompt = f.read()
response = genai.GenerativeModel(
model_name="gemini-2.0-flash-exp",
system_instruction=system_prompt
).generate_content(messages)
return response.text

Aislamiento de Datos: La Regla de Oro

Cada bot tiene su propio directorio en `/data/{project_id}/` con estructura estandarizada:

/data/
er0002/  # Bot de préstamos
conversations/
123456789.json  # Historial de chat_id específico
clientes_db.json
prestamos_db.json
pagos_db.json
system_prompt.txt
ac0001/  # Bot de chatarra
conversations/
compras_db.json
ventas_db.json
inventario_db.json
system_prompt.txt

Ventaja crítica: Si Cliente A migra a otro proveedor, simplemente copiamos `/data/clienteA/` y listo. Cero acoplamiento.

Button Automation: El Sistema de Callbacks

Los bots financieros (préstamos, chatarra) procesan ~200 callbacks/día. El patrón:

1. Usuario recibe opciones como botones Telegram (InlineKeyboard)
2. Al presionar, Telegram envía `callback_query` con `data="ver_prestamo_123"`
3. `claude_proxy.py` tiene handlers registrados:

Dual-location pattern (gotcha crítico)
if callback_data.startswith("ver_prestamo_"):
prestamo_id = callback_data.split("_")[2]
# Cargar desde DB del proyecto correcto
db_path = f"/data/{project_id}/prestamos_db.json"
prestamo = load_prestamo(db_path, prestamo_id)
# Construir respuesta dinámica
markup = [
[{"text": "💰 Registrar Pago", "callback_data": f"pago_{prestamo_id}"}],
[{"text": "📄 Ver Historial", "callback_data": f"hist_{prestamo_id}"}],
[{"text": "🔙 Volver", "callback_data": "menu_principal"}]
]
send_telegram_message(
chat_id=callback_query.from_user.id,
text=format_prestamo_details(prestamo),
reply_markup={"inline_keyboard": markup}
)

Gotcha aprendido a las malas: Los callbacks deben registrarse en DOS ubicaciones en `claude_proxy.py` (líneas ~8500 y ~12425) debido a arquitectura heredada. Omitir una ubicación = callbacks silenciosamente ignorados.

CRMs Web: Triple Sincronización

Cada bot "importante" tiene un CRM web React+Tailwind hosteado en el mismo servidor:

• ER-0002: `laguapaprestamos.com/portal` (gestión préstamos)
• AC-0001: `bot.varelainsights.com/acmetals` (chatarra)
• una clínica dental: `dashboard.una clinica dental.com.mx` (pacientes)

Regla de Triple Sync: Si actualizas lógica de negocio, DEBES sincronizar:

1. Bot Telegram (claude_proxy.py)
2. CRM Web (HTML+JS)
3. API Backend (main.py)

Ejemplo: Agregamos campo "foto de referencia" para chatarra. Tocamos:

• `claude_proxy.py`: Gemini Vision analiza fotos
• `acmetals/index.html`: Upload UI con preview
• `main.py`: Endpoint POST `/api/ac0001/upload_photo`

Seguridad: La Auditoría de Febrero 2026

Descubrimos que 11 de 12 webhooks eran vulnerables a spoofing. Solución implementada:

ANTES (vulnerable)
@app.post("/webhook/bot")
async def webhook(data: dict):
process_message(data)  # Acepta cualquier request
DESPUES (seguro)
@app.post("/webhook/bot")
async def webhook(request: Request):
# Validar secret_token (Telegram Bot API feature)
expected = os.getenv("TELEGRAM_SECRET_ER0002")
received = request.headers.get("X-Telegram-Bot-Api-Secret-Token")
if not hmac.compare_digest(expected, received):
raise HTTPException(403, "Invalid secret")
data = await request.json()
process_message(data)

Cada webhook ahora valida:

1. IP whitelist (n8n)
2. Secret token (bot-api)
3. API key (claude_proxy para endpoints de management)

Lecciones de 14 Bots en Producción

1. Nunca Compartas Base de Datos Entre Bots

Probamos SQLite compartido con tabla `bots` discriminando por `bot_id`. Pesadilla de race conditions. Migramos a un JSON por bot y paz mundial.

2. Context Window es Oro

Gemini Flash tiene 1M tokens de contexto, pero eso no significa "cargar todo el historial". Implementamos:

• Ventana deslizante: Últimos 20 mensajes
• Resúmenes periódicos: Cada 50 mensajes, Gemini resume y reinicia ventana
• Facts clave: Extraemos datos estructurados (nombre cliente, monto prestamo) a JSON separado

3. Timezone es un Bug Silencioso

Durante 2 meses, todos los timestamps estaban en UTC. Usuarios reportaban "el sistema dice que pagué a las 3am pero fue 9pm". Fix:

import pytz
MEXICO_TZ = pytz.timezone("America/Mexico_City")
def now_mexico():
return datetime.now(MEXICO_TZ)

Pero ojo: También ajustar variables de entorno Docker:

services:
bot-proxy:
environment:
- TZ=America/Mexico_City

4. Voice Messages Son un Game-Changer

Agregar TTS aumentó engagement 3x en bot de préstamos. Usuarios prefieren escuchar "Tu pago de $500 fue registrado" que leerlo. Implementación con Gemini:

def send_voice_response(chat_id, text):
# Gemini TTS
response = genai.GenerativeModel(
model_name="gemini-2.5-flash-preview-tts"
).generate_content({
"text": text,
"voice": "Zephyr"  # Voz masculina, español latino
})
# Gemini retorna PCM 24kHz, convertir a MP3
pcm_data = response.audio_data
mp3_path = convert_pcm_to_mp3(pcm_data)
# Enviar a Telegram
bot.send_voice(chat_id, open(mp3_path, "rb"))

El Costo Real de 14 Bots

Infraestructura mensual:

• VPS 8GB RAM: $20 USD (~400 MXN)
• Dominio + SSL (Cloudflare gratis): $0
• Gemini API (1500 RPM gratis): $0
• n8n self-hosted: $0
• Total: 400 MXN/mes para 14 bots

Costo por bot: ~29 MXN/mes (~$1.45 USD) RPM consumido (promedio 7 días):

• Total requests: ~45,000
• Promedio: 267 req/hora
• Pico: 890 req/hora (lunes 9am, usuarios checando saldos)

Gemini Flash gratis aguanta 1500 RPM = 90,000 req/hora. Estamos usando 0.3% de capacidad.

Conclusión: Arquitectura que Escala con el Negocio

La tentación al construir bots es "hardcodear rápido y después refactorizar". Aprendimos que "después" nunca llega cuando tienes 14 clientes en producción.

Los 3 principios que nos salvaron:

1. Aislamiento desde día 1: Un directorio, una base de datos, un system prompt por bot. Sin excepciones.
2. Capas con responsabilidades únicas: n8n orquesta, FastAPI rutea, Flask piensa, Gemini responde. Cada capa hace UNA cosa bien.
3. Monitoreo antes que features: Watchdog Docker, health checks cada 60s, logs estructurados. Si no puedes medir el uptime, no puedes venderlo.

Hoy agregamos un bot nuevo en ~45 minutos. Los primeros 3 bots nos tomaron 2 semanas cada uno. Esa es la diferencia entre arquitectura y código que "solo funciona".

¿La próxima frontera? Migrar de JSON files a PostgreSQL sin downtime, y agregar CDC (Change Data Capture) para analytics en tiempo real. Pero esa es otra historia.