{"id":24,"date":"2025-04-02T19:31:39","date_gmt":"2025-04-02T19:31:39","guid":{"rendered":"https:\/\/phonesstillexist.com\/?p=24"},"modified":"2025-04-02T19:50:24","modified_gmt":"2025-04-02T19:50:24","slug":"zoom-websockets-implementing-a-priority-incident-hotline","status":"publish","type":"post","link":"https:\/\/phonesstillexist.com\/index.php\/2025\/04\/02\/zoom-websockets-implementing-a-priority-incident-hotline\/","title":{"rendered":"Zoom Websockets: Implementing a Priority Incident Hotline"},"content":{"rendered":"\n<figure class=\"wp-block-image size-large\"><img loading=\"lazy\" decoding=\"async\" width=\"1024\" height=\"683\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/04\/assets_task_01jqw02vjze8qaztaqxhw8dyxn_img_0-1024x683.webp\" alt=\"\" class=\"wp-image-72\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/04\/assets_task_01jqw02vjze8qaztaqxhw8dyxn_img_0-1024x683.webp 1024w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/04\/assets_task_01jqw02vjze8qaztaqxhw8dyxn_img_0-300x200.webp 300w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/04\/assets_task_01jqw02vjze8qaztaqxhw8dyxn_img_0-768x512.webp 768w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/04\/assets_task_01jqw02vjze8qaztaqxhw8dyxn_img_0.webp 1536w\" sizes=\"auto, (max-width: 1024px) 100vw, 1024px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">We&#8217;ve all been there\u2014one minute you&#8217;re enjoying your coffee and planning your day, savoring a rare moment of calm, and then BAM! Suddenly, alerts start flooding in, your email goes into overdrive, and your peaceful day is completely upended. So much for getting caught up! Today, we&#8217;re diving into the world of Zoom Websockets to build a &#8220;Priority Incident Hotline.&#8221; This meeting is designed for our fictional IT team to handle high-priority incidents. When the incident manager\u2014or any authorized user\u2014joins, our system will automatically call in all the essential troubleshooting staff and send out an email with the meeting link. Let\u2019s jump right in!<\/p>\n\n\n\n<h2 class=\"wp-block-heading\">Getting Started<\/h2>\n\n\n\n<p class=\"wp-block-paragraph\">Before we begin, you&#8217;ll need a Zoom account with the privileges to create apps in the Zoom Marketplace. For this example, I\u2019m using my free Zoom account.<\/p>\n\n\n\n<h3 class=\"wp-block-heading\">Step 1: Create Your Meeting<\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">Your meeting will be a recurring one with no fixed time, so it&#8217;s always available when needed. Follow these steps:<\/p>\n\n\n\n<ol class=\"wp-block-list\">\n<li><strong>Log in to the Zoom website.<\/strong><\/li>\n\n\n\n<li><strong>Click &#8220;Meetings&#8221;<\/strong> on the left-hand menu.<\/li>\n\n\n\n<li><strong>Click &#8220;Schedule a Meeting.&#8221;<\/strong><\/li>\n\n\n\n<li><strong>Give your meeting a topic.<\/strong><\/li>\n\n\n\n<li><strong>Check the &#8220;Recurring Meeting&#8221; box<\/strong> and select <strong>&#8220;No Fixed Time&#8221;<\/strong> from the drop-down menu.<\/li>\n\n\n\n<li><strong>Click &#8220;Save.&#8221;<\/strong><\/li>\n<\/ol>\n\n\n\n<h3 class=\"wp-block-heading\">Step 2: Create Your App<\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">Now that you have your meeting set up, it&#8217;s time to create the app that will handle our WebSocket connection.<\/p>\n\n\n\n<ol class=\"wp-block-list\">\n<li><strong>Navigate to the App Marketplace:<\/strong>\n<ul class=\"wp-block-list\">\n<li>Under the Admin menu, select <strong>Advanced<\/strong> and then <strong>App Marketplace<\/strong>.<\/li>\n<\/ul>\n<\/li>\n\n\n\n<li><strong>Build a New App:<\/strong>\n<ul class=\"wp-block-list\">\n<li>Click the <strong>Develop<\/strong> dropdown in the top-right corner and select <strong>Build App<\/strong>.<\/li>\n\n\n\n<li>Choose <strong>Server-to-Server OAuth App<\/strong> and click <strong>Create<\/strong>.<\/li>\n<\/ul>\n<\/li>\n<\/ol>\n\n\n\n<figure class=\"wp-block-image size-full\"><img loading=\"lazy\" decoding=\"async\" width=\"769\" height=\"718\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/Zoom-Server-to-Server-oauth-app.jpg\" alt=\"\" class=\"wp-image-27\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/Zoom-Server-to-Server-oauth-app.jpg 769w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/Zoom-Server-to-Server-oauth-app-300x280.jpg 300w\" sizes=\"auto, (max-width: 769px) 100vw, 769px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">4.  <strong>Name Your App:<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Give your app a memorable name and click <strong>Create<\/strong>.<\/p>\n\n\n\n<figure class=\"wp-block-image size-full\"><img loading=\"lazy\" decoding=\"async\" width=\"769\" height=\"281\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Name.jpg\" alt=\"\" class=\"wp-image-29\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Name.jpg 769w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Name-300x110.jpg 300w\" sizes=\"auto, (max-width: 769px) 100vw, 769px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">5. <strong>Save Your Credentials:<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Once your app is created, make a note of the credentials. You\u2019ll need these later for API calls and establishing the Websocket connection.<\/p>\n\n\n\n<figure class=\"wp-block-image size-large\"><img loading=\"lazy\" decoding=\"async\" width=\"1024\" height=\"460\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Credentials-1024x460.jpg\" alt=\"\" class=\"wp-image-30\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Credentials-1024x460.jpg 1024w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Credentials-300x135.jpg 300w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Credentials-768x345.jpg 768w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Credentials-1536x689.jpg 1536w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/App-Credentials.jpg 1805w\" sizes=\"auto, (max-width: 1024px) 100vw, 1024px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">6. Click Continue to complete setting up your application<\/p>\n\n\n\n<p class=\"wp-block-paragraph\">7.  Complete the Basic Information section and click Continue.<\/p>\n\n\n\n<p class=\"wp-block-paragraph\">8.  <strong>Configure Websocket Connection:<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">In the Features section, note your <strong>Secret Token<\/strong> and <strong>Verification Token<\/strong>\u2014you\u2019ll need these for your future reference.<\/p>\n\n\n\n<figure class=\"wp-block-image size-large\"><img loading=\"lazy\" decoding=\"async\" width=\"1024\" height=\"317\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/Token-1024x317.jpg\" alt=\"\" class=\"wp-image-31\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/Token-1024x317.jpg 1024w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/Token-300x93.jpg 300w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/Token-768x238.jpg 768w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/Token.jpg 1461w\" sizes=\"auto, (max-width: 1024px) 100vw, 1024px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">9.  <strong>Enable Event Subscription:<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Turn on the <strong>Event Subscription Slider<\/strong> and select <strong>Websocket<\/strong>.<\/p>\n\n\n\n<figure class=\"wp-block-image size-large\"><img loading=\"lazy\" decoding=\"async\" width=\"1024\" height=\"675\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/event-subscription-1024x675.jpg\" alt=\"\" class=\"wp-image-32\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/event-subscription-1024x675.jpg 1024w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/event-subscription-300x198.jpg 300w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/event-subscription-768x506.jpg 768w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/event-subscription.jpg 1103w\" sizes=\"auto, (max-width: 1024px) 100vw, 1024px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">10. <strong>Add Your Events:<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Choose the events you want to subscribe to. Although Zoom offers many events, we\u2019re focusing on meeting events\u2014especially the &#8220;start meeting&#8221; event. For simplicity, you might select all meeting events, but for production, it&#8217;s best to subscribe only to what you need.   <\/p>\n\n\n\n<figure class=\"wp-block-image size-full\"><img loading=\"lazy\" decoding=\"async\" width=\"936\" height=\"738\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/events.jpg\" alt=\"\" class=\"wp-image-33\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/events.jpg 936w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/events-300x237.jpg 300w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/events-768x606.jpg 768w\" sizes=\"auto, (max-width: 936px) 100vw, 936px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">11.  <strong>Record the Websocket URL:<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Click <strong>Done<\/strong> when you\u2019re finished selecting your events. This action will generate a Websocket URL (wss:\/\/) that you&#8217;ll use later in your code. Make a note of this URL.<\/p>\n\n\n\n<figure class=\"wp-block-image size-full\"><img loading=\"lazy\" decoding=\"async\" width=\"1008\" height=\"645\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/endpoint-url.jpg\" alt=\"\" class=\"wp-image-34\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/endpoint-url.jpg 1008w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/endpoint-url-300x192.jpg 300w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/endpoint-url-768x491.jpg 768w\" sizes=\"auto, (max-width: 1008px) 100vw, 1008px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">12.  <strong>Configure Scopes:<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Scopes are the permissions your app token needs to perform specific tasks. Ensure you select the scopes that correspond with the events you&#8217;ve chosen.<\/p>\n\n\n\n<figure class=\"wp-block-image size-large\"><img loading=\"lazy\" decoding=\"async\" width=\"1024\" height=\"600\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/scopes-1024x600.jpg\" alt=\"\" class=\"wp-image-36\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/scopes-1024x600.jpg 1024w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/scopes-300x176.jpg 300w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/scopes-768x450.jpg 768w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/scopes.jpg 1399w\" sizes=\"auto, (max-width: 1024px) 100vw, 1024px\" \/><\/figure>\n\n\n\n<p class=\"wp-block-paragraph\">13.  <strong>Activate Your App:<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Finally, click <strong>Activate Your App<\/strong> to make it live. Congratulations\u2014you\u2019re now ready to dive into the coding part!<\/p>\n\n\n\n<figure class=\"wp-block-image size-full\"><img loading=\"lazy\" decoding=\"async\" width=\"1007\" height=\"603\" src=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/activation.jpg\" alt=\"\" class=\"wp-image-37\" srcset=\"https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/activation.jpg 1007w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/activation-300x180.jpg 300w, https:\/\/phonesstillexist.com\/wp-content\/uploads\/2025\/03\/activation-768x460.jpg 768w\" sizes=\"auto, (max-width: 1007px) 100vw, 1007px\" \/><\/figure>\n\n\n\n<h2 class=\"wp-block-heading\">The Code<\/h2>\n\n\n\n<p class=\"wp-block-paragraph\">For this project, we\u2019re harnessing the power of Python. We\u2019ll use the <code>requests<\/code> package to interact with the Zoom API, the <code>websocket<\/code> package to manage our WebSocket connections, and <code>smtplib<\/code> for email notifications. To ensure everything runs smoothly, we&#8217;ll also leverage <code>APScheduler<\/code> to manage Zoom access tokens and keep our WebSocket connection alive.<\/p>\n\n\n\n<h3 class=\"wp-block-heading\">Step 1: Prepare the Environment<\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">In this example, I\u2019m using Debian, but feel free to use whichever system you\u2019re most comfortable with.<\/p>\n\n\n\n<ol class=\"wp-block-list\">\n<li><strong>Create a Dedicated User:<\/strong>\n<ul class=\"wp-block-list\">\n<li>Log in as a user with sudo privileges and run:<code>sudo adduser zoomwebsocket<\/code><\/li>\n\n\n\n<li>Follow the prompts to set and confirm the password, and accept the defaults.<\/li>\n<\/ul>\n<\/li>\n\n\n\n<li><strong>Add the User to the Sudo Group:<\/strong>\n<ul class=\"wp-block-list\">\n<li>Execute: <code>sudo usermod -aG sudo zoomwebsocket<\/code><\/li>\n<\/ul>\n<\/li>\n\n\n\n<li><strong>Install Required Software:<\/strong>\n<ul class=\"wp-block-list\">\n<li>Update your package lists: <code>sudo apt update<\/code><\/li>\n\n\n\n<li>Install Python 3, pip, and virtualenv (if they\u2019re not already installed):<\/li>\n\n\n\n<li><code>sudo apt install python3 sudo apt install pip sudo apt install virtualenv<\/code><\/li>\n<\/ul>\n<\/li>\n\n\n\n<li><strong>Set Up a Virtual Environment:<\/strong>\n<ul class=\"wp-block-list\">\n<li>Login as your new user.<\/li>\n\n\n\n<li>Create a virtual environment in your home directory: <code>virtualenv ZoomWebSocket<\/code><\/li>\n\n\n\n<li>Change into the new directory: <code>cd ZoomWebSocket<\/code><\/li>\n\n\n\n<li>Activate the virtual environment: <code>source bin\/activate<\/code><\/li>\n\n\n\n<li>You should now see your prompt change to indicate that you\u2019re inside the virtual environment.<\/li>\n<\/ul>\n<\/li>\n\n\n\n<li><strong>Install Project Dependencies:<\/strong>\n<ul class=\"wp-block-list\">\n<li>Install the required packages using <code>requirements.txt<\/code>: <\/li>\n\n\n\n<li><code>pip install -r requirements.txt<\/code><\/li>\n<\/ul>\n<\/li>\n<\/ol>\n\n\n\n<p class=\"wp-block-paragraph\">With our environment set up and dependencies installed, we\u2019re ready to dive into the code. In this section, we\u2019ll build the core of our Zoom WebSocket Priority Incident Hotline.<\/p>\n\n\n\n<ol class=\"wp-block-list\">\n<li><strong>The Imports<\/strong><\/li>\n<\/ol>\n\n\n\n<p class=\"wp-block-paragraph\">We\u2019ll start by importing the necessary packages. These imports allow us to work with the Zoom API, handle WebSocket connections, send emails, and schedule recurring tasks.<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><code>import os<br>import sys<br>import json<br>import datetime<br>import signal<br>import logging<br>from logging.handlers import RotatingFileHandler<br>import smtplib<br>from email.message import EmailMessage<br>import requests<br>from requests.auth import HTTPBasicAuth<br>import websocket<br>from apscheduler.schedulers.background import BackgroundScheduler<br>from dotenv import load_dotenv<\/code><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">We will also need to <code>load_dotenv() <\/code>to read our environmental variables.<\/p>\n\n\n\n<p class=\"wp-block-paragraph\">2.  <strong>The Class<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Next, we define our <code>ZoomWebSocket<\/code> class. This class encapsulates our configuration, logging, API session management, and scheduling functionality.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>class ZoomWebSocket:\n    ACCOUNT_ID: str = os.getenv(\"ACCOUNT_ID\")\n    CLIENT_ID: str = os.getenv(\"CLIENT_ID\")\n    CLIENT_SECRET: str = os.getenv(\"CLIENT_SECRET\")\n    meeting_id: str = \"your meeting id\"\n\n\n# Define a list of (user, phone_number) tuples, IE numbers we want to call when the meeting starts\nparticipants = &#91;\n    (\"IT Help Desk\", \"15554560001\")\n    # (\"Network Oncall Engineer\", \"15554561000\"),\n    # (\"Server Oncall Enginner\", \"15554562000\")\n]\n\ndef __init__(self) -&gt; None:\n    \"\"\"Initialize the ZoomWebSocket instance, set up logging, session, and scheduler.\"\"\"\n    self.start_time: datetime.datetime = datetime.datetime.now()\n\n    # Set up rotating log handler\n    log_file = \"ZoomWebSocketLog.txt\"\n    log_handler = RotatingFileHandler(log_file, maxBytes=10 * 1024 * 1024, backupCount=3)\n    log_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))\n    self.logger = logging.getLogger(\"ZoomWebSocket\")\n    self.logger.setLevel(logging.INFO)\n    self.logger.addHandler(log_handler)\n    self.logger.info(f\"ZoomWebSocket started at {self.start_time}\")\n\n    # Create a persistent requests session\n    self.session = requests.Session()\n\n    # Get the access token\n    self.ACCESS_TOKEN = self.get_access_token()\n    self.logger.info(\"Access token retrieved.\")\n\n    # Set up the scheduler for heartbeat and token refresh\n    self.scheduler = BackgroundScheduler()\n    self.scheduler.add_job(self.send_heartbeat, 'interval', seconds=25)\n    self.scheduler.add_job(self.refresh_token, 'interval', minutes=55)\n    self.scheduler.start()<\/code><\/pre>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>3. Class Methods<\/strong><\/p>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>get_access_token():<\/strong> <\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">To interact with the Zoom API, we need an access token that will expire every hour. The following method retrieves this token and sets the expire time<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def get_access_token(self) -&gt; str:\n\"\"\"\nRetrieve the access token from Zoom.   \n\n Returns:\n        str: The access token if successful, otherwise an empty string.\n    \"\"\"\n    url = \"https:\/\/zoom.us\/oauth\/token\"\n    data = {\n        'grant_type': 'account_credentials',\n        'account_id': self.ACCOUNT_ID\n    }\n    try:\n        response = self.session.post(\n            url,\n            auth=HTTPBasicAuth(self.CLIENT_ID, self.CLIENT_SECRET),\n            data=data,\n            timeout=10\n        )\n        response.raise_for_status()\n        json_response = response.json()\n        self.access_token_expires = datetime.datetime.now() + datetime.timedelta(hours=1)\n        self.logger.info(f\"Access Token Expires {self.access_token_expires}\")\n        return json_response.get(\"access_token\", \"\")\n    except requests.exceptions.RequestException as e:\n        error_msg = f\"Error fetching Access Token: {str(e)}\"\n        print(error_msg)\n        self.logger.error(error_msg)\n        return \"\"<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>refresh_token():<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\"> Since the access token expires every 60 minutes we&#8217;ll need a method to refresh the access token.  The <code>BackgroundScheduler()<\/code> is setup during the class initialization to call this function every 55 minutes.  <\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def refresh_token(self) -&gt; None:\n        \"\"\"Refresh the access token using the get_access_token method.\"\"\"\n        self.logger.info(\"Refreshing access token...\")\n        new_token = self.get_access_token()\n        if new_token:\n            self.ACCESS_TOKEN = new_token\n            self.logger.info(\"New access token acquired.\")\n        else:\n            self.logger.error(\"Failed to refresh access token.\")<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>send_heartbeat()<\/strong>:  <\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">To keep the WebSocket connection alive we need to send a heartbeat message every 30 seconds. The<code> BackgroundScheduler()<\/code> is setup during the class initialization to call this function every 25 seconds.  Just a little sooner then 30 seconds to allow for any potential hang-ups along the network path.  <\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def send_heartbeat(self) -&gt; None:\n        \"\"\"Send a heartbeat message through the WebSocket to Zoom.\"\"\"\n        try:\n            if hasattr(self, 'ws'):\n                self.ws.send('{ \"module\": \"heartbeat\" }')\n                self.logger.info(\"Heartbeat sent to Zoom\")\n            else:\n                self.logger.warning(\"WebSocket not connected; heartbeat not sent.\")\n        except Exception as e:\n            self.logger.error(\"Error sending heartbeat: \" + str(e))<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>process_message()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>process_message()<\/code> method is triggered every time the WebSocket receives a message. This is where we determine the type of the incoming message and execute the appropriate action. For example, if the message type is <code>meeting.started<\/code> and it corresponds to our specific meeting ID, our system will automatically initiate a call-out to notify the relevant team. You can also add support for other events\u2014like <code>meeting.ended<\/code> or <code>meeting.participant_left<\/code>\u2014by implementing dedicated helper methods for each event type. This flexible approach allows you to tailor the response actions to suit your needs.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def process_message(self, message: str) -&gt; None:\n        \"\"\"\n        Process a message received via WebSocket.\n\n        Args:\n            message (str): The raw JSON string received.\n        \"\"\"\n        try:\n            j = json.loads(message)\n            self.logger.info(f\"{j.get('module')} message received\")\n            print(json.dumps(j, indent=4))\n            self.logger.info(json.dumps(j, indent=4))\n        except Exception as e:\n            self.logger.error(\"Error processing message JSON: \" + str(e))\n            return\n\n        if j.get('module') == 'message':\n            try:\n                content_string = j&#91;'content']\n                content = json.loads(content_string)\n            except Exception as e:\n                self.logger.error(\"Error processing content JSON: \" + str(e))\n                return\n\n            try:\n                event = content.get('event')\n                if event == \"meeting.participant_left\":\n                    self.process_leave_event(content)\n                elif event == \"meeting.ended\":\n                    self.process_meeting_end_event(content)\n                elif event == \"meeting.participant_joined\":\n                    self.process_participant_joined_event(content)\n                elif event == \"meeting.started\":\n                    self.process_meeting_started_event(content)\n            except Exception as e:\n                self.logger.error(\"Error processing event: \" + str(e))\n        elif j.get('module') == 'heartbeat':\n            self.logger.info(f\"Received heartbeat response. Success = {str(j.get('success'))}\")\n            print(f\"{datetime.datetime.now()} - Received heartbeat response. Success = {str(j.get('success'))}\")<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>process_leave_event()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>process_leave_event()<\/code> method is triggered whenever we receive a <code>meeting.participant_left<\/code> message from Zoom. I&#8217;ve observed that sometimes, even after all participants leave a meeting, Zoom keeps the meeting active for an additional five minutes with no one in attendance. While this behavior isn\u2019t problematic in itself, it poses a challenge for our application: if someone rejoins during that window, the <code>meeting.started<\/code> event won\u2019t fire, and our call-out mechanism won\u2019t be activated.<\/p>\n\n\n\n<p class=\"wp-block-paragraph\">To address this, I implemented logic within <code>process_leave_event()<\/code> that checks the meeting\u2019s participant count whenever a user leaves. If the count reaches zero, the method proactively terminates the meeting using the Zoom API. This ensures that if a new incident occurs shortly afterward, the <code>meeting.started<\/code> event will trigger again, and the necessary call-outs will be placed <\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def process_leave_event(self, content: dict) -&gt; None:\n        \"\"\"\n        Process an event when a participant leaves a meeting.\n\n        Args:\n            content (dict): The event content dictionary.\n        \"\"\"\n        try:\n            meeting = content&#91;'payload']&#91;'object']&#91;'id']\n            user = content&#91;'payload']&#91;'object']&#91;'participant']&#91;'user_name']\n            topic = content&#91;'payload']&#91;'object']&#91;'topic']\n            if meeting == self.meeting_id:\n                print(f\"{user} has left the meeting {topic}!\")\n                self.logger.info(f\"{user} has left the meeting {topic}!\")\n                details = self.get_meeting_details(self.meeting_id)\n                participant_count = details.get(\"participants\", 0)\n                if participant_count == 0:\n                    self.logger.info(\"There are zero participants in the meeting. Ending call.\")\n                    self.end_meeting(self.meeting_id)\n        except Exception as e:\n            self.logger.error(\"Error in process_leave_event: \" + str(e))<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>process_meeting_end_event()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>process_meeting_end_event()<\/code> method is invoked whenever we receive a <code>meeting.ended<\/code> message from Zoom. In this method, we simply log and print a message indicating that the meeting has ended. This helps keep our system informed of the meeting status and can be expanded later if further actions are required when a meeting concludes<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def process_meeting_end_event(self, content: dict) -&gt; None:\n        \"\"\"\n        Process an event when a meeting ends.\n\n        Args:\n            content (dict): The event content dictionary.\n        \"\"\"\n        try:\n            meeting = content&#91;'payload']&#91;'object']&#91;'id']\n            print(f\"{meeting} has ended!\")\n            self.logger.info(f\"{meeting} has ended!\")\n        except Exception as e:\n            self.logger.error(\"Error in process_meeting_end_event: \" + str(e))<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>process_participant_joined_event()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>process_participant_joined_event()<\/code> method is called when we receive a <code>meeting.participant_joined<\/code> message from Zoom. In its current form, this method simply prints and logs a message to indicate that a user has joined the meeting. This basic implementation lays the groundwork for future enhancements. For instance, if a help desk agent answers the call-out and joins the meeting, we could trigger additional notifications to keep the rest of the team informed.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def process_participant_joined_event(self, content: dict) -&gt; None:\n        \"\"\"\n        Process an event when a participant joins a meeting.\n\n        Args:\n            content (dict): The event content dictionary.\n        \"\"\"\n        try:\n            user = content&#91;'payload']&#91;'object']&#91;'participant']&#91;'user_name']\n            topic = content&#91;'payload']&#91;'object']&#91;'topic']\n            print(f\"{user} has joined the meeting {topic}!\")\n            self.logger.info(f\"{user} has joined the meeting {topic}!\")\n        except Exception as e:\n            self.logger.error(\"Error in process_participant_joined_event: \" + str(e))<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>process_meeting_started_event()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">Finally! This is the bread and butter of our application. The <code>process_meeting_started_event()<\/code> method is invoked when we receive a <code>meeting.started<\/code> event from Zoom. Within this method, we first verify that the meeting ID in the event matches the meeting ID of our &#8220;Priority Incident Hotline.&#8221; If it does, we call the <code>place_call<\/code> function, which leverages the Zoom API to automatically dial out to the users listed in our participants list. This ensures that the right team members are alerted immediately when an incident meeting is initiated. Here we can also use the <code>send_email <\/code>method to send the users the url to the meeting.  <\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def process_meeting_started_event(self, content: dict) -&gt; None:\n        \"\"\"\n        Process an event when a meeting starts.\n\n        Args:\n            content (dict): The event content dictionary.\n        \"\"\"\n        try:\n            meeting = content&#91;'payload']&#91;'object']&#91;'id']\n            print(f\"Meeting ID: {meeting} has started!\")\n            self.logger.info(f\"Meeting ID: {meeting} has started!\")\n            if meeting == self.meeting_id:\n                print(\"Meeting has started. Placing calls to participants!\")\n                self.logger.info(\"Meeting has started. Placing calls to participants!\")\n                for user, phone in self.participants:\n                    self.place_call(meeting, user, phone)\n\t\t    #Send email with invite link uncomment to use\n            '''email_msg=\"Priority Incident Meeting has started please join the meeting\\n https:\/\/zoom.us\/your meeting url\"\n            self.send_mail(to_email=&#91;'email1@example.com', 'email2@example.com'],\n                subject=\"Priority Incident Meeting has started!\", message=email_msg)'''\n        except Exception as e:\n            self.logger.error(\"Error in process_meeting_started_event: \" + str(e))<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\">place_call()<\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>place_call<\/code> method utilizes the Zoom API to instruct Zoom to place a call out to the specified phone number.  Be sure to include your current access token in the API call, as this token is required for authentication with Zoom&#8217;s API.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def place_call(self, meeting_ID: str, name: str, phone_number: str) -&gt; None:\n        \"\"\"\n        Place a call to a participant to join the meeting.\n\n        Args:\n            meeting_ID (str): The meeting ID.\n            name (str): The invitee's name.\n            phone_number (str): The invitee's phone number.\n        \"\"\"\n        meeting_options = {\n            \"method\": \"participant.invite.callout\",\n            \"params\": {\n                \"invitee_name\": str(name),\n                \"phone_number\": str(phone_number),\n                \"invite_options\": {\n                    \"require_greeting\": False,\n                    \"require_pressing_one\": False\n                },\n            }\n        }\n        headers = {\n            'authorization': 'Bearer ' + self.ACCESS_TOKEN,\n            'content-type': 'application\/json'\n        }\n        url = f'https:\/\/api.zoom.us\/v2\/live_meetings\/{meeting_ID}\/events'\n        try:\n            self.logger.info(f\"Placing call to {phone_number} to join {name} for meeting {meeting_ID}\")\n            response = self.session.patch(url, headers=headers, data=json.dumps(meeting_options))\n            print(response.text)\n            self.logger.info(f\"Place call request returned {str(response)}\")\n        except Exception as e:\n            self.logger.error(\"Error placing call: \" + str(e))<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>get_meeting_details()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>get_meeting_details()<\/code> helper method retrieves the details of the specified meeting using the Zoom API. Although this API call returns a wealth of information, in our case, we primarily use it to determine the current participant count.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def get_meeting_details(self, meeting_id: str) -&gt; dict:\n        \"\"\"\n        Retrieve meeting details.\n\n        Args:\n            meeting_id (str): The meeting ID.\n            \n        Returns:\n            dict: The meeting details or an empty dictionary if an error occurs.\n        \"\"\"\n        headers = {\n            'authorization': 'Bearer ' + self.ACCESS_TOKEN,\n            'content-type': 'application\/json'\n        }\n        try:\n            response = self.session.get(f'https:\/\/api.zoom.us\/v2\/metrics\/meetings\/{meeting_id}', headers=headers)\n            if response.status_code == 200:\n                self.logger.info(f\"Successfully retrieved meeting details for meeting {meeting_id}\")\n            else:\n                self.logger.error(\"Error retrieving meeting details: \" + response.text)\n            return response.json()\n        except Exception as e:\n            self.logger.error(\"Error in get_meeting_details: \" + str(e))\n            return {}<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>end_meeting()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>end_meeting<\/code> method uses the Zoom API to end the meeting specified.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code> def end_meeting(self, meeting_id: str) -&gt; int:\n        \"\"\"\n        End a meeting.\n\n        Args:\n            meeting_id (str): The meeting ID.\n            \n        Returns:\n            int: The HTTP status code from the request.\n        \"\"\"\n        action = {\"action\": \"end\"}\n        headers = {\n            'authorization': 'Bearer ' + self.ACCESS_TOKEN,\n            'content-type': 'application\/json'\n        }\n        try:\n            response = self.session.put(f'https:\/\/api.zoom.us\/v2\/meetings\/{meeting_id}\/status',\n                                        headers=headers, data=json.dumps(action))\n            if response.status_code == 204:\n                self.logger.info(f\"Successfully ended meeting {meeting_id}\")\n            else:\n                self.logger.error(\"Error ending meeting: \" + response.text)\n            return response.status_code\n        except Exception as e:\n            self.logger.error(\"Exception in end_meeting: \" + str(e))\n            return -1<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>send_mail()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The<code> send_mail()<\/code>method is used if you want to send an email in response to one of the Zoom events<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code># Update the IP of your SMTP server if you want to use this function to send an email\n    def send_mail(self, to_email: list&#91;str], subject: str, message: str,\n                  server: str = '1.2.3.4', from_email: str = 'critical.incident@example.com') -&gt; None:\n        \"\"\"\n        Send an email.\n\n        Args:\n            to_email (list&#91;str]): List of recipient email addresses.\n            subject (str): Subject of the email.\n            message (str): Body of the email.\n            server (str, optional): SMTP server address.\n            from_email (str, optional): Sender email address.\n        \"\"\"\n        try:\n            msg = EmailMessage()\n            msg&#91;'Subject'] = subject\n            msg&#91;'From'] = from_email\n            msg&#91;'To'] = ', '.join(to_email)\n            msg.set_content(message)\n            self.logger.info(\"Sending email.\")\n            with smtplib.SMTP(server) as smtp_server:\n                smtp_server.set_debuglevel(1)\n                smtp_server.send_message(msg)\n            print('Successfully sent the mail.')\n        except Exception as e:\n            self.logger.error(\"Error sending email: \" + str(e))\n<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>shutdown()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>shutdown<\/code> method is designed to gracefully exit the application. It accomplishes this by shutting down the scheduler\u2014which manages keep-alives and token refresh tasks\u2014and closing the WebSocket connection. This ensures that all resources are properly released when the application is stopped.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def shutdown(self) -&gt; None:\n        \"\"\"Gracefully shutdown the scheduler and close the WebSocket connection.\"\"\"\n        self.logger.info(\"Initiating graceful shutdown...\")\n        if self.scheduler:\n            self.scheduler.shutdown(wait=False)\n            self.logger.info(\"Scheduler shut down.\")\n        if hasattr(self, 'ws'):\n            self.logger.info(\"Closing WebSocket connection...\")\n            try:\n                self.ws.close()\n            except Exception as e:\n                self.logger.error(\"Error closing WebSocket: \" + str(e))\n        self.logger.info(\"Shutdown complete.\")<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>run()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>run<\/code> method is responsible for opening the WebSocket connection. It uses the WebSocket Secure (wss:\/\/) URL defined in your Zoom application\u2014ensuring that your access token is appended to the URL string (e.g., <code>\"wss:\/\/ws.zoom.us\/ws?subscriptionId=blahblahblah&amp;access_token=\"<\/code>). Additionally, this method sets up handlers for specific WebSocket events, including <code>on_open<\/code>, <code>on_message<\/code>, <code>on_error<\/code>, and <code>on_close<\/code>. These callbacks ensure that the connection is properly managed and that incoming messages trigger the appropriate actions in your application.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def run(self) -&gt; None:\n        \"\"\"Establish and maintain the WebSocket connection.\"\"\"\n        websocket.enableTrace(False)\n\t\t# Copy your wss endpoint url from your Zoom application settings.  \n\t\t# Be sure to include &amp;access_token= at the end of your url\n        ws = websocket.WebSocketApp(\n            \"wss:\/\/ws.zoom.us\/ws?subscriptionId=m98soh8RRIOsx_WeR5AIBA&amp;access_token=\" + str(self.ACCESS_TOKEN),\n            on_open=self.on_open,\n            on_message=self.on_message,\n            on_error=self.on_error,\n            on_close=self.on_close\n        )\n        try:\n            ws.run_forever(reconnect=5)\n        except Exception as e:\n            self.logger.error(\"Error in WebSocket run_forever: \" + str(e))<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>WebSocket Handlers<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">These handlers define what actions we take in regards to specific WebSocket events including <code>on_message<\/code>, <code>on_error<\/code>, <code>on_close<\/code>, and<code> on_open<\/code><\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>def on_message(self, ws: websocket.WebSocketApp, message: str) -&gt; None:\n        \"\"\"WebSocket on_message handler.\"\"\"\n        self.process_message(message)\n\n    def on_error(self, ws: websocket.WebSocketApp, error: Exception) -&gt; None:\n        \"\"\"WebSocket on_error handler.\"\"\"\n        error_msg = f\"Encountered error: {str(error)}\"\n        print(error_msg)\n        self.logger.error(error_msg)\n\n    def on_close(self, ws: websocket.WebSocketApp, close_status_code: int, close_msg: str) -&gt; None:\n        \"\"\"WebSocket on_close handler.\"\"\"\n        print(\"Connection closed\")\n        self.logger.info(\"Connection closed\")\n\n    def on_open(self, ws: websocket.WebSocketApp) -&gt; None:\n        \"\"\"WebSocket on_open handler.\"\"\"\n        print(\"Connection opened\")\n        # Save the WebSocket reference for heartbeat\n        self.ws = ws<\/code><\/pre>\n\n\n\n<h3 class=\"wp-block-heading\"><strong>main()<\/strong><\/h3>\n\n\n\n<p class=\"wp-block-paragraph\">The <code>main<\/code> method serves as the entry point of our application. It simply creates an instance of the <code>ZoomWebSocket<\/code> class and starts the WebSocket connection by calling <code>zws.run()<\/code>. Additionally, a graceful shutdown function is set up to handle exits (for example, when using Ctrl+C), ensuring that resources are properly cleaned up. <\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>if __name__ == \"__main__\":\n    zws = ZoomWebSocket()\n\n    def graceful_shutdown(signum, frame) -&gt; None:\n        \"\"\"Handle shutdown signals for graceful termination.\"\"\"\n        zws.logger.info(\"Shutdown signal received.\")\n        print(\"Shutdown signal received.\")\n        zws.shutdown()\n        sys.exit(0)\n\n    signal.signal(signal.SIGINT, graceful_shutdown)\n    signal.signal(signal.SIGTERM, graceful_shutdown)\n\n    zws.run()<\/code><\/pre>\n\n\n\n<h2 class=\"wp-block-heading\"><strong>.env File<\/strong><\/h2>\n\n\n\n<p class=\"wp-block-paragraph\">We&#8217;ll use a <code>.env<\/code> file to securely store sensitive credentials like <code>ACCOUNT_ID<\/code>, <code>CLIENT_ID<\/code>, and <code>CLIENT_SECRET<\/code>. Be sure to fill in this file with the appropriate values from your Zoom application. In the future, we could also move other configuration items\u2014such as the WebSocket URL and the meeting ID\u2014into the <code>.env<\/code> file, but that&#8217;s a project for another day.<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code># .env for ZoomWebSocket.py\nACCOUNT_ID=\"account id from your Zoom Application\"\nCLIENT_ID=\"client id from your Zoom Application\"\nCLIENT_SECRET=\"client secret from your Zoom Application\"<\/code><\/pre>\n\n\n\n<h2 class=\"wp-block-heading\"><strong>Putting It All Together<\/strong><\/h2>\n\n\n\n<p class=\"wp-block-paragraph\">Here is the complete code:<\/p>\n\n\n\n<pre class=\"wp-block-code\"><code>#!\/usr\/bin\/env python\n\"\"\"\nZoomWebSocket - A python class to interact with ZoomWebSockets.\nWritten by James Ferris\nFeel free to use as needed.\nZoom API Reference: https:\/\/developers.zoom.us\/docs\/api\/rest\/reference\/zoom-api\/methods\/#tag\/Meetings\n\"\"\"\n\nimport os\nimport sys\nimport json\nimport datetime\nimport signal\nimport logging\nfrom logging.handlers import RotatingFileHandler\nimport smtplib\nfrom email.message import EmailMessage\nimport requests\nfrom requests.auth import HTTPBasicAuth\nimport websocket\nfrom apscheduler.schedulers.background import BackgroundScheduler\nfrom dotenv import load_dotenv\n\nload_dotenv()\n\n\nclass ZoomWebSocket:\n    ACCOUNT_ID: str = os.getenv(\"ACCOUNT_ID\")\n    CLIENT_ID: str = os.getenv(\"CLIENT_ID\")\n    CLIENT_SECRET: str = os.getenv(\"CLIENT_SECRET\")\n    meeting_id: str = \"83148128109\"\n\n    # Define a list of (user, phone_number) tuples, IE numbers we want to call when the meeting starts\n    participants = &#91;\n        (\"IT Help Desk\", \"15554560001\")\n        # (\"Network Oncall Engineer\", \"15554561000\"),\n        # (\"Server Oncall Enginner\", \"15554562000\")\n    ]\n\n    def __init__(self) -&gt; None:\n        \"\"\"Initialize the ZoomWebSocket instance, set up logging, session, and scheduler.\"\"\"\n        self.start_time: datetime.datetime = datetime.datetime.now()\n\n        # Set up rotating log handler\n        log_file = \"ZoomWebSocketLog.txt\"\n        log_handler = RotatingFileHandler(log_file, maxBytes=10 * 1024 * 1024, backupCount=3)\n        log_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))\n        self.logger = logging.getLogger(\"ZoomWebSocket\")\n        self.logger.setLevel(logging.INFO)\n        self.logger.addHandler(log_handler)\n        self.logger.info(f\"ZoomWebSocket started at {self.start_time}\")\n\n        # Create a persistent requests session\n        self.session = requests.Session()\n\n        # Get the access token\n        self.ACCESS_TOKEN = self.get_access_token()\n        self.logger.info(\"Access token retrieved.\")\n\n        # Set up the scheduler for heartbeat and token refresh\n        self.scheduler = BackgroundScheduler()\n        self.scheduler.add_job(self.send_heartbeat, 'interval', seconds=20)\n        self.scheduler.add_job(self.refresh_token, 'interval', minutes=55)\n        self.scheduler.start()\n\n    def get_access_token(self) -&gt; str:\n        \"\"\"\n        Retrieve the access token from Zoom.\n        \n        Returns:\n            str: The access token if successful, otherwise an empty string.\n        \"\"\"\n        url = \"https:\/\/zoom.us\/oauth\/token\"\n        data = {\n            'grant_type': 'account_credentials',\n            'account_id': self.ACCOUNT_ID\n        }\n        try:\n            response = self.session.post(\n                url,\n                auth=HTTPBasicAuth(self.CLIENT_ID, self.CLIENT_SECRET),\n                data=data,\n                timeout=10\n            )\n            response.raise_for_status()\n            json_response = response.json()\n            self.access_token_expires = datetime.datetime.now() + datetime.timedelta(hours=1)\n            self.logger.info(f\"Access Token Expires {self.access_token_expires}\")\n            return json_response.get(\"access_token\", \"\")\n        except requests.exceptions.RequestException as e:\n            error_msg = f\"Error fetching Access Token: {str(e)}\"\n            print(error_msg)\n            self.logger.error(error_msg)\n            return \"\"\n\n    def refresh_token(self) -&gt; None:\n        \"\"\"Refresh the access token using the get_access_token method.\"\"\"\n        self.logger.info(\"Refreshing access token...\")\n        new_token = self.get_access_token()\n        if new_token:\n            self.ACCESS_TOKEN = new_token\n            self.logger.info(\"New access token acquired.\")\n        else:\n            self.logger.error(\"Failed to refresh access token.\")\n\n    def send_heartbeat(self) -&gt; None:\n        \"\"\"Send a heartbeat message through the WebSocket to Zoom.\"\"\"\n        try:\n            if hasattr(self, 'ws'):\n                self.ws.send('{ \"module\": \"heartbeat\" }')\n                self.logger.info(\"Heartbeat sent to Zoom\")\n            else:\n                self.logger.warning(\"WebSocket not connected; heartbeat not sent.\")\n        except Exception as e:\n            self.logger.error(\"Error sending heartbeat: \" + str(e))\n\n    def process_message(self, message: str) -&gt; None:\n        \"\"\"\n        Process a message received via WebSocket.\n\n        Args:\n            message (str): The raw JSON string received.\n        \"\"\"\n        try:\n            j = json.loads(message)\n            self.logger.info(f\"{j.get('module')} message received\")\n            print(json.dumps(j, indent=4))\n            self.logger.info(json.dumps(j, indent=4))\n        except Exception as e:\n            self.logger.error(\"Error processing message JSON: \" + str(e))\n            return\n\n        if j.get('module') == 'message':\n            try:\n                content_string = j&#91;'content']\n                content = json.loads(content_string)\n            except Exception as e:\n                self.logger.error(\"Error processing content JSON: \" + str(e))\n                return\n\n            try:\n                event = content.get('event')\n                if event == \"meeting.participant_left\":\n                    self.process_leave_event(content)\n                elif event == \"meeting.ended\":\n                    self.process_meeting_end_event(content)\n                elif event == \"meeting.participant_joined\":\n                    self.process_participant_joined_event(content)\n                elif event == \"meeting.started\":\n                    self.process_meeting_started_event(content)\n            except Exception as e:\n                self.logger.error(\"Error processing event: \" + str(e))\n        elif j.get('module') == 'heartbeat':\n            self.logger.info(f\"Received heartbeat response. Success = {str(j.get('success'))}\")\n            print(f\"{datetime.datetime.now()} - Received heartbeat response. Success = {str(j.get('success'))}\")\n\n    def process_leave_event(self, content: dict) -&gt; None:\n        \"\"\"\n        Process an event when a participant leaves a meeting.\n\n        Args:\n            content (dict): The event content dictionary.\n        \"\"\"\n        try:\n            meeting = content&#91;'payload']&#91;'object']&#91;'id']\n            user = content&#91;'payload']&#91;'object']&#91;'participant']&#91;'user_name']\n            topic = content&#91;'payload']&#91;'object']&#91;'topic']\n            if meeting == self.meeting_id:\n                print(f\"{user} has left the meeting {topic}!\")\n                self.logger.info(f\"{user} has left the meeting {topic}!\")\n                details = self.get_meeting_details(self.meeting_id)\n                participant_count = details.get(\"participants\", 0)\n                if participant_count == 0:\n                    self.logger.info(\"There are zero participants in the meeting. Ending call.\")\n                    self.end_meeting(self.meeting_id)\n        except Exception as e:\n            self.logger.error(\"Error in process_leave_event: \" + str(e))\n\n    def process_meeting_end_event(self, content: dict) -&gt; None:\n        \"\"\"\n        Process an event when a meeting ends.\n\n        Args:\n            content (dict): The event content dictionary.\n        \"\"\"\n        try:\n            meeting = content&#91;'payload']&#91;'object']&#91;'id']\n            print(f\"{meeting} has ended!\")\n            self.logger.info(f\"{meeting} has ended!\")\n        except Exception as e:\n            self.logger.error(\"Error in process_meeting_end_event: \" + str(e))\n\n    def process_participant_joined_event(self, content: dict) -&gt; None:\n        \"\"\"\n        Process an event when a participant joins a meeting.\n\n        Args:\n            content (dict): The event content dictionary.\n        \"\"\"\n        try:\n            user = content&#91;'payload']&#91;'object']&#91;'participant']&#91;'user_name']\n            topic = content&#91;'payload']&#91;'object']&#91;'topic']\n            print(f\"{user} has joined the meeting {topic}!\")\n            self.logger.info(f\"{user} has joined the meeting {topic}!\")\n        except Exception as e:\n            self.logger.error(\"Error in process_participant_joined_event: \" + str(e))\n\n    def process_meeting_started_event(self, content: dict) -&gt; None:\n        \"\"\"\n        Process an event when a meeting starts.\n\n        Args:\n            content (dict): The event content dictionary.\n        \"\"\"\n        try:\n            meeting = content&#91;'payload']&#91;'object']&#91;'id']\n            print(f\"Meeting ID: {meeting} has started!\")\n            self.logger.info(f\"Meeting ID: {meeting} has started!\")\n            if meeting == self.meeting_id:\n                print(\"Meeting has started. Placing calls to participants!\")\n                self.logger.info(\"Meeting has started. Placing calls to participants!\")\n                for user, phone in self.participants:\n                    self.place_call(meeting, user, phone)\n        except Exception as e:\n            self.logger.error(\"Error in process_meeting_started_event: \" + str(e))\n\n    def on_message(self, ws: websocket.WebSocketApp, message: str) -&gt; None:\n        \"\"\"WebSocket on_message handler.\"\"\"\n        self.process_message(message)\n\n    def on_error(self, ws: websocket.WebSocketApp, error: Exception) -&gt; None:\n        \"\"\"WebSocket on_error handler.\"\"\"\n        error_msg = f\"Encountered error: {str(error)}\"\n        print(error_msg)\n        self.logger.error(error_msg)\n\n    def on_close(self, ws: websocket.WebSocketApp, close_status_code: int, close_msg: str) -&gt; None:\n        \"\"\"WebSocket on_close handler.\"\"\"\n        print(\"Connection closed\")\n        self.logger.info(\"Connection closed\")\n\n    def on_open(self, ws: websocket.WebSocketApp) -&gt; None:\n        \"\"\"WebSocket on_open handler.\"\"\"\n        print(\"Connection opened\")\n        # Save the WebSocket reference for heartbeat\n        self.ws = ws\n\n    def place_call(self, meeting_ID: str, name: str, phone_number: str) -&gt; None:\n        \"\"\"\n        Place a call to a participant to join the meeting.\n\n        Args:\n            meeting_ID (str): The meeting ID.\n            name (str): The invitee's name.\n            phone_number (str): The invitee's phone number.\n        \"\"\"\n        meeting_options = {\n            \"method\": \"participant.invite.callout\",\n            \"params\": {\n                \"invitee_name\": str(name),\n                \"phone_number\": str(phone_number),\n                \"invite_options\": {\n                    \"require_greeting\": False,\n                    \"require_pressing_one\": False\n                },\n            }\n        }\n        headers = {\n            'authorization': 'Bearer ' + self.ACCESS_TOKEN,\n            'content-type': 'application\/json'\n        }\n        url = f'https:\/\/api.zoom.us\/v2\/live_meetings\/{meeting_ID}\/events'\n        try:\n            self.logger.info(f\"Placing call to {phone_number} to join {name} for meeting {meeting_ID}\")\n            response = self.session.patch(url, headers=headers, data=json.dumps(meeting_options))\n            print(response.text)\n            self.logger.info(f\"Place call request returned {str(response)}\")\n        except Exception as e:\n            self.logger.error(\"Error placing call: \" + str(e))\n\n    def get_meeting_details(self, meeting_id: str) -&gt; dict:\n        \"\"\"\n        Retrieve meeting details.\n\n        Args:\n            meeting_id (str): The meeting ID.\n            \n        Returns:\n            dict: The meeting details or an empty dictionary if an error occurs.\n        \"\"\"\n        headers = {\n            'authorization': 'Bearer ' + self.ACCESS_TOKEN,\n            'content-type': 'application\/json'\n        }\n        try:\n            response = self.session.get(f'https:\/\/api.zoom.us\/v2\/metrics\/meetings\/{meeting_id}', headers=headers)\n            if response.status_code == 200:\n                self.logger.info(f\"Successfully retrieved meeting details for meeting {meeting_id}\")\n            else:\n                self.logger.error(\"Error retrieving meeting details: \" + response.text)\n            return response.json()\n        except Exception as e:\n            self.logger.error(\"Error in get_meeting_details: \" + str(e))\n            return {}\n\n    def end_meeting(self, meeting_id: str) -&gt; int:\n        \"\"\"\n        End a meeting.\n\n        Args:\n            meeting_id (str): The meeting ID.\n            \n        Returns:\n            int: The HTTP status code from the request.\n        \"\"\"\n        action = {\"action\": \"end\"}\n        headers = {\n            'authorization': 'Bearer ' + self.ACCESS_TOKEN,\n            'content-type': 'application\/json'\n        }\n        try:\n            response = self.session.put(f'https:\/\/api.zoom.us\/v2\/meetings\/{meeting_id}\/status',\n                                        headers=headers, data=json.dumps(action))\n            if response.status_code == 204:\n                self.logger.info(f\"Successfully ended meeting {meeting_id}\")\n            else:\n                self.logger.error(\"Error ending meeting: \" + response.text)\n            return response.status_code\n        except Exception as e:\n            self.logger.error(\"Exception in end_meeting: \" + str(e))\n            return -1\n\n\t# Update the IP of your SMTP server if you want to use this function to send an email\n    def send_mail(self, to_email: list&#91;str], subject: str, message: str,\n                  server: str = '1.2.3.4', from_email: str = 'critical.incident@example.com') -&gt; None:\n        \"\"\"\n        Send an email.\n\n        Args:\n            to_email (list&#91;str]): List of recipient email addresses.\n            subject (str): Subject of the email.\n            message (str): Body of the email.\n            server (str, optional): SMTP server address.\n            from_email (str, optional): Sender email address.\n        \"\"\"\n        try:\n            msg = EmailMessage()\n            msg&#91;'Subject'] = subject\n            msg&#91;'From'] = from_email\n            msg&#91;'To'] = ', '.join(to_email)\n            msg.set_content(message)\n            self.logger.info(\"Sending email.\")\n            with smtplib.SMTP(server) as smtp_server:\n                smtp_server.set_debuglevel(1)\n                smtp_server.send_message(msg)\n            print('Successfully sent the mail.')\n        except Exception as e:\n            self.logger.error(\"Error sending email: \" + str(e))\n\n    def shutdown(self) -&gt; None:\n        \"\"\"Gracefully shutdown the scheduler and close the WebSocket connection.\"\"\"\n        self.logger.info(\"Initiating graceful shutdown...\")\n        if self.scheduler:\n            self.scheduler.shutdown(wait=False)\n            self.logger.info(\"Scheduler shut down.\")\n        if hasattr(self, 'ws'):\n            self.logger.info(\"Closing WebSocket connection...\")\n            try:\n                self.ws.close()\n            except Exception as e:\n                self.logger.error(\"Error closing WebSocket: \" + str(e))\n        self.logger.info(\"Shutdown complete.\")\n\n    def run(self) -&gt; None:\n        \"\"\"Establish and maintain the WebSocket connection.\"\"\"\n        websocket.enableTrace(False)\n\t\t# Copy your wss endpoint url from your Zoom application settings.  \n\t\t# Be sure to include &amp;access_token= at the end of your url\n        ws = websocket.WebSocketApp(\n            \"wss:\/\/ws.zoom.us\/ws?subscriptionId=blahblahblah&amp;access_token=\" + str(self.ACCESS_TOKEN),\n            on_open=self.on_open,\n            on_message=self.on_message,\n            on_error=self.on_error,\n            on_close=self.on_close\n        )\n        try:\n            ws.run_forever(reconnect=5)\n        except Exception as e:\n            self.logger.error(\"Error in WebSocket run_forever: \" + str(e))\n\n\nif __name__ == \"__main__\":\n    zws = ZoomWebSocket()\n\n    def graceful_shutdown(signum, frame) -&gt; None:\n        \"\"\"Handle shutdown signals for graceful termination.\"\"\"\n        zws.logger.info(\"Shutdown signal received.\")\n        print(\"Shutdown signal received.\")\n        zws.shutdown()\n        sys.exit(0)\n\n    signal.signal(signal.SIGINT, graceful_shutdown)\n    signal.signal(signal.SIGTERM, graceful_shutdown)\n\n    zws.run()\n<\/code><\/pre>\n\n\n\n<h2 class=\"wp-block-heading\">Helpful Links:<\/h2>\n\n\n\n<p class=\"wp-block-paragraph\">Find this code on my github<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><a href=\"https:\/\/github.com\/thevoiceguy\/ZoomWebSocket\">https:\/\/github.com\/thevoiceguy\/ZoomWebSocket<\/a><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Zoom WebSockets<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><a href=\"https:\/\/developers.zoom.us\/docs\/api\/websockets\">https:\/\/developers.zoom.us\/docs\/api\/websockets<\/a><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Zoom API Reference:<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><a href=\"https:\/\/developers.zoom.us\/docs\/api\">https:\/\/developers.zoom.us\/docs\/api<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"<p>We&#8217;ve all been there\u2014one minute you&#8217;re enjoying your coffee and planning your day, savoring a rare moment of calm, and then BAM! Suddenly, alerts start&#8230;<\/p>\n<div class=\"more-link-wrapper\"><a class=\"more-link\" href=\"https:\/\/phonesstillexist.com\/index.php\/2025\/04\/02\/zoom-websockets-implementing-a-priority-incident-hotline\/\">Continue reading<span class=\"screen-reader-text\">Zoom Websockets: Implementing a Priority Incident Hotline<\/span><\/a><\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"iawp_total_views":101,"footnotes":""},"categories":[3],"tags":[6,7,5,4],"class_list":["post-24","post","type-post","status-publish","format-standard","hentry","category-zoom","tag-api","tag-python","tag-websocket","tag-zoom","entry"],"_links":{"self":[{"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/posts\/24","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/comments?post=24"}],"version-history":[{"count":36,"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/posts\/24\/revisions"}],"predecessor-version":[{"id":73,"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/posts\/24\/revisions\/73"}],"wp:attachment":[{"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/media?parent=24"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/categories?post=24"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/phonesstillexist.com\/index.php\/wp-json\/wp\/v2\/tags?post=24"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}