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 서버 구현

Dify는 MCP 서버로 작동하여 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 서버 시작"""
        @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 클라이언트 구현

Dify는 외부 MCP 서버가 제공하는 도구를 호출할 수 있습니다.

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"affected rows: {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("deleted")

스토리지 서비스 통합

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"Uploaded to {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("Message published")
        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("No 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.cr

…

[번역을 위해 잘렸습니다]