API 統合ガイド – Chat / Completion / Workflow エンドポイント・認証・ストリーミング・Webhook

Dify で構築したアプリケーションを業務システムに統合するには、API を通じた連携が不可欠です。本記事では、Dify が提供する各種 API エンドポイントの構造・認証方式・リクエスト/レスポンス形式・ストリーミング対応・Webhook 連携を実践的に解説します。

1. Dify API の全体構造

1.1 API の分類

Dify の API は大きく2つのレイヤーに分かれます。

1.2 アプリタイプ別のエンドポイント

flowchart TD
    CLIENT[クライアントアプリ] --> AUTH{認証}
    AUTH --> |App API Key| APP[Application API]
    AUTH --> |Dataset API Key| DS[Dataset API]
    APP --> CHAT[Chat: /v1/chat-messages]
    APP --> COMP[Completion: /v1/completion-messages]
    APP --> WF[Workflow: /v1/workflows/run]
    DS --> DSOP[Dataset CRUD]
    DS --> DOC[Document CRUD]

2. 認証

2.1 API Key の取得

1. Dify コンソールでアプリを開く
2. 左メニューの 「API Access」 をクリック
3. 「API Key」 セクションで新しいキーを生成
4. キーをコピーして安全に保管

2.2 認証ヘッダーの形式

Authorization: Bearer {api_key}

リクエスト例:

curl -X POST "https://your-dify-instance.com/v1/chat-messages" \
  -H "Authorization: Bearer app-xxxxxxxxxxxxxxxxxxxx" \
  -H "Content-Type: application/json" \
  -d '{"inputs": {}, "query": "こんにちは", "user": "user-001"}'

2.3 API Key の管理ベストプラクティス

3. Chat App API

3.1 メッセージ送信

対話型アプリケーションにメッセージを送信し、AI の応答を取得します。

エンドポイント: POST /v1/chat-messages

リクエストボディ:

{
  "inputs": {
    "company_name": "株式会社Example",
    "department": "営業部"
  },
  "query": "有給休暇の申請方法を教えてください",
  "user": "user-001",
  "conversation_id": "",
  "response_mode": "blocking",
  "files": []
}

パラメータ説明:

レスポンス（blocking モード）:

{
  "event": "message",
  "message_id": "msg-xxxxxxxx",
  "conversation_id": "conv-xxxxxxxx",
  "mode": "chat",
  "answer": "有給休暇の申請は、社内ポータルの「勤怠管理」メニューから...",
  "metadata": {
    "usage": {
      "prompt_tokens": 512,
      "completion_tokens": 128,
      "total_tokens": 640,
      "total_price": "0.0064",
      "currency": "USD"
    },
    "retriever_resources": [
      {
        "dataset_id": "ds-xxxxx",
        "dataset_name": "社内規程集",
        "document_id": "doc-xxxxx",
        "document_name": "勤怠管理規程.pdf",
        "segment_id": "seg-xxxxx",
        "score": 0.92,
        "content": "有給休暇の申請は..."
      }
    ]
  },
  "created_at": 1712899200
}

3.2 会話履歴の取得

エンドポイント: GET /v1/messages

curl -X GET "https://your-dify-instance.com/v1/messages?conversation_id=conv-xxxxx&user=user-001&limit=20" \
  -H "Authorization: Bearer app-xxxxxxxxxxxxxxxxxxxx"

3.3 会話一覧の取得

エンドポイント: GET /v1/conversations

curl -X GET "https://your-dify-instance.com/v1/conversations?user=user-001&limit=20" \
  -H "Authorization: Bearer app-xxxxxxxxxxxxxxxxxxxx"

4. Completion App API

4.1 テキスト補完リクエスト

単発のテキスト生成（会話の文脈を持たない）に使用します。

エンドポイント: POST /v1/completion-messages

{
  "inputs": {
    "document_text": "本契約は、甲が乙に対して...",
    "task_type": "要約"
  },
  "user": "user-001",
  "response_mode": "blocking"
}

レスポンス:

