[PYTHON] Make a LINE WORKS bot with Amazon Lex

This is the article on the 11th day of LINEWORKS Advent Calendar 2019.

There are many services that make it easy to create chatbots, but this time I tried to develop a LINE WORKS bot using "Amazon Lex" that I had been interested in for a long time.

What is Amazon Lex?

"Amazon Lex" is an AWS service that provides a conversational interface. It seems that Lex uses the same technology used in Alexa. Not only natural language processing but also voice recognition is included.

https://aws.amazon.com/jp/lex/

Similar services include DialogFlow and Azure Bot Service and IBM. Watson Assistant

This time, I created a dialogue flow with Amazon Lex and created a bot that can talk on LINE WORKS talk.

!!!Caution!!!

__ As of 12/11/2019, Amazon Lex only supports US English, not Japanese or Tokyo region. __ This time as well, it will be a chatbot that talks in English. (When will it be available in Japanese ...) Also, the region used is Oregon.

Constitution

--A serverless configuration using AWS Lambda. --Receive callbacks from LINE WORKS via API Gateway. From there, it works with Amazon Lex for interactive processing. --LINE WORKS parameters are managed in the Systems Manager parameter store. --The LINE WORKS access token is updated by Lambda, which runs regularly.

Development environment

--Language: Python 3.7.1 --Deploy: Use Serverless Framework to deploy Lambda

Implementation

1. Create a bot with Amazon Lex

This time, I created a bot using a sample according to the official tutorial below.

Bot example: BookTrip --Amazon Lex https://docs.aws.amazon.com/ja_jp/lex/latest/dg/ex-book-trip.html

Simply put, it's a chatbot that reserves cars and hotels. I chose "Book Trip" from the sample and created it.

On such a screen, you can set the intent and the dialogue flow.

Impression that anyone who has used other chatbot creation services can use it immediately. Beginners felt that it would be difficult to set from scratch.

2. Create various keys and bots from the LINE WORKS Developer Console

Log in to the LINE WORKS Developer Console and create a key or this bot.

For details, please refer to this past article.

Let's implement LINE WORKS Talk Bot in Python ~ Part 1: API authentication ~ https://qiita.com/mmclsntr/items/1d0f520f1df5dffea24b

3. LINE WORKS bot app server creation

I configured it with Lambda and implemented the runtime in Python 3.7.

The following two Lambda functions

1. __LINE WORKS Access token regular update __ --CloudWatch Event Scheduled event (once every half day)
2. __LINE WORKS Chatbot __ --Get messages from LINE WORKS via API Gateway and work with Amazon Lex Bot.

Below is a sample code

lambda_function.py

import json
import jwt
import requests
import urllib
import boto3
import os
from datetime import datetime
from base64 import b64encode, b64decode
import hashlib
import hmac

from requests.structures import CaseInsensitiveDict

ssm = boto3.client('ssm')
lex = boto3.client('lex-runtime')

####################################
#Systems Manager parameter store#
####################################
def get_parameter(key):
    """
Get parameters from SSM parameter store
    """
    response = ssm.get_parameters(
        Names=[
            key
        ],
        WithDecryption=True
    )
    parameters = response["Parameters"]
    if len(parameters) > 0:
        return response['Parameters'][0]["Value"]
    else:
        return ""


def put_parameter(key, value):
    """
Store parameters in SSM parameter store
    """
    response = ssm.put_parameter(
        Name=key,
        Value=value,
        Type='SecureString',
        Overwrite=True
    )


##############
# Amazon Lex #
##############
def post_text_to_lex(text, user_id, bot_name, bot_alias):
    """
Send text to Amazon Lex&Get a reply
    """
    response = lex.post_text(
        botName=bot_name,
        botAlias=bot_alias,
        userId=user_id,
        inputText=text
    )

    return response["message"]


##################
# LINE WORKS API #
##################
def get_jwt(server_list_id, server_list_privatekey):
    """
Get JWT for LINE WORKS access token
    """
    current_time = datetime.now().timestamp()
    iss = server_list_id
    iat = current_time
    exp = current_time + (60 * 60) #1 hour

    secret = server_list_privatekey

    jwstoken = jwt.encode(
        {
            "iss": iss,
            "iat": iat,
            "exp": exp
        }, secret, algorithm="RS256")

    return jwstoken.decode('utf-8')


