Now that you can send messages from the portal via the Bot, you can focus on receiving messages via webhooks. In
this chapter you will:
Remove and Add Webhooks
Handle incoming Webhook messages
Configure a message response
Send an Adaptive Card
Configure Adaptive Card Response Actions
This is an area that was not covered in previous examples, because the tasks require a web server, such as the
Flask instance you are working on, to receive HTTPS POST messages from Webex. You experimented with a bot that
echoed back the webhook messages to get a feel for how they work, but now you will enable your own bot to receive
and process webhook notifications from Webex.
Step 1 - Remove and Add Webhooks
The earlier webhook section involved sending requests to create two webhooks, one for
messages, one for attachmentActions event types with an
HTTP callback destination.
Those webhooks will remain active until they are removed or they encounter enough errors to where
Webex marks them disabled (for example, after 100 failed attempts in a 5-minute period).
To make sure that you don't have multiple webhooks registered, a good housekeeping practice is to remove any
existing webhook and re-add them each time you start your application that makes use of the webhooks.
In your VS Code
tab, open up flaskr/api/v1/wbxt.py using the Explorer
If you completed earlier sections, this file will already have access to the
WBX_BOT_ACCESS_TOKEN environment variable which will be used for all webhook tasks.
Your first job is to finish out the functions that remove and re-add the webhooks. Starting with
the delete_webhooks_by_name() function. The SDK has a WebehookAPI with some methods
that mirror the capabilities from the Webex documentation.
As with other list() methods, you can iterate through until you find the "name" that matches the
WBX_WEBHOOK_NAME, a constant set to "Cisco Live LTRCOL-2574 Webhook",
and then call the
wbx_bot_api.webhook.webhook_delete()
method supplying the webhook ID of the one you found.
Remove the pass at the beginning of the function and replace it with the code
below.
defdelete_webhooks_by_name():"""
Retrieve all webhooks that are owned by the Bot, then delete the ones matching our webhook name.
"""for webhook in wbx_bot_api.webhook.list():if webhook.name == webhook_name:
wbx_bot_api.webhook.webhook_delete(webhook_id=webhook.webhook_id)
log.info(f'Webhook "{webhook.name}" for "{webhook.resource}" deleted. Callback URL="{webhook.target_url}"')
Next, create both webhooks (one for messages and one for
attachmentActions). Using the SDK, the
wbx_bot_api.webhook.create()
method is equivalent to the Webex
Create a Webhook API
that you used previously. The
Create a Webhook API
WebhookResource and WebhookEventType
models used here from the SDK ensure that the data sent is validated and normalized.
Find the create_webhooks(), remove the pass at the beginning of the
function and add the following to your file:
defcreate_webhooks():"""
Create the Webex webhooks using the Bot: one for messages, another one for the adaptive card
"""for resource_type in['messages','attachmentActions']:
webhook = wbx_bot_api.webhook.create(
resource=WebhookResource(resource_type),
event=WebhookEventType.created,
name=webhook_name,
target_url=webhook_url
)
log.info(f'Webhook "{webhook_name}" for "{webhook.resource}" created. Callback URL="{webhook_url}".')
Now just add code to call the two functions you implemented to delete the webhooks and then recreate them.
This will be called every time your application starts.
Add the following code to your file:
# Delete and recreate all webhooks with a given name every time the flask app is restarted
delete_webhooks_by_name()
create_webhooks()
In VS Code, click Save on the file then start (or restart) your Flask app by clicking
Debug > Start Debugging (or
Restart Debugging, if already started).
You should see messages that the Webhooks are deleted (the ones you created manually) and two new ones created.
Step 2 - Handling incoming webhook messages
You may have noticed that the webhook callback URL is http://collab-api-webhook.ciscolive.com:9003/api/v1/wbxt/events. This
callback destination routes from Webex directly to your lab Flask web server. The webhooks will be routed to
/events in this flaskr/api/v1/wbxt.py file, where this Namespace is configured.
Find the wbxt_events_api class under the @api.route('/events') line in the file.
The code has a few important steps:
When Flask handles a request, in this case the POST request, your code has access to an object named request
that contains all the data sent by the client. If the data received is in JSON format, it can be accessed using
request.json.The request.json data is used to create a
WebhookEvent
object using its parse_obj method. Basically this is a way to create a Python object from the JSON data using
the wxc_sdk. This makes it easier to work with the received data.
Depending on the resource, you can call the respond_to_message() or
respond_to_button_press() method. Otherwise, don't reply with any other data.
NOTE that you will have to scroll towards the end of the file to find the post() method of the wbxt_events_api
class and add the following. After doing this you'll scroll back up to implement a couple more functions in the file.
# Receive a Webex Webhook POSTdefpost(self):"""
Receives a Webex Webhook POST. Evaluates the event, in all cases, on will only process 'created' events, but
depending on the resource type (messages or attachmentActions), will respond accordingly
"""try:# Log inbound POST data
log.debug(f'Received inbound POST with data:\n'+ json.dumps(request.json, indent=4))# Parse the incoming Webhook request as a WebhookEvent
webhook_event = WebhookEvent.parse_obj(request.json)# Handle a new message eventif webhook_event.resource =='messages':
self.respond_to_message(webhook_event)# Handle an active card submission eventelif webhook_event.resource =='attachmentActions':
self.respond_to_button_press(webhook_event)except Exception as e:
log.error(f'Failed to parse: {e}')return'Ok'
Step 3 - Configure a message response
You have webhooks configured and a Flask app to receive them. You just need to decide how to handle the
message as it arrives.
In this section, you can use the information obtained from the webhook event to:
Determine the Room that originated the event
The sent Message itself
This Message, in turn, is used to look up the sender
Determine if this is a message that you want to respond to
Why do you need to look up the sender and the message for a generic response that only
requires the room ID? Consider that this Bot could have been added by
anyone using Webex. You cannot control that. But you can
control who to respond to. In this case, you might want to only reply if the sender is in the
same Webex organization as the Bot. You could also base the response on the email domain or even
specific user who sent the message. Furthermore, you never want to respond if the sender is
the Bot itself.
Based on who you decide to respond to, create your response message with
wbx_bot_api.messages.create(),
using the room id.
Scroll up a bit to find the def respond_to_message() function. Remove the pass
and replace it with the following. Note that the
"self.send_card(room_id=room.id)" line is there already but that function does not do anything yet.
defrespond_to_message(self, wh_event: WebhookEvent):"""
Respond to a message to the bot. Retrieve the Room, message, and sender information from the webhook payload and
the corresponding message id
"""# Read the Webhook event Data
wh_data = wh_event.data
# Get the room to which this active card submission was posted to from the webhook data
room = wbx_bot_api.rooms.details(room_id=wh_data.room_id)# Look up the message from the data in the webhook
message = wbx_bot_api.messages.details(message_id=wh_data.id)# Look up the sender of this message
sender = wbx_bot_api.people.details(person_id=message.person_id)# ONLY reply if the message is from someone in the same Webex org AND the message is not from the Bot itselfif(bot.org_id == sender.org_id)and(message.person_id != bot.person_id):
log.info(f'Message received from {sender.display_name}:\n{message.text}')# Send a message in reply to the same room
wbx_bot_api.messages.create(room_id=room.id, text='Your available options are:')# Send the contact card to the room which originated the active card submission
self.send_card(room_id=room.id)
In VS Code, click Save on the file then start (or restart) your Flask app by clicking
Debug > Start Debugging (or
Restart Debugging, if already started).
The Webex app should be still running. If not, you can launch it and sign in using email:
pod3wbxuser@collab-api.com
and password:
C1sco.123
Send a message and verify that you receive "Your available options are:" in reply.
Based on the message reponse, you are clearly not done. Next add an adaptive card to the response.
Step 4 - Respond with an Adaptive Card
The intricacies of adaptive card design is definitely beyond the scope of this lab. It is just a JSON-formatted
text file used implement a graphical element similar to an HTML web form, including certain graphics,
buttons and drop-downs. For this lab a simple adaptive card has been provided in the file:
flaskr/api/v1/adaptive_card.json.
It has a few actions, that will generate data that you can see when when a user clicks
a button on the card. If you like, you can copy the contents of that file and paste
it in a message to the Webhook
Echo Bot (webhook-echo@webex.bot) and see what it looks like. Shortly, your web server will be responding
with the same content, so you can see it then.
In your code, you need to do two things: send this adaptive card as part of a message, so the sender can see
it; and then handle any submissions. Start by sending the card:
The adaptive card file is already loaded for you and stored in the adaptive_card_content
variable. Expand the send_card function to send the adaptive card content as an attachment
to a message. As indicated, the text will only be visible if the client cannot render the card.
Scroll up a bit in the file to find the send_card function. Remove the pass statement and replace it with the following:
classwbxt_events_api(Resource):defsend_card(self, room_id:str):"""
Post the adaptive card to the specified room.
"""
wbx_bot_api.messages.create(
room_id=room_id,
text='Your Webex client cannot display this card',
attachments=[{'contentType':'application/vnd.microsoft.card.adaptive','content': adaptive_card_content
}])
Now we have the ability to respond with a card. Save your file in VS Code then
start (or restart) your Flask app by clicking
Debug > Start Debugging (or
Restart Debugging, if already started).
Send another message and see if you get the adaptive card in the response. Note that it might take a few seconds for the card to render.
Step 5 - Configure Response Actions
So far, you are using the webhook to respond to text-based messages. The reply now also includes an
adaptive card. The final step is to respond to submissions by that adaptive card, which triggers
a second webhook.
All the work has to be done in the respond_to_button_press() function. Similar to replying to
a message, you need to determine the room the message originated from. Instead of a message, an
attachment action must be read, which will allow you to read the submitted data as well as determine the
sender. You will again only reply to a sender within your own Webex organization.
Locate the respond_to_button_press() method, where the room,
attachement_action, and sender are determined similar to a
the code you just implemented when a message is received instead of the attachment action.
Since this is the same as the previous code, it has already been implemented for you as seen here:
defrespond_to_button_press(self, wh_event: WebhookEvent):"""
Respond to an adaptive card submission
"""# Read the Webhook event Data
wh_data = wh_event.data
# Get the room to which this message was posted to from the webhook data
room = wbx_bot_api.rooms.details(room_id=wh_data.room_id)# Look up the attachment from the data in the webhook
attachment_action = wbx_bot_api.attachment_actions.details(action_id=wh_data.id)# Look up the sender of the submission/attachment
sender = wbx_bot_api.people.details(person_id=attachment_action.person_id)# Only reply if the submission was from someone in the same Webex org as the Botif bot.org_id == sender.org_id:
log.info(f'attachmentAction received from {sender.display_name}:\n'f'{attachment_action.inputs}')try:# Evaluate the contents of the submission and take action on them
You now just need to determine the action to take based on the particular submission.
Let's take the get version request. The attachment action will have an 'action' key that, per the
adaptive card, will have have a value of get_ucm_version based on that specific
button press. You can then call the API via cucm_get_version_api.get(Resource),
convert that reply to JSON, then reply back with the version if the query was successful. Note that
you can use markdown instead of plain text, so some things can be bolded. Copy the following code
and paste it in the respond_to_button_press() function.
try:# Evaluate the contents of the submission and take action on themif attachment_action.inputs['action']=='get_ucm_version':# Retrieve the CUCM version from the API created previously
log.info('Creating response for get_ucm_version')
cucm_version_data = cucm_get_version_api.get(Resource).json
if cucm_version_data['success']:
wbx_bot_api.messages.create(
room.id,
markdown=f'The Unified CM version is **{cucm_version_data["version"]}**')else:
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'Failed to retrieve The Unified CM version\n \
Error Message:\n \
```{cucm_version_data["message"]}```'
)# Take action on the 'get_ucm_reg_phones' action
Another button action triggers the get_ucm_reg_phones action.
For this, you can leverage the cucm_perfmon_api.post() function created earlier,
passing the perfmon_counters directly to the method. Depending on the logic you
want to implement, you can just sum the values of the counters together and return the result in
the reply message.
# Take action on the 'get_ucm_reg_phones' actionelif attachment_action.inputs['action']=='get_ucm_reg_phones':# Retrieve the CUCM registered phones, in this case, we'll add up the RegisteredHardwarePhones and# RegisteredOtherStationDevices perfmon counters
log.info('Creating response for get_ucm_reg_phones')
perfmon_counters =['Cisco CallManager\RegisteredOtherStationDevices','Cisco CallManager\RegisteredHardwarePhones']
cucm_perfmon_data = cucm_perfmon_api.post(Resource, perfmon_counters=perfmon_counters).json
if cucm_perfmon_data['success']:
num_reg_devices =sum(item['Value']for item in cucm_perfmon_data['perfmon_counters_result'])
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'The number of registered devices is **{num_reg_devices}**')else:
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'Failed to retrieve the number of registered devices\n \
Error Message:\n \
```{cucm_perfmon_data["message"]}```'
)# Take action on the 'user_search' action
Lastly, you can handle the user_search action. You have to read the user
key, which stores the user ID to query. In flaskr/api/v1/core.py, you already have
a core_users_api class, which calls methods in the classes you created, namely
cucm_user_api to query CUCM, and wbxc_user_api, which queries Webex.
for user information. A determination is made as to where the user "lives" (i.e. if the CUCM user has
a Home Cluster property set, or the Webex Calling user has a
Location ID ). The result is simply which phone system the user belongs to. If
the user_detail checkbox is selected during the card submission, then the full user
details are also sent.
# Take action on the 'user_search' actionelif attachment_action.inputs['action']=='user_search':# Call the core user search, this will search both CUCM and Webex Calling
log.info(f'User Search for: {attachment_action.inputs["user"]}')
user_data = core_users_api.get(Resource, userid=attachment_action.inputs['user']).json
if user_data['success']:
wbx_bot_api.messages.create(
room.id,
text=f'User {attachment_action.inputs["user"]} is configured in {user_data["phonesystem"]}')# Output full user data if Details checkbox selectedif attachment_action.inputs['user_details'].lower()=='true':if'wbxc_user_data'in user_data:
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'Webex Calling user details:\n'f'```\n{json.dumps(user_data["wbxc_user_data"], indent=4)}\n```')if'cucm_user_data'in user_data:
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'Unified CM user details:\n'f'```\n{json.dumps(user_data["cucm_user_data"], indent=4)}\n```')else:
wbx_bot_api.messages.create(
room_id=room.id,
text=f'User {attachment_action.inputs["user"]} lookup failed: {user_data["message"]}')# Any error in looking up a value will simply be ignored.except KeyError:pass
In VS Code, start (or restart) your Flask app by clicking
Debug > Start Debugging (or
Restart Debugging, if already started).
Now try clicking the buttons, such as the Version and Get Registered Phones
under Unified CM, or try looking up
pod3wbxuser
under User Lookup.
Congratulations! You have completed have implemented several of the Python functions
necessary to create the Portal API that can now be used by the front-end of the Provisioning Portal.
The next section will take you through a tour of the portal.
Note: You will need to have a valid bot access token for this to work.
import logging
import json
from flask import request
from flask_restx import Namespace, Resource
from flaskr.api.v1.parsers import wbxt_messages_post_args
from flaskr.api.v1.cucm import cucm_get_version_api, cucm_perfmon_api
from flaskr.api.v1.core import core_users_api
from os import getenv
from wxc_sdk import WebexSimpleApi
from wxc_sdk.webhook import WebhookEvent, WebhookResource, WebhookEventType
logging.basicConfig(level=logging.INFO,format='%(asctime)s %(threadName)s %(levelname)s %(name)s %(message)s')
logging.getLogger('urllib3.connectionpool').setLevel(logging.INFO)# Change to DEBUG for detailed REST interaction output
logging.getLogger('wxc_sdk.rest').setLevel(logging.INFO)
log = logging.getLogger(__name__)
api = Namespace('wbxt', description='Webex APIs')# Create the Webex wxc_sdk API connection object for the Bot
wbx_bot_api = WebexSimpleApi(tokens=getenv('WBX_BOT_ACCESS_TOKEN'))# Get the details for the Bot account whose access token we are using
bot = wbx_bot_api.people.me()# Read an adaptive card JSON from a filewithopen('flaskr/api/v1/adaptive_card.json')as json_file:
adaptive_card_content = json.load(json_file)# Read environment variables
webhook_name = getenv('WBX_WEBHOOK_NAME')
webhook_url = getenv('WBX_WEBHOOK_URL')defdelete_webhooks_by_name():"""
Retrieve all webhooks that are owned by the Bot, then delete the ones matching our webhook name.
"""for webhook in wbx_bot_api.webhook.list():if webhook.name == webhook_name:
wbx_bot_api.webhook.webhook_delete(webhook_id=webhook.webhook_id)
log.info(f'Webhook "{webhook.name}" for "{webhook.resource}" deleted. Callback URL="{webhook.target_url}"')defcreate_webhooks():"""
Create the Webex webhooks using the Bot: one for messages, another one for the adaptive card
"""for resource_type in['messages','attachmentActions']:
webhook = wbx_bot_api.webhook.create(
resource=WebhookResource(resource_type),
event=WebhookEventType.created,
name=webhook_name,
target_url=webhook_url
)
log.info(f'Webhook "{webhook_name}" for "{webhook.resource}" created. Callback URL="{webhook_url}".')# Delete and recreate all webhooks with a given name every time the flask app is restarted
delete_webhooks_by_name()
create_webhooks()@api.route('/send_message')classwbxt_send_api(Resource):"""
Notification actions. Used for sending notifications/updates via the Bot
"""@api.expect(wbxt_messages_post_args, Validate=True)defpost(self):"""
Sends Message to a Webex Space (Room) by Room Title
"""
args = wbxt_messages_post_args.parse_args(request)# Populate the text key if not specifiedif'text'notin args:
args['text']=''# Get the Rooms list (only group type, not direct)
rooms =list(wbx_bot_api.rooms.list(type='group'))# Search through the Rooms to match the room titlefor room in rooms:if args['room_name'].strip()== room.title.strip():# Found the room, send a message
message = wbx_bot_api.messages.create(room_id=room.id, markdown=args['text'])return{'success':True,'message':f'Successfully sent message {message.id}','response':''}# Room was not foundreturn{'success':False,'message':f'Could not find Room "{args["room_name"]}"','response':''}@api.route('/events', doc=False)classwbxt_events_api(Resource):defsend_card(self, room_id:str):"""
Post the adaptive card to the specified room.
"""
wbx_bot_api.messages.create(
room_id=room_id,
text='Your Webex client cannot display this card',
attachments=[{'contentType':'application/vnd.microsoft.card.adaptive','content': adaptive_card_content
}])defrespond_to_message(self, wh_event: WebhookEvent):"""
Respond to a message to the bot. Retrieve the Room, message, and sender information from the webhook payload and
the corresponding message id
"""# Read the Webhook event Data
wh_data = wh_event.data
# Get the room to which this active card submission was posted to from the webhook data
room = wbx_bot_api.rooms.details(room_id=wh_data.room_id)# Look up the message from the data in the webhook
message = wbx_bot_api.messages.details(message_id=wh_data.id)# Look up the sender of this message
sender = wbx_bot_api.people.details(person_id=message.person_id)# ONLY reply if the message is from someone in the same Webex org AND the message is not from the Bot itselfif(bot.org_id == sender.org_id)and(message.person_id != bot.person_id):
log.info(f'Message received from {sender.display_name}:\n{message.text}')# Send a message in reply to the same room
wbx_bot_api.messages.create(room_id=room.id, text='Your available options are:')# Send the contact card to the room which originated the active card submission
self.send_card(room_id=room.id)defrespond_to_button_press(self, wh_event: WebhookEvent):"""
Respond to an adaptive card submission
"""# Read the Webhook event Data
wh_data = wh_event.data
# Get the room to which this message was posted to from the webhook data
room = wbx_bot_api.rooms.details(room_id=wh_data.room_id)# Look up the attachment from the data in the webhook
attachment_action = wbx_bot_api.attachment_actions.details(action_id=wh_data.id)# Look up the sender of the submission/attachment
sender = wbx_bot_api.people.details(person_id=attachment_action.person_id)# Only reply if the submission was from someone in the same Webex org as the Botif bot.org_id == sender.org_id:
log.info(f'attachmentAction received from {sender.display_name}:\n'f'{attachment_action.inputs}')try:# Evaluate the contents of the submission and take action on themif attachment_action.inputs['action']=='get_ucm_version':# Retrieve the CUCM version from the API created previously
log.info('Creating response for get_ucm_version')
cucm_version_data = cucm_get_version_api.get(Resource).json
if cucm_version_data['success']:
wbx_bot_api.messages.create(
room.id,
markdown=f'The Unified CM version is **{cucm_version_data["version"]}**')else:
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'Failed to retrieve The Unified CM version\n \
Error Message:\n \
```{cucm_version_data["message"]}```'
)# Take action on the 'get_ucm_reg_phones' actionelif attachment_action.inputs['action']=='get_ucm_reg_phones':# Retrieve the CUCM registered phones, in this case, we'll add up the RegisteredHardwarePhones and# RegisteredOtherStationDevices perfmon counters
log.info('Creating response for get_ucm_reg_phones')
perfmon_counters =['Cisco CallManager\RegisteredOtherStationDevices','Cisco CallManager\RegisteredHardwarePhones']
cucm_perfmon_data = cucm_perfmon_api.post(Resource, perfmon_counters=perfmon_counters).json
if cucm_perfmon_data['success']:
num_reg_devices =sum(item['Value']for item in cucm_perfmon_data['perfmon_counters_result'])
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'The number of registered devices is **{num_reg_devices}**')else:
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'Failed to retrieve the number of registered devices\n \
Error Message:\n \
```{cucm_perfmon_data["message"]}```'
)# Take action on the 'user_search' actionelif attachment_action.inputs['action']=='user_search':# Call the core user search, this will search both CUCM and Webex Calling
log.info(f'User Search for: {attachment_action.inputs["user"]}')
user_data = core_users_api.get(Resource, userid=attachment_action.inputs['user']).json
if user_data['success']:
wbx_bot_api.messages.create(
room.id,
text=f'User {attachment_action.inputs["user"]} is configured in {user_data["phonesystem"]}')# Output full user data if Details checkbox selectedif attachment_action.inputs['user_details'].lower()=='true':if'wbxc_user_data'in user_data:
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'Webex Calling user details:\n'f'```\n{json.dumps(user_data["wbxc_user_data"], indent=4)}\n```')if'cucm_user_data'in user_data:
wbx_bot_api.messages.create(
room_id=room.id,
markdown=f'Unified CM user details:\n'f'```\n{json.dumps(user_data["cucm_user_data"], indent=4)}\n```')else:
wbx_bot_api.messages.create(
room_id=room.id,
text=f'User {attachment_action.inputs["user"]} lookup failed: {user_data["message"]}')# Any error in looking up a value will simply be ignored.except KeyError:pass# Post the adaptive card again, since quite a bit of data may have been returned
self.send_card(room_id=room.id)# Receive a Webex Webhook POSTdefpost(self):"""
Receives a Webex Webhook POST. Evaluates the event, in all cases, on will only process 'created' events, but
depending on the resource type (messages or attachmentActions), will respond accordingly
"""try:# Log inbound POST data
log.debug(f'Received inbound POST with data:\n'+ json.dumps(request.json, indent=4))# Parse the incoming Webhook request as a WebhookEvent
webhook_event = WebhookEvent.parse_obj(request.json)# Handle a new message eventif webhook_event.resource =='messages':
self.respond_to_message(webhook_event)# Handle an active card submission eventelif webhook_event.resource =='attachmentActions':
self.respond_to_button_press(webhook_event)except Exception as e:
log.error(f'Failed to parse: {e}')return'Ok'
It has a few actions, that will generate data that you can see when when a user clicks
a button on the card. If you like, you can copy the contents of that file and paste
it in a message to the Webhook
Echo Bot (webhook-echo@webex.bot) and see what it looks like. Shortly, your web server will be responding
with the same content, so you can see it then.