AI Guardrails:Integrate with the Dify platform, Integrate with the Dify platform

Download the plugin

• Plugin name: AI Guardrails-Dify Plugin

• Version number: 1.0.2

• Release date: 2025-08-05

• Download link: ai_guardrails.difypkg

Install the plugin

1. On the plugin page, click Install Plugin.

2. Select Local Plugin and upload the downloaded AI Guardrails plugin.

Configure AI Guardrails in the console

Before you run the workflow in production, configure the protection features in the AI Guardrails console. For example, turn on the detection switches for sensitive data detection and prompt attack detection, or configure detailed check items for each protection dimension. For more configuration options, see Configure check items.

On the Check Item Configuration page, the protection dimensions include Content Compliance (detects non-compliant information such as politically sensitive content and violent or pornographic content), Sensitive Content (detects personal and enterprise sensitive information, and supports separate configuration management), and Prompt Attack (detects prompt manipulation). You can turn the detection switch on or off for each dimension separately.

Use the plugin in a workflow

1. Large model input risk detection: Before the large language model (LLM) node, add an AI Guardrails plugin node. Set the detection content to input, the modality type to text, and the detection type to input.

2. Large model output risk detection: After the LLM node, add an AI Guardrails node. Set the detection content to the text of the LLM, the modality type to text, and the detection type to output.

Output variables: _finalSuggestion is the overall mitigation suggestion for all check items. Valid values include block, pass, watch, and mask. You can perform subsequent processing for different suggestions as needed.

The complete set of plugin output variables also includes: contentModerationSuggestion (the content compliance suggestion), sensitiveDataSuggestion (the sensitive data suggestion), promptAttackSuggestion (the prompt attack suggestion), maliciousUrlSuggestion (the malicious URL suggestion), maliciousFileSuggestion (the malicious file suggestion), desensitization (the masked text content), and waterMark (the URL of the watermark image). You can handle the suggestion of each detection type differently.

Verify the integration

After you integrate the plugin and configure the check items, you can use the mitigation capabilities of AI Guardrails on the Dify platform. Run the workflow and check the output variables of the AI Guardrails plugin node. You can identify abnormal results in the input and output of the large model and perform subsequent processing for different suggestions as needed.

Deploy the forwarding service

Because each AI Guardrails API call supports a maximum of 2,000 characters, adapt longer input and output as follows:

• Input moderation: Split the input into multiple segments, each no longer than 2,000 characters, and call the AI Guardrails API concurrently.

• Output moderation: For each call, use the most recent 2,000 characters.

The following sample code shows the processing logic and the startup script, and consists of the following key functions:

• dify_receive: Receives API query data from Dify, verifies the Authorization header, decodes the API-Key into an AccessKey pair, and dispatches the request based on the point field.

• split_text: Chunks the text by maximum length and keeps sentences intact where possible.

• create_string_to_sign and create_signature: Construct the string to sign and generate the signature for each API request.

• request: Constructs the request parameters, sends the signed request to the AI Guardrails API, and checks the response.

• handle_app_moderation_input and handle_app_moderation_output: Parse the suggestion of each check item and build the response for Dify, including the flagged status and the action to take.

from fastapi import FastAPI, Body, HTTPException, Header
from pydantic import BaseModel
import base64
from collections.abc import Generator
from typing import Any
import hmac
import hashlib
from urllib.parse import quote
import requests
from datetime import datetime
from datetime import timezone
import uuid
import json
import re
import concurrent.futures

# You can call the service in different regions as needed. Supported regions include Shanghai (cn-shanghai), Beijing (cn-beijing), Hangzhou (cn-hangzhou), and Shenzhen (cn-shenzhen).
SERVICE_URL = "https://green-cip.cn-shanghai.aliyuncs.com"
# Chunk the text when it exceeds this length.
MAX_LENGTH = 2000
# The ServiceCode used to call the input detection and output detection of AI Guardrails.
SERVICE_INPUT = "query_security_check"
SERVICE_OUTPUT = "response_security_check"
ENCODING = "UTF-8"
ISO8601_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
ALGORITHM = "HmacSHA1"

def format_iso8601_date():
    return datetime.now(timezone.utc).strftime(ISO8601_DATE_FORMAT)

def percent_encode(value):
    if value is None:
        return ""
    return (
        quote(value.encode(ENCODING), safe="~").replace("+", "%20").replace("*", "%2A")
    )

def create_signature(string_to_sign, secret):
    secret = secret + "&"
    signature = hmac.new(
        secret.encode(ENCODING), string_to_sign.encode(ENCODING), hashlib.sha1
    ).digest()
    return base64.b64encode(signature).decode(ENCODING)

def create_string_to_sign(http_method, parameters):
    sorted_keys = sorted(parameters.keys())
    canonicalized_query_string = ""

    for key in sorted_keys:
        canonicalized_query_string += (
            "&" + percent_encode(key) + "=" + percent_encode(parameters[key])
        )

    string_to_sign = (
        http_method
        + "&"
        + percent_encode("/")
        + "&"
        + percent_encode(canonicalized_query_string[1:])
    )
    return string_to_sign

def split_text(text: str, max_length: int = 1950) -> list[str]:
    """Chunk the text by max_length and keep sentences intact where possible (recognizes multiple punctuation marks)."""
    segments = []
    while len(text) > max_length:
        # Extract a substring within the current maximum length.
        chunk = text[:max_length]

        # Use a regex to find the position of the last sentence-ending punctuation mark, such as a period, an exclamation mark, or a question mark.
        match = None
        for pattern in [r"[。！？；:\.?!]+"]:  # Match multiple end symbols.
            matches = list(re.finditer(pattern, chunk))
            if matches:
                match = matches[-1]  # Take the last matching item.

        if match:
            cut_point = match.end()  # Include the punctuation mark.
        else:
            cut_point = max_length  # If no match is found, force a cut.

        segments.append(text[:cut_point])
        text = text[cut_point:]

    if text:
        segments.append(text)
    return segments

def request(content_segment, type, aliyun_access_key, aliyun_access_secret):
    print(datetime.now(), f" [{type} request content]-> {content_segment}")
    # 3.1. Construct the request parameters.
    parameters = {
        "Action": "MultiModalGuard",
        "Version": "2022-03-02",
        "AccessKeyId": aliyun_access_key,
        "Timestamp": format_iso8601_date(),
        "SignatureMethod": "HMAC-SHA1",
        "SignatureVersion": "1.0",
        "SignatureNonce": str(uuid.uuid4()),
        "Format": "JSON",
        "Service": (
            SERVICE_INPUT if type == "input" else SERVICE_OUTPUT
        ),
        "ServiceParameters": json.dumps(
            {"content": content_segment}, ensure_ascii=False
        ),
    }

    string_to_sign = create_string_to_sign("POST", parameters)
    signature = create_signature(string_to_sign, aliyun_access_secret)
    parameters["Signature"] = signature

    # 3.2. Send the request.
    response = requests.post(SERVICE_URL, data=parameters)
    body = response.json()
    print(datetime.now(), " [response body]-> ", body)

    if response.status_code != 200:
        raise Exception(
            f"response http status_code not 200. status_code: {response.status_code}, body: {body}"
        )

    if body.get("Code") != 200:
        raise Exception(
            f"response code not 200. code: {body.get('Code')}, body: {body}"
        )

    return body

app = FastAPI()

class InputData(BaseModel):
    point: str
    params: dict = {}

@app.post("/api/dify/receive")
async def dify_receive(data: InputData = Body(...), authorization: str = Header(None)):
    """
    Receive API query data from Dify.
    """
    #print(data)

    auth_scheme, _, api_key = authorization.partition(" ")

    if auth_scheme.lower() != "bearer":
        raise HTTPException(status_code=401, detail="Unauthorized")

    # Decode the api_key.
    try:
        decoded_bytes = base64.b64decode(api_key)
        decoded_str = decoded_bytes.decode("utf-8")
        ak, sk = decoded_str.split(":", 1)
    except Exception as e:
        # If the call fails, throw an exception.
        raise HTTPException(status_code=401, detail=f"Base64 Decode AK/SK fail: {e}")

    point = data.point

    if point == "ping":
        return {"result": "pong"}
    if point == "app.moderation.input":
        return handle_app_moderation_input(params=data.params, ak=ak, sk=sk)
    elif point == "app.moderation.output":
        return handle_app_moderation_output(params=data.params, ak=ak, sk=sk)

    raise HTTPException(status_code=400, detail="Not implemented")