def get_server_token(api_id, jwttoken):
    """
Get LINE WORKS access token
    """
    url = 'https://authapi.worksmobile.com/b/{}/server/token'.format(api_id)

    headers = {
        'Content-Type' : 'application/x-www-form-urlencoded; charset=UTF-8'
    }

    params = {
        "grant_type" : urllib.parse.quote("urn:ietf:params:oauth:grant-type:jwt-bearer"),
        "assertion" : jwttoken
    }

    form_data = params

    r = requests.post(url=url, data=form_data, headers=headers)

    body = json.loads(r.text)
    access_token = body["access_token"]

    return access_token


def validate_request(body, signature, api_id):
    """
LINE WORKS request verification
    """
    #Use API ID as private key
    secretKey = api_id.encode()
    payload = body.encode()

    # HMAC-Encoded with SHA256 algorithm
    encoded_body = hmac.new(secretKey, payload, hashlib.sha256).digest()
    #BASE64 encoding
    encoded_b64_body = b64encode(encoded_body).decode()

    #Comparison
    return encoded_b64_body == signature


def send_message(content, api_id, botno, consumer_key, access_token, account_id):
    """
Send LINE WORKS message
    """
    url = 'https://apis.worksmobile.com/{}/message/sendMessage/v2'.format(api_id)

    headers = {
          'Content-Type' : 'application/json;charset=UTF-8',
          'consumerKey' : consumer_key,
          'Authorization' : "Bearer " + access_token
        }

    params = {
            "botNo" : int(botno),
            "accountId" : account_id,
            "content" : content
        }

    form_data = json.dumps(params)

    r = requests.post(url=url, data=form_data, headers=headers)

    if r.status_code == 200:
        return True

    return False

######################
#Lambda function handler#
######################
def update_token_handler(event, context):
    """
LINE WORKS access token periodic update Lambda handler function
    """
    #Get LINE WORKS parameters from the SSM parameter store
    api_id = get_parameter("lw_api_id")
    server_list_id = get_parameter("lw_server_list_id")
    server_list_privatekey = get_parameter("lw_server_list_private_key").replace("\\n", "\n")
    #Get JWT
    jwttoken = get_jwt(server_list_id, server_list_privatekey)

    #Get Server token
    access_token = get_server_token(api_id, jwttoken)

    #Set Access Token in Parameter Store
    put_parameter("lw_access_token", access_token)

    return


def chat_with_lex_handler(event, content):
    """
LINE WORKS chatbot Lambda handler function
    """
    botno = os.environ.get("BOTNO")
    lex_bot_name = os.environ.get("LEX_BOT_NAME")
    lex_bot_alias = os.environ.get("LEX_BOT_ALIAS")
    #Get LINE WORKS parameters from the SSM parameter store
    api_id = get_parameter("lw_api_id")
    consumer_key = get_parameter("lw_server_api_consumer_key")
    access_token = get_parameter("lw_access_token")

    event = CaseInsensitiveDict(event)
    headers = event["headers"]
    body = event["body"]

    #Request validation
    if not validate_request(body, headers.get("x-works-signature"), api_id):
        #Illegal request
        return

    #Perth to Json
    request = json.loads(body)

    #Get sending user
    account_id = request["source"]["accountId"]

    res_content = {
        "type" : "text",
        "text" : "Only text"
    }

    #Check the contents of the received message
    request_type = request["type"]
    ## Message
    if request_type == "message":
        content = request["content"]
        content_type = content["type"]
        ## Text
        if content_type == "text":
            text = content["text"]

            #Works with Amazon Lex
            reply_txt = post_text_to_lex(text, account_id.replace("@", "a"), lex_bot_name, lex_bot_alias)

            res_content = {
                "type" : "text",
                "text" : reply_txt
            }

    #Send
    rst = send_message(res_content, api_id, botno, consumer_key, access_token, account_id)

    res_body = {
        "code": 200,
        "message": "OK"
    }
    response = {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json"
        },
        "body": json.dumps(res_body)
    }

    return response

Please also refer to this past article. Let's implement LINE WORKS Talk Bot in Python ~ Part 2: Chatbot implementation ~ https://qiita.com/mmclsntr/items/28ba6baaf23124a53663

Source code

https://github.com/mmclsntr/lineworks-bot-with-amazon-lex

Try to move

You can talk (in English) with a chatbot created with Lex on the LINE WORKS talk as follows.

Summary

I was able to run the chatbot created with Amazon Lex with LINE WORKS without any problems. I think that simple inquiries can be easily realized. I hope you can speak Japanese. ..

After that, I will tune various dialogue settings with Lex and play with it.