Flow Action

kairon/actions/definitions/factory.py

@@ -2,6 +2,7 @@
 
 from kairon.actions.definitions.bot_response import ActionKaironBotResponse
 from kairon.actions.definitions.email import ActionEmail
+from kairon.actions.definitions.flow import ActionFlow
 from kairon.actions.definitions.form_validation import ActionFormValidation
 from kairon.actions.definitions.google import ActionGoogleSearch
 from kairon.actions.definitions.http import ActionHTTP
@@ -41,7 +42,8 @@ class ActionFactory:
         ActionType.pyscript_action.value: ActionPyscript,
         ActionType.database_action.value: ActionDatabase,
         ActionType.web_search_action.value: ActionWebSearch,
-        ActionType.live_agent_action.value: ActionLiveAgent
+        ActionType.live_agent_action.value: ActionLiveAgent,
+        ActionType.flow_action.value: ActionFlow
     }
 
     @staticmethod

kairon/actions/definitions/flow.py

@@ -0,0 +1,93 @@
+from typing import Text, Dict, Any
+
+from loguru import logger
+from mongoengine import DoesNotExist
+from rasa_sdk import Tracker
+from rasa_sdk.executor import CollectingDispatcher
+
+from kairon.actions.definitions.base import ActionsBase
+from kairon.shared.actions.data_objects import FlowActionConfig, ActionServerLogs
+from kairon.shared.actions.exception import ActionFailure
+from kairon.shared.actions.models import ActionType
+from kairon.shared.actions.utils import ActionUtility
+from kairon.shared.constants import KaironSystemSlots
+
+
+class ActionFlow(ActionsBase):
+
+    def __init__(self, bot: Text, name: Text):
+        """
+        Initialize Flow action.
+        @param bot: bot id
+        @param name: action name
+        """
+        self.bot = bot
+        self.name = name
+
+    def retrieve_config(self):
+        """
+        Fetch Flow action configuration parameters from the database
+        :return: FlowActionConfig containing configuration for the action as a dict.
+        """
+        try:
+            http_config_dict = FlowActionConfig.objects().get(bot=self.bot,
+                                                              name=self.name,
+                                                              status=True).to_mongo().to_dict()
+            logger.debug("flow_action_config: " + str(http_config_dict))
+            return http_config_dict
+        except DoesNotExist as e:
+            logger.exception(e)
+            raise ActionFailure("No Flow action found for given action and bot")
+
+    async def execute(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]):
+        """
+        Retrieves action config and executes it.
+        Information regarding the execution is logged in ActionServerLogs.
+        @param dispatcher: Client to send messages back to the user.
+        @param tracker: Tracker object to retrieve slots, events, messages and other contextual information.
+        @param domain: Bot domain
+        :return: Dict containing slot name as keys and their values.
+        """
+        bot_response = None
+        http_response = None
+        exception = None
+        status = "SUCCESS"
+        http_url = None
+        headers = {}
+        body = {}
+        dispatch_response = True
+        try:
+            flow_action_config = self.retrieve_config()
+            bot_response = flow_action_config['response']
+            dispatch_response = flow_action_config['dispatch_response']
+            body, http_url, headers = ActionUtility.prepare_flow_body(flow_action_config, tracker)
+            http_response, status_code, time_elapsed = await ActionUtility.execute_request_async(
+                headers=headers, http_url=http_url, request_method="POST", request_body=body
+            )
+            logger.info("response: " + str(http_response))
+            logger.info("status_code: " + str(status_code))
+            logger.info("time_elapsed: " + str(time_elapsed))
+        except Exception as e:
+            exception = str(e)
+            logger.exception(e)
+            status = "FAILURE"
+            bot_response = "I have failed to process your request"
+        finally:
+            if dispatch_response:
+                dispatcher.utter_message(bot_response)
+            ActionServerLogs(
+                type=ActionType.flow_action.value,
+                intent=tracker.get_intent_of_latest_message(skip_fallback_intent=False),
+                action=self.name,
+                sender=tracker.sender_id,
+                bot=tracker.get_slot("bot"),
+                url=http_url,
+                headers=headers,
+                request_params=body,
+                exception=exception,
+                api_response=str(http_response) if http_response else None,
+                bot_response=bot_response,
+                status=status,
+                user_msg=tracker.latest_message.get('text')
+            ).save()
+        return {KaironSystemSlots.kairon_action_response.value: bot_response}

