跳至主要内容
使用 Asgard Automation Tool 建立 Webhook 端點的完整範例

Webhook 整合

Asgard 的 Automation Tool 讓您可以在 Workflow 中定義 Webhook 端點，接收外部系統的事件觸發，並執行對應的自動化邏輯。常見應用場景包括：

  • 接收 GitHub / GitLab 推送事件後自動通知 Slack
  • 處理電商訂單事件並回覆客戶
  • 串接第三方 CRM / ERP 系統的資料更新

Automation Tool Workflow 概念​

典型的 Webhook Workflow 由三個階段組成：

外部系統 → Webhook 端點
              ↓
    1. Validate Payload（驗證請求格式與簽名）
              ↓
    2. Process（執行業務邏輯，如查詢資料庫、呼叫 LLM）
              ↓
    3. Response（回傳結果給呼叫方）

在 Asgard 設定 Automation Tool​

步驟一：建立 Workflow​

  1. 進入 Workflow 頁面，點擊「建立新 Workflow」
  2. 在 Workflow 中加入 Automation Tool 節點
  3. 設定觸發條件為「Webhook」

步驟二：設定 Payload 驗證​

在 Automation Tool 的設定中，您可以定義：

  • Payload Schema — 驗證傳入的 JSON 結構
  • Secret Signature — 驗證 Webhook 來源（建議啟用）
  • 允許的 Content-Type — 如 application/json

步驟三：發佈 App 並取得 Webhook URL​

發佈 App 後，Asgard 會提供一個 Webhook 端點 URL，格式如下：

POST https://api.asgard-ai.com/generic/ns/{{namespace}}/bot-provider/{{bot_provider_name}}/message/sse

範例 Webhook Payload​

以下是一個典型的電商訂單事件 Webhook Payload：

{
  "customChannelId": "order-webhook-channel",
  "customMessageId": "order-event-12345",
  "text": "新訂單通知",
  "action": "NONE",
  "payload": {
    "event": "order.created",
    "orderId": "ORD-2024-12345",
    "customer": {
      "name": "王小明",
      "email": "wang@example.com"
    },
    "items": [
      {
        "productId": "PROD-001",
        "name": "無線藍牙耳機",
        "quantity": 1,
        "price": 1290
      }
    ],
    "total": 1290,
    "currency": "TWD",
    "createdAt": "2024-01-15T10:30:00Z"
  }
}

呼叫 Webhook 端點​

cURL 範例​

curl -X POST "https://api.asgard-ai.com/generic/ns/your-namespace/bot-provider/your-bot-provider/message/sse" \
  -H "Content-Type: application/json" \
  -H "X-API-KEY: your-api-key" \
  -H "X-Webhook-Signature: sha256=your-hmac-signature" \
  -d '{
    "customChannelId": "order-webhook-channel",
    "customMessageId": "order-event-12345",
    "text": "新訂單通知",
    "action": "NONE"
  }'

JavaScript（Node.js）Webhook Server 範例​

以下展示如何在 Node.js Express 應用中接收外部 Webhook，並轉送至 Asgard 處理：

const express = require('express');
const crypto = require('crypto');
const fetch = require('node-fetch');

const app = express();
app.use(express.json());

const ASGARD_API_KEY = process.env.ASGARD_API_KEY;
const ASGARD_BASE_URL = 'https://api.asgard-ai.com';
const NAMESPACE = 'your-namespace';
const BOT_PROVIDER = 'your-bot-provider';
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

/**
 * 驗證 Webhook 簽名（HMAC-SHA256）
 */
function verifySignature(payload, signature, secret) {
  const expected = `sha256=${crypto
    .createHmac('sha256', secret)
    .update(JSON.stringify(payload))
    .digest('hex')}`;
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expected)
  );
}

/**
 * 接收 GitHub Push 事件並觸發 Asgard Workflow
 */
