Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Updates a Slack User Group with People that are on call in PagerDuty
#!/usr/bin/env python
from __future__ import print_function
import json
import logging
from urllib2 import Request, urlopen, URLError, HTTPError
from base64 import b64decode
log = logging.getLogger(__name__)
class SlackOnCall(object):
# The Slack API token to use for authentication to the Slack WebAPI
slack_token = None
# The Pager Duty API token to use for authentication into the PagerDuty API
pager_duty_token = None
# The domain prefix for the PagerDuty installation ( will be "acme")
pager_duty_domain_prefix = None
# The Slack @user-group to update (Default: oncall)
slack_user_group_handle = 'oncall'
# The maximum escalation level to add to the group
# (eg. if escalation level = 2, then levels 1 and 2 will be a part of the group
# but not any levels 3 and above.
escalation_level = 2
def __init__(self, slack_token, pager_duty_token, pager_duty_domain_prefix,
slack_user_group_handle=slack_user_group_handle, log_level='INFO',
self.slack_token = slack_token
self.pager_duty_token = pager_duty_token
self.pager_duty_domain_prefix = pager_duty_domain_prefix
self.slack_user_group_handle = slack_user_group_handle
self.escalation_level = escalation_level
self._slack_user_group = None
self._on_call_email_addresses = None
self._all_slack_users = None
def run(self):
Gets user group information and on-call information then updates the
on-call user group in slack to be the on-call users for escalation
levels 1 and 2.
slack_users = self.slack_users_by_email(self.on_call_email_addresses)
if not slack_users:
log.warning('No Slack users found for email addresses: %s', ','.join(self.on_call_email_addresses))
slack_user_ids = [u['id'] for u in slack_users]
if set(slack_user_ids) == set(self.slack_user_group['users']):'User group %s already set to %s', self.slack_user_group_handle, slack_user_ids)
self.update_on_call(slack_users)'Job Complete')
def _make_request(url, body=None, headers={}):
req = Request(url, body, headers)'Making request to %s', url)
response = urlopen(req)
body =
return json.loads(body)
except ValueError:
return body
except HTTPError as e:
log.error("Request failed: %d %s", e.code, e.reason)
except URLError as e:
log.error("Server connection failed: %s", e.reason)
def slack_user_group(self):
:return: the Slack user group matching the slack_user_group_handle
specified in the configuration
if self._slack_user_group is not None:
return self._slack_user_group
url = '{}&include_users=1'.format(self.slack_token)
groups = self._make_request(url)['usergroups']
for group in groups:
if group['handle'] == self.slack_user_group_handle:
self._slack_user_group = group
return group
raise ValueError('No user groups found that match {}'.format(self.slack_user_group_handle))
def on_call_email_addresses(self):
Hits the PagerDuty API and gets level 1 and level 2 escalation
on-call users and returns their email addresses
:return: All on-call email addresses within the escalation bounds
if self._on_call_email_addresses is not None:
return self._on_call_email_addresses
url = 'https://{}'.format(self.pager_duty_domain_prefix)
on_call = self._make_request(url, headers={'Authorization': 'Token token=' + self.pager_duty_token})
users = set() # users can be in multiple schedule, this will de-dupe
for user in on_call['users']:
for schedule in user['on_call']:
if schedule['level'] <= self.escalation_level:
users.add(user['email'])'Found %d users on-call', len(users))
self._on_call_email_addresses = users
return users
def all_slack_users(self):
if self._all_slack_users is not None:
return self._all_slack_users
url = '{}'.format(self.slack_token)
users = self._make_request(url)['members']'Found %d total Slack users', len(users))
self._all_slack_users = users
return users
def slack_users_by_email(self, emails):
Finds all slack users by their email address
:param emails: List of email address to find users
:return: List of Slack user objects found in :emails:
users = []
for user in self.all_slack_users:
if user['profile'].get('email') in emails:
return users
def update_on_call(self, slack_users):
Updates the specified user-group
:param slack_users: Slack users to modify the group with
user_ids = [u['id'] for u in slack_users]
url = '{0}&usergroup={1}&users={2}'.format(
)'Updating user group %s from %s to %s',
self.slack_user_group_handle, self.slack_user_group['users'], user_ids)
def lambda_handler(*_):
Main entry point for AWS Lambda.
Variables can not be passed in to AWS Lambda, the configuration
parameters below are encrypted using AWS IAM Keys.
# Boto is always available in AWS lambda, but may not be available in
# standalone mode
import boto3
# To generate the encrypted values, go to AWS IAM Keys and Generate a key
# Then grant decryption using the key to the IAM Role used for your lambda
# function.
# Use the command `aws kms encrypt --key-id alias/<key-alias> --plaintext <value-to-encrypt>
# Put the encrypted value in the configuration dictionary below
encrypted_config = {
'slack_token': '<ENCRYPTED VALUE>',
'pager_duty_token': '<ENCRYPTED VALUE>'
'pager_duty_domain_prefix': '<ENCRYPTED VALUE>'
kms = boto3.client('kms')
config = {x: kms.decrypt(CiphertextBlob=b64decode(y))['Plaintext'] for x, y in encrypted_config.iteritems()}
return SlackOnCall(**config).run()
def main():
Runs the Slack PagerDuty OnCall group updater as a standalone script
from argparse import ArgumentParser
parser = ArgumentParser(usage=main.__doc__)
parser.add_argument('-st', '--slack-token', required=True, dest='slack_token',
help='Slack token to use for auth into the Slack WebAPI')
parser.add_argument('-su', '--slack-user-group', dest='slack_user_group_handle', default='oncall',
help='Slack user group to add on-call users to. (Default: oncall)')
parser.add_argument('-pt', '--pager-duty-token', required=True, dest='pager_duty_token',
help='PagerDuty token to use for auth into the PagerDuty API')
parser.add_argument('-pd', '--pager-duty-domain-prefix', required=True, dest='pager_duty_domain_prefix',
help='Your domain prefix for PagerDuty')
parser.add_argument('-el', '--max-escalation-level', dest='escalation_level', default=2, type=int,
help='Max escalation level to add on-call users for group. (Default: 2)')
args = vars(parser.parse_args())
if __name__ == '__main__':

This comment has been minimized.

Copy link
Owner Author

@devdazed devdazed commented Dec 30, 2015

PagerDuty Slack OnCall Group

Use PagerDuty and Slack? Create a Slack @oncall group containing the people that are currently in specified escalation levels in PagerDuty. Just create a Slack user group called oncall then run this script.

NOTE: Users are matched by email address, the email address they use in Slack must match the one they use in PagerDuty.

./ --slack-token 'xoxp-xxxxxxxx-xxxxxxx-xxxxxxx' \
                                            --pager-duty-token 'abcdefghijklmnopqrstuvwxyz' \
                                            --pager-duty-domain-prefix 'acme'

Running as a Cron job

Just ignore, or remove, the lambda_handler function and run the above example as a scheduled cron job. Can be set to the dates/times your PagerDuty schedules change.

Running as an AWS Lambda method

Use AWS IAM keys to encrypt the configuration in the lambda_handler method and set up your AWS Lambda method to run on a schedule. The schedule should be set, just as in the cron setup, to coincide with your PagerDuty schedule changes.


This comment has been minimized.

Copy link

@ezjacob-maheim ezjacob-maheim commented May 24, 2017

awesome script. Was definitely looking for something like this.


This comment has been minimized.

Copy link

@markddavidoff markddavidoff commented Jun 19, 2019

I've updated this to work with pagerduty v2 api:

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment