サーバーサイドエンジニアの小谷です。
サーバーサイドGTMをGTM-APIを使って構築してみたので、その内容を記載していきます。

方法としては、GCPのサービスアカウントを作成し、そのキーファイルを使って、PythonでGTM-APIをたたいて構築を行いました。
また、ウェブサイトからサーバーコンテナへの送信は、Google Analytics4 のタグを含めたウェブコンテナのGTMを使っており、本記事中のPythonコードでは、ウェブコンテナとサーバーコンテナの2つのGTMコンテナを作成しています。

サーバーサイドGTMとは?従来のGTMとの違いと利点

(この段落の文章は、Gemini Flashにて生成した文章です)
ウェブサイトに様々なタグを簡単に実装できるツールとして広く知られるGoogle Tag Manager (GTM)ですが、近年注目されているのが「サーバーサイドGTM」です。従来のGTMはブラウザ上でJavaScriptを使ってタグを実装する「Web コンテナ」方式でしたが、サーバーサイドGTMはサーバー上でタグを実装する方式です。

サーバーサイドGTMでは、ブラウザではなくサーバーでタグ処理を行うため、プライバシー保護、パフォーマンス向上、クッキー管理など、従来のGTMに比べて様々なメリットがあります。

事前に必要な内容

この記事に記載のコードを実施するには次の内容が事前に必要です。

  1. Pythonの実施環境
    ローカルPCではバージョン3.12を使用していました。
    ※GCPのCloud Shell(記事記載時点のバージョンはpython 3.10.12)でも動作しました。
    requirements.txtで入れたモジュールは次の内容です。
     google-api-python-client
     google-auth
     google-auth-httplib2
     google-auth-oauthlib
  2. サーバーコンテナのURL
  3. Google Analytics4の測定ID (G-XXXXXXXX)
    取得手順:Google Analyticsのコンソール上で、画面左メニューの[管理] – [データストリーム] – 対象のストリームを選択すると、測定IDが表示されます。
  4. GTMのアカウントID
    取得手順:GTMのコンソール上で、アカウント設定のページから取得
  5. GCPのコンソール上で、Tag Manager API を有効にする。
    有効化しないで実行すると、403エラーになります。
  6. GCPのサービスアカウントを新規作成し、メールアドレスとキーファイル(json)を取得
    取得手順:GCPのコンソール上で[IAMと管理] – [サービスアカウント] – [サービスアカウントを作成]にて作成し、サービスアカウント一覧ページの該当のサービスアカウントの行の右側にある[操作]列から「鍵を管理」をクリックし、[鍵を追加]ボタンから[新しい鍵を作成]をクリックしキータイプはJSONにて発行
  7. GTMの管理者権限をサービスアカウントに付与
    GTMのコンソールのユーザー管理にて、上記6.で作成したGCPのサービスアカウントのメールアドレスを登録のうえ、管理者権限を付与して招待します。


コードに定義内容を記入

事前に必要な内容の準備が整えば、
下記のPythonコード内の所定箇所に、次の設定情報を記述します。

# 設定情報
CONFIG = {
    'gcpServiceAccountKeyFile': 'xxxxxxxx.json', # GCPのサービスアカウントキーファイル
    'webContainerName': 'WebContainer', # ウェブコンテナ名
    'serverContainerName': 'ServerContainer', # サーバーコンテナ名
    'serverContainerUrl': 'https://', # サーバーコンテナのURL
    'accountId': '', # GTMのアカウントID 数値
}
 

全体のPythonコードは下記のとおりで、実行時に必要なファイルは、下記コードのPythonファイルとGCPのサービスアカウントのキーファイル(json)1枚のみです。

import sys, os
import time
from google.oauth2 import service_account
import google.auth.transport.requests
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError



if(len(sys.argv) != 2 or (sys.argv[1] != 'web' and sys.argv[1] != 'server')):
    filename = os.path.basename(__file__)
    print('Usage:')
    print('  python '+filename+' web    # ウェブコンテナの構築')
    print('  python '+filename+' server # サーバーコンテナの構築')
    exit(0)