app.post('/webhook/github', async (req, res) => {
  const signature = req.headers['x-hub-signature-256'];

  // 驗證簽名
  if (!verifySignature(req.body, signature, WEBHOOK_SECRET)) {
    return res.status(401).json({ error: '簽名驗證失敗' });
  }

  const { repository, pusher, commits } = req.body;
  const channelId = `github-${repository.name}`;
  const messageId = `push-${Date.now()}`;

  // 組成傳送給 Asgard 的訊息
  const message = `GitHub Push 事件：${pusher.name} 推送了 ${commits.length} 個 commit 到 ${repository.full_name}`;

  try {
    const asgardUrl = `${ASGARD_BASE_URL}/generic/ns/${NAMESPACE}/bot-provider/${BOT_PROVIDER}/message/sse`;

    const asgardResponse = await fetch(asgardUrl, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'X-API-KEY': ASGARD_API_KEY,
      },
      body: JSON.stringify({
        customChannelId: channelId,
        customMessageId: messageId,
        text: message,
        action: 'NONE',
      }),
    });

    if (!asgardResponse.ok) {
      throw new Error(`Asgard API error: ${asgardResponse.status}`);
    }

    // 讀取 Asgard 的串流回應
    const reader = asgardResponse.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';
    let result = '';

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n');
      buffer = lines.pop() || '';

      for (const line of lines) {
        if (!line.startsWith('data:')) continue;
        const jsonStr = line.slice(5).trim();
        if (!jsonStr) continue;

        try {
          const event = JSON.parse(jsonStr);
          if (event.eventType === 'asgard.message.complete') {
            result = event.fact.messageComplete.message.text;
          }
          if (event.eventType === 'asgard.run.done') {
            return res.json({ success: true, result });
          }
        } catch (_) {}
      }
    }

    res.json({ success: true, result });
  } catch (error) {
    console.error('轉送 Asgard 失敗：', error);
    res.status(500).json({ error: error.message });
  }
});

app.listen(3000, () => {
  console.log('Webhook Server 運行於 http://localhost:3000');
});

Python（FastAPI）Webhook Server 範例​

import hashlib
import hmac
import json
import os
import time
from fastapi import FastAPI, Request, HTTPException, Header
from typing import Optional
import requests

app = FastAPI()

ASGARD_API_KEY = os.environ.get("ASGARD_API_KEY")
ASGARD_BASE_URL = "https://api.asgard-ai.com"
NAMESPACE = "your-namespace"
BOT_PROVIDER = "your-bot-provider"
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "").encode()


def verify_signature(payload: bytes, signature: str) -> bool:
    """驗證 HMAC-SHA256 簽名"""
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET,
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)


def call_asgard(channel_id: str, message: str) -> str:
    """呼叫 Asgard API 並取得回應"""
    url = f"{ASGARD_BASE_URL}/generic/ns/{NAMESPACE}/bot-provider/{BOT_PROVIDER}/message/sse"

    headers = {
        "Content-Type": "application/json",
        "X-API-KEY": ASGARD_API_KEY,
    }

    payload = {
        "customChannelId": channel_id,
        "customMessageId": f"webhook-{int(time.time())}",
        "text": message,
        "action": "NONE",
    }

    result = ""

    with requests.post(url, headers=headers, json=payload, stream=True) as response:
        response.raise_for_status()

        for line in response.iter_lines():
            if not line:
                continue

            decoded = line.decode("utf-8")
            if not decoded.startswith("data:"):
                continue

            json_str = decoded[5:].strip()
            if not json_str:
                continue

            try:
                event = json.loads(json_str)
                if event.get("eventType") == "asgard.message.complete":
                    result = event["fact"]["messageComplete"]["message"]["text"]
                elif event.get("eventType") == "asgard.run.done":
                    break
            except json.JSONDecodeError:
                pass

    return result


@app.post("/webhook/order")
async def handle_order_webhook(
    request: Request,
    x_webhook_signature: Optional[str] = Header(None),
):
    body = await request.body()

    # 驗證簽名
    if x_webhook_signature and WEBHOOK_SECRET:
        if not verify_signature(body, x_webhook_signature):
            raise HTTPException(status_code=401, detail="簽名驗證失敗")

    data = json.loads(body)
    order_id = data.get("orderId", "unknown")
    customer_name = data.get("customer", {}).get("name", "unknown")
    event_type = data.get("event", "unknown")

    # 組成傳送給 Asgard 的訊息
    message = f"訂單事件：{event_type}，訂單編號：{order_id}，客戶：{customer_name}"
    channel_id = f"order-{order_id}"

    result = call_asgard(channel_id, message)

    return {"success": True, "orderId": order_id, "result": result}

安全性最佳實踐​

安全提醒

建立對外 Webhook 端點時，請注意以下安全事項：

  1. 啟用簽名驗證 — 使用 HMAC-SHA256 驗證 Webhook 來源，防止偽造請求
  2. 設定 IP 白名單 — 限制只接受來自已知 IP 範圍的請求
  3. 不在回應中洩漏敏感資訊 — Webhook 回應應只包含必要的狀態資訊
  4. 設定請求逾時 — 避免長時間等待造成資源佔用
  5. 記錄所有請求 — 方便事後稽核與除錯

下一步​