Dify adapter

For Dify — but different in shape from every other adapter. Dify tools aren't in-process Python objects passed to a constructor; they're deployable plugin artifacts: a Python class subclassing dify_plugin.tool.Tool plus a manifest.yaml.

The adapter offers two paths:

Path	Use when
GovernedDifyTool base class	Building a real, deployable Dify plugin. Recommended.
DifyAdapter.dify_tools()	Adapter-level governance unit tests only — output is a list of dicts that cannot be deployed to a real Dify instance

Install

pip install kitelogik dify-plugin-sdk

dify-plugin-sdk is not a hard dependency — the module imports cleanly without it for unit testing. Real plugin deployments need it.

Path 1 — GovernedDifyTool (recommended)

Subclass GovernedDifyTool instead of dify_plugin.tool.Tool, and override _invoke_governed(tool_parameters) instead of _invoke. The base class wraps your method with the policy pipeline:

from kitelogik import OPAClient, PolicyGate, SessionContext
from kitelogik.adapters.dify import GovernedDifyTool

class GetCustomerTool(GovernedDifyTool):
    # Configure governance via class attributes (or set in __init__).
    gate = PolicyGate(opa_client=OPAClient())
    context = SessionContext(
        session_id="dify_sess_001",
        user_role="analyst",
        session_scopes=["read_customer"],
    )
    action = "get_customer"

    def _invoke_governed(self, tool_parameters):
        customer_id = tool_parameters["customer_id"]
        # Run only if governance allows
        yield self.create_text_message(f"record:{customer_id}")

Ship the resulting class as your plugin. Dify's runtime calls _invoke(...), which:

1. Builds a ToolCallInput from the registered action + tool params
2. Evaluates the policy gate
3. On deny, yields a single text message with {"blocked": true, "reason": ...} JSON
4. On allow, calls your _invoke_governed(...) and yields each result message
5. Sanitises the text payload of each yielded ToolInvokeMessage

Configuration attributes

Class attribute	Required	Default
gate	yes	None (raises RuntimeError if not set)
context	yes	None (raises RuntimeError if not set)
action	no	falls back to self.__class__.__name__
sanitize	no	True
deny_message	no	"Action blocked by governance policy."

Helper — make_governed_dify_tool

Have a plain Python function and want it deployable as a Dify plugin without writing a class? make_governed_dify_tool builds a GovernedDifyTool subclass dynamically:

from kitelogik.adapters.dify import make_governed_dify_tool

def get_customer(customer_id: str) -> str:
    return f"Customer {customer_id}: Acme Corp"

GetCustomerTool = make_governed_dify_tool(
    get_customer,
    gate=gate,
    context=context,
    action="get_customer",                   # defaults to fn.__name__
    deny_message="Customer lookup blocked.",  # optional override
)

# `GetCustomerTool` is a class. Instantiate it inside the Dify plugin runtime.

The returned class inherits from GovernedDifyTool, has the right _invoke_governed body, and is named after the function (or the name= kwarg).

Path 2 — DifyAdapter.dify_tools (testing only)

For governance unit tests where you don't want to spin up the Dify plugin runtime:

from kitelogik.adapters.dify import DifyAdapter

adapter = DifyAdapter(gate=gate, context=context)
adapter.register("get_customer", get_customer, description="Look up a customer")

descriptors = adapter.dify_tools()
# [{"name": "get_customer", "description": "...", "function": <governed callable>}]

These dicts cannot be deployed — Dify's plugin loader expects the Tool subclass shape, not dicts. Use BaseGovernedAdapter.execute() or call descriptor["function"](**args) directly in tests.

What happens on a deny

GovernedDifyTool._invoke yields a single Dify text message (ToolInvokeMessage of type TEXT) carrying the JSON payload:

{"blocked": true, "reason": "Action blocked by governance policy."}

For unit tests where dify_plugin isn't installed, the same shape is returned as a plain dict ({"type": "text", "text": "<json>"}) so test assertions don't need to depend on the SDK.

Source

kitelogik/adapters/dify.py on GitHub.