# 設定情報
CONFIG = {
    'gcpServiceAccountKeyFile': 'xxxxxxxx.json', # GCPのサービスアカウントキーファイル
    'webContainerName': 'WebContainer', # ウェブコンテナ名
    'serverContainerName': 'ServerContainer', # サーバーコンテナ名
    'serverContainerUrl': 'https://', # サーバーコンテナのURL
    'accountId': '', # GTMのアカウントID 数値
    'ga4MeasurementId': '' # GA4の測定ID G-XXXXXXXXXX
}


# 設定情報のチェック
is_config_set = True
for key in CONFIG:
    if not CONFIG[key]:
        print(f'    Error: {key} is not set in CONFIG.')
        is_config_set = False

# CONFIG['gcpServiceAccountKeyFile'] ファイルが存在するかチェック
if(CONFIG['gcpServiceAccountKeyFile'] != '' and not os.path.exists(CONFIG['gcpServiceAccountKeyFile'])):
    print(f'    Error: {CONFIG["gcpServiceAccountKeyFile"]} does not exist.')
    is_config_set = False

if not is_config_set:
    print('Please set the configuration information.')
    exit(0)



# サービスアカウントの認証情報をロード
credentials = service_account.Credentials.from_service_account_file(
    CONFIG['gcpServiceAccountKeyFile'],
    scopes=[
        'https://www.googleapis.com/auth/tagmanager.edit.containers',
        'https://www.googleapis.com/auth/tagmanager.delete.containers',
        'https://www.googleapis.com/auth/tagmanager.edit.containerversions',
        'https://www.googleapis.com/auth/tagmanager.manage.accounts',
        'https://www.googleapis.com/auth/tagmanager.publish'
    ]
)

# 認証情報を使用してHTTPクライアントを作成
request = google.auth.transport.requests.Request()
credentials.refresh(request)
service = build('tagmanager', 'v2', credentials=credentials) # GTM API サービスを作成



