SystemTce/dify-skills

Habilidade de Integração e Extensão do Dify

Guia de integração com sistemas e serviços externos

Navegação Rápida

• Integração de API
• Protocolo MCP
• Integração de Serviços Externos
• Casos de Integração
• Solução de Problemas

---

Integração de API

O Dify fornece uma API REST completa, suportando integração de aplicativos e orquestração de workflows.

Endpoints Principais da API

Endpoint	Método	Descrição
/v1/chat-messages	POST	Enviar mensagens de chat
/v1/completions	POST	Enviar requisições de completude
/v1/workflows/run	POST	Executar workflow
/v1/parameters	GET	Obter parâmetros do aplicativo
/v1/file-upload	POST	Upload de arquivo

Integração com 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):
        """Envia uma mensagem 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):
        """Executa um 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()

Exemplo de uso:

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

# Envia uma mensagem de chat
result = client.chat_message(
    query="Olá",
    user="user123",
    conversation_id=None  # Nova conversa
)
print(result)

Integração com Webhook

Webhooks são usados para receber notificações quando eventos específicos ocorrem, suportando disparo de workflows e callbacks de eventos.

Disparo de Workflow via Webhook

# Configuração do workflow
nodes:
  - id: webhook-trigger
    data:
      type: webhook-trigger
      method: POST
      path: /webhook/my-workflow
      parameters:
        - name: data
          type: string
          required: true

Tratamento de Callback Webhook

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/callback', methods=['POST'])
def handle_webhook():
    """Processa callback do Webhook do Dify"""
    payload = request.json

    # Verifica assinatura
    signature = request.headers.get('X-Dify-Signature')
    if not verify_signature(payload, signature):
        return jsonify({"error": "Assinatura inválida"}), 401

    # Processa evento
    event_type = payload.get('event')
    if event_type == 'workflow.finished':
        # Tratamento de conclusão de workflow
        result = payload.get('data', {}).get('output', {})
        process_result(result)

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

Tipos de Evento Webhook

Evento	Descrição	Momento do Disparo
workflow.started	Workflow iniciado	Quando o workflow é iniciado
workflow.finished	Workflow concluído	Quando o workflow termina normalmente
workflow.failed	Workflow falhou	Quando a execução do workflow falha
node.started	Nó iniciado	Quando um nó começa a executar
node.finished	Nó concluído	Quando a execução de um nó termina

Resposta em Streaming

Suporta Server-Sent Events (SSE) para saída em tempo real.

import requests
import json

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

# Exemplo de uso
for event in stream_chat(client, "Conte uma história", "user123"):
    if event.get('event') == 'message':
        print(event.get('data', {}).get('answer'), end='', flush=True)

Formato do Evento SSE:

{
  "event": "message",
  "task_id": "abc123",
  "message_id": "msg_xxx",
  "conversation_id": "conv_xxx",
  "answer": "Conteúdo do texto gerado",
  "created_at": 1234567890
}

---

Protocolo MCP

Model Context Protocol (MCP) é um protocolo padronizado para integração de assistentes de IA com sistemas externos.

Implementação do Servidor MCP

O Dify pode atuar como um Servidor MCP, fornecendo ferramentas e recursos para serem chamados pela 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):
        """Inicia o Servidor MCP"""
        @self.server.list_tools()
        async def list_tools() -> list[Tool]:
            return [
                Tool(
                    name="dify_chat",
                    description="Conversa com o aplicativo Dify",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "Consulta do usuário"},
                            "user": {"type": "string", "description": "Identificador do usuário"}
                        },
                        "required": ["query", "user"]
                    }
                ),
                Tool(
                    name="dify_run_workflow",
                    description="Executa um workflow do Dify",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "inputs": {"type": "object", "description": "Parâmetros de entrada do workflow"},
                            "user": {"type": "string", "description": "Identificador do usuário"}
                        },
                        "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"Ferramenta desconhecida: {name}")

        # Inicia o serviço
        await stdio_server(self.server.run())

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

Implementação do Cliente MCP

