Capítulo 8: Manejo de Sesiones y Contexto
Capítulo 8: Manejo de Sesiones y Contexto
Una sesión es la memoria de una conversación. Sin sesiones, cada llamada a query() comienza desde cero — el agente no recuerda nada de lo que ocurrió antes. Con sesiones, puedes reanudar conversaciones, construir sobre análisis previos, y crear asistentes que “conocen” el proyecto.
1. ¿Qué son las sesiones?
Una sesión en el SDK es un identificador (session_id) que apunta a un historial de conversación almacenado en el servidor de Claude. Cuando reanudas una sesión, el modelo “recuerda” todo lo que ocurrió en ella.
Diagrama: ciclo de vida de una sesión
sequenceDiagram
participant App as Aplicación
participant SDK as Claude SDK
participant API as Anthropic API
participant Store as Session Store
App->>SDK: query("Analiza este proyecto")
SDK->>API: Petición nueva (sin session_id)
API->>Store: Crear nueva sesión
API-->>SDK: SystemMessage(session_id="abc123")
SDK-->>App: Mensajes del agente...
Note over App,Store: Segunda llamada — reanuda la sesión
App->>SDK: query("Ahora refactoriza lo que encontraste", resume="abc123")
SDK->>API: Petición con session_id="abc123"
API->>Store: Cargar historial de "abc123"
API-->>SDK: Continúa donde se quedó
SDK-->>App: El agente recuerda el análisis anteriorPor qué las sesiones son fundamentales
Sin sesiones, cada query() es una conversación nueva. El agente no sabe:
- Qué archivos ya revisó
- Qué problemas encontró anteriormente
- Qué decisiones se tomaron en pasos anteriores
Con sesiones, el agente mantiene todo ese contexto, permitiendo flujos de trabajo de múltiples pasos donde cada paso construye sobre el anterior.
Diferencia entre sesión y contexto manual
| Mecanismo | Cómo funciona | Cuándo usar |
|---|---|---|
| Sesión (resume) | El servidor mantiene el historial completo | Flujos de trabajo multi-turno con continuidad |
| Contexto manual | Tu código pasa el historial en cada prompt | Control total, sin dependencia de sesiones del servidor |
| Sistema de archivos | El agente escribe notas en archivos | Persistencia entre procesos, auditoría |
2. SystemMessage de inicio: capturando el session_id
Cuando inicias una nueva sesión, el SDK emite un SystemMessage especial al principio del stream que contiene el session_id. Debes capturarlo para poder reanudar la sesión después.
Captura del session_id en Python
import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.types import SystemMessage, AssistantMessage, ResultMessage
async def query_con_sesion(
prompt: str,
opciones: ClaudeCodeOptions,
session_id: str = None
) -> tuple[str, str]:
"""
Ejecuta un query y retorna (resultado, session_id).
Si se provee session_id, reanuda esa sesión.
Si no se provee, crea una nueva sesión y retorna su ID.
"""
if session_id:
opciones = ClaudeCodeOptions(
**{k: v for k, v in vars(opciones).items() if v is not None},
resume=session_id
)
session_capturada = session_id # Puede ser None para sesiones nuevas
resultado_texto = ""
async for mensaje in query(prompt=prompt, options=opciones):
# SystemMessage contiene el session_id al inicio de la sesión
if isinstance(mensaje, SystemMessage):
if hasattr(mensaje, 'session_id') and mensaje.session_id:
session_capturada = mensaje.session_id
print(f"[Sesión] ID capturado: {session_capturada}")
# AssistantMessage contiene el texto de respuesta
elif isinstance(mensaje, AssistantMessage):
if hasattr(mensaje, 'message') and hasattr(mensaje.message, 'content'):
for bloque in mensaje.message.content:
if hasattr(bloque, 'text'):
resultado_texto = bloque.text
# ResultMessage marca el fin — puede contener session_id también
elif isinstance(mensaje, ResultMessage):
if hasattr(mensaje, 'session_id') and mensaje.session_id:
session_capturada = mensaje.session_id
return resultado_texto, session_capturada
# Uso básico
async def ejemplo_sesion_basica():
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Glob", "Grep"],
max_turns=10,
cwd="/mi/proyecto",
)
# Primera query — crea nueva sesión
resultado1, session_id = await query_con_sesion(
"Analiza la estructura general del proyecto y encuentra los archivos principales.",
opciones
)
print(f"Primer análisis completado. Session ID: {session_id}")
print(f"Resultado: {resultado1[:200]}...")
# Segunda query — continúa la misma sesión
if session_id:
resultado2, _ = await query_con_sesion(
"Basándote en el análisis anterior, ¿cuáles son los 3 problemas más críticos?",
opciones,
session_id=session_id
)
print(f"\nAnálisis profundo: {resultado2[:200]}...")
asyncio.run(ejemplo_sesion_basica())Captura del session_id en TypeScript
import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";
interface SesionResultado {
texto: string;
sessionId: string | null;
}
async function queryConSesion(
prompt: string,
opciones: ClaudeCodeOptions,
sessionId?: string
): Promise<SesionResultado> {
const opcionesSesion: ClaudeCodeOptions = sessionId
? { ...opciones, resume: sessionId }
: opciones;
let sessionCapturada: string | null = sessionId ?? null;
let resultadoTexto = "";
for await (const mensaje of query({ prompt, options: opcionesSesion })) {
// Capturar session_id del SystemMessage inicial
if (mensaje.type === "system" && "session_id" in mensaje) {
sessionCapturada = (mensaje as any).session_id ?? sessionCapturada;
console.log(`[Sesión] ID capturado: ${sessionCapturada}`);
}
// Capturar texto del AssistantMessage
if (mensaje.type === "assistant" && mensaje.message.content) {
for (const bloque of mensaje.message.content) {
if (bloque.type === "text") {
resultadoTexto = bloque.text;
}
}
}
// ResultMessage también puede tener session_id
if (mensaje.type === "result" && "session_id" in mensaje) {
sessionCapturada = (mensaje as any).session_id ?? sessionCapturada;
}
}
return { texto: resultadoTexto, sessionId: sessionCapturada };
}
// Uso
async function ejemploSesionBasica() {
const opciones: ClaudeCodeOptions = {
model: "claude-opus-4-5",
allowedTools: ["Read", "Glob", "Grep"],
maxTurns: 10,
cwd: "/mi/proyecto",
};
// Primera query — nueva sesión
const { texto: res1, sessionId } = await queryConSesion(
"Analiza la estructura del proyecto.",
opciones
);
console.log(`Session ID: ${sessionId}`);
if (sessionId) {
// Segunda query — reanuda sesión
const { texto: res2 } = await queryConSesion(
"¿Cuáles son los 3 problemas más críticos que encontraste?",
opciones,
sessionId
);
console.log(res2);
}
}
ejemploSesionBasica();3. resume — Reanudar sesiones
La opción resume en ClaudeCodeOptions indica al SDK que debe continuar una sesión existente en lugar de crear una nueva. Cuando usas resume, el modelo tiene acceso a todo el historial de la sesión anterior.
Cómo funciona internamente
graph TD
A[query con resume='abc123']
A --> B[SDK envía session_id a la API]
B --> C{¿Sesión existe?}
C -->|Sí| D[Cargar historial completo de la sesión]
C -->|No — expirada o inválida| E[Error: SessionNotFound]
D --> F[Agregar nuevo prompt al historial]
F --> G[Modelo procesa con contexto completo]
G --> H[Stream de respuesta]
H --> I[Actualizar historial de la sesión]Escenarios de uso de resume
import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
async def asistente_persistente(directorio: str):
"""
Asistente que mantiene contexto a lo largo de múltiples interacciones.
Simula una sesión de trabajo donde el agente recuerda el contexto.
"""
opciones_base = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"],
max_turns=15,
cwd=directorio,
)
session_id = None
historial_local = [] # Para debug/logging
async def ejecutar_paso(descripcion: str, prompt: str) -> str:
nonlocal session_id
opciones = ClaudeCodeOptions(
**vars(opciones_base),
resume=session_id # None en la primera llamada
)
resultado = ""
async for mensaje in query(prompt=prompt, options=opciones):
if hasattr(mensaje, 'session_id') and mensaje.session_id:
session_id = mensaje.session_id
if hasattr(mensaje, 'content'):
for bloque in mensaje.content:
if hasattr(bloque, 'text'):
resultado = bloque.text
historial_local.append({
"paso": descripcion,
"session_id": session_id,
"resultado_preview": resultado[:100]
})
print(f"[Paso completado] {descripcion} (session: {session_id})")
return resultado
# Paso 1: Exploración inicial
estructura = await ejecutar_paso(
"exploración",
"Explora el proyecto completo. Toma nota de todos los archivos importantes, "
"la arquitectura usada y las tecnologías involucradas."
)
# Paso 2: Análisis de dependencias (recuerda la exploración)
deps = await ejecutar_paso(
"análisis de dependencias",
"Basándote en lo que exploraste, analiza las dependencias entre módulos. "
"¿Hay dependencias circulares? ¿Módulos muy acoplados?"
)
# Paso 3: Identificación de mejoras (recuerda todo lo anterior)
mejoras = await ejecutar_paso(
"identificación de mejoras",
"Con toda la información que tienes del proyecto, lista las 5 mejoras "
"más importantes ordenadas por impacto. Sé específico con los archivos."
)
# Paso 4: Plan de acción (recuerda todo)
plan = await ejecutar_paso(
"plan de acción",
"Genera un plan de acción detallado para implementar las mejoras que identificaste. "
"Incluye estimación de esfuerzo y dependencias entre tareas."
)
return {
"session_id": session_id,
"estructura": estructura,
"dependencias": deps,
"mejoras": mejoras,
"plan": plan,
"historial": historial_local,
}
asyncio.run(asistente_persistente("/mi/proyecto"))Resume con verificación de sesión
from claude_code_sdk import query, ClaudeCodeOptions
import asyncio
async def verificar_y_reanudar(session_id: str, opciones: ClaudeCodeOptions) -> bool:
"""
Verifica si una sesión es válida antes de reanudarla.
Retorna True si la sesión existe y es reanudable.
"""
opciones_test = ClaudeCodeOptions(
model="claude-haiku-4-5", # Modelo más barato para la verificación
allowed_tools=[],
max_turns=1,
resume=session_id,
)
try:
async for mensaje in query(
prompt="¿Recuerdas el contexto de esta sesión? Responde solo 'sí' o 'no'.",
options=opciones_test
):
if hasattr(mensaje, 'content'):
return True # Si llegamos aquí, la sesión es válida
return True
except Exception as e:
if "session" in str(e).lower() or "not found" in str(e).lower():
return False
raise # Re-lanzar errores que no son de sesión inválida
async def reanudar_o_nueva(session_id: str, opciones: ClaudeCodeOptions) -> ClaudeCodeOptions:
"""
Retorna opciones con resume si la sesión es válida,
o opciones sin resume si la sesión expiró o no existe.
"""
if session_id and await verificar_y_reanudar(session_id, opciones):
print(f"[Sesión] Reanudando sesión: {session_id}")
return ClaudeCodeOptions(**{**vars(opciones), 'resume': session_id})
else:
print(f"[Sesión] Sesión no disponible, iniciando nueva")
return opciones4. Persistencia de sesiones
Las sesiones tienen una duración limitada en el servidor de Anthropic. Para construir sistemas robustos, debes persistir el session_id y manejar la expiración.
Estrategias de persistencia
graph TD
SID[session_id]
SID --> MEM[In-Memory - Desarrollo]
SID --> FILE[Archivo JSON - Single process]
SID --> SQLITE[SQLite - Local multi-proceso]
SID --> REDIS[Redis - Distribuido]
SID --> PG[PostgreSQL - Producción]
MEM --> |"Pérdida al reiniciar"| L1[Limitación]
FILE --> |"Sin concurrencia"| L2[Limitación]
SQLITE --> |"Un servidor"| L3[Limitación]
REDIS --> |"Infraestructura"| L4[Require]
PG --> |"Máxima fiabilidad"| OK[Producción]Implementación: SQLite para persistencia local
import sqlite3
import asyncio
import json
from datetime import datetime, timedelta
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
from typing import Optional
class SesionStore:
"""Almacén de sesiones con SQLite."""
def __init__(self, db_path: str = "~/.claude_sessions.db"):
self.db_path = Path(db_path).expanduser()
self._inicializar_db()
def _inicializar_db(self):
"""Crea las tablas necesarias si no existen."""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS sesiones (
id TEXT PRIMARY KEY,
nombre TEXT NOT NULL,
session_id TEXT NOT NULL,
creado_en TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
actualizado_en TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
expira_en TIMESTAMP,
metadatos TEXT DEFAULT '{}',
activo INTEGER DEFAULT 1
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_sesiones_nombre
ON sesiones(nombre)
""")
conn.commit()
def guardar(
self,
nombre: str,
session_id: str,
metadatos: dict = None,
ttl_horas: int = 24
) -> str:
"""Guarda o actualiza una sesión."""
expira_en = datetime.now() + timedelta(hours=ttl_horas)
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT INTO sesiones (id, nombre, session_id, expira_en, metadatos, actualizado_en)
VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(id) DO UPDATE SET
session_id = excluded.session_id,
actualizado_en = CURRENT_TIMESTAMP,
expira_en = excluded.expira_en,
metadatos = excluded.metadatos
""", (
nombre, # Usar nombre como ID para búsqueda por nombre
nombre,
session_id,
expira_en.isoformat(),
json.dumps(metadatos or {})
))
conn.commit()
return session_id
def obtener(self, nombre: str) -> Optional[str]:
"""Retorna el session_id si la sesión existe y no expiró."""
with sqlite3.connect(self.db_path) as conn:
row = conn.execute("""
SELECT session_id, expira_en
FROM sesiones
WHERE nombre = ? AND activo = 1
ORDER BY actualizado_en DESC
LIMIT 1
""", (nombre,)).fetchone()
if not row:
return None
session_id, expira_en_str = row
if expira_en_str:
expira_en = datetime.fromisoformat(expira_en_str)
if datetime.now() > expira_en:
self.invalidar(nombre)
return None
return session_id
def invalidar(self, nombre: str):
"""Marca una sesión como inactiva."""
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"UPDATE sesiones SET activo = 0 WHERE nombre = ?",
(nombre,)
)
conn.commit()
def limpiar_expiradas(self):
"""Elimina sesiones expiradas de la base de datos."""
with sqlite3.connect(self.db_path) as conn:
eliminadas = conn.execute("""
DELETE FROM sesiones
WHERE expira_en < CURRENT_TIMESTAMP OR activo = 0
""").rowcount
conn.commit()
return eliminadas
def listar(self) -> list:
"""Lista todas las sesiones activas."""
with sqlite3.connect(self.db_path) as conn:
rows = conn.execute("""
SELECT nombre, session_id, creado_en, actualizado_en, expira_en
FROM sesiones
WHERE activo = 1
ORDER BY actualizado_en DESC
""").fetchall()
return [
{
"nombre": r[0],
"session_id": r[1],
"creado_en": r[2],
"actualizado_en": r[3],
"expira_en": r[4],
}
for r in rows
]
# Uso del store
store = SesionStore()
async def asistente_con_memoria(nombre_proyecto: str, pregunta: str) -> str:
"""
Asistente que recuerda el contexto entre invocaciones.
Busca sesión existente o crea una nueva.
"""
session_id = store.obtener(nombre_proyecto)
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Glob", "Grep"],
max_turns=10,
cwd=f"/proyectos/{nombre_proyecto}",
resume=session_id, # None si es la primera vez
)
if session_id:
print(f"[Memoria] Retomando sesión existente para: {nombre_proyecto}")
else:
print(f"[Memoria] Nueva sesión para: {nombre_proyecto}")
resultado = ""
nuevo_session_id = session_id
async for mensaje in query(prompt=pregunta, options=opciones):
if hasattr(mensaje, 'session_id') and mensaje.session_id:
nuevo_session_id = mensaje.session_id
if hasattr(mensaje, 'content'):
for bloque in mensaje.content:
if hasattr(bloque, 'text'):
resultado = bloque.text
# Guardar/actualizar la sesión
if nuevo_session_id:
store.guardar(
nombre=nombre_proyecto,
session_id=nuevo_session_id,
metadatos={"ultima_pregunta": pregunta[:100]},
ttl_horas=8 # Expira en 8 horas
)
return resultado
# Uso
async def main():
# Primera invocación — nueva sesión
res1 = await asistente_con_memoria("mi-api", "¿Cuál es la arquitectura del proyecto?")
print(res1)
# Segunda invocación (horas después) — retoma la sesión
res2 = await asistente_con_memoria("mi-api", "¿Qué mejoras recomendarías basándote en lo que analizaste?")
print(res2)
asyncio.run(main())Implementación: Redis para sistemas distribuidos
import asyncio
import json
from claude_code_sdk import query, ClaudeCodeOptions
from typing import Optional
try:
import redis.asyncio as redis
REDIS_DISPONIBLE = True
except ImportError:
REDIS_DISPONIBLE = False
class SesionStoreRedis:
"""Almacén de sesiones con Redis para sistemas distribuidos."""
def __init__(self, url: str = "redis://localhost:6379", ttl: int = 86400):
self.url = url
self.ttl = ttl # TTL en segundos (default: 24 horas)
self._cliente = None
async def _cliente_redis(self):
if not self._cliente:
self._cliente = await redis.from_url(self.url)
return self._cliente
def _clave(self, nombre: str) -> str:
return f"claude_session:{nombre}"
async def guardar(self, nombre: str, session_id: str, metadatos: dict = None):
"""Guarda la sesión en Redis con TTL."""
r = await self._cliente_redis()
datos = {
"session_id": session_id,
"metadatos": json.dumps(metadatos or {}),
}
await r.hset(self._clave(nombre), mapping=datos)
await r.expire(self._clave(nombre), self.ttl)
async def obtener(self, nombre: str) -> Optional[str]:
"""Retorna el session_id si existe."""
r = await self._cliente_redis()
datos = await r.hget(self._clave(nombre), "session_id")
return datos.decode() if datos else None
async def invalidar(self, nombre: str):
"""Elimina la sesión."""
r = await self._cliente_redis()
await r.delete(self._clave(nombre))
async def extender_ttl(self, nombre: str):
"""Extiende el TTL de una sesión activa (sliding expiration)."""
r = await self._cliente_redis()
await r.expire(self._clave(nombre), self.ttl)5. Fork de sesiones
El fork permite crear una “rama” de una sesión existente, explorando alternativas sin afectar la sesión original.
graph LR
S1[Sesión original]
S1 --> F1[Fork A: estrategia conservadora]
S1 --> F2[Fork B: estrategia agresiva]
S1 --> F3[Fork C: estrategia experimental]
F1 --> R1[Resultado A]
F2 --> R2[Resultado B]
F3 --> R3[Resultado C]
R1 --> C[Comparar y elegir mejor]
R2 --> C
R3 --> CImplementación de fork manual
El SDK no tiene un comando fork explícito, pero puedes implementarlo iniciando una nueva sesión que incluye el contexto de la sesión original como parte del prompt:
import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
from typing import Optional
async def fork_sesion(
session_id_original: str,
prompt_divergencia: str,
opciones: ClaudeCodeOptions,
contexto_previo: str = ""
) -> tuple[str, str]:
"""
Crea un 'fork' de una sesión: nueva sesión con contexto de la original.
Retorna (resultado, nuevo_session_id).
"""
# El fork es una nueva sesión que incluye el contexto previo
prompt_completo = f"""Contexto de la sesión anterior:
{contexto_previo}
---
Continuando desde ese contexto, pero explorando una alternativa diferente:
{prompt_divergencia}"""
# Nueva sesión — NO usamos resume de la original
opciones_fork = ClaudeCodeOptions(
model=opciones.model,
allowed_tools=opciones.allowed_tools,
max_turns=opciones.max_turns,
cwd=opciones.cwd,
system_prompt=opciones.system_prompt,
# Sin resume — nueva sesión independiente
)
resultado = ""
nuevo_session_id = None
async for mensaje in query(prompt=prompt_completo, options=opciones_fork):
if hasattr(mensaje, 'session_id') and mensaje.session_id:
nuevo_session_id = mensaje.session_id
if hasattr(mensaje, 'content'):
for bloque in mensaje.content:
if hasattr(bloque, 'text'):
resultado = bloque.text
return resultado, nuevo_session_id
async def explorar_alternativas(
session_id_base: str,
contexto_previo: str,
alternativas: list,
opciones: ClaudeCodeOptions
) -> list:
"""
Explora múltiples alternativas en paralelo a partir de la misma base.
"""
tareas = [
fork_sesion(
session_id_base,
alternativa["prompt"],
opciones,
contexto_previo
)
for alternativa in alternativas
]
resultados = await asyncio.gather(*tareas, return_exceptions=True)
return [
{
"nombre": alternativas[i]["nombre"],
"resultado": r[0] if not isinstance(r, Exception) else "",
"session_id": r[1] if not isinstance(r, Exception) else None,
"error": str(r) if isinstance(r, Exception) else None,
}
for i, r in enumerate(resultados)
]
# Uso: explorar diferentes estrategias de refactoring
async def comparar_estrategias():
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read"],
max_turns=10,
cwd="/mi/proyecto",
)
# Sesión base: análisis del problema
resultado_base = ""
session_base = None
async for m in query(
prompt="Analiza este módulo de autenticación y describe sus problemas actuales.",
options=opciones
):
if hasattr(m, 'session_id') and m.session_id:
session_base = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado_base = b.text
# Explorar 3 estrategias de refactoring en paralelo
alternativas_resultados = await explorar_alternativas(
session_base,
resultado_base,
[
{
"nombre": "conservadora",
"prompt": "Propón una refactorización conservadora que minimice el riesgo de romper cosas existentes."
},
{
"nombre": "moderna",
"prompt": "Propón una refactorización moderna usando los últimos patrones de Python."
},
{
"nombre": "incremental",
"prompt": "Propón una refactorización incremental en 5 pasos pequeños independientes."
},
],
opciones
)
for alt in alternativas_resultados:
print(f"\n=== Estrategia: {alt['nombre']} ===")
if alt['resultado']:
print(alt['resultado'][:300])
else:
print(f"Error: {alt['error']}")
asyncio.run(comparar_estrategias())6. Sesiones largas y compactación
Las sesiones largas acumulan muchos tokens en el historial. El SDK compacta automáticamente el contexto cuando se acerca a los límites, pero hay señales que puedes detectar.
Cómo funciona la compactación automática
sequenceDiagram
participant App as Aplicación
participant SDK as SDK
participant API as Anthropic API
Note over App,API: Sesión normal
App->>SDK: query(resume=session_id) — turno 50
SDK->>API: Historial: 80K tokens
API-->>SDK: Respuesta normal
Note over App,API: Acercándose al límite
App->>SDK: query(resume=session_id) — turno 80
SDK->>API: Historial: 150K tokens
API->>API: Compactar historial antiguo
API-->>SDK: SystemMessage con is_compaction=True
SDK-->>App: Notificación de compactación
API-->>SDK: Respuesta (contexto compactado)Detectar y manejar compactación
import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.types import SystemMessage
async def query_con_deteccion_compactacion(
prompt: str,
opciones: ClaudeCodeOptions
) -> dict:
"""
Ejecuta query y detecta si ocurrió compactación del contexto.
"""
resultado = ""
session_id = None
hubo_compactacion = False
tokens_usados = 0
async for mensaje in query(prompt=prompt, options=opciones):
# Detectar SystemMessage de compactación
if isinstance(mensaje, SystemMessage):
if hasattr(mensaje, 'session_id') and mensaje.session_id:
session_id = mensaje.session_id
# La compactación se señala en el SystemMessage
if hasattr(mensaje, 'subtype') and mensaje.subtype == 'compact':
hubo_compactacion = True
print("[Sesión] Contexto compactado por el SDK")
if hasattr(mensaje, 'content'):
for bloque in mensaje.content:
if hasattr(bloque, 'text'):
resultado = bloque.text
# Rastrear uso de tokens si está disponible
if hasattr(mensaje, 'usage'):
tokens_usados = getattr(mensaje.usage, 'total_tokens', 0)
return {
"resultado": resultado,
"session_id": session_id,
"hubo_compactacion": hubo_compactacion,
"tokens_usados": tokens_usados,
}
async def sesion_con_punto_de_guardado(
sesion_nombre: str,
pasos: list,
directorio: str
):
"""
Ejecuta múltiples pasos en la misma sesión.
Guarda puntos de control por si necesitamos retomar.
"""
store = SesionStore() # Del ejemplo anterior
session_id = store.obtener(sesion_nombre)
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Write", "Edit", "Glob", "Grep"],
max_turns=15,
cwd=directorio,
resume=session_id,
)
resultados = []
for i, paso in enumerate(pasos):
print(f"\n[Paso {i+1}/{len(pasos)}] {paso['nombre']}")
res = await query_con_deteccion_compactacion(paso['prompt'], opciones)
# Guardar sesión después de cada paso exitoso
if res['session_id']:
store.guardar(
nombre=sesion_nombre,
session_id=res['session_id'],
metadatos={
"ultimo_paso": paso['nombre'],
"paso_numero": i + 1,
"total_pasos": len(pasos),
"tokens": res['tokens_usados'],
}
)
# Actualizar opciones con el session_id para el siguiente paso
opciones = ClaudeCodeOptions(**{**vars(opciones), 'resume': res['session_id']})
if res['hubo_compactacion']:
print(f"[Sesión] Nota: el contexto fue compactado en este paso")
resultados.append({
"paso": paso['nombre'],
**res
})
return resultados7. Gestión de ventana de contexto
La ventana de contexto es finita. Estrategias para optimizarla:
Estimación de tokens por tipo de contenido
def estimar_tokens(texto: str) -> int:
"""
Estimación aproximada de tokens.
Regla general: ~4 caracteres por token en inglés/español.
"""
return len(texto) // 4
def planificar_sesion(
contexto_inicial: str,
pasos: list,
limite_tokens: int = 100_000,
reserva_output: int = 4_000
) -> dict:
"""
Planifica cuántos pasos caben en una sesión dado el límite de tokens.
"""
tokens_disponibles = limite_tokens - reserva_output
tokens_usados = estimar_tokens(contexto_inicial)
pasos_posibles = []
pasos_excluidos = []
for paso in pasos:
tokens_paso = estimar_tokens(paso.get('prompt', ''))
# Cada turno acumula tokens del historial
tokens_acumulados = tokens_usados + tokens_paso + reserva_output
if tokens_acumulados < tokens_disponibles:
pasos_posibles.append(paso)
tokens_usados = tokens_acumulados
else:
pasos_excluidos.append(paso)
return {
"pasos_posibles": pasos_posibles,
"pasos_excluidos": pasos_excluidos,
"tokens_estimados": tokens_usados,
"porcentaje_ventana": round(tokens_usados / limite_tokens * 100, 1),
}
# Estrategia: dividir en sub-sesiones si hay demasiados pasos
async def ejecutar_con_division_automatica(
pasos: list,
opciones: ClaudeCodeOptions,
limite_tokens: int = 80_000
) -> list:
"""
Divide automáticamente pasos en múltiples sesiones si el contexto es muy grande.
"""
grupos = []
grupo_actual = []
tokens_grupo = 0
for paso in pasos:
tokens_paso = estimar_tokens(paso.get('prompt', ''))
if tokens_grupo + tokens_paso > limite_tokens and grupo_actual:
grupos.append(grupo_actual)
grupo_actual = [paso]
tokens_grupo = tokens_paso
else:
grupo_actual.append(paso)
tokens_grupo += tokens_paso
if grupo_actual:
grupos.append(grupo_actual)
print(f"[Contexto] Dividido en {len(grupos)} sesiones")
todos_resultados = []
for i, grupo in enumerate(grupos):
print(f"[Contexto] Ejecutando sesión {i+1}/{len(grupos)} con {len(grupo)} pasos")
resultados_grupo = await ejecutar_pasos_en_sesion(grupo, opciones)
todos_resultados.extend(resultados_grupo)
return todos_resultados
async def ejecutar_pasos_en_sesion(
pasos: list,
opciones: ClaudeCodeOptions
) -> list:
"""Ejecuta una lista de pasos en una sola sesión."""
session_id = None
resultados = []
for paso in pasos:
opts = ClaudeCodeOptions(**{**vars(opciones), 'resume': session_id})
resultado = ""
async for m in query(prompt=paso['prompt'], options=opts):
if hasattr(m, 'session_id') and m.session_id:
session_id = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
resultados.append({"paso": paso['nombre'], "resultado": resultado})
return resultados8. Sesiones conversacionales
Un chatbot con memoria que recuerda toda la conversación usando sesiones del SDK.
import asyncio
import readline # Para historial en la terminal
from claude_code_sdk import query, ClaudeCodeOptions
from datetime import datetime
class ChatbotConMemoria:
"""Chatbot que mantiene memoria de la conversación usando sesiones SDK."""
def __init__(self, nombre: str, directorio: str = None):
self.nombre = nombre
self.directorio = directorio
self.session_id = None
self.turno = 0
self.inicio = datetime.now()
@property
def opciones(self) -> ClaudeCodeOptions:
return ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Glob", "Grep"] if self.directorio else [],
max_turns=20,
cwd=self.directorio,
resume=self.session_id,
system_prompt=f"""Eres un asistente de desarrollo para el proyecto '{self.nombre}'.
Recuerdas toda la conversación y el contexto del proyecto.
Eres conciso, técnico y accionable.
Cuando el usuario hace referencia a algo mencionado antes, úsalo directamente."""
)
async def chat(self, mensaje: str) -> str:
"""Envía un mensaje y retorna la respuesta."""
self.turno += 1
resultado = ""
async for m in query(prompt=mensaje, options=self.opciones):
if hasattr(m, 'session_id') and m.session_id:
self.session_id = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
return resultado
def info_sesion(self) -> dict:
duracion = (datetime.now() - self.inicio).seconds
return {
"session_id": self.session_id,
"turnos": self.turno,
"duracion_segundos": duracion,
}
async def chat_interactivo():
"""Chatbot interactivo en terminal."""
print("=== Chatbot con Memoria (Claude Agent SDK) ===")
nombre_proyecto = input("Nombre del proyecto: ").strip() or "mi-proyecto"
directorio = input("Directorio del proyecto (Enter para omitir): ").strip() or None
bot = ChatbotConMemoria(nombre_proyecto, directorio)
print(f"\nChat iniciado. Escribe 'salir' para terminar, 'info' para ver la sesión.\n")
while True:
try:
usuario = input("Tú: ").strip()
except (KeyboardInterrupt, EOFError):
break
if not usuario:
continue
if usuario.lower() == "salir":
break
if usuario.lower() == "info":
info = bot.info_sesion()
print(f"[Sesión] ID: {info['session_id']}")
print(f"[Sesión] Turnos: {info['turnos']}, Duración: {info['duracion_segundos']}s")
continue
print("Claude: ", end="", flush=True)
respuesta = await bot.chat(usuario)
print(respuesta)
print()
info = bot.info_sesion()
print(f"\n[Sesión terminada] {info['turnos']} turnos, {info['duracion_segundos']}s")
print(f"Session ID para retomar: {info['session_id']}")
asyncio.run(chat_interactivo())Chatbot en TypeScript con historial
import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";
import * as readline from "readline";
class ChatbotConMemoria {
private sessionId: string | null = null;
private turno = 0;
private readonly nombre: string;
constructor(nombre: string) {
this.nombre = nombre;
}
private get opciones(): ClaudeCodeOptions {
return {
model: "claude-opus-4-5",
allowedTools: [],
maxTurns: 20,
resume: this.sessionId ?? undefined,
systemPrompt: `Eres un asistente llamado ${this.nombre}.
Recuerdas toda la conversación anterior.
Eres conciso y directo.`,
};
}
async chat(mensaje: string): Promise<string> {
this.turno++;
let resultado = "";
for await (const m of query({ prompt: mensaje, options: this.opciones })) {
if ("session_id" in m && (m as any).session_id) {
this.sessionId = (m as any).session_id;
}
if (m.type === "assistant" && m.message.content) {
for (const bloque of m.message.content) {
if (bloque.type === "text") resultado = bloque.text;
}
}
}
return resultado;
}
get infoSesion() {
return { sessionId: this.sessionId, turnos: this.turno };
}
}
async function chatInteractivo() {
const rl = readline.createInterface({
input: process.stdin,
output: process.stdout,
});
const bot = new ChatbotConMemoria("Asistente");
console.log('Chat iniciado. Escribe "salir" para terminar.\n');
const preguntar = () => {
rl.question("Tú: ", async (input) => {
const mensaje = input.trim();
if (!mensaje || mensaje.toLowerCase() === "salir") {
const info = bot.infoSesion;
console.log(`\nSesión: ${info.sessionId}, Turnos: ${info.turnos}`);
rl.close();
return;
}
const respuesta = await bot.chat(mensaje);
console.log(`Claude: ${respuesta}\n`);
preguntar();
});
};
preguntar();
}
chatInteractivo();9. Sesiones de desarrollo
Un asistente que recuerda el estado del proyecto entre sesiones de trabajo.
import asyncio
import json
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
class AsistenteDesarrollo:
"""
Asistente de desarrollo que persiste el contexto del proyecto.
Recuerda análisis previos, decisiones tomadas y estado del trabajo.
"""
def __init__(self, directorio_proyecto: str):
self.directorio = Path(directorio_proyecto)
self.nombre = self.directorio.name
self.archivo_estado = self.directorio / ".claude_session.json"
self.store = SesionStore() # Del ejemplo de SQLite
def _cargar_estado(self) -> dict:
"""Carga el estado guardado del proyecto."""
if self.archivo_estado.exists():
try:
with open(self.archivo_estado, encoding='utf-8') as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
pass
return {}
def _guardar_estado(self, estado: dict):
"""Guarda el estado del proyecto."""
with open(self.archivo_estado, 'w', encoding='utf-8') as f:
json.dump(estado, f, ensure_ascii=False, indent=2)
def _construir_contexto_inicial(self) -> str:
"""Construye el contexto del sistema con el historial del proyecto."""
estado = self._cargar_estado()
if not estado:
return ""
partes = [f"Estás trabajando en el proyecto '{self.nombre}'."]
if estado.get("ultimo_analisis"):
partes.append(f"\nÚltimo análisis (resumido):\n{estado['ultimo_analisis'][:500]}")
if estado.get("tareas_pendientes"):
tareas = "\n".join(f"- {t}" for t in estado["tareas_pendientes"][:5])
partes.append(f"\nTareas pendientes:\n{tareas}")
if estado.get("decisiones"):
decisiones = "\n".join(f"- {d}" for d in estado["decisiones"][-3:])
partes.append(f"\nDecisiones recientes:\n{decisiones}")
return "\n".join(partes)
async def consultar(self, pregunta: str) -> str:
"""Hace una consulta al asistente con contexto del proyecto."""
estado = self._cargar_estado()
session_id = self.store.obtener(f"dev_{self.nombre}")
# Construir system prompt con contexto histórico
contexto = self._construir_contexto_inicial()
system_prompt = f"""Eres un asistente de desarrollo experto.
{contexto}
Recuerdas toda la conversación actual y el contexto histórico del proyecto.
Cuando tomes decisiones importantes, menciónalas explícitamente."""
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"],
max_turns=15,
cwd=str(self.directorio),
resume=session_id,
system_prompt=system_prompt,
)
resultado = ""
nuevo_session_id = session_id
async for m in query(prompt=pregunta, options=opciones):
if hasattr(m, 'session_id') and m.session_id:
nuevo_session_id = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
# Actualizar estado
if nuevo_session_id:
self.store.guardar(f"dev_{self.nombre}", nuevo_session_id)
# Guardar resumen en estado local
estado["ultima_consulta"] = pregunta[:100]
if len(resultado) > 100:
estado["ultimo_analisis"] = resultado[:500]
self._guardar_estado(estado)
return resultado
async def agregar_tarea(self, tarea: str):
"""Agrega una tarea al estado del proyecto."""
estado = self._cargar_estado()
tareas = estado.get("tareas_pendientes", [])
tareas.append(tarea)
estado["tareas_pendientes"] = tareas[-10:] # Mantener solo las últimas 10
self._guardar_estado(estado)
print(f"[Asistente] Tarea agregada: {tarea}")
async def marcar_completada(self, tarea: str):
"""Marca una tarea como completada."""
estado = self._cargar_estado()
tareas = estado.get("tareas_pendientes", [])
nuevas_tareas = [t for t in tareas if tarea.lower() not in t.lower()]
estado["tareas_pendientes"] = nuevas_tareas
decisiones = estado.get("decisiones", [])
decisiones.append(f"Completado: {tarea}")
estado["decisiones"] = decisiones[-20:] # Mantener las últimas 20
self._guardar_estado(estado)
# Uso
async def main():
asistente = AsistenteDesarrollo("/mi/proyecto")
# Primera sesión de trabajo
analisis = await asistente.consultar(
"Analiza el módulo de autenticación y encuentra los problemas más urgentes."
)
print(analisis)
await asistente.agregar_tarea("Refactorizar el hash de passwords a bcrypt")
await asistente.agregar_tarea("Agregar rate limiting al endpoint de login")
# Días después — retoma el contexto
plan = await asistente.consultar(
"¿Cómo debería implementar las tareas pendientes? Empieza con la más urgente."
)
print(plan)
asyncio.run(main())10. Multi-tenant sessions
Cuando múltiples usuarios utilizan el mismo sistema, cada uno necesita su propia sesión aislada.
import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
from typing import Dict
import hashlib
class GestorSesionesTenant:
"""Gestiona sesiones aisladas por tenant/usuario."""
def __init__(self, store: SesionStore):
self.store = store
def _clave_sesion(self, tenant_id: str, contexto: str) -> str:
"""Genera una clave única para un tenant y contexto."""
hash_corto = hashlib.md5(f"{tenant_id}_{contexto}".encode()).hexdigest()[:8]
return f"tenant_{tenant_id}_{hash_corto}"
async def query_para_tenant(
self,
tenant_id: str,
prompt: str,
contexto: str = "default",
directorio: str = None,
system_prompt: str = None
) -> str:
"""
Ejecuta un query en la sesión del tenant.
Cada tenant tiene sesiones completamente aisladas.
"""
clave = self._clave_sesion(tenant_id, contexto)
session_id = self.store.obtener(clave)
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Glob"] if directorio else [],
max_turns=10,
cwd=directorio,
resume=session_id,
system_prompt=system_prompt or f"Asistes al usuario {tenant_id}.",
)
resultado = ""
nuevo_session_id = session_id
async for m in query(prompt=prompt, options=opciones):
if hasattr(m, 'session_id') and m.session_id:
nuevo_session_id = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
if nuevo_session_id:
self.store.guardar(
nombre=clave,
session_id=nuevo_session_id,
metadatos={"tenant": tenant_id, "contexto": contexto},
ttl_horas=4
)
return resultado
def invalidar_tenant(self, tenant_id: str):
"""Invalida todas las sesiones de un tenant (logout completo)."""
# En una implementación real, iterar sobre todas las claves del tenant
print(f"[Tenant] Invalidando sesiones de: {tenant_id}")
# Uso: múltiples usuarios en paralelo
async def simular_multitenant():
store = SesionStore()
gestor = GestorSesionesTenant(store)
# Múltiples usuarios haciendo consultas al mismo tiempo
usuarios = [
("user_001", "Explica el patrón Repository en Python"),
("user_002", "¿Cómo implemento autenticación JWT?"),
("user_003", "Muéstrame un ejemplo de Clean Architecture"),
]
resultados = await asyncio.gather(*[
gestor.query_para_tenant(
tenant_id=user_id,
prompt=pregunta,
contexto="general"
)
for user_id, pregunta in usuarios
])
for (user_id, pregunta), resultado in zip(usuarios, resultados):
print(f"\n{user_id}: {pregunta[:50]}")
print(f"Respuesta: {resultado[:100]}...")
# Segunda ronda — cada usuario tiene su propio contexto
seguimiento = await asyncio.gather(*[
gestor.query_para_tenant(
tenant_id="user_001",
prompt="Dame un ejemplo más completo",
contexto="general"
),
gestor.query_para_tenant(
tenant_id="user_002",
prompt="¿Y cómo manejo el refresh token?",
contexto="general"
),
])
asyncio.run(simular_multitenant())11. Session store patterns
Comparación de implementaciones
| Store | Ventajas | Desventajas | Cuándo usar |
|---|---|---|---|
| In-Memory (dict) | Simple, rápido | Pérdida al reiniciar | Tests, desarrollo |
| Archivo JSON | Simple, portátil | Sin concurrencia | Scripts single-process |
| SQLite | Local, ACID, queries | No distribuido | Apps mono-servidor |
| Redis | Rápido, TTL nativo | Infraestructura extra | Producción distribuida |
| PostgreSQL | ACID, queries complejas | Más overhead | Producción crítica |
In-Memory store para tests y desarrollo
from typing import Optional, Dict
from datetime import datetime, timedelta
import asyncio
class SesionStoreMemoria:
"""Store in-memory. Solo para tests y desarrollo."""
def __init__(self, ttl_horas: int = 8):
self._sesiones: Dict[str, dict] = {}
self.ttl = ttl_horas
def guardar(
self,
nombre: str,
session_id: str,
metadatos: dict = None,
ttl_horas: int = None
) -> str:
self._sesiones[nombre] = {
"session_id": session_id,
"creado_en": datetime.now(),
"expira_en": datetime.now() + timedelta(hours=ttl_horas or self.ttl),
"metadatos": metadatos or {},
}
return session_id
def obtener(self, nombre: str) -> Optional[str]:
sesion = self._sesiones.get(nombre)
if not sesion:
return None
if datetime.now() > sesion["expira_en"]:
del self._sesiones[nombre]
return None
return sesion["session_id"]
def invalidar(self, nombre: str):
self._sesiones.pop(nombre, None)
def limpiar_expiradas(self) -> int:
ahora = datetime.now()
antes = len(self._sesiones)
self._sesiones = {
k: v for k, v in self._sesiones.items()
if v["expira_en"] > ahora
}
return antes - len(self._sesiones)
def listar(self) -> list:
return [
{"nombre": k, "session_id": v["session_id"], "expira_en": v["expira_en"]}
for k, v in self._sesiones.items()
]
# Factory para obtener el store correcto según el entorno
def crear_store(entorno: str = "desarrollo") -> object:
"""Factory que retorna el store apropiado según el entorno."""
if entorno == "produccion":
return SesionStoreRedis()
elif entorno == "desarrollo":
return SesionStore() # SQLite
elif entorno == "test":
return SesionStoreMemoria()
else:
raise ValueError(f"Entorno desconocido: {entorno}")12. Ejemplo completo: Asistente de desarrollo persistente
Un sistema completo que recuerda análisis previos entre sesiones de trabajo.
import asyncio
import json
from pathlib import Path
from datetime import datetime
from claude_code_sdk import query, ClaudeCodeOptions
class AsistenteDesarrolloPersistente:
"""
Asistente de desarrollo completo con memoria persistente.
Recuerda: análisis previos, bugs encontrados, mejoras implementadas, arquitectura.
"""
def __init__(self, directorio: str):
self.directorio = Path(directorio)
self.nombre = self.directorio.name
self.store = SesionStore()
self.clave_sesion = f"dev_principal_{self.nombre}"
self.clave_memoria = f"memoria_{self.nombre}"
self._memoria = self._cargar_memoria()
def _cargar_memoria(self) -> dict:
"""Carga la memoria persistente del proyecto."""
archivo = self.directorio / ".claude_memory.json"
if archivo.exists():
try:
with open(archivo, encoding='utf-8') as f:
return json.load(f)
except Exception:
pass
return {
"analisis_realizados": [],
"bugs_encontrados": [],
"mejoras_implementadas": [],
"arquitectura": "",
"notas": [],
}
def _guardar_memoria(self):
"""Persiste la memoria del proyecto."""
archivo = self.directorio / ".claude_memory.json"
with open(archivo, 'w', encoding='utf-8') as f:
json.dump(self._memoria, f, ensure_ascii=False, indent=2, default=str)
def _resumen_memoria(self) -> str:
"""Genera un resumen de la memoria para el system prompt."""
partes = []
if self._memoria.get("arquitectura"):
partes.append(f"Arquitectura conocida: {self._memoria['arquitectura'][:200]}")
ultimos_analisis = self._memoria.get("analisis_realizados", [])[-3:]
if ultimos_analisis:
fechas = [a.get('fecha', 'N/A') for a in ultimos_analisis]
partes.append(f"Últimos análisis: {', '.join(fechas)}")
bugs = self._memoria.get("bugs_encontrados", [])
bugs_pendientes = [b for b in bugs if not b.get("resuelto")]
if bugs_pendientes:
descripciones = [b.get('descripcion', '')[:50] for b in bugs_pendientes[:3]]
partes.append(f"Bugs pendientes: {'; '.join(descripciones)}")
mejoras = self._memoria.get("mejoras_implementadas", [])
if mejoras:
ultimas = [m.get('descripcion', '')[:50] for m in mejoras[-3:]]
partes.append(f"Mejoras recientes: {'; '.join(ultimas)}")
return "\n".join(partes) if partes else "Sin historial previo."
async def analizar(self, aspecto: str = "general") -> str:
"""Realiza un análisis del proyecto."""
session_id = self.store.obtener(self.clave_sesion)
memoria_resumen = self._resumen_memoria()
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Glob", "Grep", "Bash"],
disallowed_tools=["Write", "Edit"],
max_turns=15,
cwd=str(self.directorio),
resume=session_id,
system_prompt=f"""Eres el asistente de desarrollo persistente del proyecto '{self.nombre}'.
Tienes memoria de sesiones anteriores:
{memoria_resumen}
Cuando encuentres nuevos bugs o mejoras, mencionarlos explícitamente con el prefijo:
- BUG: descripción del bug (archivo:linea)
- MEJORA: descripción de la mejora
- ARQUITECTURA: hallazgo arquitectural importante""",
)
resultado = ""
nuevo_session_id = session_id
async for m in query(
prompt=f"Analiza el aspecto '{aspecto}' del proyecto.",
options=opciones
):
if hasattr(m, 'session_id') and m.session_id:
nuevo_session_id = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
# Actualizar sesión
if nuevo_session_id:
self.store.guardar(self.clave_sesion, nuevo_session_id)
# Extraer y guardar hallazgos en la memoria
self._extraer_y_guardar_hallazgos(resultado, aspecto)
return resultado
def _extraer_y_guardar_hallazgos(self, resultado: str, aspecto: str):
"""Extrae hallazgos del resultado y los guarda en la memoria."""
ahora = datetime.now().isoformat()
# Registrar el análisis
self._memoria["analisis_realizados"].append({
"fecha": ahora,
"aspecto": aspecto,
"resumen": resultado[:200],
})
self._memoria["analisis_realizados"] = self._memoria["analisis_realizados"][-10:]
# Extraer bugs
lineas = resultado.split('\n')
for linea in lineas:
if linea.strip().startswith("BUG:"):
descripcion = linea.replace("BUG:", "").strip()
self._memoria["bugs_encontrados"].append({
"fecha": ahora,
"descripcion": descripcion,
"resuelto": False,
})
elif linea.strip().startswith("ARQUITECTURA:"):
info = linea.replace("ARQUITECTURA:", "").strip()
if len(info) > len(self._memoria.get("arquitectura", "")):
self._memoria["arquitectura"] = info
self._guardar_memoria()
async def implementar(self, tarea: str) -> str:
"""Implementa una mejora específica."""
session_id = self.store.obtener(self.clave_sesion)
memoria_resumen = self._resumen_memoria()
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"],
max_turns=20,
cwd=str(self.directorio),
resume=session_id,
system_prompt=f"""Eres el asistente de desarrollo del proyecto '{self.nombre}'.
Contexto histórico:
{memoria_resumen}
Implementa los cambios solicitados de forma cuidadosa y segura.
Al finalizar, confirma qué archivos modificaste y qué cambios hiciste.""",
)
resultado = ""
nuevo_session_id = session_id
async for m in query(prompt=tarea, options=opciones):
if hasattr(m, 'session_id') and m.session_id:
nuevo_session_id = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
if nuevo_session_id:
self.store.guardar(self.clave_sesion, nuevo_session_id)
# Registrar la mejora implementada
self._memoria["mejoras_implementadas"].append({
"fecha": datetime.now().isoformat(),
"descripcion": tarea[:100],
"resultado": resultado[:200],
})
self._guardar_memoria()
return resultado
# Uso del asistente persistente
async def main():
asistente = AsistenteDesarrolloPersistente("/mi/proyecto")
# Sesión 1: Análisis inicial
print("=== Análisis de seguridad ===")
analisis = await asistente.analizar("seguridad")
print(analisis[:500])
# Sesión 2 (días después): retoma el contexto
print("\n=== Implementando mejoras de seguridad ===")
resultado = await asistente.implementar(
"Implementa las mejoras de seguridad que encontramos en el análisis anterior"
)
print(resultado[:500])
asyncio.run(main())13. Ejemplo completo: Sistema de chat con historial
WebSocket-based chat server con sesiones del SDK.
import asyncio
import json
from claude_code_sdk import query, ClaudeCodeOptions
from typing import Dict
class ServidorChatSesiones:
"""
Servidor de chat simple que usa sesiones del SDK.
Cada conexión de usuario tiene su propia sesión persistente.
"""
def __init__(self):
self.sesiones_usuarios: Dict[str, str] = {} # user_id -> session_id
self.store = SesionStoreMemoria() # In-memory para este ejemplo
async def procesar_mensaje(
self,
user_id: str,
mensaje: str,
contexto_extra: str = ""
) -> str:
"""Procesa un mensaje del usuario manteniendo su sesión."""
session_id = self.store.obtener(f"chat_{user_id}")
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
allowed_tools=[], # Chat solo — sin acceso a archivos
max_turns=5,
resume=session_id,
system_prompt=f"""Eres un asistente de chat útil y conciso.
Recuerdas toda la conversación con el usuario.
{f"Contexto adicional: {contexto_extra}" if contexto_extra else ""}
Responde en máximo 3 párrafos cortos.""",
)
resultado = ""
nuevo_session_id = session_id
async for m in query(prompt=mensaje, options=opciones):
if hasattr(m, 'session_id') and m.session_id:
nuevo_session_id = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
if nuevo_session_id:
self.store.guardar(f"chat_{user_id}", nuevo_session_id, ttl_horas=2)
self.sesiones_usuarios[user_id] = nuevo_session_id
return resultado
def limpiar_sesion_usuario(self, user_id: str):
"""Limpia la sesión de un usuario (logout o reset)."""
self.store.invalidar(f"chat_{user_id}")
self.sesiones_usuarios.pop(user_id, None)
print(f"[Chat] Sesión limpiada para usuario: {user_id}")
def stats(self) -> dict:
return {
"usuarios_activos": len(self.sesiones_usuarios),
"sesiones_activas": len(self.store.listar()),
}
# Simulación de múltiples usuarios
async def simular_chat_multiusuario():
servidor = ServidorChatSesiones()
# Usuario 1 inicia conversación
res1 = await servidor.procesar_mensaje("alice", "Hola, ¿me puedes ayudar con Python?")
print(f"Alice: {res1[:100]}")
# Usuario 2 inicia conversación independiente
res2 = await servidor.procesar_mensaje("bob", "¿Qué es async/await en JavaScript?")
print(f"Bob: {res2[:100]}")
# Usuario 1 continúa (recuerda el contexto)
res3 = await servidor.procesar_mensaje(
"alice",
"¿Cómo hago una lista por comprensión con filter?"
)
print(f"Alice (cont.): {res3[:100]}")
# Usuario 2 continúa (recuerda su propio contexto, no el de Alice)
res4 = await servidor.procesar_mensaje(
"bob",
"¿Y cómo manejo errores en código async?"
)
print(f"Bob (cont.): {res4[:100]}")
print(f"\nStats: {servidor.stats()}")
asyncio.run(simular_chat_multiusuario())14. Debugging de sesiones
Inspeccionar el estado de una sesión
import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
async def inspeccionar_sesion(session_id: str) -> dict:
"""Prueba una sesión y retorna información sobre su estado."""
try:
opciones = ClaudeCodeOptions(
model="claude-haiku-4-5",
allowed_tools=[],
max_turns=2,
resume=session_id,
system_prompt="Eres un asistente de diagnóstico."
)
info = {
"session_id": session_id,
"valida": False,
"respuesta": "",
"error": None,
}
async for m in query(
prompt="Describe brevemente (máximo 2 oraciones) qué contexto recuerdas de esta sesión.",
options=opciones
):
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
info["respuesta"] = b.text
info["valida"] = True
return info
except Exception as e:
return {
"session_id": session_id,
"valida": False,
"respuesta": "",
"error": str(e),
}
async def diagnostico_store(store: SesionStore):
"""Verifica qué sesiones del store son aún válidas."""
sesiones = store.listar()
print(f"Sesiones en el store: {len(sesiones)}")
for sesion in sesiones:
info = await inspeccionar_sesion(sesion["session_id"])
estado = "VALIDA" if info["valida"] else "EXPIRADA/INVALIDA"
print(f" {sesion['nombre']}: {estado}")
if info["valida"]:
print(f" Contexto: {info['respuesta'][:80]}")
elif info["error"]:
print(f" Error: {info['error']}")
if not info["valida"]:
store.invalidar(sesion["nombre"])
print(f" -> Invalidada")
# Logging de sesiones
class SesionLogger:
"""Registra todas las interacciones con sesiones para debugging."""
def __init__(self, log_file: str = "sesiones.log"):
self.log_file = log_file
def registrar(self, evento: str, datos: dict):
import json
from datetime import datetime
linea = json.dumps({
"timestamp": datetime.now().isoformat(),
"evento": evento,
**datos
})
with open(self.log_file, 'a', encoding='utf-8') as f:
f.write(linea + '\n')
logger = SesionLogger()
async def query_con_logging(
prompt: str,
opciones: ClaudeCodeOptions,
nombre_sesion: str = "default"
) -> str:
"""Wrapper de query con logging completo de la sesión."""
import time
inicio = time.time()
logger.registrar("query_inicio", {
"sesion": nombre_sesion,
"resume": opciones.resume,
"prompt_preview": prompt[:50],
})
resultado = ""
session_id = opciones.resume
try:
async for m in query(prompt=prompt, options=opciones):
if hasattr(m, 'session_id') and m.session_id:
session_id = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
logger.registrar("query_exito", {
"sesion": nombre_sesion,
"session_id": session_id,
"duracion": round(time.time() - inicio, 2),
"resultado_chars": len(resultado),
})
except Exception as e:
logger.registrar("query_error", {
"sesion": nombre_sesion,
"error": str(e),
"duracion": round(time.time() - inicio, 2),
})
raise
return resultado15. Anti-patrones
Anti-patrón 1: Sesiones nunca limpiadas
# MAL: acumular sesiones sin límite
async def crear_sesion_y_olvidar(user_id: str, pregunta: str) -> str:
session_id = f"sesion_nueva_{time.time()}"
# Crear nueva sesión cada vez — miles de sesiones abandonadas
opciones = ClaudeCodeOptions(model="claude-haiku-4-5")
# ...
return resultado
# La sesión nunca se limpia
# BIEN: usar un store con TTL y limpiar regularmente
store = SesionStore()
async def tarea_limpieza():
"""Ejecutar periódicamente para limpiar sesiones expiradas."""
while True:
eliminadas = store.limpiar_expiradas()
if eliminadas > 0:
print(f"[Limpieza] {eliminadas} sesiones expiradas eliminadas")
await asyncio.sleep(3600) # Cada hora
asyncio.create_task(tarea_limpieza())Anti-patrón 2: Sesiones demasiado largas sin compactación manual
# MAL: sesión de 500 turnos sin control
async def sesion_interminable():
session_id = None
for i in range(500): # 500 pasos en la misma sesión
opciones = ClaudeCodeOptions(resume=session_id, max_turns=50)
# Cada turno acumula más contexto hasta sobrepasar el límite
# BIEN: dividir en sesiones lógicas con resúmenes
async def sesion_con_control(pasos: list) -> list:
PASOS_POR_SESION = 20 # Límite por sesión
resultados = []
resumen_previo = ""
for i in range(0, len(pasos), PASOS_POR_SESION):
grupo = pasos[i:i+PASOS_POR_SESION]
print(f"[Sesión] Nueva sesión para pasos {i+1}-{i+len(grupo)}")
# Iniciar nueva sesión con resumen de la anterior
opciones = ClaudeCodeOptions(
model="claude-opus-4-5",
# Sin resume — nueva sesión con contexto del resumen
system_prompt=f"Contexto anterior: {resumen_previo}" if resumen_previo else None,
)
resultado_grupo = await ejecutar_pasos_en_sesion(grupo, opciones)
resultados.extend(resultado_grupo)
# Generar resumen de esta sesión para la siguiente
if resultado_grupo:
resumen_previo = resultado_grupo[-1].get("resultado", "")[:500]
return resultadosAnti-patrón 3: Compartir sesiones entre usuarios
# MAL: múltiples usuarios comparten la misma sesión
SESION_COMPARTIDA = None
async def responder_usuario(user_id: str, pregunta: str) -> str:
global SESION_COMPARTIDA
opciones = ClaudeCodeOptions(resume=SESION_COMPARTIDA)
# ¡Todos los usuarios ven el historial de los demás! VIOLACIÓN DE PRIVACIDAD
# BIEN: una sesión por usuario
async def responder_usuario_correcto(user_id: str, pregunta: str) -> str:
store = SesionStore()
session_id = store.obtener(f"user_{user_id}")
opciones = ClaudeCodeOptions(resume=session_id)
# Cada usuario tiene su propio contexto aisladoAnti-patrón 4: No manejar sesiones expiradas
# MAL: asumir que el session_id siempre es válido
async def usar_sesion_sin_verificar(session_id: str) -> str:
opciones = ClaudeCodeOptions(resume=session_id)
async for m in query(prompt="Continúa", options=opciones):
pass # Puede fallar silenciosamente si la sesión expiró
# BIEN: manejar la posible expiración
async def usar_sesion_con_fallback(session_id: str, opciones_base: ClaudeCodeOptions) -> tuple:
try:
opciones = ClaudeCodeOptions(**{**vars(opciones_base), 'resume': session_id})
resultado = ""
nuevo_sid = session_id
async for m in query(prompt="Continúa el análisis", options=opciones):
if hasattr(m, 'session_id') and m.session_id:
nuevo_sid = m.session_id
if hasattr(m, 'content'):
for b in m.content:
if hasattr(b, 'text'):
resultado = b.text
return resultado, nuevo_sid
except Exception as e:
if "session" in str(e).lower():
print(f"[Sesión] Expirada o inválida, iniciando nueva")
return await usar_sesion_con_fallback(None, opciones_base)
raiseResumen del capítulo
Las sesiones son el mecanismo central para construir agentes con memoria y flujos de trabajo de múltiples pasos.
mindmap
root((Manejo de Sesiones))
Conceptos clave
session_id como identificador
SystemMessage con session_id
resume para reanudar
Persistencia
SQLite para local
Redis para distribuido
In-memory para tests
TTL y limpieza automática
Patrones avanzados
Fork de sesiones
Multi-tenant isolation
Sesiones conversacionales
Asistente de desarrollo
Gestión de contexto
Estimación de tokens
División en sub-sesiones
Compactación automática
Testing y debugging
Inspección de sesiones
Logging estructurado
Verificación de validez
Anti-patrones
Sesiones sin TTL
Contexto demasiado largo
Sesiones compartidas entre usuarios
No manejar sesiones expiradasPróximo capítulo: Casos de uso avanzados — implementaciones completas y funcionales para automatización real.