class GtmContainer:
    def __init__(self, service, config) -> None:
        self.service = service # GTM API サービス
        self.config = config # 設定情報
        self.container = None  # コンテナ情報
        self.latestVersionId = None  # 最新バージョンID
        self.workspaceId = None  # ワークスペースID
        self.sleep_count = 0
        self.containers = self.get_containers()


    def sleep(self, seconds=4) -> None:
        """
        RateLimit対策で、API呼び出し後に短い時間スリープする 
        :param seconds: スリープ時間(秒)
        """
        self.sleep_count += 1
        print(f'Sleep {seconds} seconds. count: {self.sleep_count}')
        time.sleep(seconds)


    def get_containers(self) -> list:
        """
        コンテナの一覧を取得
        :return: コンテナの一覧
        """
        print('Get containers')
        try:
            containers = self.service.accounts().containers().list(
                parent=f'accounts/{self.config["accountId"]}'
            ).execute()
            self.sleep()

            if 'container' not in containers:
                self.containers = []
            else:
                self.containers = containers['container']

            return self.containers

        except HttpError as error:
            print(f'Error retrieving containers: {error}')
            sys.exit(1)


    def get_container_names(self) -> list:
        """
        コンテナ名の一覧を取得
        :return: コンテナ名の一覧
        """
        print('Get container names')
        try:
            return [container['name'] for container in self.containers]
        except HttpError as error:
            print(f'Error retrieving container names: {error}')
            sys.exit(1)


    def get_or_create_container(self, container_body) -> None:
        """
        コンテナを取得または作成
        :param container_body: コンテナ情報
        """
        print('Get or create container:', container_body['name'])
        try:
            # すでにコンテナが存在する場合は取得
            self.container = self.get_container(container_body['name'])
            if self.container:
                print('Container already exists:', self.container)
            else:
                self.create_container(container_body)

            # 最新バージョンIDを取得
            # self.latestVersionId = self.get_container_latest_version_id()
            # print('latestVersionId:', self.latestVersionId)

            # デフォルトワークスペースIDを取得
            self.workspaceId = self.get_default_workspace_id()

        except HttpError as error:
            print(f'Error getting or creating container: {error}')
            sys.exit(1)


    def get_container(self, container_name) -> dict:
        """
        コンテナを取得
        :param container_name: コンテナ名
        :return: コンテナ情報
        """
        try:
            for container in self.containers:
                if container['name'] == container_name:
                    return container
        except HttpError as error:
            print(f'Error retrieving container: {error}')
        return None 


    def create_container(self, container_body) -> dict:
        """
        コンテナを作成
        :param container_body: コンテナ情報
        """
        print('Create container:', container_body['name'])
        try:
            self.container = self.service.accounts().containers().create(
                parent=f'accounts/{self.config["accountId"]}',
                body=container_body
            ).execute()
            print('Created container:', self.container)
            self.sleep()
            return self.container
        except HttpError as error:
            print(f'Error creating container: {error}')
            sys.exit(1)


    def get_container_latest_version_id(self) -> str:
        """
        コンテナの最新バージョンIDを取得
        :return: バージョンID
        """
        print('Get latest container version ID')
        try:
            response = self.service.accounts().containers().version_headers().latest(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}"
            ).execute()
            self.sleep()
            return response['containerVersionId']
        except HttpError as error:
            print(f'Error retrieving latest version ID: {error}')
            sys.exit(1)


    def get_default_workspace_id(self) -> str:
        """
        デフォルトのワークスペースIDを取得
        :return: ワークスペースID
        """
        print('Get default workspace ID')
        try:
            workspaces = self.service.accounts().containers().workspaces().list(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}"
            ).execute()
            self.sleep()
            for workspace in workspaces['workspace']:
                if workspace['name'] == 'Default Workspace':
                    return workspace['workspaceId']
        except HttpError as error:
            print(f'Error retrieving default workspace ID: {error}')
            sys.exit(1)


    def create_version(self) -> dict:
        """
        バージョンを作成
        :return: バージョン情報
        """
        print('Create version')
        try:
            version_body = {
                'name': 'New version',
                'notes': 'Published using the GTM API'
            }
            response = self.service.accounts().containers().workspaces().create_version(
                path=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}",
                body=version_body
            ).execute()
            print('Created version:', response)
            self.sleep()
            return response['containerVersion']
        except HttpError as error:
            print(f'Error creating version: {error}')
            sys.exit(1)
    

    def publish(self, version_path) -> dict:
        """
        バージョンを公開
        :param version_path: バージョンのパス
        :return: バージョン情報
        """
        print('Publish version')
        try:
            response = self.service.accounts().containers().versions().publish(
                path=version_path
            ).execute()
            print('Published version:', response)
            self.sleep()
            return response
        except HttpError as error:
            print(f'Error publishing version: {error}')
            sys.exit(1)


    def create_variable(self, variable_body: dict) -> dict:
        """
        変数を作成
        :param variable_body: 変数情報
        :return: 変数情報
        """
        print('Create variable', variable_body['name'])
        try:
            variable = self.service.accounts().containers().workspaces().variables().create(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}",
                body=variable_body
            ).execute()
            print('Created variable', variable)
            self.sleep()
            return variable
        except HttpError as error:
            print(f'Error creating variable: {error}')
            sys.exit(1)


    def create_tag(self, tag_body: dict) -> dict:
        """
        タグを作成
        :param tag_body: タグ情報
        :return: タグ情報
        """
        print('Create tag:', tag_body['name'])
        try:
            tag = self.service.accounts().containers().workspaces().tags().create(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}",
                body=tag_body
            ).execute()
            print('Created tag:', tag)
            self.sleep()
            return tag
        except HttpError as error:
            print(f'Error creating tag: {error}')
            sys.exit(1)
    

    def create_trigger(self, trigger_body: dict) -> dict:
        """
        トリガーを作成
        :param trigger_body: トリガー情報
        :return: トリガー情報
        """
        print('Create trigger:', trigger_body['name'])
        try:
            trigger = self.service.accounts().containers().workspaces().triggers().create(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}",
                body=trigger_body
            ).execute()
            print('Created trigger:', trigger)
            self.sleep()
            return trigger
        except HttpError as error:
            print(f'Error creating trigger: {error}')
            sys.exit(1)

    
    def create_client(self, client_body: dict) -> dict:
        """
        クライアントを作成
        :param client_body: クライアント情報
        :return: クライアント情報
        """
        print('Create client:', client_body['name'])
        try:
            client = self.service.accounts().containers().workspaces().clients().create(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}",
                body=client_body
            ).execute()
            print('Created client:', client)
            self.sleep()
            return client
        except HttpError as error:
            print(f'Error creating client: {error}')
            sys.exit(1)


    def get_tags(self) -> list:
        """
        タグの一覧を取得
        :return: タグの一覧
        """
        print('Get tags')
        try:
            tags = self.service.accounts().containers().workspaces().tags().list(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}"
            ).execute()
            self.sleep()

            if 'tag' not in tags:
                return []
            return tags['tag']
        except HttpError as error:
            print(f'Error retrieving tags: {error}')
            sys.exit(1)


    def get_triggers(self) -> list:
        """
        トリガーの一覧を取得
        :return: トリガーの一覧
        """
        print('Get triggers')
        try:
            triggers = self.service.accounts().containers().workspaces().triggers().list(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}"
            ).execute()
            self.sleep()
            
            if 'trigger' not in triggers:
                return []
            return triggers['trigger']
        except HttpError as error:
            print(f'Error retrieving triggers: {error}')
            sys.exit(1)
    

    def get_variables(self) -> list:
        """
        変数の一覧を取得
        :return: 変数の一覧
        """
        print('Get variables')
        try:
            variables = self.service.accounts().containers().workspaces().variables().list(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}"
            ).execute()
            self.sleep()

            if 'variable' not in variables:
                return []
            return variables['variable']
        except HttpError as error:
            print(f'Error retrieving variables: {error}')
            sys.exit(1)
    

    def get_clients(self) -> list:
        """
        クライアントの一覧を取得
        :return: クライアントの一覧
        """
        print('Get clients')
        try:
            clients = self.service.accounts().containers().workspaces().clients().list(
                parent=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}"
            ).execute()
            self.sleep()

            if 'client' not in clients:
                return []
            return clients['client']
        except HttpError as error:
            print(f'Error retrieving clients: {error}')
            sys.exit(1)

    def update_client(self, client_id, client_body):
        """
        指定されたクライアントの設定を更新する
        :param client_id: クライアント ID
        :param client_body: クライアント設定
        """
        print('Update client: ', client_id, client_body['name'])
        try:
            response = self.service.accounts().containers().workspaces().clients().update(
                path=f"accounts/{self.config['accountId']}/containers/{self.container['containerId']}/workspaces/{self.workspaceId}/clients/{client_id}",
                # path=f"accounts/{container_id}/containers/{container_id}/workspaces/{workspace_id}/clients/{client_id}",
                body=client_body
            ).execute()
            print(f'Updated client: {response}')
            self.sleep()
        except HttpError as error:
            print(f'Error updating client: {error}')
            sys.exit(1)