{
  "event": "message",
  "message_id": "msg-xxxxxxxx",
  "mode": "completion",
  "answer": "本契約は、甲（発注者）と乙（受注者）の間で...",
  "metadata": {
    "usage": {
      "prompt_tokens": 1024,
      "completion_tokens": 256,
      "total_tokens": 1280
    }
  },
  "created_at": 1712899200
}

5. Workflow App API

5.1 Workflow の実行

エンドポイント: POST /v1/workflows/run

{
  "inputs": {
    "document_list": ["契約書A.pdf", "契約書B.pdf"],
    "analysis_type": "リスク分析",
    "output_format": "json"
  },
  "user": "user-001",
  "response_mode": "blocking"
}

レスポンス:

{
  "workflow_run_id": "wr-xxxxxxxx",
  "task_id": "task-xxxxxxxx",
  "data": {
    "id": "wr-xxxxxxxx",
    "workflow_id": "wf-xxxxxxxx",
    "status": "succeeded",
    "outputs": {
      "final_result": "...",
      "metadata": {}
    },
    "elapsed_time": 12.5,
    "total_tokens": 2048,
    "total_steps": 5,
    "created_at": 1712899200,
    "finished_at": 1712899213
  }
}

5.2 Workflow 実行ステータスの確認

エンドポイント: GET /v1/workflows/run/{workflow_run_id}

curl -X GET "https://your-dify-instance.com/v1/workflows/run/wr-xxxxxxxx" \
  -H "Authorization: Bearer app-xxxxxxxxxxxxxxxxxxxx"

5.3 Workflow の停止

エンドポイント: POST /v1/workflows/run/{workflow_run_id}/stop

curl -X POST "https://your-dify-instance.com/v1/workflows/run/wr-xxxxxxxx/stop" \
  -H "Authorization: Bearer app-xxxxxxxxxxxxxxxxxxxx" \
  -H "Content-Type: application/json" \
  -d '{"user": "user-001"}'

6. ストリーミングレスポンス

6.1 ストリーミングの仕組み

response_mode: "streaming" を指定すると、Server-Sent Events (SSE) 形式でレスポンスが逐次返されます。

data: {"event": "message", "message_id": "msg-xxx", "answer": "有給", ...}

data: {"event": "message", "message_id": "msg-xxx", "answer": "休暇", ...}

data: {"event": "message", "message_id": "msg-xxx", "answer": "の", ...}

data: {"event": "message_end", "message_id": "msg-xxx", "metadata": {...}}

6.2 SSE イベントタイプ

6.3 フロントエンド実装例（JavaScript）

async function sendChatMessage(query, conversationId = "") {
  const response = await fetch("https://your-dify-instance.com/v1/chat-messages", {
    method: "POST",
    headers: {
      "Authorization": "Bearer app-xxxxxxxxxxxxxxxxxxxx",
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      inputs: {},
      query: query,
      user: "user-001",
      conversation_id: conversationId,
      response_mode: "streaming",
    }),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let fullAnswer = "";
  let newConversationId = "";

  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: ")) {
        const data = JSON.parse(line.slice(6));

        switch (data.event) {
          case "message":
            fullAnswer += data.answer;
            // UIにリアルタイム表示
            updateUI(fullAnswer);
            newConversationId = data.conversation_id;
            break;

          case "message_end":
            console.log("Token usage:", data.metadata?.usage);
            break;

          case "error":
            console.error("Error:", data.message);
            break;
        }
      }
    }
  }

  return { answer: fullAnswer, conversationId: newConversationId };
}

6.4 バックエンド実装例（Python）

import requests
import json

DIFY_BASE_URL = "https://your-dify-instance.com"
APP_API_KEY = "app-xxxxxxxxxxxxxxxxxxxx"

