Amazon Lex V2 Response Utility Class

lambda_function.py

# usage in AWS Lambda:

from lexv2_utils import IntentBase

class MyIntentClass(IntentBase):
    # Implements a behavior your bot.
    def do_fulfillment(self) -> dict:
        # Implements a your logic
        message = {
            'contentType': 'PlainText',
            'content': 'Invoked Fulfillment'}
        return self.fulfilled_close(message)

    def do_dialog(self) -> dict:
        # Implements a your logic
        return self.delegate()

def handler(event, context):
    if event['bot']['name'] == 'MyBotName':
        if event['sessionState']['intent']['name'] == 'MyIntentName':
            my_intent = MyIntentClass(event)
            return my_intent.invoke()

    raise RuntimeError('unknown event')

lexv2_utils.py

"""Amazon Lex v2 response utilities"""
import json
import base64
from typing import Any, Optional


class IntentBase(object):
    """Lex Intent Proccess Base
    event: lex event dict (see: https://docs.aws.amazon.com/lexv2/latest/dg/lambda.html#lambda-input-format)
    """
    def __init__(self, event: dict):
        if 'invocationSource' not in event:
            raise RuntimeError('event dict does not have a key: invocationSource')
        if 'sessionState' not in event:
            raise RuntimeError('event dict does not have a key: sessionState')
        if 'intent' not in event['sessionState']:
            raise RuntimeError('event dict does not have a key: sessionState.intent')
        self.__request = event

    @property
    def request(self) -> dict:
        return self.__request

    @property
    def request_attributes(self) -> dict:
        return self.request.get('requestAttributes') or {}

    @property
    def session_id(self) -> str:
        return self.request['sessionId']

    @property
    def session_state(self) -> dict:
        return self.request['sessionState']

    @property
    def intent(self) -> dict:
        return self.session_state['intent']

    @property
    def session_attributes(self) -> dict:
        if self.session_state.get('sessionAttributes') is None:
            self.session_state.update(sessionAttributes={})
        return self.session_state['sessionAttributes']

    @session_attributes.setter
    def session_attributes(self, v):
        self.session_state.update(sessionAttributes=v)

    @property
    def confirmation_state(self) -> str:
        return self.intent.get('confirmationState') or ''

    def set_session_value(self, key: str, value: str):
        self.session_attributes[key] = value

    def set_session_value_json(self, key: str, value: Any):
        # set into session with json and base64
        # see: https://docs.aws.amazon.com/lex/latest/dg/context-mgmt-complex-attributes.html
        self.set_session_value(key, base64.b64encode(json.dumps(value).encode()).decode())

    def get_session_value(self, key: str) -> Optional[str]:
        return self.session_attributes.get(key)

    def get_session_value_json(self, key: str) -> Any:
        raw = self.get_session_value(key)
        return json.loads(base64.b64decode(raw)) if raw else None

    def is_fulfillment(self) -> bool:
        return self.request.get('invocationSource') == 'FulfillmentCodeHook'

    def is_dialog(self) -> bool:
        return self.request.get('invocationSource') == 'DialogCodeHook'

    def is_confirmed(self) -> bool:
        return self.confirmation_state == 'Confirmed'

    def is_denied(self) -> bool:
        return self.confirmation_state == 'Denied'

    def invoke(self) -> dict:
        if self.is_fulfillment():
            return self.invoke_fulfillment()
        elif self.is_dialog():
            return self.invoke_dialog()
        return self.delegate()

    def is_empty_all_slots(self) -> bool:
        # When all slot elements are None...True (When even one value is set...False)
        slots = self.intent.get('slots') or dict()
        return not any(slots.values())

    def get_intent_slot_interpreted_value(self, slot_name: str) -> str:
        try:
            return self.intent['slots'][slot_name]['value']['interpretedValue']
        except Exception:
            return ''

    def invoke_fulfillment(self) -> dict:
        return self.do_fulfillment()

    def do_fulfillment(self) -> dict:
        """overrides this method"""
        if self.is_confirmed():
            return self.do_fulfillment_confirmed()
        elif self.is_denied():
            return self.do_fulfillment_denied()
        return self.do_fulfillment_confirm_none()

    def do_fulfillment_confirm_none(self) -> dict:
        return self.delegate()

    def do_fulfillment_confirmed(self) -> dict:
        return self.delegate()

    def do_fulfillment_denied(self) -> dict:
        return self.delegate()

    def invoke_dialog(self) -> dict:
        return self.do_dialog()

    def do_dialog(self) -> dict:
        """overrides this method"""
        return self.delegate()

    def _build_response_base(
            self,
            dialog_action_type: Optional[str] = None,
            message: Optional[str] = None) -> dict:
        """
        Build a Common Response (same as when Delegate)
        dialog_action_type: "Close | ConfirmIntent | Delegate | ElicitIntent | ElicitSlot"
        """
        return {
            'sessionState': {
                'sessionAttributes': self.session_attributes,
                'dialogAction': {
                    'type': dialog_action_type
                },
                'intent': self.intent
            },
            'messages': [message] if message else None,
            'sessionId': self.session_id,
            'requestAttributes': self.request_attributes
        }

    def delegate(self, message: Optional[str] = None) -> dict:
        return self._build_response_base('Delegate', message)

    def fulfilled_close(self, message=None) -> dict:
        self.intent['state'] = 'Fulfilled'
        return self._build_response_base('Close', message)

    def fulfilled_confirm(self, message=None) -> dict:
        self.intent['state'] = 'Fulfilled'
        return self._build_response_base('ConfirmIntent', message)

    def elicit_slot(self, message, slot_name) -> dict:
        self.intent['state'] = 'InProgress'
        self.intent['slots'][slot_name] = None

        response = self._build_response_base('ElicitSlot', message)
        response['sessionState']['dialogAction']['slotToElicit'] = slot_name
        return response

    def elicit_intent(self, message) -> dict:
        response = self._build_response_base('ElicitIntent', message)
        response['sessionState']['intent'] = None
        return response