OA0 = Omni AI 0
OA0 是一个探索 AI 的论坛
现在注册
已注册用户请  登录
OA0  ›  技能包  ›  n8n:通过 API 管理 n8n 工作流与自动化任务

n8n:通过 API 管理 n8n 工作流与自动化任务

 
  cloud ·  2026-02-07 06:29:55 · 3 次点击  · 0 条评论  

名称: n8n
描述: 通过 API 管理 n8n 工作流和自动化任务。适用于处理 n8n 工作流、执行记录或自动化任务时——例如列出工作流、激活/停用、检查执行状态、手动触发工作流或调试自动化问题。
元数据: {"openclaw":{"emoji":"\u2699\ufe0f","requires":{"env":["N8N_API_KEY","N8N_BASE_URL"]},"primaryEnv":"N8N_API_KEY"}}


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 UI 中:设置 → API)
- N8N_BASE_URL — 您的 n8n 实例 URL

通过 OpenClaw 设置配置凭据:

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

{
  "skills": {
    "n8n": {
      "env": {
        "N8N_API_KEY": "your-api-key-here",
        "N8N_BASE_URL": "your-n8n-url-here"
      }
    }
  }
}

或按会话设置(请勿将密钥保存在 shell rc 文件中):

export N8N_API_KEY="your-api-key-here"
export N8N_BASE_URL="your-n8n-url-here"

验证连接:

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 <workflow-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="your-api-key"

连接错误

错误: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/n8

3 次点击  ∙  0 人收藏  
登录后收藏  
目前尚无回复
0 条回复
About   ·   Help   ·    
OA0 - Omni AI 0 一个探索 AI 的社区
沪ICP备2024103595号-2
Developed with Cursor