kairon/api/app/routers/bot/action.py

@@ -7,7 +7,7 @@
     HttpActionConfigRequest, SlotSetActionRequest, EmailActionRequest, GoogleSearchActionRequest, JiraActionRequest,
     ZendeskActionRequest, PipedriveActionRequest, HubspotFormsActionRequest, TwoStageFallbackConfigRequest,
     RazorpayActionRequest, PromptActionConfigRequest, DatabaseActionRequest, PyscriptActionRequest,
-    WebSearchActionRequest, LiveAgentActionRequest
+    WebSearchActionRequest, LiveAgentActionRequest, FlowActionRequest
 )
 from kairon.shared.constants import TESTER_ACCESS, DESIGNER_ACCESS
 from kairon.shared.models import User
@@ -66,6 +66,42 @@ async def update_http_action(
     return Response(data=response, message=message)
 
 
+@router.post("/flow", response_model=Response)
+async def add_flow_action(
+        request_data: FlowActionRequest,
+        current_user: User = Security(Authentication.get_current_user_and_bot, scopes=DESIGNER_ACCESS)
+):
+    """
+    Stores the flow action config
+    """
+    flow_config_id = mongo_processor.add_flow_action(request_data.dict(), current_user.get_user(),
+                                                     current_user.get_bot())
+    return Response(data={"_id": flow_config_id}, message="Action added!")
+
+
+@router.put("/flow", response_model=Response)
+async def update_flow_action(
+        request_data: FlowActionRequest,
+        current_user: User = Security(Authentication.get_current_user_and_bot, scopes=DESIGNER_ACCESS)
+):
+    """
+    Updates the flow action config
+    """
+    action_id = mongo_processor.update_flow_action(request_data=request_data.dict(), user=current_user.get_user(),
+                                                   bot=current_user.get_bot())
+    return Response(data={"_id": action_id}, message="Action updated!")
+
+
+@router.get("/flow", response_model=Response)
+async def list_flow_actions(
+        current_user: User = Security(Authentication.get_current_user_and_bot, scopes=TESTER_ACCESS)):
+    """
+    Returns list of flow actions for bot.
+    """
+    actions = list(mongo_processor.list_flow_actions(bot=current_user.get_bot()))
+    return Response(data=actions)
+
+
 @router.post("/pyscript", response_model=Response)
 async def add_pyscript_action(
         request_data: PyscriptActionRequest,

kairon/api/models.py

@@ -22,7 +22,8 @@
     EvaluationType,
     DispatchType,
     DbQueryValueType,
-    DbActionOperationType, UserMessageType
+    DbActionOperationType, UserMessageType,
+    FlowModes, FlowActionTypes
 )
 from ..shared.constants import SLOT_SET_TYPE, FORM_SLOT_SET_TYPE
 
@@ -897,6 +898,62 @@ def validate_top_n(cls, v, values, **kwargs):
         return v
 
 
+class FlowActionRequest(BaseModel):
+    name: constr(to_lower=True, strip_whitespace=True)
+    flow_id: CustomActionParameter
+    header: str = None
+    body: str
+    footer: str = None
+    mode: FlowModes = FlowModes.published.value
+    flow_action: FlowActionTypes = FlowActionTypes.navigate.value
+    flow_token: str
+    recipient_phone: CustomActionParameter
+    initial_screen: str
+    flow_cta: str
+    dispatch_response: bool = True
+    response: str = None
+
+    @validator("name")
+    def validate_name(cls, v, values, **kwargs):
+        from kairon.shared.utils import Utility
+
+        if Utility.check_empty_string(v):
+            raise ValueError("name is required")
+        return v
+
+    @validator("body")
+    def validate_body(cls, v, values, **kwargs):
+        from kairon.shared.utils import Utility
+
+        if Utility.check_empty_string(v):
+            raise ValueError("body is required")
+        return v
+
+    @validator("initial_screen")
+    def validate_initial_screen(cls, v, values, **kwargs):
+        from kairon.shared.utils import Utility
+
+        if Utility.check_empty_string(v):
+            raise ValueError("initial_screen is required")
+        return v
+
+    @validator("flow_cta")
+    def validate_flow_cta(cls, v, values, **kwargs):
+        from kairon.shared.utils import Utility
+
+        if Utility.check_empty_string(v):
+            raise ValueError("flow_cta is required")
+        return v
+
+    @validator("flow_token")
+    def validate_flow_token(cls, v, values, **kwargs):
+        from kairon.shared.utils import Utility
+
+        if Utility.check_empty_string(v):
+            raise ValueError("flow_token is required")
+        return v
+
+
 class EmailActionRequest(BaseModel):
     action_name: constr(to_lower=True, strip_whitespace=True)
     smtp_url: str

kairon/shared/actions/data_objects.py

@@ -27,6 +27,7 @@
     DispatchType,
     DbQueryValueType,
     DbActionOperationType, UserMessageType,
+    FlowModes, FlowActionTypes
 )
 from kairon.shared.constants import SLOT_SET_TYPE, FORM_SLOT_SET_TYPE
 from kairon.shared.data.audit.data_objects import Auditlog
