SystemTce/dify-skills

Dify Integration and Extension SKILL

Integration guide with external systems and services

Quick Navigation

• API Integration
• MCP Protocol
• External Service Integration
• Integration Examples
• Troubleshooting

---

API Integration

Dify provides a complete REST API supporting application integration and workflow orchestration.

Core API Endpoints

Endpoint	Method	Description
/v1/chat-messages	POST	Send a chat message
/v1/completions	POST	Send a completion request
/v1/workflows/run	POST	Run a workflow
/v1/parameters	GET	Get application parameters
/v1/file-upload	POST	Upload a file

REST API Integration

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):
        """Send a chat message"""
        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):
        """Run a 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()

Usage Example:

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

# Send a chat message
result = client.chat_message(
    query="Hello",
    user="user123",
    conversation_id=None  # New conversation
)
print(result)

Webhook Integration

Webhooks are used to receive notifications on specific events, supporting workflow triggers and event callbacks.

Workflow Webhook Trigger

# Workflow configuration
nodes:
  - id: webhook-trigger
    data:
      type: webhook-trigger
      method: POST
      path: /webhook/my-workflow
      parameters:
        - name: data
          type: string
          required: true

Webhook Callback Handling

from flask import Flask, request, jsonify

app = Flask(__name__)

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

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

    # Process event
    event_type = payload.get('event')
    if event_type == 'workflow.finished':
        # Workflow completion handling
        result = payload.get('data', {}).get('output', {})
        process_result(result)

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

Webhook Event Types

Event	Description	Trigger Condition
workflow.started	Workflow started	When workflow starts
workflow.finished	Workflow completed	When workflow finishes normally
workflow.failed	Workflow failed	When workflow execution fails
node.started	Node started	When a node starts executing
node.finished	Node completed	When a node finishes execution

Streaming Response

Supports Server-Sent Events (SSE) for real-time streaming output.

import requests
import json

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

# Usage example
for event in stream_chat(client, "Tell a story", "user123"):
    if event.get('event') == 'message':
        print(event.get('data', {}).get('answer'), end='', flush=True)

SSE Event Format:

{
  "event": "message",
  "task_id": "abc123",
  "message_id": "msg_xxx",
  "conversation_id": "conv_xxx",
  "answer": "Generated text content",
  "created_at": 1234567890
}

---

MCP Protocol

Model Context Protocol (MCP) is a standardized protocol for integrating AI assistants with external systems.

MCP Server Implementation

Dify can act as an MCP Server, providing tools and resources for AI to call.

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):
        """Start the MCP Server"""
        @self.server.list_tools()
        async def list_tools() -> list[Tool]:
            return [
                Tool(
                    name="dify_chat",
                    description="Chat with the Dify application",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "User query"},
                            "user": {"type": "string", "description": "User identifier"}
                        },
                        "required": ["query", "user"]
                    }
                ),
                Tool(
                    name="dify_run_workflow",
                    description="Run a Dify workflow",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "inputs": {"type": "object", "description": "Workflow input parameters"},
                            "user": {"type": "string", "description": "User identifier"}
                        },
                        "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}")

        # Start the service
        await stdio_server(self.server.run())

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

MCP Client Implementation

Dify can call tools provided by external MCP Servers.

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):
        """Call an MCP tool"""
        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

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

asyncio.run(main())

Bidirectional Communication Configuration

Configure MCP nodes in the Dify workflow:

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

---

External Service Integration

Database Integration

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]:
        # Get connection parameters
        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")

        # Establish connection
        conn = psycopg2.connect(
            host=host,
            port=port,
            database=database,
            user=user,
            password=password
        )

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

            # Get results
            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")

Storage Service Integration

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})

Message Queue Integration

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.create_text_message("Message published")
        elif operation == "consume":
            consumer = KafkaConsumer(
                topic,
                bootstrap_servers=self.runtime_credentials.get("servers"),
                auto_offset_reset='earliest',
                consumer_timeout_ms=1000
            )
            messages = []
            for msg in consumer:
                messages.append(msg.value.decode('utf-8'))
            consumer.close()
            yield self.create_json_message({"messages": messages})

---

Integration Examples

Example 1: Chatbot with Knowledge Base Query

Integrate Dify with an external knowledge base:

nodes:
  - id: input
    type: start
    data:
      inputs:
        - name: query
          type: string
  - id: search-knowledge
    type: tool
    data:
      provider: external-api
      tool: search
      inputs:
        query: "{{inputs.query}}"
  - id: generate-response
    type: llm
    data:
      prompt: |
        Based on the following information, answer the user's question:
        Knowledge: {{search-knowledge.output}}
        Question: {{inputs.query}}

Example 2: Automated Data Processing Pipeline

nodes:
  - id: trigger
    type: webhook-trigger
    data:
      path: /data-pipeline
  - id: fetch-data
    type: tool
    data:
      provider: database
      tool: query
      inputs:
        query: "SELECT * FROM source_data WHERE processed = false"
  - id: transform
    type: code
    data:
      code: |
        def transform(data):
            return [{
                "id": item["id"],
                "processed_value": item["raw_value"].upper()
            } for item in data]
  - id: store
    type: tool
    data:
      provider: database
      tool: insert
      inputs:
        data: "{{transform.output}}"
        table: "processed_data"

Example 3: MCP Tool Aggregation

nodes:
  - id: user-query
    type: start
  - id: mcp-router
    type: llm
    data:
      prompt: |
        Based on the user query, decide which tool to call:
        - For weather, call the weather tool
        - For news, call the news tool
        User query: {{user-query.output}}
        Output: "tool_name" and "parameters" JSON
  - id: call-tool
    type: tool
    data:
      provider_type: mcp
      server: external-mcp-server
      tool: "{{mcp-router.output.tool_name}}"
      inputs: "{{mcp-router.output.parameters}}"

---

Troubleshooting

Common Issues and Solutions

Issue	Possible Cause	Solution
API timeout	Network latency or server overload	Increase timeout, use async requests
Invalid API key	Incorrect or expired key	Regenerate key in console
Webhook not triggering	Path mismatch or firewall	Check path, ensure network access
Streaming interrupted	Connection issues	Enable reconnection, increase buffer
MCP connection refused	Service not running	Start MCP service, check port
Database connection failed	Incorrect credentials	Verify connection parameters

Debugging Checklist

1. Check API availability

   curl -X GET "https://api.dify.ai/v1/parameters" \
     -H "Authorization: Bearer app-xxxxx"
   

2. Verify Webhook endpoint

   curl -X POST "https://your-server.com/webhook/test" \
     -H "Content-Type: application/json" \
     -d '{"test": true}'
   

3. Test MCP connectivity

   echo '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' | \
     python mcp_server.py
   

Optimization Suggestions

• Connection pooling: Reuse database connections to improve performance
• Cache strategy: Use Redis caching for frequently queried data
• Rate limiting: Control API request frequency to avoid overload
• Error handling: Implement retry logic for timeout scenarios
• Logging: Record all integration calls for auditing and debugging

---

References

• Dify API Documentation
• MCP Protocol Specification
• Plugin Development Guide