def handle_app_moderation_input(params: dict, ak: str, sk: str):
    app_id = params.get("app_id")
    inputs = params.get("inputs", {})
    query = params.get("query")

    contents = (
        [query] if len(query) <= MAX_LENGTH else split_text(query, MAX_LENGTH - 50)
    )

    # Execute concurrently.
    bodys = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(request, seg, "input", ak, sk) for seg in contents]
        for future in concurrent.futures.as_completed(futures):
            bodys.append(future.result())

    contentModerationSuggestion=""
    sensitiveDataSuggestion=""
    promptAttackSuggestion=""
    maliciousUrlSuggestion=""
    _finalSuggestion="pass"
    desensitization=""

    # Traverse bodys to parse the suggestion of each check item.
    for body in bodys:
        finalSuggestion = body.get("Data", {}).get("Suggestion", "")
        detailList = body.get("Data", {}).get("Detail", [])
        if finalSuggestion and _finalSuggestion!="block" :
            _finalSuggestion = finalSuggestion
        for detail in detailList:
            suggestion = detail.get("Suggestion", "")
            type = detail.get("Type", "")
            if type == "contentModeration":
                if suggestion and contentModerationSuggestion!="block" :
                    contentModerationSuggestion = suggestion
            elif type == "sensitiveData":
                desensitization = detail.get("Result",[])[0].get("Ext",{}).get("Desensitization","")
                if suggestion and sensitiveDataSuggestion!="block" :
                    sensitiveDataSuggestion = suggestion
            elif type == "promptAttack":
                if suggestion and promptAttackSuggestion!="block" :
                    promptAttackSuggestion = suggestion
            elif type == "maliciousUrl":
                if suggestion and maliciousUrlSuggestion!="block" :
                    maliciousUrlSuggestion = suggestion

    # You can return different responses for different scenarios.
    output_response = "Your content violates our usage policy."

    if contentModerationSuggestion=="block":
        output_response = "Your content involves content security."
    elif sensitiveDataSuggestion=="block" or sensitiveDataSuggestion=="mask":
        output_response = "Your content involves sensitive data."
    elif promptAttackSuggestion=="block":
        output_response = "Your content involves prompt attack."
    elif maliciousUrlSuggestion=="block":
        output_response = "Your content involves malicious url."

    flagged = False
    action = "direct_output"

    if _finalSuggestion == "block" :
        flagged = True
    elif sensitiveDataSuggestion=="mask":
        flagged = True
        action = "overridden"
        query = desensitization

    response = {"flagged": flagged, "action": action}

    if flagged:
        if action == "direct_output":
            response["preset_response"] = output_response
        elif action == "overridden":
            response["inputs"] = inputs
            response["query"] = query

    print(response)
    return response

def handle_app_moderation_output(params: dict, ak: str, sk: str):
    app_id = params.get("app_id")
    text = params.get("text", "")
    print(f"handle_app_moderation_output length:{len(text)}")

    # Get the last 2,000 characters. Adjust the size as needed. The size should be larger than the Dify window size.
    if len(text) > MAX_LENGTH:
        content = text[-MAX_LENGTH:]
    else:
        content = text

    # Run the detection.
    body = request(content, "output", ak, sk)

    contentModerationSuggestion=""
    sensitiveDataSuggestion=""
    promptAttackSuggestion=""
    maliciousUrlSuggestion=""
    desensitization=""
    _finalSuggestion=body.get("Data", {}).get("Suggestion", "")
    detailList = body.get("Data", {}).get("Detail", [])
    for detail in detailList:
        suggestion = detail.get("Suggestion", "")
        type = detail.get("Type", "")
        if type == "contentModeration":
            contentModerationSuggestion = suggestion
        elif type == "sensitiveData":
            desensitization = detail.get("Result",[])[0].get("Ext",{}).get("Desensitization","")
            sensitiveDataSuggestion = suggestion
        elif type == "promptAttack":
            promptAttackSuggestion = suggestion
        elif type == "maliciousUrl":
            maliciousUrlSuggestion = suggestion

    # You can return different responses for different scenarios.
    output_response = "Your content violates our usage policy."

    if contentModerationSuggestion=="block":
        output_response = "Your content involves content security."
    elif sensitiveDataSuggestion=="block":
        output_response = "Your content involves sensitive data."
    elif promptAttackSuggestion=="block":
        output_response = "Your content involves prompt attack."
    elif maliciousUrlSuggestion=="block":
        output_response = "Your content involves malicious url."

    flagged = False
    action = "direct_output"

    if _finalSuggestion == "block":
        flagged = True
    elif sensitiveDataSuggestion=="mask":
        flagged = True
        action = "overridden"

    response = {"flagged": flagged, "action": action}

    if flagged:
        if action == "direct_output":
            response["preset_response"] = output_response
        elif action == "overridden":
            response["text"] = desensitization

    print(response)
    return response

if __name__ == "__main__":
    import uvicorn
    # You can select a custom open port.
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

Save the preceding Python code to a file named main.py, install the dependencies, and start the service:

# Example startup script

pip install fastapi uvicorn requests

uvicorn main:app --reload --host 0.0.0.0

The uvicorn command starts the service defined in main.py. Alternatively, run python main.py to start the service through the built-in startup entry at the end of the script. The built-in entry uses port 8000 by default, and you can select a custom open port in the code.

In the preceding sample code for output content moderation, the default behavior is to reply with a rejection: when content is blocked, the response sets the action field to direct_output and includes a preset_response. You can change the returned action field to overridden to switch to content replacement mode, which replaces the hit keywords or sensitive data with asterisks (*).

For more information, see API guide and the Sensitive content moderation documentation of the Dify platform.

Add an API extension

Choose Settings > API Extension to add an API extension:

