AWS SDK for Python を使用してAmazon Connectのルーティングプロファイルを別のインスタンスにコピーしてみた

#Amazon Connect
#AWS
繁松昂大
2024.08.31
こんにちは。

繁松です。
 はじめにAWS SDK for Python (Boto3) を使用して、同一アカウント、同一リージョン内のAmazon Connectにルーティングプロファイルをコピーしてみました。
ルーティングプロファイルを同様の設定で作成したい場合、インポート/エクスポートの機能がないため、手作業で一つ一つ作成する必要があります。AWS SDK for Python (Boto3) を使用することで、同様の設定のルーティングプロファイルを簡単に作成できました。

ルーティングプロファイルをコピーしたいケースとしては、開発環境で作成したルーティングプロファイルを本番環境でも同じ設定で利用したい場合や、Amazon Connectのユーザー認証方式を変更したい場合等、新たに作成したインスタンスでも既存のインスタンスと同様の設定が使いたいといったケースが考えられます。
本ブログでコピーされる設定値は、「ルーティングプロファイル名」、「説明」、「デフォルトのアウトバウンドキュー」、「キューの設定」、「チャネルの設定」です。
 やってみたコピー元のインスタンスのIDと、コピー先のインスタンスのIDを確認しスクリプト内で指定します。
コピー元のインスタンスからルーティングプロファイルのリストを取得し、設定されているアウトバウンドキューの設定や説明等の詳細情報を取得します。

コピー元とコピー先に同じ名前のルーティングプロファイルがある場合は、設定値をアップデートするようにしています。
 前提条件コピー元とコピー先のインスタンスに同じ名前で作成されたキューが必要です。キューのコピーは以下のブログを参考に実施できます。

https://dev.classmethod.jp/articles/amazon-connect-queue-copy-sdk/
import boto3

