bots de Telegram con la API de Claude, de cero a producción

7234567890:AAH_abcXYZ123...

#!/usr/bin/env bash
set -euo pipefail

source ~/.config/telegram-bot.env

API="https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}"
OFFSET=0

# Función de envío (maneja el límite de 4096 caracteres)
send() {
  local chat="$1" text="$2"
  while [ -n "$text" ]; do
    local chunk="${text:0:4000}"
    text="${text:4000}"
    curl -s -X POST "${API}/sendMessage" \
      --data-urlencode "chat_id=${chat}" \
      --data-urlencode "text=${chunk}" > /dev/null
  done
}

handle_message() {
  local chat="$1" user_text="$2"

  # Solo usuarios permitidos
  if [ "$chat" != "$TELEGRAM_ALLOWED_CHAT_ID" ]; then
    return
  fi

  # ⚠️ Protección contra shell injection: stdin en lugar de argumento
  # Si user_text contiene `$(rm -rf ~)` o backticks, sin comillas
  # se ejecutaría como código. printf '%q' lo entrecomilla de forma segura, pero
  # es más confiable pasar el texto por stdin para que el shell ni lo interprete.
  local reply
  reply=$(printf '%s' "$user_text" | claude -p --model sonnet \
    --max-budget-usd 0.50 2>&1) || reply="Error: $reply"

  send "$chat" "$reply"
}

# ⚠️ Bucle de polling SIN subshell pipe
# La versión vieja `jq ... | while read` corría el while en un subshell:
# la variable OFFSET se actualizaba ahí, pero afuera seguía igual.
# La redirección con process substitution `< <(...)` deja el while
# en el shell actual: OFFSET se ve también en la siguiente iteración.
# curl --fail --max-time 30: cortamos conexiones colgadas, atrapamos HTTP 5xx.
while true; do
  UPDATES=$(curl -s --fail --max-time 35 \
    "${API}/getUpdates?offset=${OFFSET}&timeout=30") || {
    echo "Telegram API timeout/error, retrying in 5s..." >&2
    sleep 5
    continue
  }

  while read -r upd; do
    OFFSET=$(($(echo "$upd" | jq '.update_id') + 1))
    CHAT=$(echo "$upd" | jq '.message.chat.id')
    TEXT=$(echo "$upd" | jq -r '.message.text // empty')
    [ -n "$TEXT" ] && handle_message "$CHAT" "$TEXT"
  done < <(echo "$UPDATES" | jq -c '.result[]?')
done

TELEGRAM_BOT_TOKEN=7234567890:AAH_tu_token
TELEGRAM_ALLOWED_CHAT_ID=123456789   # Tu chat_id (pregúntaselo a @userinfobot)
ANTHROPIC_API_KEY=sk-ant-...

chmod 600 ~/.config/telegram-bot.env
chmod +x bot.sh
./bot.sh

pip install python-telegram-bot anthropic

import asyncio
import anthropic
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
import os

claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

# ⚠️ Memoria: un diccionario en memoria, SOLO para desarrollo y un solo proceso
#
# Problema: defaultdict(list) guarda una entrada por cada user_id nuevo para siempre.
# 10 000 usuarios × 20 mensajes × ~500 bytes = ~100 MB de RAM, y sigue creciendo.
# Al reiniciar el proceso se pierde todo el historial.
#
# Para producción: Redis / Cloudflare KV / Postgres con TTL.
# Abajo: desalojo LRU por tamaño (protección contra OOM en la versión en memoria).
from collections import OrderedDict

MAX_HISTORY = 20       # mensajes por usuario
MAX_USERS = 1000       # usuarios activos en memoria (LRU eviction)

class LRUConversations(OrderedDict):
    """Diccionario LRU: si se pasa de MAX_USERS, borra el más viejo."""
    def __getitem__(self, key):
        if key not in self:
            self[key] = []
        else:
            self.move_to_end(key)  # mark as recently used
        return super().__getitem__(key)

    def __setitem__(self, key, value):
        super().__setitem__(key, value)
        self.move_to_end(key)
        if len(self) > MAX_USERS:
            evicted_key, _ = self.popitem(last=False)  # drop oldest
            # En producción: registro + alerta "history evicted for user X"

conversation_history = LRUConversations()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    await update.message.reply_text(
        f"¡Hola, {user.first_name}! Soy un asistente de IA. Hazme cualquier pregunta."
    )

async def clear_history(update: Update, context: ContextTypes.DEFAULT_TYPE):
    conversation_history[update.effective_user.id].clear()
    await update.message.reply_text("Historial borrado.")

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    user_text = update.message.text
    
    # Mostramos "escribiendo..."
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )
    
    # Lo agregamos al historial
    conversation_history[user_id].append({
        "role": "user",
        "content": user_text
    })
    
    # Lo recortamos si es demasiado largo
    if len(conversation_history[user_id]) > MAX_HISTORY:
        conversation_history[user_id] = conversation_history[user_id][-MAX_HISTORY:]
    
    # Solicitud a Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=2000,
        system="Eres un asistente de IA útil. Responde breve y al grano.",
        messages=conversation_history[user_id]
    )
    
    reply = "".join(b.text for b in response.content if b.type == "text")
    
    # Guardamos la respuesta
    conversation_history[user_id].append({
        "role": "assistant",
        "content": reply
    })
    
    # Límite de Telegram: 4096 caracteres
    if len(reply) > 4096:
        for i in range(0, len(reply), 4096):
            await update.message.reply_text(reply[i:i+4096])
    else:
        await update.message.reply_text(reply)

