SystemTce/dify-skills

Dify 統合と拡張 SKILL

外部システムとサービスの統合ガイド

クイックナビゲーション

• API 統合
• MCP プロトコル
• 外部サービス統合
• 統合事例
• トラブルシューティング

---

API 統合

Dify は完全な REST API を提供し、アプリケーション統合とワークフロー編成をサポートします。

コア API エンドポイント

エンドポイント	メソッド	説明
/v1/chat-messages	POST	チャットメッセージの送信
/v1/completions	POST	補完リクエストの送信
/v1/workflows/run	POST	ワークフローの実行
/v1/parameters	GET	アプリケーションパラメータの取得
/v1/file-upload	POST	ファイルアップロード

REST API 統合

import requests
import json

class DifyClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip('/')
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def chat_message(self, query: str, user: str, **kwargs):
        """チャットメッセージの送信"""
        payload = {
            "query": query,
            "user": user,
            **kwargs
        }
        response = requests.post(
            f"{self.base_url}/v1/chat-messages",
            headers=self.headers,
            json=payload
        )
        return response.json()

    def run_workflow(self, workflow_id: str, inputs: dict, user: str):
        """ワークフローの実行"""
        payload = {
            "inputs": inputs,
            "user": user,
            "response_mode": "blocking"
        }
        response = requests.post(
            f"{self.base_url}/v1/workflows/run",
            headers=self.headers,
            json=payload
        )
        return response.json()

使用例:

# クライアントの初期化
client = DifyClient(
    base_url="https://api.dify.ai/v1",
    api_key="app-xxxxx"
)

# チャットメッセージの送信
result = client.chat_message(
    query="こんにちは",
    user="user123",
    conversation_id=None  # 新しいセッション
)
print(result)

Webhook 統合

Webhook は特定のイベント発生時に通知を受信するために使用され、ワークフロートリガーとイベントコールバックをサポートします。

ワークフロー Webhook トリガー

# ワークフロー設定
nodes:
  - id: webhook-trigger
    data:
      type: webhook-trigger
      method: POST
      path: /webhook/my-workflow
      parameters:
        - name: data
          type: string
          required: true

Webhook コールバック処理

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/callback', methods=['POST'])
def handle_webhook():
    """Dify Webhook コールバックの処理"""
    payload = request.json

    # 署名の検証
    signature = request.headers.get('X-Dify-Signature')
    if not verify_signature(payload, signature):
        return jsonify({"error": "Invalid signature"}), 401

    # イベントの処理
    event_type = payload.get('event')
    if event_type == 'workflow.finished':
        # ワークフロー完了時の処理
        result = payload.get('data', {}).get('output', {})
        process_result(result)

    return jsonify({"status": "success"})

Webhook イベントタイプ

イベント	説明	トリガー条件
workflow.started	ワークフロー開始	ワークフロー起動時
workflow.finished	ワークフロー完了	ワークフロー正常完了時
workflow.failed	ワークフロー失敗	ワークフロー実行失敗時
node.started	ノード開始	ノード実行開始時
node.finished	ノード完了	ノード実行完了時

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

Server-Sent Events (SSE) を使用したリアルタイムストリーミング出力をサポートします。

import requests
import json

def stream_chat(client: DifyClient, query: str, user: str):
    """ストリーミングチャットリクエスト"""
    payload = {
        "query": query,
        "user": user,
        "response_mode": "streaming"
    }

    response = requests.post(
        f"{client.base_url}/v1/chat-messages",
        headers=client.headers,
        json=payload,
        stream=True
    )

    for line in response.iter_lines():
        if line:
            # SSE 形式の解析
            data = line.decode('utf-8')
            if data.startswith('data: '):
                event = json.loads(data[6:])
                yield event

# 使用例
for event in stream_chat(client, "話をして", "user123"):
    if event.get('event') == 'message':
        print(event.get('data', {}).get('answer'), end='', flush=True)

SSE イベント形式:

