google_ads_bulk_update_ad_group_status

Instructions

Implementation Reference

• tools/ad_groups/mcp_tools_ad_groups.py:486-546 (handler)

  MCP tool handler: decorated with @mcp.tool(), accepts customer_id, ad_group_ids (list), and status (ENABLED/PAUSED/REMOVED). Calls AdGroupManager.bulk_update_ad_group_status(), logs via audit_logger, invalidates cache, and returns a formatted success/error message.

  @mcp.tool()
  def google_ads_bulk_update_ad_group_status(
      customer_id: str,
      ad_group_ids: List[str],
      status: str
  ) -> str:
      """
      Update status for multiple ad groups at once.
  
      Args:
          customer_id: Customer ID (without hyphens)
          ad_group_ids: List of ad group IDs to update
          status: New status for all ad groups (ENABLED, PAUSED, or REMOVED)
  
      Returns:
          Success message with count of updated ad groups
  
      Example:
          ad_group_ids = ["123456789", "987654321", "456789123"]
      """
      with performance_logger.track_operation('bulk_update_ad_group_status', customer_id=customer_id):
          try:
              client = get_auth_manager().get_client()
              ad_group_manager = AdGroupManager(client)
  
              if not ad_group_ids:
                  return "⚠️ No ad group IDs provided."
  
              status_upper = status.upper()
              result = ad_group_manager.bulk_update_ad_group_status(
                  customer_id,
                  ad_group_ids,
                  AdGroupStatus[status_upper]
              )
  
              # Audit log
              audit_logger.log_api_call(
                  customer_id=customer_id,
                  operation="bulk_update_ad_group_status",
                  resource_type="ad_group",
                  action="update",
                  result="success",
                  details={
                      'ad_group_count': len(ad_group_ids),
                      'new_status': status_upper
                  }
              )
  
              # Invalidate cache
              get_cache_manager().invalidate(customer_id, ResourceType.AD_GROUP)
  
              output = f"✅ Bulk status update completed!\n\n"
              output += f"**Ad Groups Updated**: {result['ad_groups_updated']}\n"
              output += f"**New Status**: {status_upper}\n\n"
              output += f"{result['message']}"
  
              return output
  
          except Exception as e:
              error_msg = ErrorHandler.handle_error(e, context="bulk_update_ad_group_status")
              return f"❌ Failed to bulk update ad group status: {error_msg}"

• managers/ad_group_manager.py:587-634 (helper)

  Helper/manager: builds AdGroupOperation objects for each ad group ID, sets the status field, applies a field mask for 'status', and sends the batch via mutate_ad_groups(). Returns count and status info.

  def bulk_update_ad_group_status(
      self,
      customer_id: str,
      ad_group_ids: List[str],
      status: AdGroupStatus
  ) -> Dict[str, Any]:
      """
      Update status for multiple ad groups at once.
  
      Args:
          customer_id: Customer ID
          ad_group_ids: List of ad group IDs
          status: New status for all ad groups
  
      Returns:
          Bulk operation result
      """
      ad_group_service = self.client.get_service("AdGroupService")
  
      operations = []
  
      for ad_group_id in ad_group_ids:
          ad_group_operation = self.client.get_type("AdGroupOperation")
          ad_group = ad_group_operation.update
  
          ad_group.resource_name = ad_group_service.ad_group_path(customer_id, ad_group_id)
          ad_group.status = self.client.enums.AdGroupStatusEnum[status.value]
  
          self.client.copy_from(
              ad_group_operation.update_mask,
              field_mask_pb2.FieldMask(paths=["status"])
          )
  
          operations.append(ad_group_operation)
  
      # Execute bulk update
      response = ad_group_service.mutate_ad_groups(
          customer_id=customer_id,
          operations=operations
      )
  
      logger.info(f"Bulk updated {len(ad_group_ids)} ad groups to {status.value}")
  
      return {
          "ad_groups_updated": len(ad_group_ids),
          "new_status": status.value,
          "message": f"Successfully updated {len(ad_group_ids)} ad groups"
      }

• managers/ad_group_manager.py:25-29 (schema)

  Schema/enum: AdGroupStatus enum with valid values ENABLED, PAUSED, REMOVED. Used by the tool handler to validate/convert the status string before passing to the manager.

  class AdGroupStatus(str, Enum):
      """Ad group status options."""
      ENABLED = "ENABLED"
      PAUSED = "PAUSED"
      REMOVED = "REMOVED"

• tools/ad_groups/mcp_tools_ad_groups.py:24-24 (registration)

  Registration entry point: register_ad_group_tools() is called with the FastMCP server instance. Inside this function, the @mcp.tool() decorator registers google_ads_bulk_update_ad_group_status.

  def register_ad_group_tools(mcp: FastMCP):

• google_ads_mcp.py:498-513 (registration)

  Top-level registration: google_ads_mcp.py imports and calls register_ad_group_tools (from the _TOOL_MODULES list at line 482) during server startup.

  def _register_all_modular_tools():
      """Import and register every modular tool module."""
      import importlib
  
      registered = 0
      for label, module_path, func_name in _TOOL_MODULES:
          try:
              mod = importlib.import_module(module_path)
              register_fn = getattr(mod, func_name)
              register_fn(mcp)
              logger.info(f"  ✓ {label}")
              registered += 1
          except Exception as exc:
              logger.error(f"  ✗ {label}: {exc}")
  
      logger.info(f"Registered {registered}/{len(_TOOL_MODULES)} modular tool modules")