智航云 开发者平台

Appearance

事件与 Webhook 指南 ​

概述 ​

智航云 AI 中台的事件系统允许你订阅平台上发生的各种事件，当事件发生时，平台会向你指定的 URL 发送 HTTP POST 请求（Webhook）。

典型应用场景：

  • 工具执行完成后更新外部系统
  • 工作流完成后发送通知
  • 监控 AI 对话量
  • 报警事件触发应急处理

支持的事件类型 ​

事件说明触发时机
app.installed应用安装用户安装你发布的应用
app.configured应用配置用户完成应用配置
tool.executed工具执行成功工具被调用并成功返回
tool.failed工具执行失败工具调用出错
knowledge.updated知识库更新知识库文档新增或修改
graph.synced图谱同步图谱数据批量同步完成
workflow.completed工作流完成工作流执行成功
workflow.failed工作流失败工作流执行出错
alert.triggered报警触发系统检测到报警条件
ai.conversationAI 对话AI 对话完成

查询所有支持的事件类型：

bash
curl https://api.invoratec.cn/v1/events/types \
  -H "X-API-Key: claw_xxxx"

注册 Webhook ​

bash
curl -X POST https://api.invoratec.cn/v1/events/webhooks \
  -H "X-API-Key: claw_xxxx" \
  -H "Content-Type: application/json" \
  -d '{
    "event": "tool.executed",
    "url": "https://my-server.com/webhooks/tool-executed",
    "secret": "my-webhook-secret-2026",
    "filter": {
      "toolName": "erp_get_inventory"
    }
  }'
python
webhook = claw.events.subscribe(
    event="tool.executed",
    url="https://my-server.com/webhooks/tool-executed",
    secret="my-webhook-secret-2026",
    filter={"toolName": "erp_get_inventory"}
)
print(webhook["id"])
javascript
const webhook = await claw.events.subscribe({
  event: 'tool.executed',
  url: 'https://my-server.com/webhooks/tool-executed',
  secret: 'my-webhook-secret-2026',
  filter: { toolName: 'erp_get_inventory' }
})
console.log(webhook.id)

参数说明：

参数类型必填说明
eventstring是事件类型
urlstring是Webhook 接收地址（HTTPS 推荐）
secretstring否用于验证 Webhook 签名的密钥
filterobject否过滤条件，只接收匹配的事件

Webhook 请求格式 ​

当事件发生时，平台向你的 URL 发送 POST 请求：

http
POST /webhooks/tool-executed HTTP/1.1
Host: my-server.com
Content-Type: application/json
X-Webhook-Signature: sha256=abc123...
X-Webhook-Event: tool.executed
X-Webhook-Timestamp: 2026-04-12T08:00:00Z

{
  "event": "tool.executed",
  "timestamp": "2026-04-12T08:00:00.000Z",
  "data": {
    "toolName": "erp_get_inventory",
    "input": { "warehouse_id": "WH-001" },
    "result": { "items": [...] },
    "elapsed_ms": 230
  }
}

验证 Webhook 签名 ​

如果注册时设置了 secret，每个 Webhook 请求的 X-Webhook-Signature 头包含 HMAC-SHA256 签名。

python
import hmac
import hashlib

def verify_webhook(payload_body, signature_header, secret):
    expected = hmac.new(
        secret.encode('utf-8'),
        payload_body,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature_header)

# 在 Flask 中使用
from flask import Flask, request
app = Flask(__name__)

@app.route('/webhooks/tool-executed', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Webhook-Signature', '')
    if not verify_webhook(request.data, signature, 'my-webhook-secret-2026'):
        return 'Invalid signature', 401

    data = request.json
    print(f"工具 {data['data']['toolName']} 执行完成")
    return 'OK', 200
javascript
const crypto = require('crypto')
const express = require('express')
const app = express()

app.use(express.json({ verify: (req, res, buf) => { req.rawBody = buf } }))

function verifyWebhook(rawBody, signatureHeader, secret) {
  const expected = crypto
    .createHmac('sha256', secret)
    .update(rawBody)
    .digest('hex')
  return signatureHeader === `sha256=${expected}`
}

app.post('/webhooks/tool-executed', (req, res) => {
  const signature = req.headers['x-webhook-signature'] || ''
  if (!verifyWebhook(req.rawBody, signature, 'my-webhook-secret-2026')) {
    return res.status(401).send('Invalid signature')
  }

  console.log(`工具 ${req.body.data.toolName} 执行完成`)
  res.status(200).send('OK')
})

管理 Webhook ​

列出所有 Webhook ​

bash
curl https://api.invoratec.cn/v1/events/webhooks \
  -H "X-API-Key: claw_xxxx"

响应包含所有 Webhook 及其状态：

json
{
  "webhooks": [
    {
      "_id": "665f...",
      "event": "tool.executed",
      "url": "https://my-server.com/webhooks/tool-executed",
      "enabled": true,
      "lastTriggered": "2026-04-12T08:00:00Z",
      "failCount": 0
    }
  ],
  "eventTypes": ["app.installed", "tool.executed", ...]
}

删除 Webhook ​

bash
curl -X DELETE https://api.invoratec.cn/v1/events/webhooks/WEBHOOK_ID \
  -H "X-API-Key: claw_xxxx"

查看事件日志 ​

查看 API 调用日志，用于调试和监控：

bash
curl "https://api.invoratec.cn/v1/events/logs?limit=50" \
  -H "X-API-Key: claw_xxxx"

实战示例：工作流完成通知到企业微信 ​

python
# 1. 注册 Webhook
claw.events.subscribe(
    event="workflow.completed",
    url="https://my-server.com/webhooks/workflow-done"
)

# 2. 在服务器端处理 Webhook
# server.py
from flask import Flask, request
import requests

app = Flask(__name__)

WECOM_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"

@app.route('/webhooks/workflow-done', methods=['POST'])
def handle_workflow_done():
    data = request.json
    workflow_name = data['data'].get('workflowName', '未知')
    status = data['data'].get('status', 'unknown')

    # 发送到企业微信
    requests.post(WECOM_WEBHOOK, json={
        "msgtype": "markdown",
        "markdown": {
            "content": f"**工作流执行完成**\n"
                       f">工作流: {workflow_name}\n"
                       f">状态: {status}\n"
                       f">时间: {data['timestamp']}"
        }
    })
    return 'OK', 200

失败重试 ​

当 Webhook 投递失败（目标服务器无响应或返回非 2xx 状态码）时：

  • 系统会记录 failCount 字段
  • 连续失败次数过多时，Webhook 可能被自动禁用
  • 建议在服务端返回 200 状态码确认收到

下一步 ​

智航云 AI 中台 — 开发者平台

© 2026 Invoratec. All rights reserved.