Gestión de Flujos de Trabajo n8n

¿Qué hace Gestión de Flujos de Trabajo n8n?

Gestión completa de automatizaciones de flujos de trabajo para la plataforma n8n, con capacidades de creación, pruebas, monitoreo de ejecuciones y optimización de rendimiento.

⚠️ CRÍTICO: Reglas de Creación de Flujos de Trabajo

Al crear flujos de trabajo n8n, SIEMPRE:

1. ✅ Genera flujos COMPLETOS con todos los nodos funcionales
2. ✅ Incluye nodos HTTP Request reales para llamadas a API (ImageFX, Gemini, Veo, Suno, etc.)
3. ✅ Agrega nodos Code para transformación de datos y lógica
4. ✅ Crea conexiones adecuadas entre todos los nodos
5. ✅ Usa tipos de nodo reales (n8n-nodes-base.httpRequest, n8n-nodes-base.code, n8n-nodes-base.set)

NUNCA:

• ❌ Crear nodos "Placeholder de Instrucciones de Configuración"
• ❌ Generar flujos con solo comentarios TODO
• ❌ Hacer flujos incompletos que requieran agregar nodos manualmente
• ❌ Usar nodos solo de texto como sustitutos de funcionalidad real

Ejemplo de flujo BUENO:

Disparador Manual → Configurar Config → HTTP Request (llamada API) → Code (analizar) → Respuesta

Ejemplo de flujo MALO:

Disparador Manual → Code ("Agregar nodos HTTP aquí, configurar APIs...")

Construye siempre el flujo de trabajo completo y funcional con todos los nodos necesarios configurados y conectados.

Configuración

Variables de entorno requeridas:

• N8N_API_KEY — Tu clave de API de n8n (Configuración → API en la interfaz de n8n)
• N8N_BASE_URL — La URL de tu instancia de n8n

Configura las credenciales a través de la configuración de OpenClaw:

Agrega a ~/.config/openclaw/settings.json:

{
  "skills": {
    "n8n": {
      "env": {
        "N8N_API_KEY": "tu-clave-api-aqui",
        "N8N_BASE_URL": "tu-url-n8n-aqui"
      }
    }
  }
}

O configúralo por sesión (no persistas secretos en archivos rc del shell):

export N8N_API_KEY="tu-clave-api-aqui"
export N8N_BASE_URL="tu-url-n8n-aqui"

Verifica la conexión:

python3 scripts/n8n_api.py list-workflows --pretty

Nota de seguridad: Nunca almacenes claves de API en texto plano en archivos de configuración del shell (~/.bashrc, ~/.zshrc). Usa el archivo de configuración de OpenClaw o un gestor de secretos seguro.

Referencia Rápida

Gestión de Flujos de Trabajo

Listar Flujos de Trabajo

python3 scripts/n8n_api.py list-workflows --pretty
python3 scripts/n8n_api.py list-workflows --active true --pretty

Obtener Detalles del Flujo de Trabajo

python3 scripts/n8n_api.py get-workflow --id <id-flujo> --pretty

Crear Flujos de Trabajo

# Desde archivo JSON
python3 scripts/n8n_api.py create --from-file workflow.json

Activar/Desactivar

python3 scripts/n8n_api.py activate --id <id-flujo>
python3 scripts/n8n_api.py deactivate --id <id-flujo>

Pruebas y Validación

Validar Estructura del Flujo

# Validar flujo existente
python3 scripts/n8n_tester.py validate --id <id-flujo>

# Validar desde archivo
python3 scripts/n8n_tester.py validate --file workflow.json --pretty

# Generar informe de validación
python3 scripts/n8n_tester.py report --id <id-flujo>

Pruebas de Ejecución en Seco (Dry Run)

# Probar con datos
python3 scripts/n8n_tester.py dry-run --id <id-flujo> --data '{"email": "[email protected]"}'

# Probar con archivo de datos
python3 scripts/n8n_tester.py dry-run --id <id-flujo> --data-file test-data.json

# Informe de prueba completo (validación + ejecución en seco)
python3 scripts/n8n_tester.py dry-run --id <id-flujo> --data-file test.json --report

Suite de Pruebas