def chat_streaming(query: str, user: str, conversation_id: str = ""):
    """ストリーミングモードでチャットメッセージを送信"""
    url = f"{DIFY_BASE_URL}/v1/chat-messages"
    headers = {
        "Authorization": f"Bearer {APP_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "inputs": {},
        "query": query,
        "user": user,
        "conversation_id": conversation_id,
        "response_mode": "streaming",
    }

    full_answer = ""
    with requests.post(url, headers=headers, json=payload, stream=True) as resp:
        resp.raise_for_status()
        for line in resp.iter_lines():
            if line:
                decoded = line.decode("utf-8")
                if decoded.startswith("data: "):
                    data = json.loads(decoded[6:])
                    event = data.get("event")

                    if event == "message":
                        chunk = data.get("answer", "")
                        full_answer += chunk
                        print(chunk, end="", flush=True)

                    elif event == "message_end":
                        print()  # 改行
                        return {
                            "answer": full_answer,
                            "conversation_id": data.get("conversation_id"),
                            "metadata": data.get("metadata"),
                        }

                    elif event == "error":
                        raise Exception(f"API Error: {data.get('message')}")

    return {"answer": full_answer}


def chat_blocking(query: str, user: str, conversation_id: str = ""):
    """ブロッキングモードでチャットメッセージを送信"""
    url = f"{DIFY_BASE_URL}/v1/chat-messages"
    headers = {
        "Authorization": f"Bearer {APP_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "inputs": {},
        "query": query,
        "user": user,
        "conversation_id": conversation_id,
        "response_mode": "blocking",
    }

    resp = requests.post(url, headers=headers, json=payload)
    resp.raise_for_status()
    return resp.json()

7. blocking vs streaming の選択基準

推奨: ユーザー向けのチャットインターフェースでは常に streaming を使用してください。バックエンドの API-to-API 連携で結果全体が必要な場合は blocking が適しています。

8. エラーハンドリング

8.1 主要なエラーコード

8.2 リトライ戦略

import time
import requests
from requests.exceptions import RequestException

def call_dify_api_with_retry(url, headers, payload, max_retries=3):
    """指数バックオフ付きリトライ"""
    for attempt in range(max_retries):
        try:
            resp = requests.post(url, headers=headers, json=payload, timeout=60)

            if resp.status_code == 429:
                retry_after = int(resp.headers.get("Retry-After", 10))
                print(f"Rate limited. Retrying after {retry_after}s...")
                time.sleep(retry_after)
                continue

            if resp.status_code >= 500:
                wait = 2 ** attempt  # 指数バックオフ: 1, 2, 4秒
                print(f"Server error {resp.status_code}. Retrying in {wait}s...")
                time.sleep(wait)
                continue

            resp.raise_for_status()
            return resp.json()

        except RequestException as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt
            print(f"Request failed: {e}. Retrying in {wait}s...")
            time.sleep(wait)

    raise Exception(f"Failed after {max_retries} retries")

9. Webhook コールバック

9.1 Workflow の Webhook 連携

Workflow の HTTP Request ノードを使って、処理結果を外部システムに通知できます。

HTTP Request ノード設定例:

method: POST
url: "https://your-system.example.com/webhook/dify-callback"
headers:
  Content-Type: "application/json"
  X-Webhook-Secret: "{{webhook_secret}}"
body:
  type: json
  content: |
    {
      "event": "workflow_completed",
      "workflow_id": "{{workflow_id}}",
      "result": {{workflow_output}},
      "timestamp": "{{current_timestamp}}",
      "status": "success"
    }

9.2 Webhook 受信側の実装例（Node.js / Express）

const express = require("express");
const crypto = require("crypto");

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

const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

function verifySignature(payload, signature) {
  const expected = crypto
    .createHmac("sha256", WEBHOOK_SECRET)
    .update(JSON.stringify(payload))
    .digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expected)
  );
}

app.post("/webhook/dify-callback", (req, res) => {
  const signature = req.headers["x-webhook-secret"];

  if (signature !== WEBHOOK_SECRET) {
    return res.status(401).json({ error: "Invalid signature" });
  }

  const { event, workflow_id, result, status } = req.body;
  console.log(`Received webhook: ${event} for workflow ${workflow_id}`);

  // ビジネスロジック: 結果をDBに保存、通知送信など
  processWorkflowResult(result, status);

  res.status(200).json({ received: true });
});

