Incoming Webhooks (ワークフロートピック & パブリッシュ API)

外部システムからCRISISにイベントを送信し、ワークフローを通じてインシデント作成・プッシュ通知・メール・Webhook配信をマネージドで実行する仕組みです。

概要

CRISISのワークフローエンジンは、イベント駆動のオートメーション基盤です。お客様側のシステムからCRISISの Topic Publish API にイベントを送信するだけで、以下のすべてをCRISISがマネージドで処理します:

• インシデント自動作成 — オペレーションブック起動による対応フロー開始
• プッシュ通知 — CRISISモバイルアプリへの即座の通知
• メール送信 — 担当グループへのメール配信
• Outgoing Webhook — 他の外部システムへの連鎖通知
• 条件フィルタリング — 閾値判定やエリア判定による不要通知の抑制
• 重複制御 — 制御ポリシーによるイベント重複排除・ピークホールド
• 拠点マッチング — 組織の登録拠点に基づく通知先の自動判定
• リトライ・配送保証 — 失敗時の指数バックオフリトライ

ユースケース

トピックの作成

CRISISコンソールから作成

1. 組織 > ワークフロー > トピック を開く
2. 「+ 新規作成」をクリック
3. 以下を入力:
   • Key: トピック識別子（例: custom.river_level）
   • Name: 表示名
   • Access Scope: private（自組織のみ）
   • Fields: フィルター条件に使うデータフィールドの定義

API経由で作成

curl -X POST https://api.crisis.jp/v1/organizations/{organizationId}/workflow_topics \
  -H "Authorization: Bearer {token}" \
  -H "Content-Type: application/json" \
  -d '{
    "key": "custom.river_level",
    "name_ja": "河川水位センサー",
    "name_en": "River Level Sensor",
    "description_ja": "IoT水位センサーからの観測データ",
    "description_en": "Observation data from IoT water level sensors",
    "access_scope": "private",
    "share_with_partner_orgs": false,
    "share_with_license_subscribers": false,
    "share_with_allowed_orgs": false,
    "share_with_sibling_orgs": false,
    "fields": [
      {
        "id": "water_level",
        "name_ja": "水位",
        "name_en": "Water Level",
        "description_ja": "観測水位（m）",
        "description_en": "Observed water level (meters)",
        "path": "level",
        "values": [{
          "value_source": "value",
          "type": "number",
          "operators": [">=", "<=", "=", "BETWEEN"]
        }]
      },
      {
        "id": "station_id",
        "name_ja": "観測所",
        "name_en": "Station",
        "description_ja": "観測所ID",
        "description_en": "Observation station ID",
        "path": "station_id",
        "values": [{
          "value_source": "value",
          "type": "string",
          "operators": ["=", "IN"]
        }]
      }
    ],
    "template_variables": [],
    "status": "ACTIVE"
  }'

トピック設定の説明

Publish API

エンドポイント

POST /v1/organizations/{organizationId}/workflow_topics/{topicId}/publish

Permission: organization.workflow_topic:publish
Response: 202 Accepted

最小リクエスト

curl -X POST https://api.crisis.jp/v1/organizations/{organizationId}/workflow_topics/{topicId}/publish \
  -H "Authorization: Bearer {token}" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "level": 3.5,
      "station_id": "ST001",
      "station_name": "多摩川水位観測所"
    },
    "status": "ACTUAL"
  }'

完全なリクエスト

curl -X POST https://api.crisis.jp/v1/organizations/{organizationId}/workflow_topics/{topicId}/publish \
  -H "Authorization: Bearer {token}" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "level": 3.5,
      "station_id": "ST001",
      "station_name": "多摩川水位観測所"
    },
    "status": "ACTUAL",
    "event": {
      "id": "evt_2026050109153000_ST001",
      "lang": "ja",
      "name": "水位超過警報",
      "description": "多摩川水位観測所で水位3.5mを記録"
    },
    "attributes": {
      "idempotency_key": "evt_2026050109153000_ST001"
    },
    "notification": {
      "lang": "ja",
      "data": {},
      "alert": {
        "options": {
          "notification": {
            "title": "水位超過警報",
            "body": "多摩川水位観測所で水位3.5mを記録しました",
            "sound": {
              "critical": 1,
              "name": "alert.aiff",
              "volume": 1.0
            }
          },
          "thread_id": "river_level_ST001",
          "collapse_id": "river_level_ST001",
          "time_to_live": 3600
        }
      }
    },
    "event_control": {
      "partition_key": "ST001",
      "metrics": {
        "level": 3.5
      }
    }
  }'

リクエストフィールド

Node.js での実装例

const axios = require('axios');

const CRISIS_API = 'https://api.crisis.jp/v1';
const ORG_ID = process.env.CRISIS_ORG_ID;
const TOPIC_ID = process.env.CRISIS_TOPIC_ID;
const TOKEN = process.env.CRISIS_API_TOKEN;