# Ejecutar múltiples casos de prueba
python3 scripts/n8n_tester.py test-suite --id <id-flujo> --test-suite test-cases.json

Monitoreo de Ejecuciones

Listar Ejecuciones

# Ejecuciones recientes (todos los flujos)
python3 scripts/n8n_api.py list-executions --limit 10 --pretty

# Ejecuciones de un flujo específico
python3 scripts/n8n_api.py list-executions --id <id-flujo> --limit 20 --pretty

Obtener Detalles de Ejecución

python3 scripts/n8n_api.py get-execution --id <id-ejecucion> --pretty

Ejecución Manual

# Activar flujo
python3 scripts/n8n_api.py execute --id <id-flujo>

# Ejecutar con datos
python3 scripts/n8n_api.py execute --id <id-flujo> --data '{"key": "value"}'

Optimización de Rendimiento

Analizar Rendimiento

# Análisis completo de rendimiento
python3 scripts/n8n_optimizer.py analyze --id <id-flujo> --pretty

# Analizar período específico
python3 scripts/n8n_optimizer.py analyze --id <id-flujo> --days 30 --pretty

Obtener Sugerencias de Optimización

# Sugerencias ordenadas por prioridad
python3 scripts/n8n_optimizer.py suggest --id <id-flujo> --pretty

Generar Informe de Optimización

# Informe legible con métricas, cuellos de botella y sugerencias
python3 scripts/n8n_optimizer.py report --id <id-flujo>

Obtener Estadísticas del Flujo

# Estadísticas de ejecución
python3 scripts/n8n_api.py stats --id <id-flujo> --days 7 --pretty

API de Python

Uso Básico

from scripts.n8n_api import N8nClient

client = N8nClient()

# Listar flujos de trabajo
workflows = client.list_workflows(active=True)

# Obtener flujo de trabajo
workflow = client.get_workflow('id-flujo')

# Crear flujo de trabajo
new_workflow = client.create_workflow({
    'name': 'Mi Flujo',
    'nodes': [...],
    'connections': {...}
})

# Activar/desactivar
client.activate_workflow('id-flujo')
client.deactivate_workflow('id-flujo')

# Ejecuciones
executions = client.list_executions(workflow_id='id-flujo', limit=10)
execution = client.get_execution('id-ejecucion')

# Ejecutar flujo de trabajo
result = client.execute_workflow('id-flujo', data={'key': 'value'})

Validación y Pruebas

from scripts.n8n_api import N8nClient
from scripts.n8n_tester import WorkflowTester

client = N8nClient()
tester = WorkflowTester(client)

# Validar flujo de trabajo
validation = tester.validate_workflow(workflow_id='123')
print(f"Válido: {validation['valid']}")
print(f"Errores: {validation['errors']}")
print(f"Advertencias: {validation['warnings']}")

# Ejecución en seco
result = tester.dry_run(
    workflow_id='123',
    test_data={'email': '[email protected]'}
)
print(f"Estado: {result['status']}")

# Suite de pruebas
test_cases = [
    {'name': 'Prueba 1', 'input': {...}, 'expected': {...}},
    {'name': 'Prueba 2', 'input': {...}, 'expected': {...}}
]
results = tester.test_suite('123', test_cases)
print(f"Aprobadas: {results['passed']}/{results['total_tests']}")

# Generar informe
report = tester.generate_test_report(validation, result)
print(report)

Optimización de Rendimiento

from scripts.n8n_optimizer import WorkflowOptimizer

optimizer = WorkflowOptimizer()

# Analizar rendimiento
analysis = optimizer.analyze_performance('id-flujo', days=7)
print(f"Puntuación de Rendimiento: {analysis['performance_score']}/100")
print(f"Salud: {analysis['execution_metrics']['health']}")

# Obtener sugerencias
suggestions = optimizer.suggest_optimizations('id-flujo')
print(f"Acciones Prioritarias: {len(suggestions['priority_actions'])}")
print(f"Victorias Rápidas: {len(suggestions['quick_wins'])}")

# Generar informe
report = optimizer.generate_optimization_report(analysis)
print(report)

Flujos de Trabajo Comunes

1. Validar y Probar Flujo

