n8n工作流管理

为n8n平台提供全面的工作流自动化管理，包括创建、测试、执行监控和性能优化功能。

⚠️ 关键：工作流创建规则

创建n8n工作流时，务必：

1. ✅生成完整的工作流包含所有功能节点
2. ✅包含实际的HTTP请求节点用于API调用（ImageFX、Gemini、Veo、Suno等）
3. ✅添加代码节点用于数据转换和逻辑处理
4. ✅创建正确的连接在所有节点之间
5. ✅使用真实的节点类型（n8n-nodes-base.httpRequest、n8n-nodes-base.code、n8n-nodes-base.set）

切勿：

• ❌ 创建“设置说明”占位符节点
• ❌ 生成仅包含TODO注释的工作流
• ❌ 制作不完整的工作流，需要手动添加节点
• ❌ 使用纯文本节点替代实际功能

良好工作流示例：

手动触发器 → 设置配置 → HTTP请求（API调用） → 代码（解析） → 响应

不良工作流示例：

手动触发器 → 代码（“在此处添加HTTP节点，配置API...”）

始终构建完整的、功能齐全的工作流，所有必要的节点都已配置并连接好。

设置

必需的环境变量：

• N8N_API_KEY— 您的n8n API密钥（在n8n用户界面中：设置 → API）
• N8N_BASE_URL— 您的n8n实例URL

通过OpenClaw设置配置凭据：

添加到~/.config/openclaw/settings.json：

{
  "技能": {
    "n8n": {
      "环境": {
        "N8N_API_KEY": "你的API密钥在此",
        "N8N_BASE_URL": "你的n8n网址在此"
      }
    }
  }
}

或者按会话设置（请勿将密钥保存在shell配置文件中）：

export N8N_API_KEY="你的API密钥在此"
export N8N_BASE_URL="你的n8n网址在此"

验证连接：

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

安全须知：切勿将API密钥以明文形式存储在shell配置文件中（~/.bashrc、~/.zshrc）。请使用OpenClaw设置文件或安全的密钥管理器。

快速参考

工作流管理

列出工作流

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

获取工作流详情

python3 scripts/n8n_api.py get-workflow --id <工作流ID> --pretty

创建工作流程

# 从 JSON 文件创建
python3 scripts/n8n_api.py create --from-file workflow.json

激活/停用

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

测试与验证

验证工作流结构

# 验证现有工作流
python3 scripts/n8n_tester.py validate --id <workflow-id>

# 从文件验证
python3 scripts/n8n_tester.py validate --file workflow.json --pretty

# 生成验证报告
python3 scripts/n8n_tester.py report --id <workflow-id>

模拟运行测试

# 使用数据测试
python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data '{"email": "test@example.com"}'

# 使用数据文件测试
python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test-data.json

# 完整测试报告（验证 + 模拟运行）
python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test.json --report

测试套件

# 运行多个测试用例
python3 scripts/n8n_tester.py test-suite --id <workflow-id> --test-suite test-cases.json

执行监控

列出执行记录

# 近期执行记录（所有工作流）
python3 scripts/n8n_api.py list-executions --limit 10 --pretty

# 特定工作流的执行记录
python3 scripts/n8n_api.py list-executions --id <workflow-id> --limit 20 --pretty

获取执行详情

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

手动执行

# 触发工作流
python3 scripts/n8n_api.py execute --id <workflow-id>

# 使用数据执行
python3 scripts/n8n_api.py execute --id <workflow-id> --data '{"key": "value"}'

性能优化

分析性能

# 完整性能分析
python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --pretty

# 分析特定时间段
python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --days 30 --pretty

获取优化建议

# 按优先级排序的建议
python3 scripts/n8n_optimizer.py suggest --id <workflow-id> --pretty

生成优化报告

# 包含指标、瓶颈和建议的人类可读报告
python3 scripts/n8n_optimizer.py report --id <workflow-id>

获取工作流统计信息

# 执行统计
python3 scripts/n8n_api.py stats --id <workflow-id> --days 7 --pretty

Python API

基本用法

from scripts.n8n_api import N8nClient

client = N8nClient()

# 列出工作流
workflows = client.list_workflows(active=True)

# 获取工作流
workflow = client.get_workflow('workflow-id')

# 创建工作流
new_workflow = client.create_workflow({
    'name': '我的工作流',
    'nodes': [...],
    'connections': {...}
})