async function publishEvent(level, stationId, stationName) {
  const response = await axios.post(
    `${CRISIS_API}/organizations/${ORG_ID}/workflow_topics/${TOPIC_ID}/publish`,
    {
      data: { level, station_id: stationId, station_name: stationName },
      status: 'ACTUAL',
      event: {
        id: `${stationId}_${Date.now()}`,
        lang: 'ja',
        name: `水位警報: ${stationName}`,
      },
      notification: {
        lang: 'ja',
        data: {},
        alert: {
          options: {
            notification: {
              title: `水位警報: ${stationName}`,
              body: `${stationName}で水位${level}mを記録しました`,
              sound: { critical: 1, name: 'alert.aiff', volume: 1.0 },
            },
            thread_id: `river_level_${stationId}`,
            collapse_id: `river_level_${stationId}`,
            time_to_live: 3600,
          },
        },
      },
      event_control: {
        partition_key: stationId,
        metrics: { level },
      },
    },
    { headers: { Authorization: `Bearer ${TOKEN}` } }
  );
  console.log('Published:', response.status);
}

// センサーから水位を受信したとき
publishEvent(3.5, 'ST001', '多摩川水位観測所');

Go での実装例

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
    "os"
    "time"
)

type PublishRequest struct {
    Data         map[string]any `json:"data"`
    Status       string         `json:"status"`
    Event        *Event         `json:"event,omitempty"`
    Notification *Notification  `json:"notification,omitempty"`
    EventControl *EventControl  `json:"event_control,omitempty"`
}

type Event struct {
    ID   string `json:"id"`
    Lang string `json:"lang"`
    Name string `json:"name"`
}

type NotificationContent struct {
    Title string `json:"title"`
    Body  string `json:"body"`
    Sound any    `json:"sound,omitempty"`
}

type AlertOptions struct {
    Notification *NotificationContent `json:"notification,omitempty"`
    ThreadID     string               `json:"thread_id,omitempty"`
    CollapseID   string               `json:"collapse_id,omitempty"`
    TimeToLive   int                  `json:"time_to_live,omitempty"`
}

type Alert struct {
    Options AlertOptions `json:"options"`
}

type Notification struct {
    Alert *Alert         `json:"alert,omitempty"`
    Data  map[string]any `json:"data"`
    Lang  string         `json:"lang"`
}

type EventControl struct {
    PartitionKey string             `json:"partition_key"`
    Metrics      map[string]float64 `json:"metrics"`
}

func publish(level float64, stationID, stationName string) error {
    req := PublishRequest{
        Data:   map[string]any{"level": level, "station_id": stationID, "station_name": stationName},
        Status: "ACTUAL",
        Event:  &Event{ID: fmt.Sprintf("%s_%d", stationID, time.Now().Unix()), Lang: "ja", Name: "水位警報: " + stationName},
        Notification: &Notification{
            Lang: "ja",
            Data: map[string]any{},
            Alert: &Alert{Options: AlertOptions{
                Notification: &NotificationContent{
                    Title: "水位警報: " + stationName,
                    Body:  fmt.Sprintf("%sで水位%.1fmを記録しました", stationName, level),
                    Sound: map[string]any{"critical": 1, "name": "alert.aiff", "volume": 1.0},
                },
                ThreadID:   "river_level_" + stationID,
                CollapseID: "river_level_" + stationID,
                TimeToLive: 3600,
            }},
        },
        EventControl: &EventControl{
            PartitionKey: stationID,
            Metrics:      map[string]float64{"level": level},
        },
    }

    body, _ := json.Marshal(req)
    url := fmt.Sprintf("%s/organizations/%s/workflow_topics/%s/publish",
        os.Getenv("CRISIS_API"), os.Getenv("CRISIS_ORG_ID"), os.Getenv("CRISIS_TOPIC_ID"))

    httpReq, _ := http.NewRequest("POST", url, bytes.NewBuffer(body))
    httpReq.Header.Set("Content-Type", "application/json")
    httpReq.Header.Set("Authorization", "Bearer "+os.Getenv("CRISIS_API_TOKEN"))

    resp, err := http.DefaultClient.Do(httpReq)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    fmt.Printf("Published: %d\n", resp.StatusCode)
    return nil
}

notification（通知設定）

notification フィールドを指定すると、ワークフローのプッシュ通知アクションで使用される通知内容（タイトル、本文、サウンド等）をPublish側から制御できます。

構造

{
  "notification": {
    "lang": "ja",
    "data": {},
    "alert": {
      "options": {
        "notification": {
          "title": "水位超過警報",
          "body": "多摩川水位観測所で水位3.5mを記録しました",
          "sound": {
            "critical": 1,
            "name": "alert.aiff",
            "volume": 1.0
          }
        },
        "thread_id": "river_level_ST001",
        "collapse_id": "river_level_ST001",
        "time_to_live": 3600
      }
    }
  }
}

フィールド詳細

alert.options

sound フィールド

sound は文字列またはオブジェクトで指定できます:

オブジェクト形式のフィールド:

ワークフロー条件（フィルター）