# Validar estructura del flujo
python3 scripts/n8n_tester.py validate --id <id-flujo> --pretty

# Probar con datos de muestra
python3 scripts/n8n_tester.py dry-run --id <id-flujo> \
  --data '{"email": "[email protected]", "name": "Usuario de Prueba"}'

# Si las pruebas pasan, activar
python3 scripts/n8n_api.py activate --id <id-flujo>

2. Depurar Flujo Fallido

# Verificar ejecuciones recientes
python3 scripts/n8n_api.py list-executions --id <id-flujo> --limit 10 --pretty

# Obtener detalles de ejecución específica
python3 scripts/n8n_api.py get-execution --id <id-ejecucion> --pretty

# Validar estructura del flujo
python3 scripts/n8n_tester.py validate --id <id-flujo>

# Generar informe de prueba
python3 scripts/n8n_tester.py report --id <id-flujo>

# Verificar problemas de optimización
python3 scripts/n8n_optimizer.py report --id <id-flujo>

3. Optimizar Rendimiento del Flujo

# Analizar rendimiento actual
python3 scripts/n8n_optimizer.py analyze --id <id-flujo> --days 30 --pretty

# Obtener sugerencias accionables
python3 scripts/n8n_optimizer.py suggest --id <id-flujo> --pretty

# Generar informe completo
python3 scripts/n8n_optimizer.py report --id <id-flujo>

# Revisar estadísticas de ejecución
python3 scripts/n8n_api.py stats --id <id-flujo> --days 30 --pretty

# Probar optimizaciones con ejecución en seco
python3 scripts/n8n_tester.py dry-run --id <id-flujo> --data-file test-data.json

4. Monitorear Salud del Flujo

# Verificar flujos activos
python3 scripts/n8n_api.py list-workflows --active true --pretty

# Revisar estado de ejecuciones recientes
python3 scripts/n8n_api.py list-executions --limit 20 --pretty

# Obtener estadísticas de cada flujo crítico
python3 scripts/n8n_api.py stats --id <id-flujo> --pretty

# Generar informes de salud
python3 scripts/n8n_optimizer.py report --id <id-flujo>

Verificaciones de Validación

El módulo de pruebas realiza una validación exhaustiva:

Validación de Estructura

• ✓ Campos requeridos presentes (nodos, conexiones)
• ✓ Todos los nodos tienen nombre y tipo
• ✓ Los destinos de conexión existen
• ✓ Sin nodos desconectados (advertencia)

Validación de Configuración

• ✓ Nodos que requieren credenciales están configurados
• ✓ Parámetros requeridos están establecidos
• ✓ Nodos HTTP tienen URL
• ✓ Nodos Webhook tienen rutas
• ✓ Nodos de correo electrónico tienen contenido

Validación de Flujo

• ✓ El flujo tiene nodos de disparo
• ✓ Flujo de ejecución adecuado
• ✓ Sin dependencias circulares
• ✓ Nodos finales identificados

Análisis de Optimización

El optimizador analiza múltiples dimensiones:

Métricas de Ejecución

• Total de ejecuciones
• Tasas de éxito/fracaso
• Estado de salud (excelente/bueno/regular/malo)
• Patrones de error

Métricas de Rendimiento

• Conteo de nodos y complejidad
• Patrones de conexión
• Operaciones costosas (llamadas API, consultas a bases de datos)
• Oportunidades de ejecución en paralelo

Detección de Cuellos de Botella

• Operaciones costosas secuenciales
• Altas tasas de fracaso
• Falta de manejo de errores
• Problemas de límites de tasa

Oportunidades de Optimización

• Ejecución en Paralelo: Identificar nodos que pueden ejecutarse concurrentemente
• Caché: Sugerir almacenamiento en caché para llamadas API repetidas
• Procesamiento por Lotes: Recomendar lotes para grandes conjuntos de datos
• Manejo de Errores: Agregar mecanismos de recuperación de errores
• Reducción de Complejidad: Dividir flujos complejos
• Configuración de Tiempos de Espera: Configurar límites de ejecución

Puntuación de Rendimiento

Los flujos reciben una puntuación de rendimiento (0-100) basada en:

