GCP MCP

Instructions

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)}"