最近Cisco Webex Teamsを活用したシステム開発を行っている際、Webex Teamsのチャットボットを作る機会があり、自分用メモを残しておく。

また、今回はデモのため、サーバー公開にはngrokを用いる。

今回の流れは、以下の通りである。
・Cisco Webex Developersのアカウント取得
・アクセストークンの取得
・各種Pythonライブラリのインストール
・ngrokの準備
・実装
・動作確認

なお今回用いた開発環境は、以下の通りである。
・Raspberry Pi3 B +
・Python3
・ngrok
・Flask

Cisco Webex Developersのアカウント取得

まずはこのサイトにアクセス。
https://developer.webex.com/

右上に表示されているSign upをクリックして登録を進めていく。

アクセストークンの取得

https://developer.webex.com/docs/api/getting-started
getting-startedのページにアクセスし、トークンを取得する

デモの段階では12時間ごとにこのトークンを取り直さなければならない。

My Webex Teams Appsをクリック。

Botを選択する。

各種フォームを埋めたらAddbotをクリック。

各種Pythonライブラリのインストール

terminal

$ pip3 install requests

terminal

$ pip3 install flask

ngrokの準備

https://qiita.com/kaba/items/82de7e53d99ad9c74cc0
このサイトを参考にした。

実装

今回は例として、Hiとボットに話しかけるとhello world!を返すだけのボットを作ってみることにする。
実装したコードを貼っておく。

まずは先ほど取得したトークンをYourTokenに代入する。

token = 'YourToken'

メッセージをポストするために関数を定義する。

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

この際、メッセージを検知する際の挙動によって複数回ハンドルされてしまうので、一度この関数が呼び出されたら、一定時間パスするようにしている。
この挙動に関してご存知の方がいればご教授願います。

ここでは、/panelにアクセスするとmainpage.htmlを表示する部分である。
使い方の説明などを書いておくと良いだろう。

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

message ID, person ID, email and room IDを受け付けたメッセージ情報から回収

@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)

ここでは、もし、メッセージを受信したユーザーIDがボット自身かどうかをみている。
また、Hiを受信したらhello world!を返すよう、先ほど定義したpost_messageを呼び出している。

    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"

また、ngrokは起動するごとにIPアドレスが変わってしまうので、そのIPアドレスを読み取っている。

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']

エラー処置。

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

最終的に以下のようなる。

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記録
header = {"Authorization": "Bearer %s" % token, "content-type": "application/json"}
requests.packages.urllib3.disable_warnings() #SSL警告の削除
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記録

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

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

動作確認

terminal

$ ngrok http 5000

terminal

$ python3 main.py