Webhook実装ガイド

概要

PigeonCloudのWebhookを使用すると、データの作成・更新・削除などのイベントをリアルタイムで外部システムに通知できます。通知設定で指定したURLに対して、POSTリクエストでイベント情報がJSON形式で送信されます。

トリガーとなるアクション

Webhookは以下のアクションをトリガーとして発火します。通知設定画面で必要なアクションを選択してください。

アクション	説明
作成（create）	レコードが新規作成されたとき
更新（update）	レコードが更新されたとき
削除（delete）	レコードが削除されたとき
コメント追加	レコードにコメントが追加されたとき
ワークフローステータス変更	ワークフローのステータスが変更されたとき

通知設定の詳細は通知設定を参照してください。

受信するデータ形式

WebhookはPOSTリクエストで以下のJSONをbodyに送信します。

{
  "notification_log_id": 123,
  "notification_text": "新規レコードが作成されました",
  "action": "create",
  "table": "dataset__2",
  "table_name": "顧客管理",
  "data_id": 456,
  "records": [
    {
      "raw_data": {
        "ID": 456,
        "項目1": "値",
        "作成者": 1,
        "作成日時": "2026-02-16 10:00:00"
      },
      "view_data": {
        "ID": "456",
        "項目1": "値",
        "作成者": "田中 太郎",
        "作成日時": "2026年02月16日 10:00"
      }
    }
  ]
}

raw_data：データベースに保存されている生の値（IDは数値、日時はISO形式）
view_data：画面表示用にフォーマットされた値（IDは文字列、日時は表示形式）
フィールド名は管理画面で設定した項目名（ラベル）が使用されます

HTTPヘッダー

署名検証が有効な場合、以下のヘッダーが付与されます。

ヘッダー	説明
`X-Webhook-Signature`	RSA-SHA256署名（Base64エンコード）
`X-Webhook-Timestamp`	送信時のUnixタイムスタンプ
`X-Webhook-Request-Id`	リクエスト固有のID（べき等性チェック用）

基本的な受信実装

最もシンプルなWebhook受信サーバーの例です（Python / Flask）。

from flask import Flask, request

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.get_json()
    action = data.get('action')
    table = data.get('table')
    records = data.get('records', [])

    if records:
        record = records[0]
        raw_data = record.get('raw_data', {})
        view_data = record.get('view_data', {})
        print(f"Action: {action}, Table: {table}")
        print(f"Raw data: {raw_data}")

    return 'OK', 200

if __name__ == '__main__':
    app.run(port=5000)

署名検証付き実装（推奨）

なりすましやデータ改ざんを防ぐため、署名検証の実装を強く推奨します。

通知設定画面の「公開鍵ダウンロード」ボタンから公開鍵をダウンロード
以下のコードで署名検証を実装

from flask import Flask, request
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.serialization import load_pem_public_key
import time
import base64

app = Flask(__name__)

# 公開鍵の読み込み
with open('webhook_public_key.pem', 'rb') as f:
    PUBLIC_KEY = load_pem_public_key(f.read())

def verify_signature(timestamp, request_id, body, signature):
    try:
        # タイムスタンプの検証（5分以内）
        if abs(int(time.time()) - int(timestamp)) > 300:
            return False
        # 署名の検証
        data = f"{timestamp}.{request_id}.{body}".encode()
        PUBLIC_KEY.verify(
            base64.b64decode(signature),
            data,
            padding.PKCS1v15(),
            hashes.SHA256()
        )
        return True
    except Exception:
        return False

@app.route('/webhook', methods=['POST'])
def webhook():
    signature = request.headers.get('X-Webhook-Signature')
    timestamp = request.headers.get('X-Webhook-Timestamp')
    request_id = request.headers.get('X-Webhook-Request-Id')

    if not all([signature, timestamp, request_id]):
        return 'Missing required headers', 400

    body = request.get_data(as_text=True)
    if not verify_signature(timestamp, request_id, body, signature):
        return 'Invalid signature', 401

    data = request.get_json()
    records = data.get('records', [])
    if records:
        print(f"Verified webhook - ID: {request_id}")
        print(f"Data: {records[0].get('view_data')}")

    return 'OK', 200

if __name__ == '__main__':
    app.run(port=5000, ssl_context='adhoc')

署名検証の仕組み

署名は以下の形式の文字列に対してRSA-SHA256で生成されます。

{timestamp}.{request_id}.{body}

検証手順：

X-Webhook-Timestampが現在時刻から5分以内であることを確認（リプレイ攻撃防止）
上記形式の文字列を構築し、公開鍵でX-Webhook-Signatureを検証
X-Webhook-Request-Idを記録し、同一IDの二重処理を防止（べき等性）

注意事項

受信URLはHTTPSを使用してください（本番環境では必須）
署名検証は任意ですが、セキュリティのため強く推奨します
CSVアップロードや編集モードで大量データを更新した場合、まとめて1回の通知として送信されます
Webhook URLは通知設定画面から複数指定可能です
受信サーバーはHTTP 200を返すようにしてください

概要