# 激活/停用
client.activate_workflow('workflow-id')
client.deactivate_workflow('workflow-id')

# 执行记录
executions = client.list_executions(workflow_id='workflow-id', limit=10)
execution = client.get_execution('execution-id')

# 执行工作流
result = client.execute_workflow('workflow-id', data={'key': 'value'})

验证与测试

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

client = N8nClient()
tester = WorkflowTester(client)

# 验证工作流
validation = tester.validate_workflow(workflow_id='123')
print(f"有效: {validation['valid']}")
print(f"错误: {validation['errors']}")
print(f"警告: {validation['warnings']}")

# 试运行
result = tester.dry_run(
    workflow_id='123',
    test_data={'email': 'test@example.com'}
)
print(f"状态: {result['status']}")

# 测试套件
test_cases = [
    {'name': '测试 1', 'input': {...}, 'expected': {...}},
    {'name': '测试 2', 'input': {...}, 'expected': {...}}
]
results = tester.test_suite('123', test_cases)
print(f"通过: {results['passed']}/{results['total_tests']}")

# 生成报告
report = tester.generate_test_report(validation, result)
print(report)

性能优化

from scripts.n8n_optimizer import WorkflowOptimizer

optimizer = WorkflowOptimizer()

# 分析性能
analysis = optimizer.analyze_performance('workflow-id', days=7)
print(f"性能得分: {analysis['performance_score']}/100")
print(f"健康度: {analysis['execution_metrics']['health']}")

# 获取建议
suggestions = optimizer.suggest_optimizations('workflow-id')
print(f"优先操作: {len(suggestions['priority_actions'])}")
print(f"快速优化: {len(suggestions['quick_wins'])}")

# 生成报告
report = optimizer.generate_optimization_report(analysis)
print(report)

常见工作流程

1. 验证和测试工作流

# 验证工作流结构
python3 scripts/n8n_tester.py validate --id <workflow-id> --pretty

# 使用样本数据测试
python3 scripts/n8n_tester.py dry-run --id <workflow-id> \
  --data '{"email": "test@example.com", "name": "测试用户"}'

# 如果测试通过，则激活
python3 scripts/n8n_api.py activate --id <workflow-id>

2. 调试失败的工作流

# 检查最近的执行记录
python3 scripts/n8n_api.py list-executions --id <workflow-id> --limit 10 --pretty

# 获取特定执行详情
python3 scripts/n8n_api.py get-execution --id <execution-id> --pretty

# 验证工作流结构
python3 scripts/n8n_tester.py validate --id <workflow-id>

# 生成测试报告
python3 scripts/n8n_tester.py report --id <workflow-id>

# 检查优化问题
python3 scripts/n8n_optimizer.py report --id <workflow-id>

3. 优化工作流性能

# 分析当前性能
python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --days 30 --pretty

# 获取可操作建议
python3 scripts/n8n_optimizer.py suggest --id <workflow-id> --pretty

# 生成综合报告
python3 scripts/n8n_optimizer.py report --id <workflow-id>

# 查看执行统计数据
python3 scripts/n8n_api.py stats --id <workflow-id> --days 30 --pretty

# 使用空运行测试优化
python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test-data.json

4. 监控工作流健康状态

# 检查活跃工作流
python3 scripts/n8n_api.py list-workflows --active true --pretty

# 查看近期执行状态
python3 scripts/n8n_api.py list-executions --limit 20 --pretty

# 获取每个关键工作流的统计数据
python3 scripts/n8n_api.py stats --id <workflow-id> --pretty

# 生成健康报告
python3 scripts/n8n_optimizer.py report --id <workflow-id>

验证检查

测试模块执行全面的验证：

结构验证

• ✓ 存在必填字段（节点、连接）
• ✓ 所有节点均有名称和类型
• ✓ 连接目标存在
• ✓ 无孤立节点（警告）

配置验证

• ✓ 需要凭据的节点已配置
• ✓ 必填参数已设置
• ✓ HTTP 节点具有 URL
• ✓ Webhook 节点具有路径
• ✓ 邮件节点具有内容

流程验证

• ✓ 工作流包含触发器节点
• ✓ 执行流程正确
• ✓ 无循环依赖
• ✓ 终端节点已识别

优化分析

优化器从多个维度进行分析：

执行指标

• 总执行次数
• 成功率/失败率
• 健康状况（优秀/良好/一般/差）
• 错误模式

性能指标

