Pagerduty API

PagerDuty API Integration

Implements production-grade PagerDuty API integration using the pdpyras Python SDK and direct HTTP API calls. When loaded, this skill makes the model implement Event API v2 ingestion (trigger/acknowledge/resolve), incident querying and management, on-call schedule lookup, escalation policy management, alert grouping, and maintenance window creation. All implementations follow PagerDuty best practices: use Events API for alert ingestion (not REST API), always include dedup_key for deduplication, implement exponential backoff for rate limits, and validate API connectivity on startup.

TL;DR Checklist

• Use pdpyras SDK or direct HTTP calls with PAGERDUTY_API_KEY from env var
• Use Events API v2 (https://events.pagerduty.com/v2/enqueue) for alert ingestion
• Use REST API (https://api.pagerduty.com) for management operations
• Always include dedup_key in Events API calls for deduplication
• Use routing_key (integration key) for Events API, api_key for REST API
• Include required headers: Authorization: Token token=<key>, Accept: application/vnd.pagerduty+json;version=2
• Add explicit From email header for user context operations
• Implement exponential backoff for 429 rate limit responses
• Use event_action: trigger, event_action: acknowledge, event_action: resolve consistently
• Always include payload.summary, payload.source, payload.severity in trigger events

When to Use

Use this skill when:

• Triggering alerts from monitoring systems (Prometheus, Datadog, New Relic)
• Acknowledging and resolving incidents programmatically
• Looking up who is currently on-call
• Managing escalation policies via API
• Creating and managing incident responders
• Setting up maintenance windows to suppress alerts
• Querying incident history for reporting
• Automating incident response workflows
• Building chatops integrations (Slack, Mattermost)
• Syncing on-call schedules with internal systems

When NOT to Use

• For Datadog-specific alerting — use coding-datadog-api instead
• For Opsgenie (Atlassian competitor) — use dedicated Opsgenie patterns
• For internal-only alerting without on-call management — consider simpler solutions
• When you need push notification only (SMS, email) without escalation
• For incident response in non-24/7 environments (just use email)

Core Workflow

1. Initialize Configuration — Configure PagerDuty client using PAGERDUTY_API_KEY (REST API) and PAGERDUTY_ROUTING_KEY (Events API integration key) from environment variables. Determine correct endpoint: api.pagerduty.com for REST, events.pagerduty.com for Events. Checkpoint: Validate connectivity with GET /users or send a test resolve event.

2. Choose API for Task — Decide between:

   • Events API v2: Ingestion only (trigger/ack/resolve), uses routing key (integration key), lower rate limits, higher reliability for alerts
   • REST API: All management operations, uses API key, full CRUD on incidents, schedules, policies, users

   Checkpoint: Never use REST API for alert ingestion; Events API is designed for that.

3. Implement Events API Ingestion — Build event payloads with required fields: routing_key, event_action, dedup_key, payload.summary, payload.severity, payload.source. Add optional: payload.custom_details, links, images. Checkpoint: Every trigger event has unique dedup_key for deduplication and resolution.

4. Query & Manage Incidents — Use REST API to list incidents with filters (statuses, service_ids, urgency), get single incident, update status, add notes, assign responders. Use since/until for historical queries. Checkpoint: All list operations have appropriate time bounds to avoid full scans.

5. On-Call & Schedules — Query on-call users via GET /oncalls with schedule_ids, time_zone, since/until. Get schedule details, overrides, and final schedule. Checkpoint: Always specify time_zone to avoid UTC confusion.

6. Maintenance Windows — Create maintenance windows via POST /maintenance_windows with start_time, end_time, services, description. Checkpoint: Maintenance windows have timezone-aware timestamps in ISO 8601 format.

Implementation Patterns

Pattern 1: PagerDuty Client Initialization (BAD vs GOOD)

"""PagerDuty client initialization patterns.

Two primary APIs:
1. Events API v2: Alert ingestion (trigger/ack/resolve)
   - Endpoint: https://events.pagerduty.com/v2/enqueue
   - Uses: routing_key (integration key from service integration)
   - Lower rate limits, higher reliability for alert paths

2. REST API v2: Management operations
   - Endpoint: https://api.pagerduty.com
   - Uses: api_key (Personal API token or OAuth)
   - Full CRUD on incidents, schedules, policies, users

Headers for REST API:
- Authorization: Token token=<api_key>
- Accept: application/vnd.pagerduty+json;version=2
- Content-Type: application/json
- From: user@example.com (for operations requiring user context)

SDK: pdpyras is the official Python SDK
"""

from __future__ import annotations

import os
import json
import time
import logging
import random
from typing import Any, Optional, Literal
from datetime import datetime, timedelta, timezone
from dataclasses import dataclass, field

import requests

logger = logging.getLogger(__name__)

EventAction = Literal["trigger", "acknowledge", "resolve"]
Severity = Literal["critical", "warning", "error", "info"]
IncidentStatus = Literal["triggered", "acknowledged", "resolved"]


# ===================================================================
# ❌ BAD — hardcoded keys, wrong API choice, missing headers
# ===================================================================

def bad_pagerduty_init_bad() -> dict[str, Any]:
    """❌ BAD: Hardcoded keys, wrong API, missing headers."""
    
    # ❌ Hardcoded! Should come from env
    api_key = "u+xxxxxxxxx_xxxxx_xx"
    
    # ❌ Using REST API for alert ingestion (wrong!)
    # Should use Events API for triggering/acking/resolving
    endpoint = "https://api.pagerduty.com/incidents"
    
    # ❌ Missing required headers:
    # - Accept with version specifier
    # - Content-Type
    
    return {"api_key": api_key, "endpoint": endpoint}


# ===================================================================
# ✅ GOOD — env-based config, correct APIs, typed errors
# ===================================================================


class PagerDutyError(Exception):
    """Base exception for PagerDuty client errors."""
    pass


class PagerDutyAuthError(PagerDutyError):
    """Authentication/API key is invalid."""
    pass


class PagerDutyRateLimitError(PagerDutyError):
    """Rate limit exceeded."""
    
    def __init__(self, message: str, retry_after: Optional[int] = None):
        super().__init__(message)
        self.retry_after = retry_after


@dataclass
class PagerDutyConfig:
    """PagerDuty configuration from environment variables.
    
    Environment variables:
        PAGERDUTY_API_KEY: REST API Personal Token
        PAGERDUTY_ROUTING_KEY: Events API integration key (per-service)
        PAGERDUTY_FROM_EMAIL: User email for 'From' header
        PAGERDUTY_TIMEOUT: Request timeout in seconds
    """
    
    # REST API config
    api_key: Optional[str] = None
    rest_api_endpoint: str = "https://api.pagerduty.com"
    
    # Events API config
    routing_key: Optional[str] = None
    events_api_endpoint: str = "https://events.pagerduty.com/v2/enqueue"
    
    # Common
    from_email: Optional[str] = None
    timeout: float = 30.0
    max_retries: int = 3
    initial_retry_delay: float = 1.0
    
    @classmethod
    def from_env(cls) -> "PagerDutyConfig":
        """Load from environment variables."""
        timeout_str = os.environ.get("PAGERDUTY_TIMEOUT", "30")
        try:
            timeout = float(timeout_str)
        except ValueError:
            timeout = 30.0
        
        return cls(
            api_key=os.environ.get("PAGERDUTY_API_KEY"),
            routing_key=os.environ.get("PAGERDUTY_ROUTING_KEY"),
            from_email=os.environ.get("PAGERDUTY_FROM_EMAIL"),
            timeout=timeout,
        )
    
    def validate(self, require_api_key: bool = False, require_routing_key: bool = False) -> bool:
        """Validate configuration.
        
        Args:
            require_api_key: If True, REST API key must be present
            require_routing_key: If True, Events API routing key must be present
        
        Returns:
            True if valid
            
        Raises:
            ValueError: If validation fails
        """
        if require_api_key and not self.api_key:
            raise ValueError("PAGERDUTY_API_KEY required for REST API operations")
        
        if require_routing_key and not self.routing_key:
            raise ValueError("PAGERDUTY_ROUTING_KEY required for Events API operations")
        
        return True


class PagerDutyClient:
    """Client for both PagerDuty Events API and REST API.
    
    Features:
    - Events API: trigger, acknowledge, resolve
    - REST API: incidents, oncalls, schedules, policies
    - Automatic retries with exponential backoff and jitter
    - Rate limit handling
    """
    
    def __init__(self, config: PagerDutyConfig) -> None:
        self._config = config
        self._session = requests.Session()
    
    def _get_rest_headers(self) -> dict[str, str]:
        """Get headers for REST API calls."""
        if not self._config.api_key:
            raise ValueError("PAGERDUTY_API_KEY not configured")
        
        headers: dict[str, str] = {
            "Authorization": f"Token token={self._config.api_key}",
            "Accept": "application/vnd.pagerduty+json;version=2",
            "Content-Type": "application/json",
        }
        
        # Some operations require From header (user context)
        if self._config.from_email:
            headers["From"] = self._config.from_email
        
        return headers
    
    def _get_events_headers(self) -> dict[str, str]:
        """Get headers for Events API calls."""
        return {
            "Content-Type": "application/json",
        }
    
    def _calculate_delay(self, attempt: int) -> float:
        """Calculate exponential backoff with jitter.
        
        Args:
            attempt: 0-based attempt number
        
        Returns:
            Delay in seconds
        """
        delay = self._config.initial_retry_delay * (2 ** attempt)
        # Add jitter ±20%
        jitter = random.uniform(0.8, 1.2)
        return delay * jitter
    
    def _request_with_retry(
        self,
        method: str,
        url: str,
        headers: dict[str, str],
        json_data: Optional[dict[str, Any]] = None,
        params: Optional[dict[str, Any]] = None,
        is_events_api: bool = False,
    ) -> dict[str, Any]:
        """Make HTTP request with automatic retries.
        
        Args:
            method: HTTP method
            url: Full URL
            headers: Request headers
            json_data: Optional JSON body
            params: Optional query params
            is_events_api: If True, use Events API error handling
        
        Returns:
            Parsed JSON response
        
        Raises:
            PagerDutyAuthError: If 401/403
            PagerDutyRateLimitError: If 429
            PagerDutyError: For other errors
        """
        last_exception: Optional[Exception] = None
        
        for attempt in range(self._config.max_retries):
            try:
                response = self._session.request(
                    method=method,
                    url=url,
                    headers=headers,
                    json=json_data,
                    params=params,
                    timeout=self._config.timeout,
                )
                
                # Handle response
                if response.status_code == 204:
                    return {}
                
                try:
                    result = response.json()
                except json.JSONDecodeError:
                    result = {"raw": response.text[:500]}
                
                # Check status codes
                if response.status_code in (200, 201, 202):
                    return result
                
                elif response.status_code == 401:
                    raise PagerDutyAuthError("PagerDuty API key invalid or missing")
                
                elif response.status_code == 403:
                    raise PagerDutyAuthError(f"PagerDuty forbidden: {result}")
                
                elif response.status_code == 429:
                    retry_after = response.headers.get("Retry-After")
                    delay = int(retry_after) if retry_after else self._calculate_delay(attempt)
                    
                    if attempt < self._config.max_retries - 1:
                        logger.warning(
                            "PagerDuty rate limited (429), retrying in %.1fs",
                            delay
                        )
                        time.sleep(delay)
                        continue
                    else:
                        raise PagerDutyRateLimitError(
                            f"PagerDuty rate limit exceeded after {self._config.max_retries} retries",
                            retry_after=int(retry_after) if retry_after else None
                        )
                
                else:
                    # Other error - don't retry
                    raise PagerDutyError(
                        f"PagerDuty HTTP {response.status_code}: {result}"
                    )
                
            except requests.RequestException as e:
                last_exception = e
                if attempt < self._config.max_retries - 1:
                    delay = self._calculate_delay(attempt)
                    logger.warning("PagerDuty request failed, retrying in %.1fs: %s", delay, e)
                    time.sleep(delay)
                continue
        
        # If we get here, all retries failed
        raise PagerDutyError(
            f"PagerDuty request failed after {self._config.max_retries} retries: {last_exception}"
        ) from last_exception
    
    # ===================================================================
    # Events API v2 (Alert Ingestion)
    # ===================================================================
    
    def send_event(
        self,
        event_action: EventAction,
        dedup_key: str,
        summary: str,
        severity: Severity = "critical",
        source: Optional[str] = None,
        custom_details: Optional[dict[str, Any]] = None,
        routing_key_override: Optional[str] = None,
        links: Optional[list[dict[str, str]]] = None,
        images: Optional[list[dict[str, str]]] = None,
    ) -> dict[str, Any]:
        """Send an event to PagerDuty Events API v2.
        
        Use this to trigger, acknowledge, or resolve incidents.
        
        Args:
            event_action: "trigger", "acknowledge", or "resolve"
            dedup_key: Unique key for deduplication (required for all actions)
            summary: Brief summary (required for trigger)
            severity: "critical", "warning", "error", "info" (trigger only)
            source: Location affected (e.g., hostname, service name)
            custom_details: Additional context data
            routing_key_override: Override default routing key
            links: List of {href, text} links
            images: List of {src, href, alt} images
        
        Returns:
            Response from Events API: {status, message, dedup_key}
        
        Example:
            client.send_event(
                event_action="trigger",
                dedup_key="checkout-service-high-errors-20240115",
                summary="High error rate (>5%) on checkout service",
                severity="critical",
                source="checkout-service-prod",
                custom_details={
                    "error_rate_pct": 8.7,
                    "total_requests": 1245,
                    "error_count": 108,
                },
                links=[
                    {"href": "https://grafana.example.com/d/checkout", "text": "Dashboard"},
                    {"href": "https://runbooks.example.com/high-errors", "text": "Runbook"},
                ],
            )
        """
        routing_key = routing_key_override or self._config.routing_key
        if not routing_key:
            raise ValueError("PAGERDUTY_ROUTING_KEY not configured")
        
        # Build payload
        payload: dict[str, Any] = {
            "routing_key": routing_key,
            "event_action": event_action,
            "dedup_key": dedup_key,
        }
        
        if event_action == "trigger":
            # Trigger requires full payload
            trigger_payload: dict[str, Any] = {
                "summary": summary,
                "severity": severity,
            }
            
            if source:
                trigger_payload["source"] = source
            else:
                trigger_payload["source"] = "unknown"  # Required field
            
            if custom_details:
                trigger_payload["custom_details"] = custom_details
            
            payload["payload"] = trigger_payload
            
            # Optional fields for trigger
            if links:
                payload["links"] = links
            if images:
                payload["images"] = images
        
        elif event_action in ("acknowledge", "resolve"):
            # Ack/resolve can have optional payload
            if summary:
                payload["payload"] = {"summary": summary}
        
        headers = self._get_events_headers()
        
        response = self._request_with_retry(
            method="POST",
            url=self._config.events_api_endpoint,
            headers=headers,
            json_data=payload,
            is_events_api=True,
        )
        
        logger.info(
            "PagerDuty %s event sent: dedup_key=%s, status=%s",
            event_action,
            dedup_key,
            response.get("status", "unknown"),
        )
        
        return response
    
    def trigger_incident(
        self,
        dedup_key: str,
        summary: str,
        severity: Severity = "critical",
        source: Optional[str] = None,
        custom_details: Optional[dict[str, Any]] = None,
        runbook_url: Optional[str] = None,
        dashboard_url: Optional[str] = None,
    ) -> dict[str, Any]:
        """Convenience method to trigger an incident with standard links.
        
        Args:
            dedup_key: Unique deduplication key
            summary: Brief summary
            severity: Severity level
            source: Source service/host
            custom_details: Additional context
            runbook_url: Optional runbook link
            dashboard_url: Optional dashboard link
        
        Returns:
            Events API response
        """
        links: list[dict[str, str]] = []
        
        if runbook_url:
            links.append({"href": runbook_url, "text": "Runbook"})
        if dashboard_url:
            links.append({"href": dashboard_url, "text": "Dashboard"})
        
        return self.send_event(
            event_action="trigger",
            dedup_key=dedup_key,
            summary=summary,
            severity=severity,
            source=source,
            custom_details=custom_details,
            links=links if links else None,
        )
    
    def resolve_incident(
        self,
        dedup_key: str,
        summary: str = "Incident resolved",
    ) -> dict[str, Any]:
        """Resolve an incident via Events API.
        
        Args:
            dedup_key: Same key used when triggering
            summary: Resolution message
        
        Returns:
            Events API response
        """
        return self.send_event(
            event_action="resolve",
            dedup_key=dedup_key,
            summary=summary,
        )
    
    # ===================================================================
    # REST API - Incidents
    # ===================================================================
    
    def list_incidents(
        self,
        statuses: Optional[list[IncidentStatus]] = None,
        service_ids: Optional[list[str]] = None,
        urgency: Optional[Literal["high", "low"]] = None,
        since: Optional[str] = None,  # ISO 8601
        until: Optional[str] = None,  # ISO 8601
        limit: int = 25,
        offset: int = 0,
        sort_by: Optional[str] = None,  # e.g., "created_at:desc"
    ) -> dict[str, Any]:
        """List incidents with filtering.
        
        Args:
            statuses: Filter by status: ["triggered", "acknowledged", "resolved"]
            service_ids: Filter by service IDs
            urgency: "high" or "low"
            since: ISO 8601 start time
            until: ISO 8601 end time
            limit: Results per page (max 100)
            offset: Pagination offset
            sort_by: Sort field and direction
        
        Returns:
            Dict with incidents, limit, offset, total
        """
        params: dict[str, Any] = {
            "limit": min(limit, 100),
            "offset": offset,
        }
        
        if statuses:
            params["statuses[]"] = statuses
        if service_ids:
            params["service_ids[]"] = service_ids
        if urgency:
            params["urgencies[]"] = [urgency]
        if since:
            params["since"] = since
        if until:
            params["until"] = until
        if sort_by:
            params["sort_by"] = sort_by
        
        headers = self._get_rest_headers()
        
        return self._request_with_retry(
            method="GET",
            url=f"{self._config.rest_api_endpoint}/incidents",
            headers=headers,
            params=params,
        )
    
    def get_incident(self, incident_id: str) -> dict[str, Any]:
        """Get a single incident by ID.
        
        Args:
            incident_id: Incident ID (e.g., "Q022JNK45AB8")
        
        Returns:
            Incident details
        """
        headers = self._get_rest_headers()
        
        return self._request_with_retry(
            method="GET",
            url=f"{self._config.rest_api_endpoint}/incidents/{incident_id}",
            headers=headers,
        )
    
    def update_incident_status(
        self,
        incident_id: str,
        status: IncidentStatus,
        resolution: Optional[str] = None,
    ) -> dict[str, Any]:
        """Update incident status.
        
        Args:
            incident_id: Incident ID
            status: New status: "acknowledged" or "resolved"
            resolution: Resolution message for resolved status
        
        Returns:
            Updated incident
        """
        incident: dict[str, Any] = {
            "type": "incident_reference",
            "status": status,
        }
        
        if status == "resolved" and resolution:
            incident["resolution"] = resolution
        
        payload = {"incident": incident}
        headers = self._get_rest_headers()
        
        return self._request_with_retry(
            method="PUT",
            url=f"{self._config.rest_api_endpoint}/incidents/{incident_id}",
            headers=headers,
            json_data=payload,
        )
    
    # ===================================================================
    # REST API - On-Calls & Schedules
    # ===================================================================
    
    def get_oncalls(
        self,
        schedule_ids: Optional[list[str]] = None,
        escalation_policy_ids: Optional[list[str]] = None,
        time_zone: Optional[str] = None,
        since: Optional[str] = None,
        until: Optional[str] = None,
        include: Optional[list[str]] = None,  # users, schedules, escalation_policies
    ) -> dict[str, Any]:
        """Get current on-call users.
        
        Args:
            schedule_ids: Filter by schedule IDs
            escalation_policy_ids: Filter by policy IDs
            time_zone: Time zone (e.g., "America/New_York", "UTC")
            since: ISO 8601 start
            until: ISO 8601 end
            include: Related objects to include
        
        Returns:
            Oncall entries with users
        """
        params: dict[str, Any] = {}
        
        if schedule_ids:
            params["schedule_ids[]"] = schedule_ids
        if escalation_policy_ids:
            params["escalation_policy_ids[]"] = escalation_policy_ids
        if time_zone:
            params["time_zone"] = time_zone
        if since:
            params["since"] = since
        if until:
            params["until"] = until
        if include:
            params["include[]"] = include
        
        headers = self._get_rest_headers()
        
        return self._request_with_retry(
            method="GET",
            url=f"{self._config.rest_api_endpoint}/oncalls",
            headers=headers,
            params=params,
        )
    
    def get_schedule(
        self,
        schedule_id: str,
        time_zone: Optional[str] = None,
        since: Optional[str] = None,
        until: Optional[str] = None,
    ) -> dict[str, Any]:
        """Get a schedule with rendered on-call layers.
        
        Args:
            schedule_id: Schedule ID
            time_zone: Time zone for rendering
            since: Start time
            until: End time
        
        Returns:
            Schedule details with final_schedule
        """
        params: dict[str, Any] = {}
        
        if time_zone:
            params["time_zone"] = time_zone
        if since:
            params["since"] = since
        if until:
            params["until"] = until
        
        headers = self._get_rest_headers()
        
        return self._request_with_retry(
            method="GET",
            url=f"{self._config.rest_api_endpoint}/schedules/{schedule_id}",
            headers=headers,
            params=params,
        )
    
    # ===================================================================
    # REST API - Maintenance Windows
    # ===================================================================
    
    def create_maintenance_window(
        self,
        start_time: str,  # ISO 8601
        end_time: str,    # ISO 8601
        service_ids: list[str],
        description: str,
    ) -> dict[str, Any]:
        """Create a maintenance window to suppress notifications.
        
        Args:
            start_time: ISO 8601 start time (e.g., "2024-01-15T10:00:00Z")
            end_time: ISO 8601 end time
            service_ids: List of service IDs
            description: Description
        
        Returns:
            Created maintenance window
        """
        # Build services references
        services = [
            {"id": sid, "type": "service_reference"}
            for sid in service_ids
        ]
        
        window = {
            "type": "maintenance_window",
            "start_time": start_time,
            "end_time": end_time,
            "description": description,
            "services": services,
        }
        
        payload = {"maintenance_window": window}
        headers = self._get_rest_headers()
        
        return self._request_with_retry(
            method="POST",
            url=f"{self._config.rest_api_endpoint}/maintenance_windows",
            headers=headers,
            json_data=payload,
        )
    
    def validate(self) -> bool:
        """Validate connectivity by listing users or sending test event.
        
        Returns:
            True if connection works
        """
        # Try REST API first if key available
        if self._config.api_key:
            try:
                headers = self._get_rest_headers()
                self._request_with_retry(
                    "GET",
                    f"{self._config.rest_api_endpoint}/users",
                    headers=headers,
                    params={"limit": 1},
                )
                logger.info("PagerDuty REST API connectivity validated")
                return True
            except Exception as e:
                logger.warning("REST API validation failed: %s", e)
        
        # Try Events API if routing key available
        if self._config.routing_key:
            try:
                # Send a resolve with a unique key (safe, won't trigger)
                dedup_key = f"validation-test-{int(time.time())}"
                response = self.resolve_incident(dedup_key, "Validation test")
                logger.info("PagerDuty Events API connectivity validated")
                return True
            except Exception as e:
                logger.warning("Events API validation failed: %s", e)
        
        raise PagerDutyError("No valid PagerDuty API keys configured")


# Global client (lazy-loaded)
_global_client: Optional[PagerDutyClient] = None


def get_pagerduty_client() -> PagerDutyClient:
    """Get or create global PagerDutyClient."""
    global _global_client
    if _global_client is None:
        config = PagerDutyConfig.from_env()
        _global_client = PagerDutyClient(config)
    return _global_client

Pattern 2: Alert Deduplication & Correlation

"""Deduplication key strategies for PagerDuty Events API.

The dedup_key is critical:
- Same key = same incident (for ack/resolve)
- Used for deduplication of repeated events
- MUST be consistent across trigger/ack/resolve

Good strategies:
1. Service + Alert Type: "checkout-service-high-error-rate"
2. Service + Alert Type + Date: "checkout-service-high-errors-20240115"
3. Service + Specific Component: "checkout-service-db-connection-failed"
4. Include entity ID for per-resource alerts: "host-web-01-cpu-high"

Never use:
- Random UUIDs (can't resolve later)
- Timestamps alone (every event is unique)
- Very short or generic keys ("error", "alert")
"""

from __future__ import annotations

import hashlib
import logging
from typing import Any, Optional
from datetime import datetime, timezone

logger = logging.getLogger(__name__)


class DedupKeyStrategy:
    """Generator for consistent deduplication keys.
    
    Provides strategies for generating deterministic dedup_key values.
    """
    
    @staticmethod
    def by_service_and_type(
        service: str,
        alert_type: str,
        include_date: bool = True,
    ) -> str:
        """Generate key by service and alert type.
        
        Args:
            service: Service name (e.g., "checkout-service")
            alert_type: Alert type (e.g., "high-error-rate")
            include_date: Include YYYYMMDD for daily uniqueness
        
        Returns:
            dedup_key string
        """
        parts = [service, alert_type]
        
        if include_date:
            today = datetime.now(timezone.utc).strftime("%Y%m%d")
            parts.append(today)
        
        return "-".join(parts)
    
    @staticmethod
    def by_entity(
        service: str,
        entity_type: str,
        entity_id: str,
        alert_type: str,
    ) -> str:
        """Generate key including specific entity.
        
        Use for per-resource alerts like:
        - Specific host down
        - Specific database connection issue
        - Specific queue backlog
        
        Args:
            service: Service name
            entity_type: Type (host, queue, database, etc.)
            entity_id: Entity identifier
            alert_type: Alert type
        
        Returns:
            dedup_key string
        """
        return f"{service}-{entity_type}-{entity_id}-{alert_type}"
    
    @staticmethod
    def by_content_hash(
        service: str,
        alert_type: str,
        content: str,
    ) -> str:
        """Generate key by hashing alert content.
        
        Use when same exact error should deduplicate.
        
        Args:
            service: Service name
            alert_type: Alert type
            content: Content to hash (error message, stack trace, etc.)
        
        Returns:
            dedup_key with hash suffix
        """
        content_hash = hashlib.md5(content.encode()).hexdigest()[:12]
        return f"{service}-{alert_type}-{content_hash}"


# Alert correlation helpers

class AlertCorrelator:
    """Correlates related alerts to reduce noise.
    
    Use before sending to PagerDuty to:
    - Group related alerts
    - Determine severity escalation
    - Add context links
    """
    
    def __init__(self, client: Any) -> None:
        self._client = client
        self._recent_alerts: dict[str, float] = {}  # dedup_key -> timestamp
    
    def should_trigger(
        self,
        dedup_key: str,
        threshold_seconds: float = 300.0,  # 5 minutes
    ) -> tuple[bool, Optional[str]]:
        """Check if alert should trigger or is recent duplicate.
        
        Args:
            dedup_key: Deduplication key
            threshold_seconds: Suppress if triggered within this window
        
        Returns:
            (should_trigger, reason) tuple
        """
        now = datetime.now(timezone.utc).timestamp()
        
        if dedup_key in self._recent_alerts:
            last_trigger = self._recent_alerts[dedup_key]
            elapsed = now - last_trigger
            
            if elapsed < threshold_seconds:
                return (
                    False,
                    f"Suppressed: triggered {elapsed:.0f}s ago (threshold={threshold_seconds}s)"
                )
        
        # Record this trigger
        self._recent_alerts[dedup_key] = now
        
        # Clean up old entries
        self._cleanup_old_entries(now, threshold_seconds * 2)
        
        return (True, "First occurrence or outside suppression window")
    
    def _cleanup_old_entries(self, now: float, max_age: float) -> None:
        """Remove entries older than max_age."""
        keys_to_remove = [
            k for k, v in self._recent_alerts.items()
            if now - v > max_age
        ]
        for k in keys_to_remove:
            del self._recent_alerts[k]


# Example: Standard alerting workflow

def alert_on_high_error_rate(
    client: Any,
    correlator: AlertCorrelator,
    service: str,
    error_rate_pct: float,
    threshold_pct: float = 5.0,
) -> dict[str, Any]:
    """Standard workflow for high error rate alert.
    
    Steps:
    1. Check if rate exceeds threshold
    2. Check correlation/suppression window
    3. Build rich context
    4. Send to PagerDuty
    
    Returns:
        Result dict
    """
    if error_rate_pct < threshold_pct:
        return {"action": "suppressed", "reason": f"Below threshold ({error_rate_pct}% < {threshold_pct}%)"}
    
    # Determine severity
    if error_rate_pct >= 20.0:
        severity = "critical"
    elif error_rate_pct >= 10.0:
        severity = "error"
    else:
        severity = "warning"
    
    # Generate consistent dedup key
    dedup_key = DedupKeyStrategy.by_service_and_type(
        service=service,
        alert_type="high-error-rate",
        include_date=True,
    )
    
    # Check correlation
    should_trigger, reason = correlator.should_trigger(dedup_key)
    if not should_trigger:
        return {"action": "suppressed", "reason": reason, "dedup_key": dedup_key}
    
    # Build context
    summary = (
        f"High error rate on {service}: {error_rate_pct:.1f}% "
        f"(threshold: {threshold_pct}%)"
    )
    
    custom_details = {
        "service": service,
        "error_rate_pct": round(error_rate_pct, 2),
        "threshold_pct": threshold_pct,
        "severity": severity,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }
    
    # Send to PagerDuty
    response = client.trigger_incident(
        dedup_key=dedup_key,
        summary=summary,
        severity=severity,
        source=service,
        custom_details=custom_details,
        runbook_url=f"https://runbooks.example.com/high-error-rate",
        dashboard_url=f"https://grafana.example.com/d/{service}",
    )
    
    return {
        "action": "triggered",
        "dedup_key": dedup_key,
        "severity": severity,
        "response": response,
    }

Constraints

MUST DO

• Use Events API v2 for alert ingestion (trigger/ack/resolve), NOT REST API
• Always include dedup_key in all Events API calls
• Use routing_key for Events API, api_key for REST API
• Include required REST API headers: Authorization: Token token=..., Accept: application/vnd.pagerduty+json;version=2
• Add From header for user context operations
• Implement exponential backoff with jitter for 429 responses
• Include payload.summary, payload.severity, payload.source in all trigger events
• Use consistent dedup_key generation strategy across trigger/ack/resolve
• Include runbook_url and dashboard_url as links when triggering
• Set appropriate maintenance windows during deploys

MUST NOT DO

• NEVER use REST API for alert ingestion (Events API is designed for this)
• NEVER generate random dedup_key (can't resolve later)
• NEVER use timestamps in dedup_key (every event becomes unique)
• NEVER omit payload.source in trigger events (required field)
• NEVER send PII or sensitive data in custom_details
• NEVER ignore 429 responses without backoff
• NEVER create maintenance windows without timezone-aware timestamps
• NEVER resolve incidents with wrong dedup_key
• NEVER use event_action: trigger when you mean acknowledge or resolve
• NEVER hardcode API keys or routing keys in source code

Output Template

When implementing PagerDuty integrations, produce:

1. Client Initialization — PagerDutyConfig and PagerDutyClient reading from env vars
2. Deduplication Strategy — DedupKeyStrategy with service+alert_type+date pattern
3. Alert Correlation — AlertCorrelator for noise reduction and suppression windows
4. Standard Alert Workflow — Threshold check → severity determination → context building → trigger
5. Events API Usage — trigger_incident(), resolve_incident() with proper fields
6. REST API Helpers — list_incidents(), get_oncalls(), create_maintenance_window()
7. Rate Limit Handling — Exponential backoff with jitter, Retry-After parsing

Related Skills

Live References

📎 API Keys vs Routing Keys

…(truncated)