def main():
    app = Application.builder().token(os.environ["TELEGRAM_BOT_TOKEN"]).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("clear", clear_history))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    print("Bot en marcha...")
    app.run_polling()

if __name__ == "__main__":
    main()

import "dotenv/config";
import { Bot } from "grammy";
import Anthropic from "@anthropic-ai/sdk";

const bot = new Bot(process.env.TELEGRAM_BOT_TOKEN!);
const claude = new Anthropic();

// ⚠️ Historial de diálogos: un Map en memoria, SOLO para desarrollo
//
// Problema: Map<number, ...> guarda una entrada por cada user_id para siempre.
// Con 10k usuarios el proceso se come toda la RAM y se cae (OOM).
// Al reiniciar se pierde todo el historial (no es persistente).
//
// Para producción usa Redis o Cloudflare KV con TTL:
//   await env.KV.put(`chat:${userId}`, JSON.stringify(history),
//                    { expirationTtl: 60 * 60 * 24 * 7 }); // 7 días
//
// Abajo: desalojo LRU por tamaño (protección contra OOM en un solo proceso).
const MAX_USERS = 1000;
const conversations = new Map<number, Array<{role: string, content: string}>>();

function getHistory(userId: number): Array<{role: string, content: string}> {
  if (conversations.has(userId)) {
    // LRU touch: lo quitamos y lo regresamos al final (Map conserva el orden de inserción)
    const h = conversations.get(userId)!;
    conversations.delete(userId);
    conversations.set(userId, h);
    return h;
  }
  // Desalojamos el más viejo si se llena
  if (conversations.size >= MAX_USERS) {
    const oldestKey = conversations.keys().next().value;
    if (oldestKey !== undefined) conversations.delete(oldestKey);
  }
  const fresh: Array<{role: string, content: string}> = [];
  conversations.set(userId, fresh);
  return fresh;
}

bot.on("message:text", async (ctx) => {
  const userId = ctx.from.id;
  const userText = ctx.message.text;
  
  const history = getHistory(userId);  // LRU-touch + lo crea si no existe
  history.push({ role: "user", content: userText });
  
  // Mensaje provisional
  const placeholder = await ctx.reply("...");
  
  let buffer = "";
  let lastEdit = Date.now();
  
  // Streaming desde Claude
  const stream = claude.messages.stream({
    model: "claude-sonnet-5-5",
    max_tokens: 2000,
    system: "Eres un asistente de IA útil. Responde en español.",
    messages: history as any,
  });
  
  for await (const chunk of stream) {
    if (chunk.type === "content_block_delta" && 
        chunk.delta.type === "text_delta") {
      buffer += chunk.delta.text;
      
      // Actualizamos cada 800 ms (límite de la API de Telegram)
      if (Date.now() - lastEdit > 800 && buffer.length > 20) {
        try {
          await ctx.api.editMessageText(
            ctx.chat.id, placeholder.message_id, buffer.slice(0, 4000)
          );
          lastEdit = Date.now();
        } catch (e) { /* el mensaje no cambió */ }
      }
    }
  }
  
  // Actualización final
  await ctx.api.editMessageText(ctx.chat.id, placeholder.message_id, buffer.slice(0, 4000));
  
  history.push({ role: "assistant", content: buffer });
  if (history.length > 20) conversations.set(userId, history.slice(-20));
});

bot.command("clear", async (ctx) => {
  conversations.delete(ctx.from.id);
  await ctx.reply("Historial borrado.");
});

bot.start();

import io
import openai
import aiohttp
from telegram.ext import MessageHandler, filters

openai_client = openai.AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

async def handle_voice(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await context.bot.send_chat_action(update.effective_chat.id, "typing")
    
    # Paso 1: descargamos el archivo OGG
    file = await context.bot.get_file(update.message.voice.file_id)
    async with aiohttp.ClientSession() as session:
        async with session.get(file.file_path) as resp:
            audio_bytes = await resp.read()
    
    # Paso 2: transcribimos con Whisper
    audio_file = io.BytesIO(audio_bytes)
    audio_file.name = "voice.ogg"
    
    # el nombre del modelo de reconocimiento es un ejemplo: revisa los modelos vigentes en la documentación de OpenAI
    transcription = await openai_client.audio.transcriptions.create(
        model="whisper-1",
        file=audio_file,
        language="es"
    )
    
    transcript = transcription.text
    
    # Paso 3: mostramos lo que reconocimos
    await update.message.reply_text(f"🎙️ Entendí: _{transcript}_", parse_mode="Markdown")
    
    # Paso 4: se lo pasamos a Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1000,
        messages=[{"role": "user", "content": transcript}]
    )
    await update.message.reply_text("".join(b.text for b in response.content if b.type == "text"))

