Dify 集成和扩展技能

Dify 集成和扩展 SKILL

Guía de integración con sistemas y servicios externos

Navegación rápida

• Integración de API
• Protocolo MCP
• Integración de servicios externos
• Casos de integración
• Solución de problemas

---

Integración de API

Dify ofrece una API REST completa que admite la integración de aplicaciones y la orquestación de flujos de trabajo.

Puntos finales principales de la API

Punto final	Método	Descripción
/v1/chat-messages	POST	Enviar mensajes de chat
/v1/completions	POST	Enviar solicitudes de finalización
/v1/workflows/run	POST	Ejecutar un flujo de trabajo
/v1/parameters	GET	Obtener parámetros de la aplicación
/v1/file-upload	POST	Cargar archivos

Integración 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):
        """Enviar un mensaje 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):
        """Ejecutar un flujo de trabajo"""
        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()

Ejemplo de uso:

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

# Enviar un mensaje de chat
result = client.chat_message(
    query="Hola",
    user="user123",
    conversation_id=None  # Nueva conversación
)
print(result)

Integración Webhook

Los webhooks se utilizan para recibir notificaciones cuando ocurren eventos específicos, admiten la activación de flujos de trabajo y las devoluciones de llamada de eventos.

Activación de Webhook de flujo de trabajo

# Configuración del flujo de trabajo
nodes:
  - id: webhook-trigger
    data:
      type: webhook-trigger
      method: POST
      path: /webhook/my-workflow
      parameters:
        - name: data
          type: string
          required: true

Manejo de devolución de llamada de Webhook

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/callback', methods=['POST'])
def handle_webhook():
    """Manejar la devolución de llamada del Webhook de Dify"""
    payload = request.json

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

    # Procesar el evento
    event_type = payload.get('event')
    if event_type == 'workflow.finished':
        # Manejar la finalización del flujo de trabajo
        result = payload.get('data', {}).get('output', {})
        process_result(result)

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

Tipos de eventos de Webhook

Evento	Descripción	Momento de activación
workflow.started	El flujo de trabajo comienza	Cuando se inicia el flujo de trabajo
workflow.finished	El flujo de trabajo se completa	Cuando el flujo de trabajo finaliza normalmente
workflow.failed	El flujo de trabajo falla	Cuando la ejecución del flujo de trabajo falla
node.started	El nodo comienza	Cuando comienza la ejecución de un nodo
node.finished	El nodo se completa	Cuando la ejecución de un nodo finaliza

Respuesta en streaming

Admite Server-Sent Events (SSE) para salida en streaming en tiempo real.

import requests
import json

def stream_chat(client: DifyClient, query: str, user: str):
    """Solicitud de chat en streaming"""
    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:
            # Analizar formato SSE
            data = line.decode('utf-8')
            if data.startswith('data: '):
                event = json.loads(data[6:])
                yield event

# Ejemplo de uso
for event in stream_chat(client, "Cuenta una historia", "user123"):
    if event.get('event') == 'message':
        print(event.get('data', {}).get('answer'), end='', flush=True)

Formato de evento SSE:

{
  "event": "message",
  "task_id": "abc123",
  "message_id": "msg_xxx",
  "conversation_id": "conv_xxx",
  "answer": "Contenido del texto generado",
  "created_at": 1234567890
}

---

Protocolo MCP

Model Context Protocol (MCP) es un protocolo estandarizado para la integración entre asistentes de IA y sistemas externos.

Implementación del Servidor MCP

Dify puede actuar como un servidor MCP, proporcionando herramientas y recursos para que la IA los llame.

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):
        """Iniciar el servidor MCP"""
        @self.server.list_tools()
        async def list_tools() -> list[Tool]:
            return [
                Tool(
                    name="dify_chat",
                    description="Mantener una conversación con la aplicación Dify",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "Consulta del usuario"},
                            "user": {"type": "string", "description": "Identificador del usuario"}
                        },
                        "required": ["query", "user"]
                    }
                ),
                Tool(
                    name="dify_run_workflow",
                    description="Ejecutar un flujo de trabajo de Dify",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "inputs": {"type": "object", "description": "Parámetros de entrada del flujo de trabajo"},
                            "user": {"type": "string", "description": "Identificador del usuario"}
                        },
                        "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"Herramienta desconocida: {name}")

        # Iniciar el servicio
        await stdio_server(self.server.run())

# Ejecutar el servidor
if __name__ == "__main__":
    client = DifyClient("https://api.dify.ai/v1", "app-xxxxx")
    server = DifyMCPServer(client)
    asyncio.run(server.start())

Implementación del Cliente MCP

Dify puede llamar a herramientas proporcionadas por servidores MCP externos.

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):
        """Llamar a una herramienta 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

# Ejemplo de uso
async def main():
    client = DifyMCPClient(
        server_command="node",
        server_args=["mcp-server.js"]
    )
    result = await client.call_tool("get_weather", {"city": "Madrid"})
    print(result)

asyncio.run(main())

Configuración de comunicación bidireccional

Configurar un nodo MCP en un flujo de trabajo de 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: Herramienta de búsqueda
          input_schema:
            type: object
            properties:
              query:
                type: string

---

Integración de servicios externos

Integración de bases de datos

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]:
        # Obtener parámetros de conexión
        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")

        # Establecer conexión
        conn = psycopg2.connect(
            host=host,
            port=port,
            database=database,
            user=user,
            password=password
        )

        try:
            # Ejecutar consulta
            query = tool_parameters.get("query")
            cursor = conn.cursor()
            cursor.execute(query)

            # Obtener resultados
            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"Filas afectadas: {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("eliminado")

Integración de servicios de almacenamiento

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"Subido a {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})

Integración de colas de mensajes

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("Mensaje publicado")
        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 hay mensaje")

        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