get_escalation_policies

Instructions

Implementation Reference

• src/pagerduty_mcp_server/server.py:38-87 (handler)

  The MCP tool handler function `get_escalation_policies` decorated with @mcp.tool(). It dispatches to either `show_escalation_policy` (if policy_id is provided) or `list_escalation_policies` (with optional user context via `build_user_context`).

  @mcp.tool()
  def get_escalation_policies(
      *,
      policy_id: Optional[str] = None,
      current_user_context: bool = True,
      query: Optional[str] = None,
      user_ids: Optional[List[str]] = None,
      team_ids: Optional[List[str]] = None,
      limit: Optional[int] = None,
  ) -> Dict[str, Any]:
      """Get PagerDuty escalation policies by filters or get details for a specific policy ID.
  
      Args:
          policy_id (str): The escalation policy ID to retrieve (optional, cannot be used with any other filters).
          current_user_context (bool): Use current user's ID/team IDs context (default: True). Not used if `policy_id` is provided.
          query (str): Policies whose names contain the search query (optional). Not used if `policy_id` is provided.
          user_ids (List[str]): Policies that include these user IDs (optional, excludes current_user_context). Not used if `policy_id` is provided.
          team_ids (List[str]): Policies assigned to these team IDs (optional, excludes current_user_context). Not used if `policy_id` is provided.
          limit (int): Limit the number of results (optional). Not used if `policy_id` is provided.
      """
      if policy_id is not None:
          disallowed_filters_present = (
              query is not None
              or user_ids is not None
              or team_ids is not None
              or limit is not None
          )
          if disallowed_filters_present:
              raise ValueError(
                  "When `policy_id` is provided, other filters (like query, user_ids, team_ids, limit) cannot be used. See `docs://tools` for more information."
              )
  
          return escalation_policies.show_escalation_policy(policy_id=policy_id)
  
      if current_user_context:
          if user_ids is not None or team_ids is not None:
              raise ValueError(
                  "Cannot specify user_ids or team_ids when current_user_context is True. See `docs://tools` for more information."
              )
          user_context = users.build_user_context()
          user_ids = [user_context["user_id"]]
          team_ids = user_context["team_ids"]
      elif not (user_ids or team_ids):
          raise ValueError(
              "Must specify at least user_ids or team_ids when current_user_context is False. See `docs://tools` for more information."
          )
  
      return escalation_policies.list_escalation_policies(
          query=query, user_ids=user_ids, team_ids=team_ids, limit=limit
      )

• src/pagerduty_mcp_server/parsers/escalation_policy_parser.py:6-38 (schema)

  The `parse_escalation_policy` function that transforms raw PagerDuty API escalation policy responses into a structured format with consistent field names (id, name, description, escalation_rules, services, teams).

  def parse_escalation_policy(*, result: Dict[str, Any]) -> Dict[str, Any]:
      """Parses a raw escalation policy API response into a structured format without unneeded fields.
  
      Args:
          result (Dict[str, Any]): The raw escalation policy API response
  
      Returns:
          Dict[str, Any]: A dictionary containing:
              - id (str): The policy ID
              - name (str): The policy name
              - escalation_rules (List[Dict]): List of escalation rules, each containing:
                  - id (str): Rule ID
                  - escalation_delay_in_minutes (int): Delay before escalation
                  - targets (List[Dict]): List of targets with id, type, and summary
              - services (List[Dict]): List of services with id
              - teams (List[Dict]): List of teams with id and summary
              - description (str): Policy description
  
      Note:
          If the input is None or not a dictionary, returns an empty dictionary.
          All fields are optional and will be None if not present in the input.
  
      Raises:
          KeyError: If accessing nested dictionary fields fails
      """
  
      if not result:
          return {}
  
      parsed_policy = {}
  
      # Simple fields
      simple_fields = ["id", "name", "description"]

• src/pagerduty_mcp_server/server.py:38-39 (registration)

  The `@mcp.tool()` decorator on line 38 registers `get_escalation_policies` as an MCP tool on the FastMCP server instance.

  @mcp.tool()
  def get_escalation_policies(

• src/pagerduty_mcp_server/escalation_policies.py:19-62 (helper)

  The `list_escalation_policies` helper function that queries the PagerDuty API /escalation_policies endpoint with optional filters (query, user_ids, team_ids, limit) and returns parsed results.

  def list_escalation_policies(
      *,
      query: Optional[str] = None,
      user_ids: Optional[List[str]] = None,
      team_ids: Optional[List[str]] = None,
      limit: Optional[int] = None,
  ) -> Dict[str, Any]:
      """List escalation policies based on the given criteria. Exposed in `get_escalation_policies`.
  
      Args:
          query (str): Filter escalation policies whose names contain the search query (optional)
          user_ids (List[str]): Filter results to only escalation policies that include the given user IDs (optional)
          team_ids (List[str]): Filter results to only escalation policies assigned to teams with the given IDs (optional)
          limit (int): Limit the number of results returned (optional)
  
      Returns:
          See the "Standard Response Format" section in `tools.md` for the complete standard response structure.
          The response will contain a list of escalation policies in the standard format.
      Raises:
          See the "Error Handling" section in `tools.md` for common error scenarios.
      """
  
      pd_client = create_client()
  
      params = {}
      if query:
          params["query"] = query
      if user_ids:
          params["user_ids[]"] = user_ids
      if team_ids:
          params["team_ids[]"] = team_ids
      if limit:
          params["limit"] = limit
  
      try:
          response = pd_client.list_all(ESCALATION_POLICIES_URL, params=params)
          parsed_response = [
              parse_escalation_policy(result=result) for result in response
          ]
          return utils.api_response_handler(
              results=parsed_response, resource_name="escalation_policies"
          )
      except Exception as e:
          utils.handle_api_error(e)

• src/pagerduty_mcp_server/escalation_policies.py:65-97 (helper)

  The `show_escalation_policy` helper function that fetches a single escalation policy by ID from the PagerDuty API and returns parsed details.

  def show_escalation_policy(*, policy_id: str) -> Dict[str, Any]:
      """Get detailed information about a given escalation policy. Exposed in `get_escalation_policies`.
  
      Args:
          policy_id (str): The ID of the escalation policy to get
  
      Returns:
          See the "Standard Response Format" section in `tools.md` for the complete standard response structure.
          The response will contain a single escalation policy in the standard format.
      Raises:
          See the "Error Handling" section in `tools.md` for common error scenarios.
      """
  
      if not policy_id:
          raise ValueError("policy_id cannot be empty")
  
      pd_client = create_client()
  
      try:
          response = pd_client.jget(f"{ESCALATION_POLICIES_URL}/{policy_id}")
          try:
              policy_data = response["escalation_policy"]
          except KeyError:
              raise RuntimeError(
                  f"Failed to fetch escalation policy {policy_id}: Response missing 'escalation_policy' field"
              )
  
          return utils.api_response_handler(
              results=parse_escalation_policy(result=policy_data),
              resource_name="escalation_policy",
          )
      except Exception as e:
          utils.handle_api_error(e)