Step-by-Step: Configure PagerDuty 2026.1 and Slack 5.0 for Real-Time Production Alerts

import requests
import os
import json
import sys
import logging
import time
from typing import Dict, Any, Optional

# Configure logging for audit trail
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

class PagerDuty2026Setup:
    \"\"\"Wrapper for PagerDuty 2026.1 REST API v3 (released Q1 2026)\"\"\"

    BASE_URL = \"https://api.pagerduty.com\"

    def __init__(self, api_key: str):
        if not api_key:
            raise ValueError(\"PagerDuty API key cannot be empty\")
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            \"Authorization\": f\"Token {self.api_key}\",
            \"Content-Type\": \"application/json\",
            \"Accept\": \"application/json\"
        })

    def _make_request(self, method: str, endpoint: str, payload: Optional[Dict] = None) -> Dict[str, Any]:
        \"\"\"Handle API requests with retry logic for rate limits (429) and server errors (5xx)\"\"\"
        url = f\"{self.BASE_URL}{endpoint}\"
        max_retries = 3
        retry_delay = 1  # seconds

        for attempt in range(max_retries):
            try:
                response = self.session.request(method, url, json=payload, timeout=10)
                if response.status_code == 429:
                    retry_after = int(response.headers.get(\"Retry-After\", retry_delay))
                    logger.warning(f\"Rate limited. Retrying after {retry_after}s (attempt {attempt+1}/{max_retries})\")
                    time.sleep(retry_after)
                    continue
                response.raise_for_status()
                return response.json()
            except requests.exceptions.RequestException as e:
                logger.error(f\"Request failed: {str(e)}\")
                if attempt == max_retries -1:
                    raise
                time.sleep(retry_delay * (2 ** attempt))  # Exponential backoff
        raise RuntimeError(\"Max retries exceeded for PagerDuty API request\")

    def create_service(self, service_name: str, description: str = \"Production Alert Service\") -> Dict[str, Any]:
        \"\"\"Create a new PagerDuty service with 2026.1 default incident urgency (high)\"\"\"
        payload = {
            \"service\": {
                \"name\": service_name,
                \"description\": description,
                \"alert_creation\": \"create_alerts_and_incidents\",  # New in 2026.1: unified alert/incident creation
                \"incident_urgency\": \"high\",
                \"addons\": [],  # Disable default addons to reduce latency
                \"support_hours\": {
                    \"type\": \"support_hours\",
                    \"time_zone\": \"UTC\",
                    \"days\": [{\"start_time\": \"00:00:00\", \"end_time\": \"24:00:00\", \"day\": day} for day in range(7)]  # 24/7 support
                }
            }
        }
        logger.info(f\"Creating PagerDuty service: {service_name}\")
        return self._make_request(\"POST\", \"/services\", payload)

    def configure_event_orchestration(self, service_id: str) -> Dict[str, Any]:
        \"\"\"Set up 2026.1 Event Orchestration to route alerts to Slack webhook\"\"\"
        payload = {
            \"orchestration\": {
                \"name\": f\"prod-alert-orchestration-{service_id[:8]}\",
                \"integration_type\": \"generic\",  # Supports custom webhooks from Slack
                \"rules\": [
                    {
                        \"condition\": \"alert.severity == 'critical'\",
                        \"action\": \"trigger_incident\",
                        \"incident\": {
                            \"title\": \"Critical Production Alert: {{alert.title}}\",
                            \"body\": \"{{alert.body}}\",
                            \"urgency\": \"high\"
                        }
                    },
                    {
                        \"condition\": \"alert.severity == 'warning'\",
                        \"action\": \"trigger_incident\",
                        \"incident\": {
                            \"title\": \"Warning: {{alert.title}}\",
                            \"urgency\": \"low\"
                        }
                    }
                ]
            }
        }
        logger.info(f\"Configuring Event Orchestration for service {service_id}\")
        return self._make_request(\"POST\", f\"/services/{service_id}/orchestrations\", payload)

    def add_slack_notification_rule(self, service_id: str, slack_webhook_url: str) -> Dict[str, Any]:
        \"\"\"Add notification rule to forward incidents to Slack via webhook\"\"\"
        payload = {
            \"notification_rule\": {
                \"type\": \"notification_rule\",
                \"start_delay_in_minutes\": 0,  # Immediate notification
                \"contact_method\": {
                    \"type\": \"webhook_contact_method\",
                    \"name\": \"Slack Production Alerts\",
                    \"address\": slack_webhook_url
                },
                \"filter\": {
                    \"type\": \"incident_filter\",
                    \"conditions\": [{\"field\": \"urgency\", \"operator\": \"equals\", \"value\": \"high\"}]
                }
            }
        }
        logger.info(f\"Adding Slack notification rule for service {service_id}\")
        return self._make_request(\"POST\", f\"/services/{service_id}/notification_rules\", payload)

if __name__ == \"__main__\":
    # Load API key from environment variable (never hardcode keys!)
    pd_api_key = os.getenv(\"PAGERDUTY_API_KEY\")
    if not pd_api_key:
        logger.error(\"Missing PAGERDUTY_API_KEY environment variable\")
        sys.exit(1)

    # Slack webhook URL (replace with your own after Step 3)
    slack_webhook = os.getenv(\"SLACK_WEBHOOK_URL\", \"https://hooks.slack.com/services/placeholder\")

    setup = PagerDuty2026Setup(pd_api_key)

    try:
        # Step 1: Create PagerDuty service
        service_resp = setup.create_service(\"prod-core-alerts-2026\")
        service_id = service_resp[\"service\"][\"id\"]
        logger.info(f\"Created service with ID: {service_id}\")

        # Step 2: Configure Event Orchestration
        orch_resp = setup.configure_event_orchestration(service_id)
        orch_id = orch_resp[\"orchestration\"][\"id\"]
        logger.info(f\"Configured Event Orchestration with ID: {orch_id}\")

        # Step 3: Add Slack notification rule
        notif_resp = setup.add_slack_notification_rule(service_id, slack_webhook)
        notif_id = notif_resp[\"notification_rule\"][\"id\"]
        logger.info(f\"Added notification rule with ID: {notif_id}\")

        # Output IDs for later use
        print(json.dumps({
            \"service_id\": service_id,
            \"orchestration_id\": orch_id,
            \"notification_rule_id\": notif_id
        }, indent=2))

    except Exception as e:
        logger.error(f\"Setup failed: {str(e)}\")
        sys.exit(1)

const { WebClient } = require('@slack/web-api');
const { createServer } = require('http');
const dotenv = require('dotenv');
const fs = require('fs');
const path = require('path');

// Load environment variables from .env file
dotenv.config();

// Configure logging
const logger = {
    info: (msg) => console.log(`[INFO] ${new Date().toISOString()}: ${msg}`),
    error: (msg) => console.error(`[ERROR] ${new Date().toISOString()}: ${msg}`),
    warn: (msg) => console.warn(`[WARN] ${new Date().toISOString()}: ${msg}`)
};

class Slack50WorkflowSetup {
    constructor(token) {
        if (!token) {
            throw new Error('Slack bot token is required');
        }
        this.client = new WebClient(token);
        this.workspaceId = null;
    }

    /**
     * Initialize workspace ID (required for Workflow API calls in Slack 5.0)
     */
    async initWorkspace() {
        try {
            const resp = await this.client.auth.test();
            this.workspaceId = resp.team_id;
            logger.info(`Connected to Slack workspace: ${resp.team} (ID: ${this.workspaceId})`);
            return this.workspaceId;
        } catch (err) {
            logger.error(`Failed to initialize workspace: ${err.message}`);
            throw err;
        }
    }

    /**
     * Create a Slack 5.0 Workflow for PagerDuty incident handling
     * Workflow triggers on incoming webhook from PagerDuty, posts to #incidents channel
     */
    async createPagerDutyWorkflow(channelId, webhookUrl) {
        const workflowPayload = {
            name: 'PagerDuty Incident Handler',
            description: 'Automatically post PagerDuty incidents to Slack and add action buttons',
            trigger: {
                type: 'webhook_trigger',
                webhook_url: webhookUrl,
                filters: [
                    {
                        field: 'incident.status',
                        operator: 'not_equals',
                        value: 'resolved'
                    }
                ]
            },
            actions: [
                {
                    type: 'send_message_action',
                    channel: channelId,
                    message: {
                        text: `🚨 *New PagerDuty Incident*: ${webhookUrl.incident.title}`,
                        blocks: [
                            {
                                type: 'header',
                                text: {
                                    type: 'plain_text',
                                    text: `🚨 Incident ${webhookUrl.incident.id}: ${webhookUrl.incident.title}`
                                }
                            },
                            {
                                type: 'section',
                                fields: [
                                    { type: 'mrkdwn', text: `*Severity*: ${webhookUrl.incident.severity}` },
                                    { type: 'mrkdwn', text: `*Status*: ${webhookUrl.incident.status}` },
                                    { type: 'mrkdwn', text: `*Urgency*: ${webhookUrl.incident.urgency}` },
                                    { type: 'mrkdwn', text: `*Created*: ${new Date(webhookUrl.incident.created_at).toLocaleString()}` }
                                ]
                            },
                            {
                                type: 'section',
                                text: {
                                    type: 'mrkdwn',
                                    text: `*Description*: ${webhookUrl.incident.body || 'No description provided'}`
                                }
                            },
                            {
                                type: 'actions',
                                elements: [
                                    {
                                        type: 'button',
                                        text: { type: 'plain_text', text: 'Acknowledge' },
                                        style: 'primary',
                                        value: `ack_${webhookUrl.incident.id}`
                                    },
                                    {
                                        type: 'button',
                                        text: { type: 'plain_text', text: 'Resolve' },
                                        style: 'danger',
                                        value: `resolve_${webhookUrl.incident.id}`
                                    },
                                    {
                                        type: 'button',
                                        text: { type: 'plain_text', text: 'Escalate' },
                                        value: `escalate_${webhookUrl.incident.id}`
                                    }
                                ]
                            }
                        ]
                    }
                }
            ]
        };

        try {
            logger.info('Creating Slack 5.0 Workflow for PagerDuty incidents');
            const resp = await this.client.workflows.create(workflowPayload);
            logger.info(`Workflow created with ID: ${resp.workflow_id}`);
            return resp.workflow_id;
        } catch (err) {
            logger.error(`Failed to create workflow: ${err.message}`);
            throw err;
        }
    }

    /**
     * Create a Slack app webhook to receive PagerDuty alerts
     */
    async createIncomingWebhook(channelId) {
        try {
            const resp = await this.client.webhooks.create({
                channel: channelId,
                description: 'PagerDuty 2026.1 Incoming Webhook'
            });
            logger.info(`Incoming webhook created: ${resp.url}`);
            return resp.url;
        } catch (err) {
            logger.error(`Failed to create webhook: ${err.message}`);
            throw err;
        }
    }

    /**
     * Start a local server to handle Slack button clicks (action endpoints)
     */
    startActionServer(port = 3000) {
        const server = createServer(async (req, res) => {
            if (req.method !== 'POST') {
                res.writeHead(405);
                return res.end('Method not allowed');
            }

            let body = '';
            req.on('data', chunk => { body += chunk; });
            req.on('end', async () => {
                try {
                    const payload = JSON.parse(body);
                    const action = payload.actions[0].value;
                    const incidentId = action.split('_')[1];
                    const actionType = action.split('_')[0];

                    logger.info(`Received ${actionType} action for incident ${incidentId}`);

                    // TODO: Call PagerDuty API to update incident status (Step 4)
                    // For now, return success
                    res.writeHead(200, { 'Content-Type': 'application/json' });
                    res.end(JSON.stringify({ success: true }));

                } catch (err) {
                    logger.error(`Action server error: ${err.message}`);
                    res.writeHead(500);
                    res.end('Internal server error');
                }
            });
        });

        server.listen(port, () => {
            logger.info(`Slack action server running on port ${port}`);
        });
    }
}

// Main execution
(async () => {
    const slackToken = process.env.SLACK_BOT_TOKEN;
    if (!slackToken) {
        logger.error('Missing SLACK_BOT_TOKEN environment variable');
        process.exit(1);
    }

    const channelId = process.env.SLACK_CHANNEL_ID;
    if (!channelId) {
        logger.error('Missing SLACK_CHANNEL_ID environment variable (e.g., C1234567890)');
        process.exit(1);
    }

    const setup = new Slack50WorkflowSetup(slackToken);

    try {
        await setup.initWorkspace();
        const webhookUrl = await setup.createIncomingWebhook(channelId);
        const workflowId = await setup.createPagerDutyWorkflow(channelId, webhookUrl);

        // Save IDs to file for later use
        fs.writeFileSync(
            path.join(__dirname, 'slack-config.json'),
            JSON.stringify({ webhookUrl, workflowId, channelId }, null, 2)
        );
        logger.info('Saved Slack config to slack-config.json');

        // Start action server to handle button clicks
        setup.startActionServer(3000);

    } catch (err) {
        logger.error(`Slack setup failed: ${err.message}`);
        process.exit(1);
    }
})();

import os
import json
import logging
import requests
from flask import Flask, request, jsonify
from typing import Dict, Any, Optional
import time

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\"
)
logger = logging.getLogger(__name__)

class BidirectionalSync:
    \"\"\"Handle bidirectional sync between PagerDuty 2026.1 and Slack 5.0\"\"\"

    def __init__(self, pd_api_key: str, slack_bot_token: str, slack_channel_id: str):
        self.pd_api_key = pd_api_key
        self.slack_token = slack_bot_token
        self.slack_channel_id = slack_channel_id

        # PagerDuty session
        self.pd_session = requests.Session()
        self.pd_session.headers.update({
            \"Authorization\": f\"Token {self.pd_api_key}\",
            \"Content-Type\": \"application/json\"
        })

        # Slack session
        self.slack_session = requests.Session()
        self.slack_session.headers.update({
            \"Authorization\": f\"Bearer {self.slack_token}\",
            \"Content-Type\": \"application/json\"
        })

        self.slack_message_ts_map = {}  # Map PagerDuty incident ID to Slack message timestamp

    def _pd_request(self, method: str, endpoint: str, payload: Optional[Dict] = None) -> Dict[str, Any]:
        \"\"\"Make PagerDuty API request with retry logic\"\"\"
        url = f\"https://api.pagerduty.com{endpoint}\"
        max_retries = 3

        for attempt in range(max_retries):
            try:
                resp = self.pd_session.request(method, url, json=payload, timeout=10)
                resp.raise_for_status()
                return resp.json()
            except requests.exceptions.RequestException as e:
                logger.error(f\"PagerDuty request failed: {str(e)}\")
                if attempt == max_retries -1:
                    raise
                time.sleep(2 ** attempt)
        raise RuntimeError(\"Max retries exceeded for PagerDuty request\")

    def _slack_request(self, method: str, endpoint: str, payload: Optional[Dict] = None) -> Dict[str, Any]:
        \"\"\"Make Slack API request with retry logic\"\"\"
        url = f\"https://slack.com/api{endpoint}\"
        max_retries = 3

        for attempt in range(max_retries):
            try:
                resp = self.slack_session.request(method, url, json=payload, timeout=10)
                resp.raise_for_status()
                return resp.json()
            except requests.exceptions.RequestException as e:
                logger.error(f\"Slack request failed: {str(e)}\")
                if attempt == max_retries -1:
                    raise
                time.sleep(2 ** attempt)
        raise RuntimeError(\"Max retries exceeded for Slack request\")

    def handle_slack_action(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        \"\"\"Handle Slack button click actions (acknowledge, resolve, escalate)\"\"\"
        try:
            action_value = payload['actions'][0]['value']
            action_type, incident_id = action_value.split('_', 1)

            logger.info(f\"Handling Slack action: {action_type} for incident {incident_id}\")

            if action_type == 'ack':
                # Acknowledge incident in PagerDuty
                pd_payload = {
                    \"incident\": {
                        \"type\": \"incident\",
                        \"status\": \"acknowledged\"
                    }
                }
                self._pd_request(\"PUT\", f\"/incidents/{incident_id}\", pd_payload)
                logger.info(f\"Acknowledged PagerDuty incident {incident_id}\")

                # Update Slack message to reflect status
                ts = self.slack_message_ts_map.get(incident_id)
                if ts:
                    slack_payload = {
                        \"channel\": self.slack_channel_id,
                        \"ts\": ts,
                        \"text\": f\"✅ Incident {incident_id} acknowledged\"
                    }
                    self._slack_request(\"POST\", \"/chat.update\", slack_payload)

                return {\"success\": True, \"action\": \"acknowledged\"}

            elif action_type == 'resolve':
                # Resolve incident in PagerDuty
                pd_payload = {
                    \"incident\": {
                        \"type\": \"incident\",
                        \"status\": \"resolved\"
                    }
                }
                self._pd_request(\"PUT\", f\"/incidents/{incident_id}\", pd_payload)
                logger.info(f\"Resolved PagerDuty incident {incident_id}\")

                # Update Slack message
                ts = self.slack_message_ts_map.get(incident_id)
                if ts:
                    slack_payload = {
                        \"channel\": self.slack_channel_id,
                        \"ts\": ts,
                        \"text\": f\"✅ Incident {incident_id} resolved\"
                    }
                    self._slack_request(\"POST\", \"/chat.update\", slack_payload)

                return {\"success\": True, \"action\": \"resolved\"}

            elif action_type == 'escalate':
                # Escalate incident to next on-call tier
                pd_payload = {
                    \"escalation\": {
                        \"incident_id\": incident_id,
                        \"escalation_policy_id\": os.getenv(\"PD_ESCALATION_POLICY_ID\")
                    }
                }
                self._pd_request(\"POST\", \"/escalations\", pd_payload)
                logger.info(f\"Escalated PagerDuty incident {incident_id}\")

                return {\"success\": True, \"action\": \"escalated\"}

            else:
                return {\"success\": False, \"error\": \"Unknown action type\"}

        except Exception as e:
            logger.error(f\"Failed to handle Slack action: {str(e)}\")
            return {\"success\": False, \"error\": str(e)}

    def handle_pagerduty_webhook(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        \"\"\"Handle incoming PagerDuty webhook (incident status updates)\"\"\"
        try:
            incident = payload['incident']
            incident_id = incident['id']
            status = incident['status']

            logger.info(f\"Handling PagerDuty webhook for incident {incident_id}: {status}\")

            # Post or update Slack message
            if status == 'triggered':
                # New incident: post to Slack
                slack_payload = {
                    \"channel\": self.slack_channel_id,
                    \"text\": f\"🚨 New Incident: {incident['title']}\",
                    \"blocks\": [
                        {
                            \"type\": \"header\",
                            \"text\": {\"type\": \"plain_text\", \"text\": f\"🚨 {incident['title']}\"}
                        },
                        {
                            \"type\": \"section\",
                            \"fields\": [
                                {\"type\": \"mrkdwn\", \"text\": f\"*Status*: {status}\"},
                                {\"type\": \"mrkdwn\", \"text\": f\"*Severity*: {incident['severity']}\"},
                                {\"type\": \"mrkdwn\", \"text\": f\"*Urgency*: {incident['urgency']}\"}
                            ]
                        }
                    ]
                }
                resp = self._slack_request(\"POST\", \"/chat.postMessage\", slack_payload)
                self.slack_message_ts_map[incident_id] = resp['ts']
                logger.info(f\"Posted Slack message for incident {incident_id}\")

            elif status in ['acknowledged', 'resolved']:
                # Update existing Slack message
                ts = self.slack_message_ts_map.get(incident_id)
                if ts:
                    slack_payload = {
                        \"channel\": self.slack_channel_id,
                        \"ts\": ts,
                        \"text\": f\"Incident {incident_id} {status}\",
                        \"blocks\": [
                            {
                                \"type\": \"header\",
                                \"text\": {\"type\": \"plain_text\", \"text\": f\"✅ {incident['title']} ({status})\"}
                            }
                        ]
                    }
                    self._slack_request(\"POST\", \"/chat.update\", slack_payload)
                    logger.info(f\"Updated Slack message for incident {incident_id}\")

            return {\"success\": True}

        except Exception as e:
            logger.error(f\"Failed to handle PagerDuty webhook: {str(e)}\")
            return {\"success\": False, \"error\": str(e)}

# Initialize Flask app
app = Flask(__name__)

# Load config from environment
PD_API_KEY = os.getenv(\"PAGERDUTY_API_KEY\")
SLACK_BOT_TOKEN = os.getenv(\"SLACK_BOT_TOKEN\")
SLACK_CHANNEL_ID = os.getenv(\"SLACK_CHANNEL_ID\")

if not all([PD_API_KEY, SLACK_BOT_TOKEN, SLACK_CHANNEL_ID]):
    logger.error(\"Missing required environment variables\")
    exit(1)

sync = BidirectionalSync(PD_API_KEY, SLACK_BOT_TOKEN, SLACK_CHANNEL_ID)

@app.route('/slack/actions', methods=['POST'])
def slack_actions():
    \"\"\"Endpoint to handle Slack button clicks\"\"\"
    payload = json.loads(request.form['payload'])
    result = sync.handle_slack_action(payload)
    return jsonify(result)

@app.route('/pagerduty/webhook', methods=['POST'])
def pagerduty_webhook():
    \"\"\"Endpoint to receive PagerDuty incident updates\"\"\"
    payload = request.json
    result = sync.handle_pagerduty_webhook(payload)
    return jsonify(result)

if __name__ == '__main__':
    logger.info(\"Starting bidirectional sync server on port 5000\")
    app.run(host='0.0.0.0', port=5000, debug=False)

{
  \"rules\": [
    {
      \"condition\": \"alert.service == 'payment-service' && alert.severity == 'critical'\",
      \"action\": \"trigger_incident\",
      \"incident\": {
        \"title\": \"CRITICAL: Payment Service Down - {{alert.title}}\",
        \"body\": \"Runbook: https://wiki.example.com/runbooks/payment-outage\",
        \"urgency\": \"high\",
        \"assigned_to\": [\"team_payments_oncall\"]
      }
    }
  ]
}

{
  \"blocks\": [
    {
      \"type\": \"section\",
      \"fields\": [
        { \"type\": \"mrkdwn\", \"text\": \"*Incident*: <${pagerduty.incident.html_url}|${pagerduty.incident.id}>\" },
        { \"type\": \"mrkdwn\", \"text\": \"*On-Call*: <@${pagerduty.oncall.user.slack_id}>\" },
        { \"type\": \"mrkdwn\", \"text\": \"*Runbook*: <${pagerduty.incident.runbook_url}|Open Runbook>\" },
        { \"type\": \"mrkdwn\", \"text\": \"*War Room*: \" }
      ]
    }
  ]
}

import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1):
    \"\"\"Retry a function with exponential backoff and jitter\"\"\"
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries -1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            logger.warning(f\"Retry attempt {attempt+1} after {delay:.2f}s: {str(e)}\")
            time.sleep(delay)
    raise RuntimeError(\"Max retries exceeded\")