automation-workflows

automation-workflows

用途

此技能使用 IFTTT 风格的触发器（例如，基于事件或时间的触发器）自动执行工作流，通过 Webhook 链接服务以实现集成，并构建数据转换管道，以高效处理和路由数据。

适用场景

使用此技能自动执行重复性任务，例如在应用之间同步数据、响应 Web 事件（例如，API 调用或文件更改），或创建用于数据处理的自定义管道。它非常适合开发、物联网或业务自动化中的集成场景，这些场景需要串联触发器和操作。

主要功能

• 通过简单的配置定义触发器，例如使用如下格式的 JSON 文件：{"trigger": "http-post", "url": "https://example.com/hook"}。
• 串联 Webhook 以按顺序执行操作，例如在 GitHub 事件发生时触发脚本，然后向 Slack 发送消息。
• 构建数据管道以进行转换，例如使用内置函数 JSON.parse() 或自定义 JS 代码片段映射数据。
• 支持工作流中的条件逻辑，例如根据载荷数据设置 if-then 规则。
• 处理异步操作，例如使用类 cron 表达式（例如，每日运行的 "0 0 * * *"）安排任务。

使用模式

要设置工作流，首先在 YAML 配置文件中定义工作流（例如，包含 triggers、actions 和 transforms 键的 workflow.yaml）。通过 CLI 或 API 加载该文件，然后模拟触发器进行测试。持续使用时，可通过系统 cron 或此技能的内置调度器安排工作流。敏感数据始终应使用环境变量，例如在配置中使用 $WORKFLOW_API_KEY。监控执行日志以便调试，并通过将输出作为输入来串联多个工作流。

常用命令/API

大多数操作都使用 OpenClaw CLI；通过环境变量使用 $WORKFLOW_API_KEY 进行身份验证。主要命令：

• 创建工作流：claw workflow create --file workflow.yaml --key $WORKFLOW_API_KEY
• 运行工作流：claw workflow run --id 123 --trigger manual
• 列出工作流：claw workflow list --cluster community API 端点位于 https://api.openclaw.com/workflows：
• POST /workflows 用于创建：发送如下 JSON 请求正文：{"name": "my-flow", "trigger": {"type": "webhook", "url": "https://hook.example.com"}}。
• GET /workflows/{id} 用于获取详细信息。 Python API 调用代码片段：

import requests; import os
key = os.environ.get('WORKFLOW_API_KEY')
response = requests.post('https://api.openclaw.com/workflows', json={"name": "test"}, headers={'Authorization': f'Bearer {key}'})
print(response.json())

进行数据转换时，可在配置中使用内联 JS：例如，使用 {"transform": "return data.map(item => item * 2);"} 处理数组。

集成说明

通过此技能的服务端公开 Webhook，以进行集成（例如，运行 claw server start --port 8080 以监听端口）。对于外部服务，请在配置中添加身份验证信息，例如 {"action": "webhook", "url": "https://api.service.com/endpoint", "headers": {"Authorization": "Bearer $SERVICE_API_KEY"}}。载荷格式使用 JSON，例如 {"event": "push", "data": {...}}。通过引用先前的输出串联集成，例如将 GitHub Webhook 的输出传递给数据库插入操作。确保服务支持 HTTP 回调；使用 Postman 等工具进行测试。

错误处理

始终检查 API 响应状态码（例如，如果 response.status_code != 200，则记录错误）。在 CLI 命令中使用 --debug 标志以显示详细输出，例如 claw workflow run --id 123 --debug。处理密钥无效等常见错误时，先验证环境变量（例如，在脚本中：if not os.environ.get('WORKFLOW_API_KEY'): raise ValueError("Key missing")）。对于工作流，可在配置中定义错误处理器，例如 {"onError": "retry 3 times"} 或 {"onError": "notify slack"}。错误包装代码片段：

try:
    response = requests.post(url, headers=headers)
    response.raise_for_status()
except requests.exceptions.HTTPError as e:
    print(f"Error: {e} - Retrying...")

将所有故障记录到文件或服务中，以便审计。

图关系

• 相关：integration（共享 Webhook 处理功能）、automation（使用常见触发器）、community 集群中的 workflow-management。
• 依赖：用于外部触发器的 webhook-services。
• 关联：用于高级转换的 data-pipelines 技能。