• Tasa de Éxito: Más alta es mejor (50% de peso)
• Complejidad: Más baja es mejor (30% de peso)
• Cuellos de Botella: Menos es mejor (crítico: -20, alto: -10, medio: -5)
• Optimizaciones: Mejores prácticas implementadas (+5 cada una)

Interpretación de la puntuación:

• 90-100: Excelente - Bien optimizado
• 70-89: Bueno - Mejoras menores posibles
• 50-69: Regular - Optimización recomendada
• 0-49: Pobre - Problemas significativos

Mejores Prácticas

Desarrollo

1. Planificar Estructura: Diseñar nodos y conexiones del flujo antes de construir
2. Validar Primero: Siempre validar antes del despliegue
3. Probar Exhaustivamente: Usar ejecución en seco con múltiples casos de prueba
4. Manejo de Errores: Agregar nodos de error para fiabilidad
5. Documentación: Comentar lógica compleja en nodos Code

Pruebas

1. Datos de Muestra: Crear archivos de datos de prueba realistas
2. Casos Límite: Probar condiciones límite y errores
3. Incremental: Probar cada adición de nodo
4. Regresión: Volver a probar después de cambios
5. Similar a Producción: Usar un entorno de pruebas que refleje producción

Despliegue

1. Inactivo Primero: Desplegar flujos en estado inactivo
2. Despliegue Gradual: Probar con tráfico limitado inicialmente
3. Monitorear de Cerca: Observar las primeras ejecuciones cuidadosamente
4. Retroceso Rápido: Estar listo para desactivar si surgen problemas
5. Documentar Cambios: Mantener un registro de cambios de las modificaciones

Optimización

1. Métricas Base: Capturar rendimiento antes de cambios
2. Un Cambio a la Vez: Aislar impactos de optimización
3. Medir Resultados: Comparar métricas antes/después
4. Revisiones Regulares: Programar revisiones de optimización mensuales
5. Conciencia de Costos: Monitorear uso de API y costos de ejecución

Mantenimiento

1. Verificaciones de Salud: Revisión semanal de estadísticas de ejecución
2. Análisis de Errores: Investigar patrones de fallos
3. Monitoreo de Rendimiento: Rastrear tiempos de ejecución
4. Rotación de Credenciales: Actualizar credenciales regularmente
5. Limpieza: Archivar o eliminar flujos no utilizados

Solución de Problemas

Error de Autenticación

Error: N8N_API_KEY not found in environment

Solución: Establecer la variable de entorno:

export N8N_API_KEY="tu-clave-api"

Error de Conexión

Error: HTTP 401: Unauthorized

Solución:

1. Verificar que la clave API sea correcta
2. Comprobar que N8N_BASE_URL esté configurada correctamente
3. Confirmar que el acceso a la API esté habilitado en n8n

Errores de Validación

Validation failed: Node missing 'name' field

Solución: Revisar la estructura JSON del flujo, asegurarse de que todos los campos requeridos estén presentes

Tiempo de Espera de Ejecución

Status: timeout - Execution did not complete

Solución:

1. Revisar si hay bucles infinitos en el flujo
2. Reducir el tamaño del conjunto de datos para pruebas
3. Optimizar operaciones costosas
4. Establecer tiempo de espera de ejecución en la configuración del flujo

Límite de Tasa

Error: HTTP 429: Too Many Requests

Solución:

1. Agregar nodos Wait entre llamadas API
2. Implementar retroceso exponencial
3. Usar procesamiento por lotes
4. Verificar límites de tasa de la API

Credenciales Faltantes

Warning: Node 'HTTP_Request' may require credentials

Solución:

1. Configurar credenciales en la interfaz de n8n
2. Asignar credenciales al nodo
3. Probar conexión antes de activar

Estructura de Archivos

~/clawd/skills/n8n/
├── SKILL.md                    # Este archivo
├── scripts/
│   ├── n8n_api.py             # Cliente API central (extendido)
│   ├── n8n_tester.py          # Pruebas y validación
│   └── n8n_optimizer.py       # Optimización de rendimiento
└── references/
    └── api.md                 # Referencia de API n8n

Referencia de API

Para documentación detallada de la API REST de n8n, consulta references/api.md o visita: https://docs.n8n.io/api/