{
  "event": "message",
  "task_id": "abc123",
  "message_id": "msg_xxx",
  "conversation_id": "conv_xxx",
  "answer": "生成されたテキスト内容",
  "created_at": 1234567890
}

---

MCP プロトコル

Model Context Protocol (MCP) は、AI アシスタントと外部システムの統合のための標準化されたプロトコルです。

MCP Server の実装

Dify は MCP Server として機能し、AI が呼び出すためのツールとリソースを提供できます。

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import asyncio

class DifyMCPServer:
    def __init__(self, dify_client: DifyClient):
        self.client = dify_client
        self.server = Server("dify-mcp")

    async def start(self):
        """MCP Server の起動"""
        @self.server.list_tools()
        async def list_tools() -> list[Tool]:
            return [
                Tool(
                    name="dify_chat",
                    description="Dify アプリとの対話",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "ユーザークエリ"},
                            "user": {"type": "string", "description": "ユーザー識別子"}
                        },
                        "required": ["query", "user"]
                    }
                ),
                Tool(
                    name="dify_run_workflow",
                    description="Dify ワークフローの実行",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "inputs": {"type": "object", "description": "ワークフロー入力パラメータ"},
                            "user": {"type": "string", "description": "ユーザー識別子"}
                        },
                        "required": ["inputs", "user"]
                    }
                )
            ]

        @self.server.call_tool()
        async def call_tool(name: str, arguments: dict) -> list[TextContent]:
            if name == "dify_chat":
                result = self.client.chat_message(
                    query=arguments["query"],
                    user=arguments["user"]
                )
                return [TextContent(type="text", text=json.dumps(result))]
            elif name == "dify_run_workflow":
                result = self.client.run_workflow(
                    workflow_id=arguments.get("workflow_id", ""),
                    inputs=arguments["inputs"],
                    user=arguments["user"]
                )
                return [TextContent(type="text", text=json.dumps(result))]
            else:
                raise ValueError(f"Unknown tool: {name}")

        # サービスの起動
        await stdio_server(self.server.run())

# サーバーの実行
if __name__ == "__main__":
    client = DifyClient("https://api.dify.ai/v1", "app-xxxxx")
    server = DifyMCPServer(client)
    asyncio.run(server.start())

MCP Client の実装

Dify は外部の MCP Server が提供するツールを呼び出すことができます。

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

