PagerDuty MCP Server

Instructions

Implementation Reference

• src/pagerduty_mcp_server/server.py:40-90 (handler)

  The primary handler for the 'get_escalation_policies' tool. Registered via @mcp.tool() decorator. Dispatches to helper functions based on whether a specific policy_id is provided or list filters are used. Includes input validation and user context building.

  @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/escalation_policies.py:19-63 (helper)

  Helper function to list escalation policies using PagerDuty API. Called by the main handler when no specific policy_id is provided. Handles filtering params, API call, parsing with parse_escalation_policy, and response formatting.

  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-98 (helper)

  Helper function to retrieve a single escalation policy by ID using PagerDuty API. Called by the main handler when policy_id is provided. Handles API call, parsing, and response formatting.

  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)