app.add_handler(MessageHandler(filters.VOICE, handle_voice))

from telegram import InlineKeyboardButton, InlineKeyboardMarkup

async def show_menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [
            InlineKeyboardButton("📝 Escribir un post", callback_data="mode_content"),
            InlineKeyboardButton("📊 Analítica", callback_data="mode_analytics"),
        ],
        [InlineKeyboardButton("🗑️ Borrar historial", callback_data="clear_history")],
    ]
    await update.message.reply_text(
        "Elige un modo:", reply_markup=InlineKeyboardMarkup(keyboard)
    )

async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()  # OBLIGATORIO: quita el "relojito" del botón
    
    if query.data == "mode_content":
        context.user_data["mode"] = "content"
        await query.edit_message_text("Modo: asistente de contenido. Escribe el tema 👇")
    elif query.data == "clear_history":
        conversation_history[query.from_user.id].clear()
        await query.answer("¡Historial borrado!", show_alert=True)

from telegram.ext import CallbackQueryHandler
app.add_handler(CallbackQueryHandler(handle_callback))

# Instalación
npm install -g wrangler
wrangler login

# Proyecto nuevo
mkdir my-telegram-bot && cd my-telegram-bot
npm install grammy @anthropic-ai/sdk

# Secretos (¡no en el código!)
wrangler secret put BOT_TOKEN
wrangler secret put ANTHROPIC_API_KEY
wrangler secret put WEBHOOK_SECRET

name = "my-telegram-bot"
main = "src/index.ts"
compatibility_date = "2026-10-01"  # pon la fecha vigente cuando crees el proyecto

import { Bot, webhookCallback } from "grammy";
import Anthropic from "@anthropic-ai/sdk";

export interface Env {
  BOT_TOKEN: string;
  ANTHROPIC_API_KEY: string;
  WEBHOOK_SECRET: string;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    // Revisión del secret_token
    const secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token");
    if (secret !== env.WEBHOOK_SECRET) {
      return new Response("Unauthorized", { status: 401 });
    }
    
    const bot = new Bot(env.BOT_TOKEN);
    const claude = new Anthropic({ apiKey: env.ANTHROPIC_API_KEY });
    
    bot.on("message:text", async (ctx) => {
      const response = await claude.messages.create({
        model: "claude-haiku-4-5",  // Haiku: ahorro en Workers
        max_tokens: 1024,
        messages: [{ role: "user", content: ctx.message.text }],
      });
      await ctx.reply(response.content.map((b) => (b.type === "text" ? b.text : "")).join("") as string);
    });
    
    return webhookCallback(bot, "cloudflare-mod")(request);
  },
};

# Despliegue
wrangler deploy
# Salida: https://my-telegram-bot.YOUR-USERNAME.workers.dev

# Registro del webhook (una sola vez)
curl "https://api.telegram.org/bot<BOT_TOKEN>/setWebhook" \
  -d "url=https://my-telegram-bot.YOUR-USERNAME.workers.dev" \
  -d "secret_token=<WEBHOOK_SECRET>"

# Estructura del proyecto
my-bot/
├── bot.py
├── requirements.txt   # python-telegram-bot y anthropic (usa las versiones vigentes)
└── Procfile           # web: python bot.py

# Despliegue
git init && git add . && git commit -m "initial"
# → Railway.com → New Project → Deploy from GitHub
# Agregar las variables: TELEGRAM_BOT_TOKEN, ANTHROPIC_API_KEY

# ❌ PROHIBIDO
bot = Bot(token="7234567890:AAH...")

# ✅ CORRECTO
bot = Bot(token=os.environ["TELEGRAM_BOT_TOKEN"])

ALLOWED_USERS = {123456789}  # Tus chat_id

async def check_access(update: Update) -> bool:
    if update.effective_user.id not in ALLOWED_USERS:
        await update.message.reply_text("Sin acceso.")
        return False
    return True

from collections import defaultdict
from datetime import datetime, timedelta

requests = defaultdict(list)

def check_rate_limit(user_id: int, max_req=5, window_sec=60) -> bool:
    now = datetime.now()
    cutoff = now - timedelta(seconds=window_sec)
    requests[user_id] = [ts for ts in requests[user_id] if ts > cutoff]
    if len(requests[user_id]) >= max_req:
        return False
    requests[user_id].append(now)
    return True

SUPPORT_PROMPT = """Eres agente de soporte de la empresa Acme Realty.
Ayudas a clientes hispanohablantes a encontrar propiedades en Ecuador.
Si preguntan por precios, pregunta su presupuesto y preferencias.
Si hay una queja, discúlpate y pide un dato de contacto.
Si no sabes algo, dilo con honestidad y ofrece contactar a un asesor."""

Presionó un botón → eligió su interés (comprar/rentar)
→ escribió su presupuesto → dejó su contacto
→ el asesor recibió el prospecto en un chat aparte

MODES = {
    "telegram": "Post para Telegram, máx. 800 caracteres, 3-5 emojis",
    "instagram": "Instagram con 10-15 hashtags al final",
    "blog": "Artículo de al menos 800 palabras con subtítulos",
}