""""
ウェブコンテナ側の設定

  コンテナ名: WebContainer
  ターゲット プラットフォーム: ウェブ

  変数: ユーザー定義変数
     変数名: GA Measurement ID
     タイプ: 定数
     値: G-XXXXXXXXXX ※GAの測定ID

  変数: ユーザー定義変数
     変数名: Server Container URL
     タイプ: Googleタグ設定:
     構成パラメータ:
      名前: server_container_url 値: https://

  タグ: タグ名: GA4 Event
     タイプ: Google アナリティクス: GA4イベント
     測定ID: {{GA Measurement ID}} 変数に設定したGA Measurement ID
     イベント名: GA4Event
     トリガー: Initailization All Pages

  タグ: タグ名: Server Container URL
     タイプ: Googleタグ
     タグID: {{GA Measurement ID}} 変数に設定したGA Measurement ID
     設定
      構成の設定変数: {{Server Container URL}} 変数に設定したServer Container URL
     トリガー: Initailization All Pages
"""


if 'web' == sys.argv[1]:
    web_container = GtmContainer(service, CONFIG)
    # ウェブコンテナを取得または作成
    web_container.get_or_create_container({'name': CONFIG['webContainerName'], 'usageContext': ['web']})

    # 変数を作成 GA4の測定IDを設定
    web_container.create_variable({'name': 'GA Measurement ID', 'type': 'c', # type:c定数
                        'parameter': [{'type': 'template', 'key': 'value', 'value': CONFIG['ga4MeasurementId']}]})

    # 変数を作成 サーバーコンテナのURLを設定
    variable_body = {'name': 'Server Container URL','type': 'gtcs', # type:gtcs Googleタグ設定
        'parameter': [{'type': 'list','key': 'configSettingsTable',
                'list': [{'type': 'map',
                        'map': [
                            {'type': 'template','key': 'parameter','value': 'server_container_url'},
                            {'type': 'template','key': 'parameterValue','value': CONFIG['serverContainerUrl']}
                        ]}]}]}
    web_container.create_variable(variable_body)

    # タグを作成 GA4イベント
    tag_body = {
        'name': 'GA4 Event', 'type': 'gaawe',  # Google Analytics: GA4イベント
        'parameter': [
            {'type': 'boolean', 'key': 'sendEcommerceData', 'value': 'false'},
            {'type': 'boolean', 'key': 'enhancedUserId', 'value': 'false'},
            {'type': 'template', 'key': 'eventName', 'value': 'GA4Event'},
            {'type': 'template', 'key': 'measurementIdOverride', 'value': '{{GA Measurement ID}}'} # 変数を指定
        ],
        'firingTriggerId': ['2147479573'], # トリガーIDを指定 2147479573 は Initialization All Pages
        'tagFiringOption': 'oncePerEvent',
        'monitoringMetadata': {'type': 'map'},
        'consentSettings': {'consentStatus': 'notSet'}
    }
    web_container.create_tag(tag_body)

    # タグを作成 サーバーコンテナURL
    tag_body = {
        'name': 'Server Container URL', 'type': 'googtag',
        'parameter': [
            {'type': 'template', 'key': 'tagId', 'value': '{{GA Measurement ID}}'},
            {'type': 'template', 'key': 'configSettingsVariable', 'value': '{{Server Container URL}}'}
        ],
        'firingTriggerId': ['2147479573'], 'tagFiringOption': 'oncePerEvent',
        'monitoringMetadata': {'type': 'map'}, 'consentSettings': {'consentStatus': 'notSet'}
    }
    web_container.create_tag(tag_body)

    # バージョン作成と公開
    version = web_container.create_version()
    web_container.publish(version['path'])
    print('Web container created and published successfully.')



