← Volver al listado de tecnologías

Capítulo 6: Sistema de Hooks

Por: Artiko
claudeagent-sdkhooksvalidacion

Capítulo 6: Sistema de Hooks

¿Qué son los Hooks?

Los hooks son funciones que se ejecutan en puntos específicos del ciclo del agente. Permiten:

flowchart TD
    SS["SessionStart"] -->|Hook al iniciar sesión| UP
    UP["UserPromptSubmit"] -->|Hook al recibir prompt| PRE
    PRE["PreToolUse"] -->|Hook ANTES de ejecutar| TOOL
    TOOL["Ejecutar Tool"] --> POST
    POST["PostToolUse"] -->|Hook DESPUÉS de ejecutar| STOP
    STOP["Stop"] -->|Hook al finalizar| SE
    SE["SessionEnd"] -->|Hook al cerrar sesión| FIN((Fin))

Hooks Disponibles

HookMomentoUso común
SessionStartAl iniciarConfiguración
SessionEndAl finalizarCleanup
UserPromptSubmitAl recibir promptValidación
PreToolUseAntes de toolBloqueo/validación
PostToolUseDespués de toolLogging
StopAl terminarResumen

Estructura de un Hook

async def mi_hook(input_data, tool_use_id, context):
    # input_data: datos de entrada del tool
    # tool_use_id: ID único de esta ejecución
    # context: contexto de la sesión

    # Retorna {} para permitir la acción
    return {}

    # O retorna una decisión de permiso
    return {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",
            "permissionDecisionReason": "Razón del bloqueo"
        }
    }

Hook PreToolUse: Bloquear Comandos

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher

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

    if tool_name != "Bash":
        return {}

    command = tool_input.get("command", "")
    comandos_peligrosos = ["rm -rf", "rm -r /", "sudo rm"]

    for peligroso in comandos_peligrosos:
        if peligroso in command:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Comando bloqueado: {peligroso}"
                }
            }

    return {}

options = ClaudeAgentOptions(
    allowed_tools=["Bash"],
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[bloquear_rm])
        ]
    }
)

Hook PostToolUse: Auditoría

from datetime import datetime

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

    if tool_name in ["Write", "Edit"]:
        file_path = tool_input.get("file_path", "desconocido")
        with open("./audit.log", "a") as f:
            timestamp = datetime.now().isoformat()
            f.write(f"{timestamp}: {tool_name} -> {file_path}\n")

    return {}

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Edit"],
    permission_mode="acceptEdits",
    hooks={
        "PostToolUse": [
            HookMatcher(matcher="Write|Edit", hooks=[log_cambios])
        ]
    }
)

HookMatcher: Filtrar por Tool

# Match exacto
HookMatcher(matcher="Bash", hooks=[mi_hook])

# Match con regex
HookMatcher(matcher="Edit|Write", hooks=[mi_hook])

# Match todos los tools
HookMatcher(matcher=".*", hooks=[mi_hook])

# Múltiples hooks para el mismo matcher
HookMatcher(matcher="Bash", hooks=[validar, loggear, notificar])

Ejemplo: Validación de Rutas

import os

async def validar_ruta(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 ["Read", "Write", "Edit"]:
        return {}

    file_path = tool_input.get("file_path", "")
    ruta_permitida = "/home/usuario/proyecto"

    # Verifica que la ruta esté dentro del proyecto
    ruta_absoluta = os.path.abspath(file_path)
    if not ruta_absoluta.startswith(ruta_permitida):
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": f"Ruta fuera del proyecto: {file_path}"
            }
        }

    return {}

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Edit"],
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Read|Write|Edit", hooks=[validar_ruta])
        ]
    }
)

Ejemplo: Rate Limiting

from collections import defaultdict
from datetime import datetime, timedelta

llamadas = defaultdict(list)

async def rate_limit(input_data, tool_use_id, context):
    tool_name = input_data.get("tool_name", "")
    ahora = datetime.now()

    # Limpia llamadas antiguas (más de 1 minuto)
    llamadas[tool_name] = [
        t for t in llamadas[tool_name]
        if ahora - t < timedelta(minutes=1)
    ]

    # Máximo 10 llamadas por minuto
    if len(llamadas[tool_name]) >= 10:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": f"Rate limit excedido para {tool_name}"
            }
        }

    llamadas[tool_name].append(ahora)
    return {}

Ejemplo: Notificación por Slack

import httpx

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

    if tool_name in ["Write", "Edit"]:
        file_path = tool_input.get("file_path", "")

        async with httpx.AsyncClient() as client:
            await client.post(
                "https://hooks.slack.com/services/xxx",
                json={
                    "text": f"Agente modificó: {file_path}"
                }
            )

    return {}

Combinando Múltiples Hooks

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Edit", "Bash"],
    permission_mode="acceptEdits",
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[bloquear_rm, rate_limit]),
            HookMatcher(matcher="Read|Write|Edit", hooks=[validar_ruta])
        ],
        "PostToolUse": [
            HookMatcher(matcher="Write|Edit", hooks=[log_cambios, notificar_cambios])
        ]
    }
)

Resumen