← Volver al listado de tecnologías

Capítulo 10: Mejores Prácticas

Por: Artiko
claudeagent-sdkmejores-practicaspatrones

Capítulo 10: Mejores Prácticas

1. Principio de Mínimo Privilegio

Otorga solo las herramientas necesarias.

# MAL: Todas las herramientas
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep", "WebSearch"]
)

# BIEN: Solo lo necesario para la tarea
# Para análisis (solo lectura)
options = ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])

# Para edición controlada
options = ClaudeAgentOptions(allowed_tools=["Read", "Edit"])

2. Prompts Específicos y Estructurados

# MAL: Prompt vago
system_prompt = "Ayuda con código"

# BIEN: Prompt estructurado
system_prompt = """Eres un revisor de código Python especializado.

TAREAS:
1. Buscar bugs potenciales
2. Identificar vulnerabilidades
3. Sugerir mejoras de rendimiento

FORMATO DE RESPUESTA:
## [SEVERIDAD] Archivo:Línea
**Problema**: Descripción clara
**Impacto**: Qué puede causar
**Solución**: Código corregido

SEVERIDADES:
- CRÍTICO: Seguridad o pérdida de datos
- ALTO: Bugs que rompen funcionalidad
- MEDIO: Código subóptimo
- BAJO: Estilo y convenciones"""

3. Validación con Hooks

Siempre valida acciones sensibles.

async def validar_escritura(input_data, tool_use_id, context):
    tool_name = input_data.get("tool_name", "")
    tool_input = input_data.get("tool_input", {})

    if tool_name not in ["Write", "Edit"]:
        return {}

    file_path = tool_input.get("file_path", "")

    # Bloquea archivos sensibles
    archivos_protegidos = [".env", "secrets", "credentials", ".git"]
    for protegido in archivos_protegidos:
        if protegido in file_path:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Archivo protegido: {file_path}"
                }
            }

    return {}

4. Logging y Auditoría

Registra todas las acciones para debugging.

import logging
from datetime import datetime

logging.basicConfig(
    filename='agent.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

async def log_acciones(input_data, tool_use_id, context):
    tool_name = input_data.get("tool_name", "")
    tool_input = input_data.get("tool_input", {})

    logging.info(f"Tool: {tool_name}, Input: {tool_input}")
    return {}

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [HookMatcher(matcher=".*", hooks=[log_acciones])]
    }
)

5. Manejo Robusto de Errores

from claude_agent_sdk import (
    query,
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError
)

async def ejecutar_seguro(prompt: str, options: ClaudeAgentOptions):
    max_reintentos = 3

    for intento in range(max_reintentos):
        try:
            async for msg in query(prompt=prompt, options=options):
                yield msg
            return

        except CLIConnectionError:
            logging.warning(f"Conexión fallida, intento {intento + 1}")
            await asyncio.sleep(2 ** intento)  # Backoff exponencial

        except ProcessError as e:
            logging.error(f"Proceso falló: {e.exit_code}")
            raise

        except CLINotFoundError:
            logging.error("Claude Code no instalado")
            raise

    raise Exception("Máximo de reintentos alcanzado")

6. Timeouts y Límites

import asyncio

async def query_con_timeout(prompt: str, timeout_segundos: int = 300):
    options = ClaudeAgentOptions(
        max_turns=10,  # Límite de turnos
        allowed_tools=["Read", "Glob"]
    )

    try:
        resultado = None
        async for msg in asyncio.wait_for(
            query(prompt=prompt, options=options).__aiter__().__anext__(),
            timeout=timeout_segundos
        ):
            if hasattr(msg, 'result'):
                resultado = msg.result
        return resultado

    except asyncio.TimeoutError:
        logging.error(f"Timeout después de {timeout_segundos}s")
        raise

7. Separación de Responsabilidades

Usa subagentes para tareas específicas.

options = ClaudeAgentOptions(
    allowed_tools=["Task"],  # Solo delegación
    agents={
        "lector": AgentDefinition(
            description="Lee y analiza código",
            tools=["Read", "Glob", "Grep"]
        ),
        "escritor": AgentDefinition(
            description="Escribe y edita código",
            tools=["Write", "Edit"]
        ),
        "ejecutor": AgentDefinition(
            description="Ejecuta comandos",
            tools=["Bash"]
        )
    }
)

8. Testing de Agentes

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_agente_no_modifica_archivos_protegidos():
    with patch('claude_agent_sdk.query') as mock_query:
        # Simula que el agente intenta editar .env
        mock_query.return_value = AsyncMock()

        resultado = await mi_agente("Edita el archivo .env")

        # Verifica que el hook bloqueó la acción
        assert "bloqueado" in resultado or resultado is None

@pytest.mark.asyncio
async def test_agente_responde_correctamente():
    resultado = await mi_agente("¿Cuántos archivos .py hay?")
    assert isinstance(resultado, str)
    assert len(resultado) > 0

9. Evaluación y Métricas

class MetricasAgente:
    def __init__(self):
        self.total_queries = 0
        self.exitosos = 0
        self.fallidos = 0
        self.tiempo_total = 0

    async def ejecutar_con_metricas(self, prompt: str):
        self.total_queries += 1
        inicio = datetime.now()

        try:
            resultado = await mi_agente(prompt)
            self.exitosos += 1
            return resultado
        except Exception:
            self.fallidos += 1
            raise
        finally:
            self.tiempo_total += (datetime.now() - inicio).seconds

    def reporte(self):
        return {
            "total": self.total_queries,
            "exitosos": self.exitosos,
            "fallidos": self.fallidos,
            "tasa_exito": self.exitosos / self.total_queries if self.total_queries > 0 else 0,
            "tiempo_promedio": self.tiempo_total / self.total_queries if self.total_queries > 0 else 0
        }

10. Patrones de Diseño

Patrón: Agente Supervisor

flowchart TD
    subgraph Supervisor["SUPERVISOR"]
        Recibe["Recibe Tarea"]
        Recibe --> Planifica["Planifica"]
        Planifica --> Delega["Delega a Workers"]
        Delega --> W1["Worker 1"]
        Delega --> W2["Worker 2"]
        Delega --> W3["Worker 3"]
        W1 --> Combina["Combina Resultados"]
        W2 --> Combina
        W3 --> Combina
    end

Patrón: Pipeline

async def pipeline(archivo: str):
    # Etapa 1: Análisis
    analisis = await agente_analisis(archivo)

    # Etapa 2: Transformación
    if analisis['necesita_cambios']:
        await agente_transformacion(archivo, analisis['cambios'])

    # Etapa 3: Validación
    validacion = await agente_validacion(archivo)

    # Etapa 4: Reporte
    return await agente_reporte(analisis, validacion)

Checklist de Producción

Recursos Adicionales