""""
サーバーコンテナ側の設定

  コンテナ名: ServerContainer
  ターゲット プラットフォーム: Server

  コンテナの設定
   ※[管理] - [コンテナの設定] - [URLを追加]
   サーバーコンテナのURL: https://
   タグ設定サーバー
    コンテナの設定(CONTAINER_CONFIG): aWQ9Rxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx== 

  変数: ユーザー定義変数
     タイプ: 定数
     変数名: GA Measurement ID
     値: G-XXXXXXXXXX ※GAの測定ID

  トリガー: トリガーのタイプ: カスタム
       トリガー名: GA4 Event
       このトリガーの発生場所: 一部のイベント [Event Name] [等しい] [GA4Event]

  クライアント
     タイプ: Google アナリティクス: GA4(ウェブ)
     クライアント名: GA4 Client
     優先度: 0
     有効化の条件
      ・デフォルトのGA4パス [チェック有り]
      ・特定のID向けのデフォルトのgtag.jsパス [チェック有り]
       測定ID: {{GA Measurement ID}}
      ・依存関係にあるすべてのGoogleスクリプトを自動で配信する [チェック有り]
      ・HTTPレスポンスを圧縮する [チェック有り]
      ・地域ごとの設定を有効にする [チェック無し]
     詳細設定: 初期設定のまま

  クライアント
     タイプ: Google タグマネージャー: ウェブコンテナ
     クライアント名: Web Container Client
     優先度: 0
     指定された ID に gtm.js を配信する
      許可されているコンテナ ID: GTM-XXXXXXX1 ※ウェブコンテナのID
      ・依存関係にあるすべての Google スクリプトを自動で配信する [チェック有り]
      ・HTTP レスポンスを圧縮する [チェック有り]
      ・地域ごとの設定を有効にする [チェック無し]

  タグ: タイプ: Google アナリティクス: GA4
     タグ名: GA4
     測定ID: {{GA Measurement ID}}
     Google シグナルを有効にする。 [チェック無し]
     ユーザーの IP アドレスを削除する: true
     トリガー: GA4 Event
"""
if 'server' == sys.argv[1]:
    server_container = GtmContainer(service, CONFIG)

    # ウェブコンテナが無ければエラー ※ウェブコンテナのIDが必要なためエラーにしている。
    container_names = server_container.get_container_names()
    if CONFIG['webContainerName'] not in container_names:
        print(f'Error: Web container {CONFIG["webContainerName"]} does not exist.')
        exit(0)
    web_container = server_container.get_container(CONFIG["webContainerName"])

    # サーバーコンテナを取得または作成
    server_container.get_or_create_container({'name': CONFIG['serverContainerName'], 'usageContext': ['server']})
    
    # 変数を作成 GA4の測定IDを設定
    server_container.create_variable({'name': 'GA Measurement ID', 'type': 'c', # type:c定数
        'parameter': [{'type': 'template', 'key': 'value', 'value': CONFIG['ga4MeasurementId']}]})

    # トリガーを作成 GA4イベント
    trigger_body = {
        'name': 'GA4 Event', 'type': 'always',
        'filter': [{'type': 'equals',
                'parameter': [
                    {'type': 'template', 'key': 'arg0', 'value': '{{Event Name}}'},
                    {'type': 'template', 'key': 'arg1', 'value': 'GA4Event'}
                ]}]}
    trigger = server_container.create_trigger(trigger_body)
    trigger_id = trigger.get('triggerId', '')

    # タグを作成 GA4
    tag_body = {
        'name': 'GA4', 'type': 'sgtmgaaw', # type:sgtmgaaw Google アナリティクス: GA4
        'parameter': [
            {'type': 'boolean', 'key': 'redactVisitorIp', 'value': 'true'},
            {'type': 'template', 'key': 'epToIncludeDropdown', 'value': 'all'},
            {'type': 'template', 'key': 'upToIncludeDropdown', 'value': 'all'},
            {'type': 'template', 'key': 'measurementId', 'value': '{{GA Measurement ID}}'}
        ],
        'firingTriggerId': [trigger_id], 'tagFiringOption': 'oncePerEvent',
    }
    server_container.create_tag(tag_body)

    # クライアントを作成 GA4 Client
    client_body = {
        'name': 'GA4 Client', 'type': 'gaaw_client', # type:gaaw_client Google アナリティクス: GA4(ウェブ)
        'parameter': [
            {'type': 'template', 'key': 'cookieDomain', 'value': 'auto'},
            {'type': 'boolean', 'key': 'activateResponseCompression', 'value': 'true'},
            {'type': 'template', 'key': 'cookieMaxAgeInSec', 'value': '63072000'},
            {'type': 'boolean', 'key': 'activateGeoResolution', 'value': 'false'},
            {'type': 'boolean', 'key': 'activateGtagSupport', 'value': 'true'},
            {'type': 'boolean', 'key': 'activateDependencyServing', 'value': 'true'},
            {'type': 'boolean', 'key': 'activateDefaultPaths', 'value': 'true'},
            {'type': 'template', 'key': 'cookiePath', 'value': '/'},
            {'type': 'boolean', 'key': 'migrateFromJsClientId', 'value': 'false'},
            {'type': 'template', 'key': 'cookieManagement', 'value': 'server'},
            {'type': 'template', 'key': 'cookieName', 'value': 'FPID'},
            {'type': 'list', 'key': 'gtagMeasurementIds', 'list': [{'type': 'map', 'map': [{'type': 'template', 'key': 'measurementId', 'value': '{{GA Measurement ID}}'}]}]}
        ]
    }

    clients = server_container.get_clients()
    is_client_exist = False
    for client in clients:
        if client['name'] == 'GA4' and client['type'] == 'gaaw_client':
            is_client_exist = True
            # 初期クライアントのGA4があれば更新する。
            server_container.update_client(client['clientId'], client_body)
            break
    if(not is_client_exist):
        # クライアントを作成 GA4 Client
        server_container.create_client(client_body)


    # クライアントを作成 Web Container Client
    client_body = {
        'name': 'Web Container Client', 'type': 'gtm_client', # type:gtm_client Google タグ マネージャー: ウェブコンテナ
        'parameter': [
            {'type': 'boolean', 'key': 'activateResponseCompression', 'value': 'true'},
            {'type': 'boolean', 'key': 'activateGeoResolution', 'value': 'false'},
            {'type': 'boolean', 'key': 'activateDependencyServing', 'value': 'true'},
            {'type': 'list', 'key': 'allowedContainerIds', 
            'list': [{'type': 'map', 'map': [{'type': 'template', 'key': 'containerId', 'value': web_container["publicId"]}]}]}
        ]
    }
    server_container.create_client(client_body)

    # バージョン作成と公開
    version = server_container.create_version()
    server_container.publish(version['path'])
    print('Server container created and published successfully.')

