Integración de Claude API con Python Flask: Tutorial Paso a Paso

4. Endpoint básico de chat

Agrega el endpoint principal al app.py:

@app.route("/api/chat", methods=["POST"])
def chat():
    """
    Endpoint de chat básico.
    Recibe una lista de mensajes y devuelve la respuesta de Claude.
    
    Body esperado:
    {
        "messages": [
            {"role": "user", "content": "Hola"},
            {"role": "assistant", "content": "Hola, ¿en qué te ayudo?"},
            {"role": "user", "content": "¿Qué es Flask?"}
        ]
    }
    """
    data = request.get_json()


    if not data or "messages" not in data:
        return jsonify({"error": "Se requiere el campo 'messages'"}), 400


    messages = data["messages"]


    # Validación básica: el último mensaje debe ser del usuario
    if not messages or messages[-1]["role"] != "user":
        return jsonify({"error": "El último mensaje debe ser del usuario"}), 400


    try:
        response = client.messages.create(
            model=MODEL,
            max_tokens=2048,
            messages=messages
        )


        assistant_message = response.content[0].text


        return jsonify({
            "response": assistant_message,
            "usage": {
                "input_tokens": response.usage.input_tokens,
                "output_tokens": response.usage.output_tokens
            }
        })


    except RateLimitError:
        return jsonify({"error": "Límite de requests alcanzado. Intenta más tarde."}), 429
    except APIConnectionError:
        return jsonify({"error": "No se pudo conectar con la API de Anthropic."}), 503
    except APIError as e:
        return jsonify({"error": f"Error de API: {str(e)}"}), 500

Prueba el endpoint con curl:

curl -X POST http://localhost:5000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"messages": [{"role": "user", "content": "¿Qué es el patrón MVC?"}]}'

5. Streaming de respuestas

El streaming es fundamental para una buena experiencia de usuario: en vez de esperar varios segundos a que Claude termine de generar la respuesta completa, el texto aparece progresivamente como si estuviera escribiendo. Esto es especialmente útil para respuestas largas.

@app.route("/api/chat/stream", methods=["POST"])
def chat_stream():
    """
    Endpoint de chat con streaming usando Server-Sent Events (SSE).
    El cliente recibe tokens a medida que Claude los genera.
    """
    data = request.get_json()


    if not data or "messages" not in data:
        return jsonify({"error": "Se requiere el campo 'messages'"}), 400


    messages = data["messages"]


    def generate():
        try:
            with client.messages.stream(
                model=MODEL,
                max_tokens=2048,
                messages=messages
            ) as stream:
                for text_chunk in stream.text_stream:
                    # Formato SSE: "data: <contenido>\n\n"
                    payload = json.dumps({"chunk": text_chunk})
                    yield f"data: {payload}\n\n"


                # Señal de fin del stream
                yield "data: [DONE]\n\n"


        except RateLimitError:
            error_payload = json.dumps({"error": "Rate limit alcanzado"})
            yield f"data: {error_payload}\n\n"
        except APIError as e:
            error_payload = json.dumps({"error": str(e)})
            yield f"data: {error_payload}\n\n"


    return Response(
        generate(),
        mimetype="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"  # Importante para Nginx
        }
    )

Actualiza el JavaScript en index.html para consumir el stream:

async function sendMessageStream() {
    const message = userInput.value.trim();
    if (!message) return;


    appendMessage('user', message);
    conversationHistory.push({ role: 'user', content: message });
    userInput.value = '';


    // Crear div vacío para la respuesta que se irá llenando
    const claudeDiv = appendMessage('claude', '');
    let fullResponse = '';


    const response = await fetch('/api/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ messages: conversationHistory })
    });


    const reader = response.body.getReader();
    const decoder = new TextDecoder();


    while (true) {
        const { done, value } = await reader.read();
        if (done) break;


        const lines = decoder.decode(value).split('\n');
        for (const line of lines) {
            if (!line.startsWith('data: ')) continue;
            const raw = line.slice(6).trim();
            if (raw === '[DONE]') {
                conversationHistory.push({ role: 'assistant', content: fullResponse });
                break;
            }
            try {
                const parsed = JSON.parse(raw);
                if (parsed.chunk) {
                    fullResponse += parsed.chunk;
                    claudeDiv.textContent = fullResponse;
                    chatBox.scrollTop = chatBox.scrollHeight;
                }
            } catch {}
        }
    }
}

6. Manejo de conversaciones multi-turno

Para mantener conversaciones coherentes, necesitas enviar el historial completo en cada request. Aquí un módulo para gestionar conversaciones del lado del servidor con Flask sessions:

# Agrega flask-session para persistencia server-side
# pip install Flask-Session


from flask import session
import uuid


app.secret_key = os.environ.get("SECRET_KEY", "dev-secret-key-cambiar-en-produccion")


