Cap 4: Integración con Claude API

Arquitectura de la integración

sequenceDiagram
    participant U as Usuario
    participant TG as Telegram
    participant BOT as Bot Python
    participant API as Anthropic API

    U->>TG: mensaje
    TG->>BOT: Update
    BOT->>BOT: recuperar historial del usuario
    BOT->>API: messages.create(historial + nuevo mensaje)
    API-->>BOT: respuesta de Claude
    BOT->>BOT: guardar en historial
    BOT->>TG: respuesta
    TG->>U: mensaje

services/claude_api.py

import anthropic
from config import ANTHROPIC_API_KEY, CLAUDE_MODEL

client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)

SYSTEM_PROMPT = """Eres un asistente de programación experto.
Respondes de forma concisa y precisa.
Cuando muestres código, usa bloques de código con el lenguaje correcto.
Si no sabes algo, dilo claramente."""

def ask_claude(messages: list[dict]) -> str:
    """
    Envía el historial de mensajes a Claude y retorna la respuesta.
    
    messages: lista de dicts {"role": "user"|"assistant", "content": str}
    """
    response = client.messages.create(
        model=CLAUDE_MODEL,
        max_tokens=2048,
        system=SYSTEM_PROMPT,
        messages=messages,
    )
    return response.content[0].text

storage/history.py — Historial en memoria

Almacenamos el historial por user_id. En el capítulo 6 lo hacemos persistente:

from collections import defaultdict

# Historial en memoria: {user_id: [{"role": ..., "content": ...}]}
_history: dict[int, list[dict]] = defaultdict(list)

MAX_MESSAGES = 20  # máximo de turnos a recordar

def get_history(user_id: int) -> list[dict]:
    return _history[user_id]

def add_message(user_id: int, role: str, content: str):
    _history[user_id].append({"role": role, "content": content})
    # Mantener solo los últimos MAX_MESSAGES mensajes
    if len(_history[user_id]) > MAX_MESSAGES:
        _history[user_id] = _history[user_id][-MAX_MESSAGES:]

def clear_history(user_id: int):
    _history[user_id] = []

handlers/messages.py actualizado

import asyncio
from telegram import Update
from telegram.constants import ChatAction
from telegram.ext import ContextTypes
from services.claude_api import ask_claude
from storage.history import get_history, add_message

async def typing_loop(bot, chat_id: int, stop_event: asyncio.Event):
    while not stop_event.is_set():
        await bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING)
        await asyncio.sleep(4)

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    user_text = update.message.text
    chat_id = update.effective_chat.id

    # Indicador de escritura activo durante la llamada a Claude
    stop = asyncio.Event()
    typing_task = asyncio.create_task(typing_loop(context.bot, chat_id, stop))

    try:
        # Agregar mensaje del usuario al historial
        add_message(user_id, "user", user_text)

        # Llamar a Claude con el historial completo
        history = get_history(user_id)
        respuesta = ask_claude(history)

        # Guardar respuesta en el historial
        add_message(user_id, "assistant", respuesta)

    except Exception as e:
        respuesta = f"Error al contactar Claude: {e}"
    finally:
        stop.set()
        typing_task.cancel()

    # Enviar respuesta (máx 4096 chars por mensaje)
    await send_long_message(update, respuesta)

async def send_long_message(update: Update, text: str):
    """Divide mensajes largos en partes de máx 4096 caracteres."""
    MAX_LEN = 4000  # margen de seguridad
    if len(text) <= MAX_LEN:
        await update.message.reply_text(text)
        return
    partes = [text[i:i+MAX_LEN] for i in range(0, len(text), MAX_LEN)]
    for parte in partes:
        await update.message.reply_text(parte)

handlers/commands.py — /reset integrado

from storage.history import clear_history

async def reset(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    clear_history(user_id)
    await update.message.reply_text(
        "Conversación reiniciada. Puedo ayudarte con algo más."
    )

Streaming (opcional)

Para respuestas en tiempo real, usa el streaming de Anthropic. Edita el mensaje de Telegram a medida que llega texto:

async def ask_claude_streaming(messages: list[dict], update: Update, context) -> str:
    """Envía respuesta en tiempo real editando el mensaje."""
    # Enviar mensaje inicial
    sent = await update.message.reply_text("...")
    acumulado = ""
    ultimo_update = 0

    with client.messages.stream(
        model=CLAUDE_MODEL,
        max_tokens=2048,
        system=SYSTEM_PROMPT,
        messages=messages,
    ) as stream:
        for chunk in stream.text_stream:
            acumulado += chunk
            # Actualizar cada 30 chars para no exceder rate limits de Telegram
            if len(acumulado) - ultimo_update >= 30:
                try:
                    await sent.edit_text(acumulado)
                    ultimo_update = len(acumulado)
                except Exception:
                    pass  # ignorar errores de edición (mensaje idéntico, etc.)

    # Edición final con texto completo
    await sent.edit_text(acumulado)
    return acumulado

Nota: Telegram tiene rate limit de ~1 edición/segundo por mensaje. El check de 30 caracteres ayuda, pero en producción considera una cola con asyncio.Queue.

Formatear respuestas con Markdown

Claude usa Markdown en sus respuestas. Para que Telegram lo renderice:

import re

def preparar_markdown(text: str) -> str:
    """
    Convierte el Markdown de Claude a MarkdownV2 de Telegram.
    Alternativa: enviar como HTML.
    """
    # Opción más simple: usar parse_mode="Markdown" (v1)
    # Limitación: no soporta tablas ni listas anidadas
    return text

# En el handler:
await update.message.reply_text(
    preparar_markdown(respuesta),
    parse_mode="Markdown"
)

Alternativa práctica: Enviar sin parse_mode para texto plano simple, y con parse_mode solo para bloques de código explícitos.

Flujo completo con historial

sequenceDiagram
    participant U as Usuario
    participant BOT as Bot
    participant MEM as Historia
    participant CL as Claude API

    U->>BOT: "¿Qué es un closure en JS?"
    BOT->>MEM: add(user, "¿Qué es un closure en JS?")
    BOT->>CL: messages=[{user: "¿Qué es..."}]
    CL-->>BOT: "Un closure es..."
    BOT->>MEM: add(assistant, "Un closure es...")
    BOT->>U: "Un closure es..."

    U->>BOT: "Dame un ejemplo"
    BOT->>MEM: add(user, "Dame un ejemplo")
    BOT->>CL: messages=[{user: "¿Qué es..."}, {assistant: "Un closure..."}, {user: "Dame..."}]
    CL-->>BOT: "const counter = () => { let n = 0..."
    BOT->>MEM: add(assistant, "const counter...")
    BOT->>U: "const counter..."

El historial hace que “Dame un ejemplo” tenga contexto del closure, sin que el usuario tenga que repetir la pregunta.