コードを実行してコンテナを構築

上記のPythonコードをcraete_gtm_container.py(任意名可)と保存し、次のコマンドを実行すると、GTMのコンテナが構築されます。

# ウェブコンテナ作成コマンド
python create_gtm_container.py web

# サーバーコンテナ作成コマンド
python create_gtm_container.py server

サーバーコンテナの構築時に、ウェブコンテナのコンテナID(GTM-XXXXXXXX)を必要としているので、先にウェブコンテナから作成してください。
GTM-APIの利用に際し、100秒あたり25件のリクエストの制限があり、APIを利用するたびに4秒スリープさせているので、終了までに40~50秒程度時間がかかります。
Tag Manager API の制限と割り当て

コンテナ構築後

ウェブコンテナとサーバーコンテナの構築後、GTMのコンソール上で指定コンテナ内に入ると、ページ上部にGTM-XXXXXXXXと記載されたコンテナIDが表示されます。

このコンテナIDの箇所をクリックすると、ウェブコンテナならhtmlに埋め込むGTMタグ内容が表示され、サーバーコンテナならタグ設定サーバーのプロビジョニング方法を選択する画面が表示されます。

ウェブコンテナのコンテナIDをクリック後の表示内容

ウェブコンテナのコンテナIDをクリック後に表示される画面で、コードを2つ取得します。1つは<head>タグ内の上部に貼り付けるコードで、もう1つは<body>タグ直後に貼り付けるコードです。
この2つのコードをそのままhtmlに貼り付けて展開しても稼働はしますが、2つのコード内のwww.googletagmanager.comと書かれた箇所をサーバーサイドコンテナのURLのドメインに書き換えると、gtm.jsとga4のgtag/jsの読み込みもサーバーコンテナ側から読み込まれます。