@app.route("/api/conversation/new", methods=["POST"])
def new_conversation():
    """Crea una nueva conversación y devuelve su ID."""
    conversation_id = str(uuid.uuid4())
    session[f"conv_{conversation_id}"] = []
    return jsonify({"conversation_id": conversation_id})




@app.route("/api/conversation/<conversation_id>/message", methods=["POST"])
def send_message_to_conversation(conversation_id):
    """
    Envía un mensaje a una conversación existente.
    El servidor mantiene el historial.
    """
    session_key = f"conv_{conversation_id}"
    
    if session_key not in session:
        return jsonify({"error": "Conversación no encontrada"}), 404


    data = request.get_json()
    user_message = data.get("message", "").strip()


    if not user_message:
        return jsonify({"error": "El mensaje no puede estar vacío"}), 400


    # Recuperar historial y agregar el nuevo mensaje del usuario
    history = session[session_key]
    history.append({"role": "user", "content": user_message})


    try:
        response = client.messages.create(
            model=MODEL,
            max_tokens=2048,
            system="Eres un asistente útil y preciso. Responde siempre en español.",
            messages=history
        )


        assistant_message = response.content[0].text
        
        # Guardar la respuesta en el historial
        history.append({"role": "assistant", "content": assistant_message})
        session[session_key] = history  # Actualizar la sesión
        session.modified = True


        return jsonify({
            "response": assistant_message,
            "turn": len(history) // 2,  # Número de intercambios
            "usage": {
                "input_tokens": response.usage.input_tokens,
                "output_tokens": response.usage.output_tokens
            }
        })


    except APIError as e:
        # Revertir el mensaje del usuario si hubo error
        history.pop()
        session[session_key] = history
        return jsonify({"error": str(e)}), 500




@app.route("/api/conversation/<conversation_id>/history", methods=["GET"])
def get_history(conversation_id):
    """Devuelve el historial completo de una conversación."""
    session_key = f"conv_{conversation_id}"
    if session_key not in session:
        return jsonify({"error": "Conversación no encontrada"}), 404
    return jsonify({"history": session[session_key]})

7. System prompts y personalización

El system prompt define la personalidad y el contexto de Claude para toda la conversación. Es una de las herramientas más poderosas de la API:

# Ejemplos de system prompts especializados


SYSTEM_PROMPTS = {
    "asistente_general": """
        Eres un asistente útil y preciso que responde siempre en español.
        Sé conciso pero completo. Si no sabes algo, dilo abiertamente.
    """,
    
    "experto_python": """
        Eres un experto en Python con 15 años de experiencia.
        Siempre provees código limpio, bien documentado y siguiendo PEP 8.
        Cuando expliques conceptos, usa ejemplos prácticos y concisos.
        Menciona las mejores prácticas y posibles errores comunes.
    """,
    
    "soporte_tecnico": """
        Eres un agente de soporte técnico amigable y paciente.
        Tu objetivo es resolver problemas técnicos paso a paso.
        Haz preguntas clarificadoras cuando sea necesario.
        Explica en términos simples cuando el usuario no sea técnico.
    """
}




@app.route("/api/chat/specialized", methods=["POST"])
def chat_specialized():
    """Chat con un system prompt especializado."""
    data = request.get_json()
    messages = data.get("messages", [])
    persona = data.get("persona", "asistente_general")


    system_prompt = SYSTEM_PROMPTS.get(persona, SYSTEM_PROMPTS["asistente_general"])


    response = client.messages.create(
        model=MODEL,
        max_tokens=2048,
        system=system_prompt.strip(),
        messages=messages
    )


    return jsonify({"response": response.content[0].text})

Inyección dinámica de contexto

Una técnica muy útil es enriquecer el system prompt con información dinámica:

def build_system_prompt(user_name: str, user_plan: str, current_date: str) -> str:
    """Construye un system prompt personalizado con contexto dinámico."""
    return f"""
    Eres el asistente de soporte de MiApp.
    
    CONTEXTO DEL USUARIO:
    - Nombre: {user_name}
    - Plan: {user_plan}
    - Fecha actual: {current_date}
    
    INSTRUCCIONES:
    - Saluda al usuario por su nombre en el primer mensaje.
    - Si el usuario tiene plan "free", sugiere el plan "pro" cuando sea relevante.
    - Responde siempre en español con un tono profesional pero amigable.
    """.strip()

8. Manejo de errores robusto

Un manejo de errores apropiado es crítico en producción. El SDK de Anthropic tiene excepciones específicas para cada situación:

from anthropic import (
    Anthropic,
    APIError,
    APIConnectionError,
    APITimeoutError,
    AuthenticationError,
    BadRequestError,
    NotFoundError,
    PermissionDeniedError,
    RateLimitError,
)
import time
import logging


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)




