GCP MCP

Instructions

Implementation Reference

• src/gcp_mcp/gcp_modules/monitoring/tools.py:186-284 (handler)

  The core handler function for the 'create_alert_policy' MCP tool. It uses the Google Cloud Monitoring API to create an alert policy with specified metric, threshold, duration, and optional notification channels. The @mcp.tool() decorator registers it as an MCP tool.
  @mcp.tool() def create_alert_policy(project_id: str, display_name: str, metric_type: str, filter_str: str, duration_seconds: int = 60, threshold_value: float = 0.0, comparison: str = "COMPARISON_GT", notification_channels: Optional[List[str]] = None) -> str: """ Create a new alert policy in a GCP project. Args: project_id: The ID of the GCP project display_name: The display name for the alert policy metric_type: The metric type to monitor (e.g., "compute.googleapis.com/instance/cpu/utilization") filter_str: The filter for the metric data duration_seconds: The duration in seconds over which to evaluate the condition (default: 60) threshold_value: The threshold value for the condition (default: 0.0) comparison: The comparison type (COMPARISON_GT, COMPARISON_LT, etc.) (default: COMPARISON_GT) notification_channels: Optional list of notification channel IDs Returns: Result of the alert policy creation """ try: from google.cloud import monitoring_v3 from google.protobuf import duration_pb2 # Initialize the Alert Policy Service client client = monitoring_v3.AlertPolicyServiceClient() # Format the project name project_name = f"projects/{project_id}" # Create a duration object duration = duration_pb2.Duration(seconds=duration_seconds) # Create the alert condition condition = monitoring_v3.AlertPolicy.Condition( display_name=f"Condition for {display_name}", condition_threshold=monitoring_v3.AlertPolicy.Condition.MetricThreshold( filter=filter_str, comparison=getattr(monitoring_v3.ComparisonType, comparison), threshold_value=threshold_value, duration=duration, trigger=monitoring_v3.AlertPolicy.Condition.Trigger( count=1 ), aggregations=[ monitoring_v3.Aggregation( alignment_period=duration_pb2.Duration(seconds=60), per_series_aligner=monitoring_v3.Aggregation.Aligner.ALIGN_MEAN, cross_series_reducer=monitoring_v3.Aggregation.Reducer.REDUCE_MEAN ) ] ) ) # Create the alert policy alert_policy = monitoring_v3.AlertPolicy( display_name=display_name, conditions=[condition], combiner=monitoring_v3.AlertPolicy.ConditionCombinerType.OR ) # Add notification channels if provided if notification_channels: alert_policy.notification_channels = [ f"projects/{project_id}/notificationChannels/{channel_id}" for channel_id in notification_channels ] # Create the policy policy = client.create_alert_policy(name=project_name, alert_policy=alert_policy) # Format response conditions_str = "\n".join([ f"- {c.display_name}: {c.condition_threshold.filter}" for c in policy.conditions ]) notifications_str = "None" if policy.notification_channels: notifications_str = "\n".join([ f"- {channel.split('/')[-1]}" for channel in policy.notification_channels ]) return f""" Alert Policy created successfully: - Name: {policy.display_name} - Policy ID: {policy.name.split('/')[-1]} - Combiner: {policy.combiner.name} Conditions: {conditions_str} Notification Channels: {notifications_str} """ except Exception as e: return f"Error creating alert policy: {str(e)}"
• src/gcp_mcp/server.py:54-54 (registration)

  Invocation of register_tools from the monitoring module, which defines and registers the 'create_alert_policy' tool using the @mcp.tool() decorator.
  monitoring_tools.register_tools(mcp)
• src/gcp_mcp/server.py:13-13 (registration)

  Import of the monitoring tools module containing the 'create_alert_policy' implementation and registration.
  from .gcp_modules.monitoring import tools as monitoring_tools