• API Endpoint: Enter the accessible endpoint of the deployed forwarding service script.

• API-Key: Enter the Base64 string generated by concatenating your Alibaba Cloud AccessKey ID and AccessKey secret with a colon (:). The pseudocode is base64({aliyun_accessKey_id}:{aliyun_accessKey_secret}).

You can use the following sample code to generate the Base64 string:

import base64

# AccessKey ID and AccessKey secret
access_key_id = ""
access_key_secret = ""

# Concatenate and encode
auth_str = f"{access_key_id}:{access_key_secret}"
encoded_auth = base64.b64encode(auth_str.encode('utf-8')).decode('utf-8')

print(encoded_auth)

Configure the API extension in the Agent

Configure the API extension in the Agent to complete the integration:

1. At the bottom of the Agent page, find the Feature Enabled status bar and click the Manage entry on its right side to configure content moderation.

2. Select API Extension.

3. Select the API extension that you created for AI Guardrails.

4. Turn the switches for input and output content on or off based on your business requirements. In the dialog box, you can set both Moderate Input Content and Moderate Output Content to the Return Preset Response via API mode.

5. After you complete the configuration, click Save.

Verify the integration

When a conversation runs in the Agent, the forwarding service logs each moderation request and response. The following example shows the result of integrating AI Guardrails into a Dify Agent by extending the content moderation API:

point='app.moderation.input' params={'app_id': '4a1a3c82-...', 'inputs': {}, 'query': 'Who are you'}
2025-05-29 11:57:05  [input request content]-> Who are you
2025-05-29 11:57:05  [response body]-> {'RequestId': '...', 'Message': 'OK', 'Data': {'Suggestion': 'pass', 'SensitiveLevel': 'S8'}, 'Code': 200}
{'flagged': False, 'action': 'direct_output'}
INFO:     30.221.145.163:00695 - "POST /api/dify/receive HTTP/1.1" 200 OK

point='app.moderation.output' params={'app_id': '4a1a3c82-...', 'text': '<think>xxx</think> ...'}
handle_app_moderation_output length:297
2025-05-29 11:57:18  [output request content] => <think>...
2025-05-29 11:57:18  [response body]-> {'RequestId': '...', 'Message': 'OK', 'Data': {'Suggestion': 'pass'}, 'Code': 200}
{'flagged': False, 'action': 'direct_output'}
INFO:     30.221.145.163:00695 - "POST /api/dify/receive HTTP/1.1" 200 OK

In this example, the AI Guardrails API returns Code 200 and Suggestionpass for both the input and the output. The forwarding service responds with flagged set to false, which means that the content passes moderation and the conversation continues.

If the Authorization header is not in the Bearer format, or if the API-Key cannot be Base64-decoded into an AccessKey pair, the forwarding service returns a 401 status code. In this case, check the API-Key that you configured for the API extension, and make sure that your Dify instance can access the endpoint of the forwarding service.

Dify Community Edition installation fails

If you encounter the following error when you install Dify Community Edition or another private deployment version, resolve it by disabling signature verification or by using a signed version.

The page displays an Upload Failed prompt, and the error message is PluginDaemonBadRequestError: plugin verification has been enabled, and the plugin you want to install has a bad signature.

Solution 1: Disable signature verification

cd ${dify_path}/docker
# Stop
docker compose down
vi .env
# Change FORCE_VERIFYING_SIGNATURE to false
FORCE_VERIFYING_SIGNATURE=false
# Restart
docker compose up -d

Solution 2: Use the signed version

For more information, see the Third-party signature topic in the official Dify platform documentation.

1. Download the signed version and the public key used for signing:

   • ai_guardrails.signed.difypkg

   • dify_sign.public.pem

2. Place the public key in a location that the plugin daemon can access. For example, create a public_keys directory under docker/volumes/plugin_daemon and copy the public key file to the corresponding path:

   mkdir docker/volumes/plugin_daemon/public_keys
   cp dify_sign.public.pem docker/volumes/plugin_daemon/public_keys

3. Modify docker-compose.yaml:

   services:
     plugin_daemon:
       environment:
         FORCE_VERIFYING_SIGNATURE: true
         THIRD_PARTY_SIGNATURE_VERIFICATION_ENABLED: true
         THIRD_PARTY_SIGNATURE_VERIFICATION_PUBLIC_KEYS: /app/storage/public_keys/dify_sign.public.pem

   docker/volumes/plugin_daemon is mounted to /app/storage in the plugin_daemon container. Make sure that the path specified in THIRD_PARTY_SIGNATURE_VERIFICATION_PUBLIC_KEYS corresponds to the path inside the container.

4. Restart the container:

   cd docker
   docker compose down
   docker compose up -d

   After the service is restarted, third-party signature verification is enabled in the current Community Edition environment.