O Dify pode chamar ferramentas fornecidas por um Servidor MCP externo.

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):
        """Chama uma ferramenta 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

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

asyncio.run(main())

Configuração de Comunicação Bidirecional

Configure um nó MCP em um workflow do 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: Ferramenta de busca
          input_schema:
            type: object
            properties:
              query:
                type: string

---

Integração de Serviços Externos

Integração com Banco de Dados

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]:
        # Obtém parâmetros de conexão
        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")

        # Estabelece conexão
        conn = psycopg2.connect(
            host=host,
            port=port,
            database=database,
            user=user,
            password=password
        )

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

            # Obtém 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"linhas afetadas: {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("deletado")

Integração com Serviço de Armazenamento

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"Enviado para {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})

Integração com Fila de Mensagens

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("Mensagem publicada")
        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("Nenhuma mensagem")

        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("Mensagem publicada")
        elif operation == "consume":
            consumer = KafkaConsumer(
                topic,
                bootstrap_servers=self.runtime_credentials.get("servers"),
                auto_offset_reset='earliest',
                consumer_timeout_ms=5000
            )
            messages = []
            for msg in consumer:
                messages.append({
                    "value": msg.value.decode('utf-8'),
                    "key": msg.key.decode('utf-8') if msg.key else None,
                    "offset": msg.offset
                })
            consumer.close()
            yield self.create_json_message({"messages": messages})

---

Casos de Integração

Assistente Inteligente com Banco de Dados

Integre dados empresariais com um assistente de IA do Dify:

# Configuração do workflow
nodes:
  - id: user-query
    type: start
    data:
      inputs:
        - name: question
          type: string
          required: true

  - id: database-lookup
    type: tool
    data:
      provider: database
      tool: DatabaseTool
      inputs:
        query: |-
          SELECT * FROM products
          WHERE description ILIKE '%{{question}}%'

  - id: generate-response
    type: llm
    data:
      prompt: |-
        Dados da consulta:
        {{#database-lookup.output#}}
        
        Com base nos dados acima, responda: {{question}}

Pipeline de Processamento de Arquivos

Use armazenamento em nuvem e filas para construir um pipeline:

nodes:
  - id: file-upload
    type: tool
    data:
      provider: s3
      tool: S3Tool
      inputs:
        operation: upload
        bucket: uploads
        key: "{{$context.date}}/{{$context.file.name}}"

  - id: send-task
    type: tool
    data:
      provider: rabbitmq
      tool: RabbitMQTool
      inputs:
        operation: publish
        queue: file-processing
        message: |-
          {
            "bucket": "uploads",
            "key": "{{file-upload.output.key}}",
            "timestamp": "{{$context.date}}"
          }

Configuração de Ferramentas MCP

Conecte-se a ferramentas MCP externas:

nodes:
  - id: mcp-search
    type: tool
    data:
      provider_type: mcp
      server_config:
        type: stdio
        command: python
        args: ["mcp_search_server.py"]
      tools:
        - name: web_search
          inputs:
            query: "{{user_query}}"
            max_results: 5

---

Solução de Problemas

Problema	Causa Possível	Solução
Falha na conexão da API	Chave de API incorreta	Verifique a chave de API e as permissões
Timeout do Webhook	Rede instável	Aumente o tempo limite ou adicione mecanismo de retry
Ferramenta MCP não encontrada	Nome da ferramenta incorreto	Verifique a lista de ferramentas do servidor MCP
Conexão com banco de dados recusada	Configuração de segurança incorreta	Verifique a lista de permissões de IP e as configurações de firewall

Problemas Comuns

Erro na Integração de API

erro: requests.exceptions.ConnectionError
solução: Verifique se a URL de API está correta e se a rede está acessível

Falha no Webhook

{
  "error": "Assinatura inválida",
  "solution": "Verifique se o token do Webhook está configurado e passado corretamente"
}

Problema no MCP

# erro: mcp.client.session.McpError: Ferramenta não encontrada
# solução: Confirme que o nome da ferramenta no servidor MCP corresponde ao nome usado na chamada

Depuração

Ative logs detalhados para depuração:

import logging
logging.basicConfig(level=logging.DEBUG)

# Inicia o servidor MCP com logs
server = DifyMCPServer(client)
logging.debug("Servidor MCP iniciado")

---

Licença

Este skill está sob a licença MIT. Consulte o arquivo LICENSE para mais detalhes.