function processWorkflowResult(result, status) {
  // 例: Slack通知、DB保存、後続処理のトリガーなど
  console.log("Processing result:", result);
}

app.listen(3000, () => console.log("Webhook server running on port 3000"));

10. 実装パターン集

10.1 LINE Bot 連携

# LINE Bot から Dify Chat API を呼び出すパターン
from flask import Flask, request
from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import requests

app = Flask(__name__)
line_bot_api = LineBotApi("LINE_CHANNEL_ACCESS_TOKEN")
handler = WebhookHandler("LINE_CHANNEL_SECRET")

DIFY_API_URL = "https://your-dify-instance.com/v1/chat-messages"
DIFY_API_KEY = "app-xxxxxxxxxxxxxxxxxxxx"

# ユーザーごとの会話IDを管理
conversation_store = {}

@app.route("/callback", methods=["POST"])
def callback():
    handler.handle(request.get_data(as_text=True),
                   request.headers["X-Line-Signature"])
    return "OK"

@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
    user_id = event.source.user_id
    conv_id = conversation_store.get(user_id, "")

    resp = requests.post(
        DIFY_API_URL,
        headers={
            "Authorization": f"Bearer {DIFY_API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "inputs": {},
            "query": event.message.text,
            "user": user_id,
            "conversation_id": conv_id,
            "response_mode": "blocking",
        },
    )
    data = resp.json()

    conversation_store[user_id] = data.get("conversation_id", "")

    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=data["answer"])
    )

10.2 バッチ処理パターン

import csv
import json
import time
import requests

DIFY_API_URL = "https://your-dify-instance.com/v1/workflows/run"
DIFY_API_KEY = "app-xxxxxxxxxxxxxxxxxxxx"

def process_batch(input_csv: str, output_csv: str):
    """CSVファイルの各行に対してWorkflowを実行するバッチ処理"""
    results = []

    with open(input_csv, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader):
            print(f"Processing row {i + 1}: {row.get('title', 'N/A')}")

            resp = requests.post(
                DIFY_API_URL,
                headers={
                    "Authorization": f"Bearer {DIFY_API_KEY}",
                    "Content-Type": "application/json",
                },
                json={
                    "inputs": {
                        "document_text": row["content"],
                        "analysis_type": row.get("type", "要約"),
                    },
                    "user": "batch-processor",
                    "response_mode": "blocking",
                },
            )

            if resp.status_code == 200:
                data = resp.json()
                results.append({
                    "input_title": row.get("title", ""),
                    "output": data["data"]["outputs"].get("final_result", ""),
                    "status": "success",
                    "tokens": data["data"].get("total_tokens", 0),
                })
            else:
                results.append({
                    "input_title": row.get("title", ""),
                    "output": "",
                    "status": f"error: {resp.status_code}",
                    "tokens": 0,
                })

            # レート制限対策: リクエスト間に待機
            time.sleep(1)

    # 結果をCSVに出力
    with open(output_csv, "w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["input_title", "output", "status", "tokens"])
        writer.writeheader()
        writer.writerows(results)

    print(f"Batch complete. Results saved to {output_csv}")

11. セキュリティと運用

11.1 セキュリティチェックリスト

• API Key がソースコードにハードコーディングされていない
• HTTPS 通信が使用されている
• CORS 設定が適切に構成されている
• API Key のローテーションポリシーが策定されている
• レート制限が設定されている
• 入力バリデーションが実装されている
• エラーレスポンスに内部情報が漏洩していない

11.2 モニタリング項目

12. まとめ

Dify API を業務システムに統合する際のポイントを以下にまとめます。

API 統合の目標は「curl コマンドが打てる」ことではなく、Dify アプリケーションが実際のビジネスシステムの一部として安定稼働することです。上記の実装パターンとベストプラクティスを参考に、堅牢な統合を実現してください。