[PYTHON] How to make a Cisco Webex Teams BOT with Flask

Recently, when developing a system using Cisco Webex Teams, I had the opportunity to create a chatbot for Webex Teams and leave a note for myself.

Also, for the demonstration this time, ngrok will be used to publish the server.

The flow this time is as follows. · Get a Cisco Webex Developers account ・ Acquisition of access token ・ Installation of various Python libraries ・ Preparation of ngrok ·Implementation ・ Operation check

The development environment used this time is as follows. ・ Raspberry Pi3 B + ・ Python3 ・ Ngrok ・ Flask

Get a Cisco Webex Developers account

First, access this site. https://developer.webex.com/

Click Sign up displayed in the upper right to proceed with registration.

Obtaining an access token

https://developer.webex.com/docs/api/getting-started Go to the getting-started page and get a token

During the demo stage, you will have to retake this token every 12 hours.

Click My Webex Teams Apps.

Select a bot.

After filling in the various forms, click Addbot.

Installation of various Python libraries

terminal

$ pip3 install requests

terminal

$ pip3 install flask

Preparing ngrok

https://qiita.com/kaba/items/82de7e53d99ad9c74cc0 I referred to this site.

Implementation

This time, as an example, let's create a bot that only returns hello world! When talking to Hi and the bot. Paste the implemented code.

First, substitute the token you obtained earlier into YourToken.

token = 'YourToken'

Define a function to post a message.

def post_message(room_id,txt,token):
    global ms_flag
    if ms_flag == True:
        ms_flag = False
        m = MultipartEncoder({'roomId': room_id,'text': txt})
        r = requests.post('https://api.ciscospark.com/v1/messages', data=m,headers={'Authorization': 'Bearer %s' % token,'Content-Type': m.content_type})
    else:
        time.sleep(5)
        ms_flag = True

At this time, it is handled multiple times depending on the behavior when detecting the message, so once this function is called, it passes for a certain period of time. If anyone knows about this behavior, please teach me.

Here, it is the part that displays mainpage.html when accessing / panel. It's a good idea to write an explanation of how to use it.

@app.route("/panel",methods=['GET'])    
def main_page():
    return render_template("mainpage.html")

Collect message ID, person ID, email and room ID from the received message information

@app.route("/",methods=['POST'])   
def handle_message():
    json = request.json
    
    message_id = json["data"]["id"]
    user_id = json["data"]["personId"]
    email = json["data"]["personEmail"]
    room_id = json["data"]["roomId"]
    bot_id = "yourbotid"

    print(message_id, file = sys.stdout)
    print(user_id, file=sys.stdout)
    print(email, file=sys.stdout)
    print(room_id, file=sys.stdout)

Here, we are looking at whether the user ID that received the message is the bot itself. Also, the post_message defined earlier is called so that when Hi is received, hello world! Is returned.

    if user_id != bot_id:
        global token
        header = {"Authorization": "Bearer %s" % token}
        get_rooms_url = "https://api.ciscospark.com/v1/messages/" + message_id
        api_response = requests.get(get_rooms_url, headers=header, verify=False)
        response_json = api_response.json()
        message = response_json["text"]
        print(message, file= sys.stdout)
        if message == "Hi" or message == "bot Hi":
            post_message(room_id,"hello world!",token)
       
        return "Success"
    else:
        return "Pass"

Also, since the IP address of ngrok changes every time it is started, it reads the IP address.

term_output_json = os.popen('curl http://127.0.0.1:4040/api/tunnels').read() 
tunnel_info = json.loads(term_output_json)
public_url = tunnel_info['tunnels'][0]['public_url']

Error handling.

if api_response.status_code != 200:
    print('Webhook registration Error !')
    exit(0)

The final result is as follows.

main.py

from __future__ import print_function 
import requests
import sys
import json
import os
import time
from flask import *
from requests_toolbelt.multipart.encoder import MultipartEncoder
import functools

token = 'YourToken'
ms_flag = True

app = Flask(__name__)

def post_message(room_id,txt,token):
    global ms_flag
    if ms_flag == True:
        ms_flag = False
        m = MultipartEncoder({'roomId': room_id,'text': txt})
        r = requests.post('https://api.ciscospark.com/v1/messages', data=m,headers={'Authorization': 'Bearer %s' % token,'Content-Type': m.content_type})
    else:
        time.sleep(5)
        ms_flag = True

@app.route("/panel",methods=['GET'])    
def main_page():
    return render_template("mainpage.html")


@app.route("/",methods=['POST'])   
def handle_message():
    json = request.json
    
    message_id = json["data"]["id"]
    user_id    = json["data"]["personId"]
    email      = json["data"]["personEmail"]
    room_id    = json["data"]["roomId"]
    bot_id     = "yourbotid"

    print(message_id, file = sys.stdout)
    print(user_id, file=sys.stdout)
    print(email, file=sys.stdout)
    print(room_id, file=sys.stdout)


    if user_id != bot_id:
        global token
        header = {"Authorization": "Bearer %s" % token}
        get_rooms_url = "https://api.ciscospark.com/v1/messages/" + message_id
        api_response = requests.get(get_rooms_url, headers=header, verify=False)
        response_json = api_response.json()
        message = response_json["text"]
        print(message, file= sys.stdout)
        if message == "Hi" or message == "bot Hi":
            post_message(room_id,"hello world!",token)
       
        return "Success"
    else:
        return "Pass"


term_output_json = os.popen('curl http://127.0.0.1:4040/api/tunnels').read() 
tunnel_info = json.loads(term_output_json)
public_url = tunnel_info['tunnels'][0]['public_url']

#Webhook record
header = {"Authorization": "Bearer %s" % token, "content-type": "application/json"}
requests.packages.urllib3.disable_warnings() #Remove SSL warning
post_message_url = "https://api.ciscospark.com/v1/webhooks"

payload = {
    "resource": "messages",
    "event": "all",
    "targetUrl": public_url,
    "name": "BotDemoWebHook"
}

api_response = requests.post(post_message_url, json=payload, headers=header, verify=False) #webhook record

if api_response.status_code != 200:
    print('Webhook registration Error !')
    exit(0)

if __name__ == '__main__':
    app.run(host='localhost', use_reloader=True, debug=True)

Operation check

terminal

$ ngrok http 5000

terminal

$ python3 main.py