class DifyMCPClient:
    def __init__(self, server_command: str, server_args: list[str]):
        self.server_params = StdioServerParameters(
            command=server_command,
            args=server_args
        )

    async def call_tool(self, tool_name: str, arguments: dict):
        """MCP ツールの呼び出し"""
        async with stdio_client(self.server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                result = await session.call_tool(tool_name, arguments)
                return result

# 使用例
async def main():
    client = DifyMCPClient(
        server_command="node",
        server_args=["mcp-server.js"]
    )
    result = await client.call_tool("get_weather", {"city": "東京"})
    print(result)

asyncio.run(main())

双方向通信設定

Dify ワークフローでの MCP ノード設定：

nodes:
  - id: mcp-tool
    data:
      type: tool
      provider_type: mcp
      server_config:
        type: stdio
        command: python
        args: ["mcp_server.py"]
      tools:
        - name: search
          description: 検索ツール
          input_schema:
            type: object
            properties:
              query:
                type: string

---

外部サービス統合

データベース統合

PostgreSQL/MySQL

import psycopg2
from dify_plugin import Tool
from typing import Generator

class DatabaseTool(Tool):
    def _invoke(
        self,
        tool_parameters: dict
    ) -> Generator[ToolInvokeMessage, None, None]:
        # 接続パラメータの取得
        host = self.runtime_credentials.get("host")
        port = self.runtime_credentials.get("port")
        database = self.runtime_credentials.get("database")
        user = self.runtime_credentials.get("user")
        password = self.runtime_credentials.get("password")

        # 接続の確立
        conn = psycopg2.connect(
            host=host,
            port=port,
            database=database,
            user=user,
            password=password
        )

        try:
            # クエリの実行
            query = tool_parameters.get("query")
            cursor = conn.cursor()
            cursor.execute(query)

            # 結果の取得
            if query.strip().upper().startswith("SELECT"):
                columns = [desc[0] for desc in cursor.description]
                results = cursor.fetchall()
                yield self.create_json_message({
                    "columns": columns,
                    "data": results
                })
            else:
                conn.commit()
                yield self.create_text_message(f"影響を受けた行数: {cursor.rowcount}")
        finally:
            conn.close()

Redis

import redis
from dify_plugin import Tool

class RedisTool(Tool):
    def _invoke(self, tool_parameters: dict) -> Generator[ToolInvokeMessage, None, None]:
        r = redis.Redis(
            host=self.runtime_credentials.get("host"),
            port=self.runtime_credentials.get("port"),
            password=self.runtime_credentials.get("password"),
            db=self.runtime_credentials.get("db", 0)
        )

        operation = tool_parameters.get("operation")
        key = tool_parameters.get("key")

        if operation == "get":
            value = r.get(key)
            yield self.create_text_message(value.decode() if value else "")
        elif operation == "set":
            value = tool_parameters.get("value")
            r.set(key, value)
            yield self.create_text_message("OK")
        elif operation == "delete":
            r.delete(key)
            yield self.create_text_message("削除されました")

ストレージサービス統合

S3/MinIO

import boto3
from dify_plugin import Tool

class S3Tool(Tool):
    def _invoke(self, tool_parameters: dict) -> Generator[ToolInvokeMessage, None, None]:
        s3 = boto3.client(
            's3',
            endpoint_url=self.runtime_credentials.get("endpoint"),
            aws_access_key_id=self.runtime_credentials.get("access_key"),
            aws_secret_access_key=self.runtime_credentials.get("secret_key")
        )

        operation = tool_parameters.get("operation")
        bucket = tool_parameters.get("bucket")
        key = tool_parameters.get("key")

        if operation == "upload":
            content = tool_parameters.get("content")
            s3.put_object(Bucket=bucket, Key=key, Body=content)
            yield self.create_text_message(f"アップロード完了: {bucket}/{key}")
        elif operation == "download":
            response = s3.get_object(Bucket=bucket, Key=key)
            content = response['Body'].read()
            yield self.create_binary_message(content)
        elif operation == "list":
            response = s3.list_objects_v2(Bucket=bucket)
            keys = [obj['Key'] for obj in response.get('Contents', [])]
            yield self.create_json_message({"keys": keys})

メッセージキュー統合

RabbitMQ

import pika
from dify_plugin import Tool

class RabbitMQTool(Tool):
    def _invoke(self, tool_parameters: dict) -> Generator[ToolInvokeMessage, None, None]:
        credentials = pika.PlainCredentials(
            self.runtime_credentials.get("username"),
            self.runtime_credentials.get("password")
        )
        connection = pika.BlockingConnection(
            pika.ConnectionParameters(
                host=self.runtime_credentials.get("host"),
                port=self.runtime_credentials.get("port"),
                credentials=credentials
            )
        )
        channel = connection.channel()

        operation = tool_parameters.get("operation")
        queue = tool_parameters.get("queue")

        if operation == "publish":
            message = tool_parameters.get("message")
            channel.queue_declare(queue=queue, durable=True)
            channel.basic_publish(
                exchange='',
                routing_key=queue,
                body=message,
                properties=pika.BasicProperties(delivery_mode=2)
            )
            yield self.create_text_message("メッセージが公開されました")
        elif operation == "consume":
            method, properties, body = channel.basic_get(queue=queue)
            if body:
                yield self.create_text_message(body.decode())
                channel.basic_ack(method.delivery_tag)
            else:
                yield self.create_text_message("メッセージがありません")

        connection.close()

Kafka

from kafka import KafkaProducer, KafkaConsumer
from dify_plugin import Tool
import json

class KafkaTool(Tool):
    def _invoke(self, tool_parameters: dict) -> Generator[ToolInvokeMessage, None, None]:
        operation = tool_parameters.get("operation")
        topic = tool_parameters.get("topic")

        if operation == "publish":
            message = tool_parameters.get("message")
            producer = KafkaProducer(
                bootstrap_servers=self.runtime_credentials.get("servers"),
                value_serializer=lambda v: json.dumps(v).encode('utf-8')
            )
            producer.send(topic, value={"message": message})
            producer.flush()
            producer.close()
            yield self.create_text_message("メッセージが公開されました")
        elif operation == "consume":
            consumer = KafkaConsumer(
                topic,
                bootstrap_servers=self.runtime_credentials.get("servers"),
                auto_offset_reset='earliest',
                consumer_timeout_ms=1000
            )
            for msg in consumer:
                yield self.create_text_message(msg.value.decode('utf-8'))
                break
            else:
                yield self.create_text_message("メッセージがありません")
            consumer.close()

---

統合事例

AI 搭載カスタマーサポートボット

# 複数サービスを統合したワークフロー例
def customer_support_workflow(user_query: str, user_id: str):
    # 1. Dify チャットを使用した問い合わせ理解
    intent = dify_client.chat_message(
        query=f"ユーザーの意図を特定してください: {user_query}",
        user=user_id
    )

    # 2. データベースからのユーザー情報取得
    user_info = database_client.query(
        f"SELECT * FROM users WHERE user_id = '{user_id}'"
    )

    # 3. Redis からのセッション履歴取得
    session_history = redis_client.get(f"session:{user_id}")

    # 4. 応答の生成のために Dify ワークフローを実行
    response = dify_client.run_workflow(
        workflow_id="customer-support",
        inputs={
            "user_query": user_query,
            "user_info": user_info,
            "session_history": session_history
        },
        user=user_id
    )

    return response

データパイプライン処理

# Kafka からデータを消費し、データベースに保存するワークフロー
nodes:
  - id: kafka-consumer
    type: tool
    provider_type: kafka
    config:
      topic: raw-data
      group_id: dify-processor

  - id: data-transform
    type: code
    config:
      code: |
        def transform(data):
            # データ変換ロジック
            return {
                "processed": True,
                "timestamp": datetime.now().isoformat(),
                "content": data
            }

  - id: database-saver
    type: tool
    provider_type: postgresql
    config:
      query: |
        INSERT INTO processed_data (content, processed_at)
        VALUES (%(content)s, %(timestamp)s)

---

トラブルシューティング

一般的な問題

問題	原因	解決策
API 認証エラー	API キーが無効または期限切れ	API キーを確認し、必要に応じて再生成
接続タイムアウト	ネットワークの問題またはファイアウォール	エンドポイントがアクセス可能か、ファイアウォール設定を確認
レート制限超過	短時間にリクエストが多すぎる	レート制限を確認し、リクエスト間隔を調整
MCP 接続失敗	MCP Server が利用できない	MCP Server が実行中であることを確認
データベース接続エラー	認証情報が間違っている	接続パラメータを確認

デバッグモード

# 環境変数でデバッグモードを有効化
export DIFY_DEBUG=true

# 詳細なログ出力
export LOG_LEVEL=DEBUG

パフォーマンス最適化

• 接続プール: データベースや Redis など、高コストな接続にはプールを使用
• キャッシュ戦略: 頻繁にアクセスされるデータには Redis キャッシュを実装
• 非同期処理: 同期呼び出しの代わりに非同期モードによるワークフロー実行
• バッチ処理: 大量データ処理にはバッチ操作を使用

ヘルスチェック

def health_check():
    """全統合サービスの健全性チェック"""
    services = {
        "dify_api": check_dify_api(),
        "database": check_database(),
        "redis": check_redis(),
        "message_queue": check_message_queue()
    }

    status = all(services.values())
    return {
        "status": "healthy" if status else "degraded",
        "services": services
    }

---

参考資料

• Dify 公式ドキュメント
• MCP プロトコル仕様
• Dify GitHub リポジトリ