get_monitoring_alerts

Instructions

    Get active monitoring alerts for a GCP project.
    
    Args:
        project_id: The ID of the GCP project to get alerts for
    
    Returns:
        Active alerts for the specified GCP project
    

Implementation Reference

• src/gcp_mcp/gcp_modules/monitoring/tools.py:67-184 (handler)

  The handler function for the 'get_monitoring_alerts' MCP tool. It lists enabled alert policies in the GCP project and checks for active incidents in the last hour by querying time series data for incident counts. Returns formatted list of active alerts or error message. Registered directly via @mcp.tool() decorator.

      @mcp.tool()
      def get_monitoring_alerts(project_id: str) -> str:
          """
          Get active monitoring alerts for a GCP project.
          
          Args:
              project_id: The ID of the GCP project to get alerts for
          
          Returns:
              Active alerts for the specified GCP project
          """
          try:
              from google.cloud import monitoring_v3
              from google.protobuf.json_format import MessageToDict
              
              # Initialize the Alert Policy Service client
              alert_client = monitoring_v3.AlertPolicyServiceClient()
              
              # Format the project name
              project_name = f"projects/{project_id}"
              
              # Create the request object
              request = monitoring_v3.ListAlertPoliciesRequest(
                  name=project_name
              )
              
              # List all alert policies
              alert_policies = alert_client.list_alert_policies(request=request)
              
              # Initialize the Metric Service client for metric data
              metric_client = monitoring_v3.MetricServiceClient()
              
              # Format the response
              active_alerts = []
              
              for policy in alert_policies:
                  # Check if the policy is enabled
                  if not policy.enabled:
                      continue
                  
                  # Check for active incidents
                  filter_str = f'resource.type="alerting_policy" AND metric.type="monitoring.googleapis.com/alert_policy/incident_count" AND metric.label.policy_name="{policy.name.split("/")[-1]}"'
                  
                  # Create a time interval for the last hour
                  import datetime
                  from google.protobuf import timestamp_pb2
                  
                  now = datetime.datetime.utcnow()
                  seconds = int(now.timestamp())
                  end_time = timestamp_pb2.Timestamp(seconds=seconds)
                  
                  start_time = datetime.datetime.utcnow() - datetime.timedelta(hours=1)
                  seconds = int(start_time.timestamp())
                  start_time_proto = timestamp_pb2.Timestamp(seconds=seconds)
                  
                  # Create the time interval
                  interval = monitoring_v3.TimeInterval(
                      start_time=start_time_proto,
                      end_time=end_time
                  )
                  
                  # List the time series
                  try:
                      # Create the request object
                      request = monitoring_v3.ListTimeSeriesRequest(
                          name=project_name,
                          filter=filter_str,
                          interval=interval,
                          view=monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL
                      )
                      
                      # List the time series
                      time_series = metric_client.list_time_series(request=request)
                      
                      is_active = False
                      for series in time_series:
                          # Check if there's a non-zero value in the time series
                          for point in series.points:
                              if point.value.int64_value > 0:
                                  is_active = True
                                  break
                          if is_active:
                              break
                      
                      if is_active:
                          # Format conditions
                          conditions = []
                          for condition in policy.conditions:
                              if condition.display_name:
                                  conditions.append(f"    - {condition.display_name}: {condition.condition_threshold.filter}")
                          
                          # Add to active alerts
                          alert_info = [
                              f"- {policy.display_name} (ID: {policy.name.split('/')[-1]})",
                              f"  Description: {policy.documentation.content if policy.documentation else 'No description'}",
                              f"  Severity: {policy.alert_strategy.notification_rate_limit.period.seconds}s between notifications" if policy.alert_strategy.notification_rate_limit else "  No rate limiting"
                          ]
                          
                          if conditions:
                              alert_info.append("  Conditions:")
                              alert_info.extend(conditions)
                          
                          active_alerts.append("\n".join(alert_info))
                  except Exception:
                      # Skip if we can't check for active incidents
                      continue
              
              if not active_alerts:
                  return f"No active alerts found for project {project_id}."
              
              alerts_str = "\n".join(active_alerts)
              
              return f"""
  Active Monitoring Alerts in GCP Project {project_id}:
  {alerts_str}
  """
          except Exception as e:
              return f"Error getting monitoring alerts: {str(e)}"