Integrar inteligencia artificial en tus aplicaciones web ya no es cosa de grandes empresas. En este tutorial construirás desde cero una aplicación Flask que se comunica con la API de Claude de Anthropic, cubriendo desde el “Hola Mundo” hasta streaming, manejo de conversaciones y buenas prácticas de producción.
Nivel: Intermedio | Tiempo estimado: 60–90 minutos
Stack: Python 3.11+, Flask 3.x, Anthropic SDK
Requisitos previos
- Python 3.11 o superior instalado
- Conocimientos básicos de Flask
- Una API Key de Anthropic (puedes obtenerla en console.anthropic.com)
Índice
- Configuración del entorno
- Tu primera llamada a Claude
- Creando la aplicación Flask base
- Endpoint básico de chat
- Streaming de respuestas
- Manejo de conversaciones multi-turno
- System prompts y personalización
- Manejo de errores robusto
- Estructura de proyecto para producción
- Buenas prácticas y consideraciones finales
1. Configuración del entorno
Primero, crea un entorno virtual e instala las dependencias necesarias.
# Crear y activar entorno virtual
python -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
# Instalar dependencias
pip install flask anthropic python-dotenv
Crea un archivo .env en la raíz de tu proyecto para guardar tu API Key de forma segura:
# .env
ANTHROPIC_API_KEY=sk-ant-api03-tu-clave-aqui
FLASK_ENV=development
FLASK_DEBUG=1
⚠️ Importante: Agrega .env a tu .gitignore para nunca subir tu clave a un repositorio.
echo ".env" >> .gitignore
echo "venv/" >> .gitignore
2. Tu primera llamada a Claude
Antes de integrar Flask, verifica que el SDK funciona correctamente con un script simple:
# test_claude.py
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv()
client = Anthropic() # Lee ANTHROPIC_API_KEY automáticamente del entorno
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{
"role": "user",
"content": "¿Qué es Flask y por qué es popular en Python?"
}
]
)
print(response.content[0].text)
Ejecútalo:
python test_claude.py
Si ves una respuesta coherente de Claude, ¡todo está en orden!
Anatomía de la respuesta
El objeto response que devuelve la API tiene esta estructura relevante:
response.id # ID único del mensaje
response.model # Modelo utilizado
response.stop_reason # "end_turn", "max_tokens", etc.
response.content # Lista de bloques de contenido
response.content[0].text # El texto de la respuesta
response.usage.input_tokens # Tokens consumidos en el prompt
response.usage.output_tokens # Tokens generados
3. Creando la aplicación Flask base
Crea la estructura básica del proyecto:
mi-app-claude/
├── app.py
├── .env
├── .gitignore
├── requirements.txt
└── templates/
└── index.html
# app.py
from flask import Flask, request, jsonify, render_template, Response
from anthropic import Anthropic, APIError, APIConnectionError, RateLimitError
from dotenv import load_dotenv
import os
import json
load_dotenv()
app = Flask(__name__)
# Inicializar el cliente de Anthropic una sola vez (es thread-safe)
client = Anthropic()
# Modelo a utilizar (centralizado para fácil cambio)
MODEL = "claude-sonnet-4-6"
@app.route("/")
def index():
return render_template("index.html")
if __name__ == "__main__":
app.run(debug=True)
Y una plantilla HTML sencilla para probar:
<!-- templates/index.html -->
<!DOCTYPE html>
<html lang="es">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Chat con Claude</title>
<style>
body { font-family: sans-serif; max-width: 800px; margin: 40px auto; padding: 0 20px; }
#chat-box { border: 1px solid #ddd; height: 400px; overflow-y: auto; padding: 15px; border-radius: 8px; margin-bottom: 15px; }
.user-msg { text-align: right; color: #2563eb; margin: 8px 0; }
.claude-msg { text-align: left; color: #374151; margin: 8px 0; }
#input-area { display: flex; gap: 10px; }
#user-input { flex: 1; padding: 10px; border: 1px solid #ddd; border-radius: 6px; font-size: 14px; }
button { padding: 10px 20px; background: #2563eb; color: white; border: none; border-radius: 6px; cursor: pointer; }
button:hover { background: #1d4ed8; }
</style>
</head>
<body>
<h1>🤖 Chat con Claude</h1>
<div id="chat-box"></div>
<div id="input-area">
<input type="text" id="user-input" placeholder="Escribe tu mensaje..." />
<button onclick="sendMessage()">Enviar</button>
</div>
<script>
const chatBox = document.getElementById('chat-box');
const userInput = document.getElementById('user-input');
let conversationHistory = [];
async function sendMessage() {
const message = userInput.value.trim();
if (!message) return;
appendMessage('user', message);
conversationHistory.push({ role: 'user', content: message });
userInput.value = '';
const claudeDiv = appendMessage('claude', '...');
try {
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ messages: conversationHistory })
});
const data = await response.json();
claudeDiv.textContent = data.response;
conversationHistory.push({ role: 'assistant', content: data.response });
} catch (error) {
claudeDiv.textContent = 'Error al conectar con Claude.';
}
}
function appendMessage(role, text) {
const div = document.createElement('div');
div.className = role === 'user' ? 'user-msg' : 'claude-msg';
div.textContent = text;
chatBox.appendChild(div);
chatBox.scrollTop = chatBox.scrollHeight;
return div;
}
userInput.addEventListener('keypress', (e) => {
if (e.key === 'Enter') sendMessage();
});
</script>
</body>
</html>
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_tokenssegú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.
Visited 2 times, 1 visit(s) today