@@ -448,6 +449,45 @@ def clean(self):
             self.custom_text.key = "custom_text"
 
 
+@auditlogger.log
+@push_notification.apply
+class FlowActionConfig(Auditlog):
+    name = StringField(required=True)
+    flow_id = EmbeddedDocumentField(CustomActionRequestParameters)
+    header = StringField(default=None)
+    body = StringField(required=True)
+    footer = StringField(default=None)
+    mode = StringField(default=FlowModes.published.value,
+                       choices=[p_type.value for p_type in FlowModes])
+    flow_action = StringField(default=FlowActionTypes.navigate.value,
+                              choices=[p_type.value for p_type in FlowModes])
+    flow_token = StringField(required=True)
+    recipient_phone = EmbeddedDocumentField(CustomActionRequestParameters)
+    initial_screen = StringField(required=True)
+    flow_cta = StringField(required=True)
+    dispatch_response = BooleanField(default=True)
+    response = StringField(default=None)
+    bot = StringField(required=True)
+    user = StringField(required=True)
+    timestamp = DateTimeField(default=datetime.utcnow)
+    status = BooleanField(default=True)
+
+    def validate(self, clean=True):
+        if clean:
+            self.clean()
+
+        if Utility.check_empty_string(self.name):
+            raise ValidationError("Action name cannot be empty")
+        if Utility.check_empty_string(self.body):
+            raise ValidationError("body cannot be empty")
+        if Utility.check_empty_string(self.initial_screen):
+            raise ValidationError("initial_screen cannot be empty")
+        if Utility.check_empty_string(self.flow_cta):
+            raise ValidationError("flow_cta cannot be empty")
+        if Utility.check_empty_string(self.flow_token):
+            raise ValidationError("flow_token cannot be empty")
+
+
 @auditlogger.log
 @push_notification.apply
 class GoogleSearchAction(Auditlog):

kairon/shared/actions/models.py

@@ -20,6 +20,16 @@ class ActionParameterType(str, Enum):
     key_vault = "key_vault"
 
 