Publish APIで送信した data は、ワークフローの条件設定（フィルター）で評価されます。

仕組み

1. トピックの fields でデータ構造を宣言
2. CRISISコンソールのワークフローエディタで条件を設定
3. Publish時に data がフィルター条件と照合され、マッチしたワークフローのみ実行

例: 水位3.0m以上のときだけ通知

トピック fields で path: "level" を定義済みの場合:

条件: level >= 3.0

• Publish: {"data": {"level": 2.5}, "status": "ACTUAL"} → 条件不一致 → ワークフロー実行されない
• Publish: {"data": {"level": 3.5}, "status": "ACTUAL"} → 条件一致 → ワークフロー実行

サポートされる演算子

イベント制御ポリシー（Control Policy）

連続的にイベントが発生するセンサーデータなどで、不要な重複通知を防ぐための仕組みです。

仕組み

水位センサーが1分ごとにデータを送信する場合:

• 水位 3.5m → 通知
• 水位 3.6m → また通知？ ← これを抑制したい
• 水位 4.0m → ピーク更新 → 通知すべき
• 水位 2.0m → 解除通知

制御ポリシーは前回の通知状態を記憶し、本当に通知すべきタイミングだけワークフローを実行します。

ポリシータイプ

METRIC タイプの詳細

{
  "type": "METRIC",
  "rules": [
    {
      "metric_name": "level",
      "mode": "ABSOLUTE",
      "threshold": 3.0
    }
  ],
  "peak_hold_duration": 30,
  "exit_stable_limit": 3,
  "auto_resolve_duration": 120,
  "status": "ACTIVE"
}

METRIC動作フロー

値 3.5 (≥ 3.0, 新規) → NOTIFY ✅ (ピーク: 3.5)
値 3.6 (≥ 3.0, ピーク更新) → NOTIFY ✅ (ピーク: 3.6)
値 3.4 (≥ 3.0, ピーク未更新) → SUPPRESS ❌
値 4.0 (≥ 3.0, ピーク更新) → NOTIFY ✅ (ピーク: 4.0)
値 2.5 (< 3.0, 安定1回目) → SUPPRESS ❌
値 2.3 (< 3.0, 安定2回目) → SUPPRESS ❌
値 2.0 (< 3.0, 安定3回目) → RESOLVE ✅ (解除通知)

SET タイプの詳細

新しい要素（エリアコード等）が追加された時のみ通知:

{
  "type": "SET",
  "auto_resolve_duration": 60,
  "status": "ACTIVE"
}

event_control.areas に指定した文字列配列を前回通知済みセットと比較し、新規要素があれば通知。

status フィールド（ACTUAL / DRILL / TEST）

ワークフローには動作モード設定があり、status と照合されます。

ワークフロー側で「ACTUALモードで動作」と設定されている場合、status: "DRILL" のPublishでは実行されません。

エンドツーエンドの例

シナリオ: 水位センサー → インシデント自動作成

1. トピック作成（一度だけ）

• Key: custom.river_level
• Fields: level (number), station_id (string)
• ControlPolicy: METRIC, threshold=3.0

2. ワークフロー作成（CRISISコンソール）

• トリガー: custom.river_level
• 条件: level >= 3.0
• アクション:
  • オペレーションブック起動（→ インシデント作成）
  • プッシュ通知送信

3. センサーがPublish

curl -X POST https://api.crisis.jp/v1/organizations/{orgId}/workflow_topics/{topicId}/publish \
  -H "Authorization: Bearer {token}" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {"level": 3.5, "station_id": "ST001", "station_name": "多摩川"},
    "status": "ACTUAL",
    "event_control": {"partition_key": "ST001", "metrics": {"level": 3.5}}
  }'

4. CRISISが自動実行

Publish受信
  → 制御ポリシー: 閾値3.0超え + 新規 → NOTIFY
  → フィルター: level(3.5) >= 3.0 → マッチ
  → アクション実行:
    → RUN_OPERATION_BOOK → インシデント作成 ✅
    → SEND_PUSH_NOTIFICATION → モバイルアプリ通知 ✅

5. 2回目のPublish（水位上昇）

{"data": {"level": 4.2, ...}, "event_control": {"partition_key": "ST001", "metrics": {"level": 4.2}}}

→ 制御ポリシー: ピーク更新(3.5→4.2) → NOTIFY → 再度アクション実行

6. 水位が下がったPublish

{"data": {"level": 2.5, ...}, "event_control": {"partition_key": "ST001", "metrics": {"level": 2.5}}}

→ 制御ポリシー: 閾値未満 → SUPPRESS（exit_stable_limit回に達するまで）

認証

Publish APIへのアクセスにはサービスアカウントを使用します。

1. 組織 > サービスアカウント でサービスアカウントを作成
2. ワークフロートピック発行者 権限ロールを付与
3. アクセストークンを発行、もしくはフェデレーション資格を作成
4. リクエストヘッダーに Authorization: Bearer {access_token} を設定

ベストプラクティス