サーバーコンテナのコンテナIDをクリック後の表示内容



サーバーコンテナのコンテナIDをクリック後、サーバーコンテナ側でタグ設定サーバーを自動か手動で設定する画面が表示されます。
手動でプロビジョニングしている場合は、「タグ設定サーバーを手動でプロビジョニングする」を選択すると、コンテナ設定の文字列が表示されます。この文字列をCloudRun等の実行環境に割り当てます。

サーバーコンテナの実行環境となるCloudRun等の構築についてはここでは記載しませんが、ここまでが、GCPのサービスアカウントでGTM-APIを使ってウェブコンテナとサーバーコンテナを構築する内容になります。

構成や仕様の変更

2024年8月にこの記事のPythonコードを書いてましたが、2024年9月下旬に実行すると、サーバーコンテナ側のクライアントがなぜかもう1つ増えており、調べると、サーバーコンテナを構築した時点でGAという名前で、Google アナリティクス: GA4(ウェブ)のタイプのクライアントが構築されるようになっていました。これはGTM-API経由で構築したサーバーコンテナだけでなく、GTMのコンソール上で手動で構築した場合でも同クライアントがサーバーコンテナ内に構築されていました。
上述のPythonコードではnameがGAでtypeがgaaw_clientのクライアントがあれば新規に作成せず、内容を更新するよう対応しました。
また、ここ数年でもGTM関連では下記等の変更がありました。

  • 2023年7月でユニバーサルアナリティクス(UA)は終了し、UAではなくGoogle Analytics 4(GA4)に変更
  • サーバーコンテナのURLの設定が、UAではtransport_urlの項目名を使っていたが、GA4ではserver_container_urlに変更
  • [Googleアナリティクス:GA4 設定]のタグがGoogleタグに変更

今後もこういった内容等の変更が生じる可能性もあり、そのうちこの記事のコードにおいても、実行するともう古い記述のためエラーになったりするケースがあるかもしれませんが、参考にしていただければ幸いです。



■ワントゥーテンでは中途採用募集中です!

1→10(ワントゥーテン)のカルチャーや、作品のクリエイティブに共感し、自身のより高い成長を求めている方からのご応募をお待ちしています!