+class FlowModes(str, Enum):
+    draft = "draft"
+    published = "published"
+
+
+class FlowActionTypes(str, Enum):
+    navigate = "navigate"
+    data_exchange = "data_exchange"
+
+
 class EvaluationType(str, Enum):
     expression = "expression"
     script = "script"
@@ -48,6 +58,7 @@ class ActionType(str, Enum):
     database_action = "database_action"
     web_search_action = "web_search_action"
     live_agent_action = "live_agent_action"
+    flow_action = "flow_action"
 
 
 class HttpRequestContentType(str, Enum):

kairon/shared/actions/utils.py

@@ -941,6 +941,57 @@ def prepare_hubspot_form_request(tracker, fields: list, bot: Text):
     def get_basic_auth_str(username: Text, password: Text):
         return requests.auth._basic_auth_str(username, password)
 
+    @staticmethod
+    def prepare_flow_body(flow_action_config: dict, tracker: Tracker):
+        api_key = Sysadmin.get_bot_secret(flow_action_config['bot'], BotSecretType.d360_api_key.value, raise_err=False)
+        http_url = Utility.environment["flow"]["url"]
+        header_key = Utility.environment["flow"]["headers"]["key"]
+        headers = {header_key: api_key}
+        flow_body = {
+            "recipient_type": "individual",
+            "messaging_product": "whatsapp",
+            "type": "interactive",
+            "interactive": {
+                "type": "flow",
+                "action": {
+                    "name": "flow",
+                    "parameters": {
+                        "mode": "published",
+                        "flow_message_version": "3",
+                        "flow_token": "AQAAAAACS5FpgQ_cAAAAAD0QI3s.",
+                                      "flow_action": "navigate",
+                    }
+                }
+            }
+        }
+        parameter_type = flow_action_config['recipient_phone']['parameter_type']
+        recipient_phone = tracker.get_slot(flow_action_config['recipient_phone']['value']) \
+            if parameter_type == 'slot' else flow_action_config['recipient_phone']['value']
+        parameter_type = flow_action_config['flow_id']['parameter_type']
+        flow_id = tracker.get_slot(flow_action_config['flow_id']['value']) \
+            if parameter_type == 'slot' else flow_action_config['flow_id']['value']
+        header = flow_action_config.get('header')
+        body = flow_action_config['body']
+        footer = flow_action_config.get('footer')
+        mode = flow_action_config['mode']
+        flow_action = flow_action_config['flow_action']
+        flow_token = flow_action_config['flow_token']
+        initial_screen = flow_action_config['initial_screen']
+        flow_cta = flow_action_config['flow_cta']
+        flow_body["to"] = recipient_phone
+        if header:
+            flow_body["interactive"]["header"] = {"type": "text", "text": header}
+        flow_body["interactive"]["body"] = {"text": body}
+        if footer:
+            flow_body["interactive"]["footer"] = {"text": footer}
+        flow_body["interactive"]["action"]["parameters"]["mode"] = mode
+        flow_body["interactive"]["action"]["parameters"]["flow_action"] = flow_action
+        flow_body["interactive"]["action"]["parameters"]["flow_token"] = flow_token
+        flow_body["interactive"]["action"]["parameters"]["flow_cta"] = flow_cta
+        flow_body["interactive"]["action"]["parameters"]["flow_id"] = flow_id
+        flow_body["interactive"]["action"]["parameters"]["flow_action_payload"] = {"screen": initial_screen}
+        return flow_body, http_url, headers
+
     @staticmethod
     def evaluate_script(script: Text, data: Any, raise_err_on_failure: bool = True):
         log = [f"evaluation_type: script", f"script: {script}", f"data: {data}", f"raise_err_on_failure: {raise_err_on_failure}"]

kairon/shared/admin/constants.py

@@ -3,3 +3,4 @@
 
 class BotSecretType(str, Enum):
     gpt_key = "gpt_key"
+    d360_api_key = "d360_api_key"