Reverse invoking an App means that a plugin can access data from an App within Dify. This module supports both streaming and non-streaming App calls. If you are unfamiliar with the basic concepts of reverse invocation, please first read Reverse Invocation of Dify Services.

Interface Types:

  • For Chatbot/Agent/Chatflow type applications, they are all chat-based applications and thus share the same input and output parameter types. Therefore, they can be uniformly treated as the Chat Interface.
  • For Workflow applications, they occupy a separate Workflow Interface.
  • For Completion (text generation application) applications, they occupy a separate Completion Interface.

Please note that plugins are only allowed to access Apps within the Workspace where the plugin resides.

​
Calling the Chat Interface

​
Entry Point

    self.session.app.chat

​
Interface Specification

    def invoke(
        self,
        app_id: str,
        inputs: dict,
        response_mode: Literal["streaming", "blocking"],
        conversation_id: str,
        files: list,
    ) -> Generator[dict, None, None] | dict:
        pass

When response_mode is streaming, this interface will directly return Generator[dict]. Otherwise, it returns dict. For specific interface fields, please refer to the return results of ServiceApi.

​
Use Case

We can call a Chat type App within an Endpoint and return the result directly.

import json
from typing import Mapping
from werkzeug import Request, Response
from dify_plugin import Endpoint

class Duck(Endpoint):
    def _invoke(self, r: Request, values: Mapping, settings: Mapping) -> Response:
        """
        Invokes the endpoint with the given request.
        """
        app_id = values["app_id"]

        def generator():
            # Note: The original example incorrectly called self.session.app.workflow.invoke
            # It should call self.session.app.chat.invoke for a chat app.
            # Assuming a chat app is intended here based on the section title.
            response = self.session.app.chat.invoke(
                app_id=app_id, 
                inputs={}, # Provide actual inputs as needed
                response_mode="streaming", 
                conversation_id="some-conversation-id", # Provide a conversation ID if needed
                files=[]
            )

            for data in response:
                yield f"{json.dumps(data)} <br>"

        return Response(generator(), status=200, content_type="text/html")

​
Calling the Workflow Interface

​
Entry Point

    self.session.app.workflow

​
Interface Specification

    def invoke(
        self,
        app_id: str,
        inputs: dict,
        response_mode: Literal["streaming", "blocking"],
        files: list,
    ) -> Generator[dict, None, None] | dict:
        pass

​
Calling the Completion Interface

​
Entry Point

    self.session.app.completion

Interface Specification

    def invoke(
        self,
        app_id: str,
        inputs: dict,
        response_mode: Literal["streaming", "blocking"],
        files: list,
    ) -> Generator[dict, None, None] | dict:
        pass

Edit this page | Report an issue