Skip to content

Reverse Invocation App

Reverse invoking an App means that a plugin can access data from an App within FlexAI. 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 FlexAI 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.

from typing import Mapping
from werkzeug import Request, Response
from flexai_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

{/ Contributing Section DO NOT edit this section! It will be automatically generated by the script. /}

Edit this page | Report an issue