def copy_routing_profile(source_instance_id, target_instance_id, region_name='ap-northeast-1'):
    connect = boto3.client('connect', region_name=region_name)

    # コピー元のインスタンスからルーティングプロファイルのリストを取得
    list_routing_profiles = connect.list_routing_profiles(
        InstanceId=source_instance_id
    )['RoutingProfileSummaryList']

    for routing_profiles in list_routing_profiles:
        routing_profiles_id = routing_profiles['Id']

        # 詳細なルーティングプロファイル情報を取得
        routing_profiles_details = connect.describe_routing_profile(
            InstanceId=source_instance_id,
            RoutingProfileId=routing_profiles_id
        )['RoutingProfile']

        # コピー元のルーティングプロファイルに設定されたアウトバウンドキュー名を取得
        outbound_queue_name = connect.describe_queue(
            InstanceId=source_instance_id,
            QueueId=routing_profiles_details['DefaultOutboundQueueId']
        )['Queue']['Name']

        # コピー先のインスタンスで同じ名前のキューのIDを検索
        outbound_queue_id = connect.search_queues(
            InstanceId=target_instance_id,
            SearchCriteria={
                'StringCondition': {
                    'FieldName': 'Name',
                    'Value': outbound_queue_name,
                    'ComparisonType': 'EXACT'
                }
            }
        )['Queues'][0]['QueueId']

        # チャンネルの設定情報を整形
        media_concurrencies_list = routing_profiles_details['MediaConcurrencies']

        media_concurrencies = []
        for media in media_concurrencies_list:
            media_concurrencies.append({
                'Channel': media['Channel'],
                'Concurrency': media['Concurrency'],
                'CrossChannelBehavior': {
                    'BehaviorType': media['CrossChannelBehavior']['BehaviorType']
                },
            })

        # ルーティングプロファイルに紐づくキュー情報を取得
        routing_profiles_queues = connect.list_routing_profile_queues(
            InstanceId=source_instance_id,
            RoutingProfileId=routing_profiles_id
        )['RoutingProfileQueueConfigSummaryList']

        # キュー情報を整形
        queue_configs = []
        for queue in routing_profiles_queues:
            # コピー先のインスタンスで同じ名前のキューのIDを検索
            queue_id = connect.search_queues(
                InstanceId=target_instance_id,
                SearchCriteria={
                    'StringCondition': {
                        'FieldName': 'Name',
                        'Value': queue['QueueName'],
                        'ComparisonType': 'EXACT'
                    }
                }
            )['Queues'][0]['QueueId']

            queue_configs.append({
                'QueueReference': {
                    'QueueId': queue_id,
                    'Channel': queue['Channel']
                },
                'Priority': queue['Priority'],
                'Delay': queue['Delay']
            })

        existing_routing_profiles = connect.list_routing_profiles(
            InstanceId=target_instance_id
        )['RoutingProfileSummaryList']

        for existing_routing_profile in existing_routing_profiles:
            if existing_routing_profile['Name'] == routing_profiles['Name']:
                update_params = {
                    'InstanceId': target_instance_id,
                    'RoutingProfileId': existing_routing_profile['Id']
                }
                # ルーティングプロファイルの説明を更新
                update_params['Description'] = routing_profiles_details['Description']
                connect.update_routing_profile_name(**update_params)
                del update_params['Description']

                # アウトバウンドキューを更新
                update_params['DefaultOutboundQueueId'] = outbound_queue_id
                connect.update_routing_profile_default_outbound_queue(**update_params)
                del update_params['DefaultOutboundQueueId']

                # チャンネル設定を更新
                update_params['MediaConcurrencies'] = media_concurrencies
                connect.update_routing_profile_concurrency(**update_params)
                del update_params['MediaConcurrencies']

                # コピー先のインスタンスのルーティングプロファイルに紐づくキュー設定を解除
                routing_profiles_queues_target = connect.list_routing_profile_queues(**update_params)['RoutingProfileQueueConfigSummaryList']

                disassociate_routing_profile = []
                for routing_profiles_queue_target in routing_profiles_queues_target:
                    disassociate_routing_profile.append(
                        {
                            'QueueId': routing_profiles_queue_target['QueueId'],
                            'Channel': routing_profiles_queue_target['Channel']
                        }
                    )
                    if disassociate_routing_profile:
                        update_params['QueueReferences'] = disassociate_routing_profile
                        connect.disassociate_routing_profile_queues(**update_params)
                        del update_params['QueueReferences']

                # キュー設定を更新
                if queue_configs:
                # コピー先インスタンスのルーティングプロファイルにキュー設定を紐づけ
                    update_params['QueueConfigs'] = queue_configs
                    connect.associate_routing_profile_queues(**update_params)
                break 

        else:
            # ルーティングプロファイルを作成
            create_params = {
                'InstanceId': target_instance_id,
                'Name': routing_profiles_details['Name'],
                'Description': routing_profiles_details['Description'],
                'DefaultOutboundQueueId': outbound_queue_id,
                'MediaConcurrencies': media_concurrencies,
                'Tags': routing_profiles_details.get('Tags', {})
            }
            if queue_configs:
                create_params['QueueConfigs'] = queue_configs

            connect.create_routing_profile(**create_params)
            print(f"Copied Routing Profile: {routing_profiles_details['Name']}")

# コピー元とコピー先のインスタンスIDを指定
source_instance_id = 'source_instance_idを指定'
target_instance_id = 'target_instance_idを指定'

copy_routing_profile(source_instance_id, target_instance_id)
注意点

同じ名前のルーティングプロファイルがあった場合、設定値のアップデートでtagの項目は含まれません。
 実行前コピー元インスタンス

コピー先インスタンス

 実行後コピー先インスタンス

test1のルーティングプロファイルが追加されました。

 さいごにAWS SDK for Python (Boto3) を使ったルーティングプロファイルのコピーをやってみました。

コンソールを確認しながら実施するよりも簡単に同様のルーティングプロファイルの作成が実施できると思います。

コピー元とコピー先の指定を間違えないように気を付けてください。
次回はセキュリティプロファイルのコピーについて書きます。