Exam Topics Covered: 3.1 Construct API requests to implement chatops with Webex Teams API
This section shows how to build a ChatBot in WebEx Teams. The overall work flow is the following:
The below application runs on Python Flask which must be installed with pip install flask. In order to run the application an environment variable must be set: FLASK_APP=chatbot_server.py. It can then be run with flask run. In addition, in order for the application to receive the webhook from Webex teams it must be reachable over the public Internet. I used an online proxy called NGrok (https://ngrok.com/) to receive the webook from WebEx teams and tunnel it to my local workstation. When running NGrok it provides a URL which is used as the targetUrl in the webhook. Note that the URL changes each time you restart NGrok, so this is only useable for testing. NGrok also provides a static URL for a small fee.|
The application programs a webhook using the API upon first startup. It then waits for a webhook, which is triggered by a message in the Teams room with the bot mentioned. When the webhook is received, the application connects back to the API to retrieve the message based on the message ID that was provided in the webhook. It then parses the text of the message, and posts a reply to the room.
from flask import Flask, request, render_template, Response
import requests
import os
from datetime import datetime
import re
import json
# Create the web application instance
flask_app = Flask(__name__)
# Grab the API token from the environment. This is the bot token.
token = os.environ.get('TEAMS_ACCESS_TOKEN')
# This is the base_url for all API calls
base_url = "https://webexapis.com"
# An Authorization header must be set with the bot token
headers = {'Content-Type':'application/json',
'Accept': 'application/json',
'Authorization': f'Bearer {token}'}
# Get rooms to determine RoomID
resp = requests.get(f"{base_url}/v1/rooms", headers=headers)
room_id = [item['id'] for item in resp.json()['items'] if "Bot_Testing" == item['title']][0]
# Webhook payload
# The resource is messages being sent to the room
# The filter indicates the webhook should trigger when messages are sent to
# "me", which indicates the bot
webhook_data = dict(name="chatbot_webhook",
targetUrl="http://eb54696397db.ngrok.io/events",
resource="messages",
filter=f"roomId={room_id}&mentionedPeople=me",
event="created")
# Check if webhook already created. Prevents a new webhook from
# being created every time the program is run
try:
resp = requests.get(f"{base_url}/v1/webhooks", headers=headers)
except requests.exceptions.RequestException as e:
print(e)
# Generate list of current webhook names
webhooks = [webhook.get('name') for webhook in resp.json()['items']]
# Create webhook, if not already created
if "chatbot_webhook" not in webhooks:
try:
resp = requests.post(f"{base_url}/v1/webhooks", headers=headers, json=webhook_data)
resp.raise_for_status()
print("Created webhook: chatbot_webhook")
except Exception as e:
print(e)
else:
print("Webhook already configured, skipping")
# This is a function that grabs current Covid cases in a specified
# NC zip code. The user can send a message to the bot with their
# zip code and the bot will reply with the current number of cases
def get_cases(zip_code):
url = f"https://services.arcgis.com/iFBq2AW9XO0jYYF7/" \
f"arcgis/rest/services/Covid19byZIPnew/FeatureServer" \
f"/0/query?where=ZIPCode={zip_code}&outFields=*&f=json"
headers = {'Accept': 'application/json'}
resp = requests.get(url, headers=headers)
cases = resp.json()['features'][0]['attributes']['Cases']
return cases
@flask_app.route('/', methods=['GET', 'POST'])
def home_page():
return render_template('index.html')
@flask_app.route('/events', methods=['GET', 'POST'])
def webex_teams_webhook_events():
# App doesn't do anything on a GET
if request.method == 'GET':
return f"Nothing to do here!"
# The webhook issues a POST to the TargetURL
elif request.method == 'POST':
# Respond to incoming webhook from Webex Teams
json_data = request.get_json()
# This information will be displayed on the console where
# flask run was executed
print(f"TIME: {datetime.now().strftime('%H:%M:%S')}")
print("\nWEBHOOK POST RECEIVED:")
print(f"{json_data}\n")
# Parse the json for the message ID and then retrieve the message
msg_id = json_data['data']['id']
# Retrieve the message
resp = requests.get(f"{base_url}/v1/messages/{msg_id}", headers=headers)
# Parse the response
msg = resp.json()
print(f"Message from: {msg['personEmail']}")
print(f"Message: {msg['text']}")
print(msg)
if 'cases' in msg['text'].lower():
m = re.search('[0-9]{5}', msg['text'])
if m:
zip = m.group()
cases = get_cases(zip)
msg_text = f"The number of cases that have been reported in {zip} is {cases}"
else:
msg_text = 'Please give me a 5-digit NC zip code'
else:
msg_text = "Not sure what you're asking, you can ask me things like 'how many cases in zip code 27502?'"
# Respond to the message
msg_data = dict(roomId=msg['roomId'],
parentId=msg['id'],
text=msg_text)
resp = requests.post(f"{base_url}/v1/messages", headers=headers, json=msg_data)
print(resp.status_code, resp.content)
return "OK"
if __name__ == '__main__':
# Start the flask web server
flask_app.run(host='0.0.0.0', port=5001)
Below is the index.html file that should be placed in the templates directory.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Webex Teams Bot served via Flask</title>
</head>
<body>
<p>
<strong>Flask web server is up and running!</strong>
</p>
</body>
</html>
For comparison, here is the exact same application written using the Webex Teams SDK. The Webex Teams SDK can be installed by running pip install webexteamssdk.
from flask import Flask, request, render_template
from webexteamssdk import WebexTeamsAPI
from datetime import datetime
import requests
import os
import re
# Create the web application instance
flask_app = Flask(__name__)
token = os.environ.get('TEAMS_ACCESS_TOKEN')
api = WebexTeamsAPI(access_token=token)
room_id = [room.id for room in api.rooms.list() if room.title == 'Bot_Testing'][0]
webhooks_list = [webhook.name for webhook in api.webhooks.list()]
if 'chatbot_webhook' not in webhooks_list:
api.webhooks.create(name='chatbot_webhook2',
targetUrl='http://eb54696397db.ngrok.io/events',
resource='messages',
event='created',
filter=f"roomId={room_id}&mentionedPeople=me")
else:
print('chatbot_webhook already configured, skipping')
def get_cases(zip_code):
url = f"https://services.arcgis.com/iFBq2AW9XO0jYYF7/" \
f"arcgis/rest/services/Covid19byZIPnew/FeatureServer" \
f"/0/query?where=ZIPCode={zip_code}&outFields=*&f=json"
headers = {'Accept': 'application/json'}
resp = requests.get(url, headers=headers)
cases = resp.json()['features'][0]['attributes']['Cases']
return cases
@flask_app.route('/', methods=['GET', 'POST'])
def home_page():
return render_template('index.html')
@flask_app.route('/events', methods=['GET', 'POST'])
def webex_teams_webhook_events():
if request.method == 'GET':
return f"Nothing to do here!"
elif request.method == 'POST':
"""Respond to incoming webhook from Webex Teams"""
json_data = request.get_json()
print(f"TIME: {datetime.now().strftime('%H:%M:%S')}")
print("\nWEBHOOK POST RECEIVED:")
print(f"{json_data}\n")
# Parse the json for the message ID and then retrieve the message
msg_id = json_data['data']['id']
# Get the posted message
message = api.messages.get(msg_id)
print(message)
if 'cases' in message.text.lower():
m = re.search("[0-9]{5}", message.text)
if m:
zipcode = m.group()
cases = get_cases(zipcode)
msg_text = f"The number of cases in {zipcode} is {cases}"
else:
msg_text = f"Please include the 5-digit NC zip code you would like me to search in"
else:
msg_text = f"I didn't understand. You can ask me things like 'How many cases are in zip code 27502?'"
api.messages.create(roomId=message.roomId,
parentId=message.id,
text=msg_text)
return "OK"
if __name__ == '__main__':
# Start the flask web server
flask_app.run(host='0.0.0.0', port=5001)