← Volver al listado de tecnologías
Capítulo 6: Sistema de Hooks
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:
- Validar acciones antes de ejecutarlas
- Registrar logs de auditoría
- Bloquear comandos peligrosos
- Transformar inputs/outputs
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
| Hook | Momento | Uso común |
|---|---|---|
SessionStart | Al iniciar | Configuración |
SessionEnd | Al finalizar | Cleanup |
UserPromptSubmit | Al recibir prompt | Validación |
PreToolUse | Antes de tool | Bloqueo/validación |
PostToolUse | Después de tool | Logging |
Stop | Al terminar | Resumen |
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
- Hooks interceptan el ciclo de vida del agente
PreToolUsepara validar/bloquear antes de ejecutarPostToolUsepara logging después de ejecutarHookMatcherfiltra qué tools activan el hook- Combina hooks para seguridad en capas