def call_claude_with_retry(messages: list, system: str = None, max_retries: int = 3) -> str:
    """
    Llama a Claude con reintentos automáticos para errores transitorios.
    
    Args:
        messages: Historial de mensajes
        system: System prompt opcional
        max_retries: Número máximo de reintentos
        
    Returns:
        Texto de la respuesta de Claude
        
    Raises:
        Exception: Si se agotan los reintentos o hay un error no recuperable
    """
    kwargs = {
        "model": MODEL,
        "max_tokens": 2048,
        "messages": messages
    }
    if system:
        kwargs["system"] = system


    for attempt in range(max_retries):
        try:
            response = client.messages.create(**kwargs)
            return response.content[0].text


        except AuthenticationError:
            # Error crítico, no tiene sentido reintentar
            logger.error("API Key inválida o expirada")
            raise Exception("Error de autenticación con la API de Anthropic")


        except BadRequestError as e:
            # Error en los datos enviados, no reintentar
            logger.error(f"Request inválido: {e}")
            raise Exception(f"Datos inválidos en la request: {str(e)}")


        except RateLimitError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Backoff exponencial: 1s, 2s, 4s
                logger.warning(f"Rate limit alcanzado. Esperando {wait_time}s... (intento {attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                raise Exception("Se alcanzó el límite de requests. Intenta más tarde.")


        except (APIConnectionError, APITimeoutError):
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                logger.warning(f"Error de conexión. Reintentando en {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise Exception("No se pudo conectar con Anthropic. Verifica tu conexión.")


        except APIError as e:
            logger.error(f"Error inesperado de la API: {e}")
            raise Exception(f"Error de API: {str(e)}")


    raise Exception("Se agotaron los reintentos")




# Manejador de errores global para Flask
@app.errorhandler(Exception)
def handle_exception(e):
    logger.error(f"Error no manejado: {e}", exc_info=True)
    return jsonify({
        "error": "Error interno del servidor",
        "message": str(e)
    }), 500

9. Estructura de proyecto para producción

Cuando tu aplicación crece, conviene organizarla mejor:

mi-app-claude/
├── app/
│   ├── __init__.py          # Factory de la app Flask
│   ├── config.py            # Configuración por entornos
│   ├── claude_client.py     # Cliente de Anthropic centralizado
│   └── routes/
│       ├── __init__.py
│       ├── chat.py          # Endpoints de chat
│       └── conversations.py # Endpoints de conversaciones
├── tests/
│   ├── test_chat.py
│   └── test_conversations.py
├── .env
├── .env.example
├── requirements.txt
└── wsgi.py

# app/config.py
import os


class Config:
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-key-insegura")
    ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")
    CLAUDE_MODEL = os.environ.get("CLAUDE_MODEL", "claude-sonnet-4-6")
    CLAUDE_MAX_TOKENS = int(os.environ.get("CLAUDE_MAX_TOKENS", "2048"))


class DevelopmentConfig(Config):
    DEBUG = True


class ProductionConfig(Config):
    DEBUG = False
    # En producción, usa SESSION_TYPE = "redis" o similar


config = {
    "development": DevelopmentConfig,
    "production": ProductionConfig,
    "default": DevelopmentConfig
}

# app/__init__.py
from flask import Flask
from .config import config


def create_app(config_name="default"):
    app = Flask(__name__)
    app.config.from_object(config[config_name])


    from .routes.chat import chat_bp
    from .routes.conversations import conversations_bp
    
    app.register_blueprint(chat_bp, url_prefix="/api/chat")
    app.register_blueprint(conversations_bp, url_prefix="/api/conversations")


    return app

# app/claude_client.py — Cliente singleton
from anthropic import Anthropic
import os


_client = None


def get_client() -> Anthropic:
    """Retorna el cliente de Anthropic (singleton)."""
    global _client
    if _client is None:
        _client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
    return _client

10. Buenas prácticas y consideraciones finales

✅ Seguridad

• Nunca expongas tu API Key en el frontend. Toda llamada a Anthropic debe ir desde tu backend.
• Valida y sanitiza los inputs de usuarios antes de enviarlos a la API.
• Implementa autenticación en tus endpoints para evitar abuso.
• Usa variables de entorno, nunca hardcodees credenciales.

✅ Rendimiento

• El cliente Anthropic() es thread-safe: instáncialo una vez al arrancar la app, no por request.
• Para aplicaciones con muchos usuarios simultáneos, considera usar Quart (la versión async de Flask) junto con AsyncAnthropic.
• Cachea respuestas cuando sea posible para consultas repetitivas.

✅ Costos

• Monitorea el uso de tokens con response.usage. Cada token cuenta.
• Ajusta max_tokens según lo que realmente necesita tu caso de uso.
• Los modelos más pequeños (claude-haiku-4-5) son mucho más económicos para tareas simples de clasificación o extracción.