• 节点数量与复杂度
• 连接模式
• 高成本操作（API调用、数据库查询）
• 并行执行机会

瓶颈检测

• 连续高成本操作
• 高失败率
• 缺少错误处理
• 速率限制问题

优化机会

• 并行执行：识别可并发运行的节点
• 缓存：建议对重复的API调用进行缓存
• 批量处理：建议对大型数据集进行批量处理
• 错误处理：添加错误恢复机制
• 复杂度降低：拆分复杂工作流
• 超时设置：配置执行限制

性能评分

工作流基于以下因素获得性能评分（0-100）：

• 成功率：越高越好（权重50%）
• 复杂度：越低越好（权重30%）
• 瓶颈：越少越好（严重：-20，高：-10，中：-5）
• 优化措施：已实施最佳实践（每项+5分）

评分解读：

• 90-100分：优秀 - 优化完善
• 70-89分：良好 - 存在小幅改进空间
• 50-69分：合格 - 建议进行优化
• 0-49分：较差 - 存在显著问题

最佳实践

开发阶段

1. 方案结构：构建前先设计工作流节点与连接关系
2. 先行验证：部署前务必执行验证
3. 全面测试：采用多测试用例的试运行机制
4. 错误处理：添加错误处理节点保障可靠性
5. 文档规范：在代码节点中对复杂逻辑添加注释

测试阶段

1. 样本数据：创建符合实际的测试数据文件
2. 边界情况：测试边界条件与异常场景
3. 增量测试：每新增节点后立即测试
4. 回归测试：变更后重新测试
5. 类生产环境：使用与生产环境一致的预发布环境

部署

1. 先设为非活跃：以非活跃状态部署工作流
2. 渐进式发布：初始阶段使用有限流量测试
3. 密切监控：仔细监控首次执行
4. 快速回滚：出现问题时立即准备停用
5. 变更记录：维护修改的变更日志

优化

1. 基准指标：记录变更前的性能表现
2. 单次变更原则：隔离优化措施的影响
3. 效果评估：对比优化前后的指标
4. 定期评审：安排月度优化评审会议
5. 成本意识：监控API使用量与执行成本

维护

1. 健康检查：每周审查执行统计数据
2. 错误分析：调查故障模式
3. 性能监控：追踪执行时间
4. 凭证轮换：定期更新凭证
5. 清理：归档或删除未使用的工作流

故障排除

认证错误

错误：未在环境中找到 N8N_API_KEY

解决方案：设置环境变量：

export N8N_API_KEY="你的-api-密钥"

连接错误

错误：HTTP 401：未授权

解决方案：

1. 验证 API 密钥是否正确
2. 检查 N8N_BASE_URL 是否正确设置
3. 确认在 n8n 中已启用 API 访问

验证错误

验证失败：节点缺少 'name' 字段

解决方案：检查工作流 JSON 结构，确保所有必填字段都存在

执行超时

状态：超时 - 执行未完成

解决方案：

1. 检查工作流是否存在无限循环
2. 减少测试数据集的大小
3. 优化高开销操作
4. 在工作流设置中设置执行超时

速率限制

错误：HTTP 429：请求过多

解决方案：

1. 在 API 调用之间添加等待节点
2. 实施指数退避
3. 使用批处理
4. 检查API速率限制

缺少凭据

警告：节点 'HTTP_Request' 可能需要凭据

解决方案：

1. 在 n8n UI 中配置凭据
2. 为节点分配凭据
3. 激活前测试连接

文件结构

~/clawd/skills/n8n/
├── SKILL.md                    # 此文件
├── scripts/
│   ├── n8n_api.py             # 核心 API 客户端（扩展版）
│   ├── n8n_tester.py          # 测试与验证
│   └── n8n_optimizer.py       # 性能优化
└── references/
    └── api.md                 # n8n API 参考

API 参考

详细的 n8n REST API 文档，请参阅references/api.md或访问：https://docs.n8n.io/api/

支持

文档：

• n8n 官方文档：https://docs.n8n.io
• n8n 社区论坛：https://community.n8n.io
• n8n API 参考：https://docs.n8n.io/api/

调试：

1. 使用验证：python3 scripts/n8n_tester.py validate --id <workflow-id>
2. 检查执行日志：python3 scripts/n8n_api.py get-execution --id <execution-id>
3. 查看优化报告：python3 scripts/n8n_optimizer.py report --id <workflow-id>
4. 使用试运行测试：python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test.json