Dify Integrations- und Erweiterungs-SKILL
Leitfaden zur Integration mit externen Systemen und Diensten
Schnellnavigation
API-Integration
Dify bietet eine vollständige REST-API, die Anwendungsintegration und Workflow-Orchestrierung unterstützt.
Wichtige API-Endpunkte
| Endpunkt | Methode | Beschreibung |
|---|---|---|
/v1/chat-messages | POST | Senden einer Chat-Nachricht |
/v1/completions | POST | Senden einer Vervollständigungsanfrage |
/v1/workflows/run | POST | Ausführen eines Workflows |
/v1/parameters | GET | Abrufen von Anwendungsparametern |
/v1/file-upload | POST | Datei-Upload |
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):
"""Senden einer Chat-Nachricht"""
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):
"""Ausführen eines Workflows"""
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()
Anwendungsbeispiel:
# Initialisieren des Clients
client = DifyClient(
base_url="https://api.dify.ai/v1",
api_key="app-xxxxx"
)
# Senden einer Chat-Nachricht
result = client.chat_message(
query="Hallo",
user="user123",
conversation_id=None # Neue Sitzung
)
print(result)
Webhook-Integration
Webhooks werden verwendet, um bei bestimmten Ereignissen Benachrichtigungen zu empfangen. Sie unterstützen Workflow-Trigger und Ereignis-Rückrufe.
Workflow-Webhook-Trigger
# Workflow-Konfiguration
nodes:
- id: webhook-trigger
data:
type: webhook-trigger
method: POST
path: /webhook/my-workflow
parameters:
- name: data
type: string
required: true
Webhook-Rückrufverarbeitung
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/webhook/callback', methods=['POST'])
def handle_webhook():
"""Verarbeitung eines Dify-Webhook-Rückrufs"""
payload = request.json
# Signatur überprüfen
signature = request.headers.get('X-Dify-Signature')
if not verify_signature(payload, signature):
return jsonify({"error": "Ungültige Signatur"}), 401
# Ereignis verarbeiten
event_type = payload.get('event')
if event_type == 'workflow.finished':
# Workflow-Abschluss verarbeiten
result = payload.get('data', {}).get('output', {})
process_result(result)
return jsonify({"status": "Erfolg"})
Webhook-Ereignistypen
| Ereignis | Beschreibung | Auslösezeitpunkt |
|---|---|---|
workflow.started | Workflow gestartet | Beim Start des Workflows |
workflow.finished | Workflow abgeschlossen | Bei normalem Abschluss des Workflows |
workflow.failed | Workflow fehlgeschlagen | Bei fehlgeschlagener Workflow-Ausführung |
node.started | Knoten gestartet | Beim Start der Knotenausführung |
node.finished | Knoten abgeschlossen | Bei abgeschlossener Knotenausführung |
Streaming-Antworten
Unterstützt Server-Sent Events (SSE) für Echtzeit-Streaming-Ausgaben.
import requests
import json
def stream_chat(client: DifyClient, query: str, user: str):
"""Streaming-Chat-Anfrage"""
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-Format analysieren
data = line.decode('utf-8')
if data.startswith('data: '):
event = json.loads(data[6:])
yield event
# Anwendungsbeispiel
for event in stream_chat(client, "Erzähl eine Geschichte", "user123"):
if event.get('event') == 'message':
print(event.get('data', {}).get('answer'), end='', flush=True)
SSE-Ereignisformat:
{
"event": "message",
"task_id": "abc123",
"message_id": "msg_xxx",
"conversation_id": "conv_xxx",
"answer": "Generierter Textinhalt",
"created_at": 1234567890
}
MCP-Protokoll
Das Model Context Protocol (MCP) ist ein standardisiertes Protokoll für die Integration von KI-Assistenten mit externen Systemen.
MCP-Server-Implementierung
Dify kann als MCP-Server fungieren und Werkzeuge sowie Ressourcen für KI-Aufrufe bereitstellen.
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-Server starten"""
@self.server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="dify_chat",
description="Führt einen Chat mit der Dify-Anwendung",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Benutzerabfrage"},
"user": {"type": "string", "description": "Benutzerkennung"}
},
"required": ["query", "user"]
}
),
Tool(
name="dify_run_workflow",
description="Führt einen Dify-Workflow aus",
inputSchema={
"type": "object",
"properties": {
"inputs": {"type": "object", "description": "Workflow-Eingabeparameter"},
"user": {"type": "string", "description": "Benutzerkennung"}
},
"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"Unbekanntes Werkzeug: {name}")
# Dienst starten
await stdio_server(self.server.run())
# Server ausführen
if __name__ == "__main__":
client = DifyClient("https://api.dify.ai/v1", "app-xxxxx")
server = DifyMCPServer(client)
asyncio.run(server.start())
MCP-Client-Implementierung
Dify kann Werkzeuge von externen MCP-Servern aufrufen.
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-Werkzeug aufrufen"""
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
# Anwendungsbeispiel
async def main():
client = DifyMCPClient(
server_command="node",
server_args=["mcp-server.js"]
)
result = await client.call_tool("get_weather", {"city": "Berlin"})
print(result)
asyncio.run(main())
Bidirektionale Kommunikationskonfiguration
Konfiguration eines MCP-Knotens im 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: Suchwerkzeug
input_schema:
type: object
properties:
query:
type: string
Externe Diensteinbindung
Datenbankintegration
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]:
# Verbindungsparameter abrufen
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")
# Verbindung herstellen
conn = psycopg2.connect(
host=host,
port=port,
database=database,
user=user,
password=password
)
try:
# Abfrage ausführen
query = tool_parameters.get("query")
cursor = conn.cursor()
cursor.execute(query)
# Ergebnis abrufen
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"betroffene Zeilen: {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("gelöscht")
Speicherdienstintegration
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"Hochgeladen nach {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})
Nachrichtenwarteschlangenintegration
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("Nachricht veröffentlicht")
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("Keine Nachricht")
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("Nachricht veröffentlicht")
elif operation == "consume":
consumer = KafkaConsumer(
topic,
bootstrap_servers=self.runtime_credentials.get("servers"),
auto_offset_reset='earliest',
enable_auto_commit=True,
group_id='dify-consumer'
)
for message in consumer:
value = json.loads(message.value.decode('utf-8'))
yield self.create_json_message(value)
break
consumer.close()