SKILL d'intégration et d'extension Dify

Guide d'intégration avec les systèmes et services externes

Navigation rapide

Intégration API
Protocole MCP
Intégration de services externes
Cas d'intégration
Dépannage

Intégration API

Dify fournit une API REST complète, prenant en charge l'intégration d'applications et l'orchestration de workflows.

Points de terminaison API principaux

Point de terminaison	Méthode	Description
`/v1/chat-messages`	POST	Envoyer un message de chat
`/v1/completions`	POST	Envoyer une demande de complétion
`/v1/workflows/run`	POST	Exécuter un workflow
`/v1/parameters`	GET	Obtenir les paramètres de l'application
`/v1/file-upload`	POST	Téléchargement de fichier

Intégration API REST

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):
        """Envoyer un message de chat"""
        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):
        """Exécuter un workflow"""
        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()

Exemple d'utilisation :

# Initialiser le client
client = DifyClient(
    base_url="https://api.dify.ai/v1",
    api_key="app-xxxxx"
)

# Envoyer un message de chat
result = client.chat_message(
    query="Bonjour",
    user="user123",
    conversation_id=None  # Nouvelle conversation
)
print(result)

Intégration Webhook

Les webhooks sont utilisés pour recevoir des notifications lors d'événements spécifiques, et prennent en charge le déclenchement de workflows et les rappels d'événements.

Déclenchement de workflow par Webhook

# Configuration du workflow
nodes:
  - id: webhook-trigger
    data:
      type: webhook-trigger
      method: POST
      path: /webhook/mon-workflow
      parameters:
        - name: data
          type: string
          required: true

Gestion des rappels Webhook

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/callback', methods=['POST'])
def handle_webhook():
    """Gérer un rappel Webhook Dify"""
    payload = request.json

    # Vérifier la signature
    signature = request.headers.get('X-Dify-Signature')
    if not verify_signature(payload, signature):
        return jsonify({"error": "Signature invalide"}), 401

    # Traiter l'événement
    event_type = payload.get('event')
    if event_type == 'workflow.finished':
        # Traitement de la fin du workflow
        result = payload.get('data', {}).get('output', {})
        process_result(result)

    return jsonify({"status": "succès"})

Types d'événements Webhook

Événement	Description	Moment du déclenchement
`workflow.started`	Début du workflow	Au démarrage du workflow
`workflow.finished`	Fin du workflow	Lorsque le workflow se termine normalement
`workflow.failed`	Échec du workflow	Lorsque l'exécution du workflow échoue
`node.started`	Début du nœud	Au début de l'exécution du nœud
`node.finished`	Fin du nœud	Lorsque le nœud termine son exécution

Réponses en continu

Prise en charge des Server-Sent Events (SSE) pour une sortie en continu en temps réel.

import requests
import json

def stream_chat(client: DifyClient, query: str, user: str):
    """Requête de chat en continu"""
    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:
            # Analyser le format SSE
            data = line.decode('utf-8')
            if data.startswith('data: '):
                event = json.loads(data[6:])
                yield event

# Exemple d'utilisation
for event in stream_chat(client, "Raconte une histoire", "user123"):
    if event.get('event') == 'message':
        print(event.get('data', {}).get('answer'), end='', flush=True)

Format de l'événement SSE :

{
  "event": "message",
  "task_id": "abc123",
  "message_id": "msg_xxx",
  "conversation_id": "conv_xxx",
  "answer": "Contenu du texte généré",
  "created_at": 1234567890
}

Protocole MCP

Model Context Protocol (MCP) est un protocole standardisé pour l'intégration des assistants IA avec les systèmes externes.

Implémentation du serveur MCP

Dify peut agir comme un serveur MCP, fournissant des outils et des ressources pour les appels IA.

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):
        """Démarrer le serveur MCP"""
        @self.server.list_tools()
        async def list_tools() -> list[Tool]:
            return [
                Tool(
                    name="dify_chat",
                    description="Discuter avec l'application Dify",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "Requête utilisateur"},
                            "user": {"type": "string", "description": "Identifiant utilisateur"}
                        },
                        "required": ["query", "user"]
                    }
                ),
                Tool(
                    name="dify_run_workflow",
                    description="Exécuter un workflow Dify",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "inputs": {"type": "object", "description": "Paramètres d'entrée du workflow"},
                            "user": {"type": "string", "description": "Identifiant utilisateur"}
                        },
                        "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"Outil inconnu : {name}")

        # Démarrer le service
        await stdio_server(self.server.run())

# Exécuter le serveur
if __name__ == "__main__":
    client = DifyClient("https://api.dify.ai/v1", "app-xxxxx")
    server = DifyMCPServer(client)
    asyncio.run(server.start())

Implémentation du client MCP

Dify peut appeler des outils fournis par des serveurs MCP externes.

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):
        """Appeler un outil 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

# Exemple d'utilisation
async def main():
    client = DifyMCPClient(
        server_command="node",
        server_args=["mcp-server.js"]
    )
    result = await client.call_tool("get_weather", {"city": "Pékin"})
    print(result)

asyncio.run(main())

Configuration de la communication bidirectionnelle

Configuration d'un nœud MCP dans un workflow Dify :

nodes:
  - id: mcp-tool
    data:
      type: tool
      provider_type: mcp
      server_config:
        type: stdio
        command: python
        args: ["mcp_server.py"]
      tools:
        - name: search
          description: Outil de recherche
          input_schema:
            type: object
            properties:
              query:
                type: string

Intégration de services externes

Intégration de bases de données

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]:
        # Obtenir les paramètres de connexion
        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")

        # Établir la connexion
        conn = psycopg2.connect(
            host=host,
            port=port,
            database=database,
            user=user,
            password=password
        )

        try:
            # Exécuter la requête
            query = tool_parameters.get("query")
            cursor = conn.cursor()
            cursor.execute(query)

            # Obtenir le résultat
            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"lignes affectées : {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("supprimé")

Intégration de services de stockage

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"Téléchargé vers {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})

Intégration de files d'attente de messages

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 publié")
        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("Aucun 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("Message publié")

SystemTce/dify-skills

Qu'est-ce que dify-skills ?

Demander à votre IA préférée

Documentation