【AWS】AWS CDKでマルチエージェントを構築する

今回はAWS CDKを使用して、複数のエージェントとMCPサーバーを一括デプロイした方法についてまとめます。

AWS CDKを使用した Amazon Bedrock AgentCore Gateway × Lambda によるMCPサーバー構築については以下の記事を参照してください。

この記事で学べること

いずれもAWS CDKを使った構築方法です。

• エージェントのツールとして使用するLambdaをVPC内へデプロイする方法
• Bedrock AgentCore Gatewayのターゲットに作成したLambdaを割り当ててデプロイする方法
• Bedrock AgentCore Runtimeへ複数のエージェントをデプロイする方法
• awslabsのAWS Cost Explorer MCPサーバーをローカルMCPとしてBedrock AgentCore Runtimeへデプロイする方法

マルチエージェントを構築しようとした理由

• 社内で「今の時代、AIエージェントは自前で構築できる」ということを伝えたい
• Amazon Bedrock AgentCoreを使用してエージェントを構築したという前例を作りたい
• 技術検証のため
• 構築したエージェントを拡張していくことで、社内の業務効率化につなげたい
• 社内のAI活用に対するハードルを下げたい

エージェントとMCPサーバーの構成

以下のように設計しました。

User（社員）
↓
社内業務用システム（Ruby on Rails）
↓
チェアマンエージェント
|
|- 自社プロダクトマニュアル検索エージェント
|   |- Tool: Bedrock KnowledgeBasesへのRAG検索ツール
|
|- 自社プロダクトAPI実行エージェント
|   |- RemoteMCP: APIを実行するMCPサーバー
|
|- 社内業務用システムAPI実行エージェント
|   |- RemoteMCP: APIを実行するMCPサーバー
|
|- Amazon Connect通話履歴分析エージェント
|   |- Tool: 通話履歴を検索するツール, キューを検索するツールなど
|
|- AWSコスト分析エージェント
    |- LocalMCP: AWS Cost Explorer MCPサーバー

社内業務用システムのAIチャット機能はClaude Codeを用いて別途実装済みです。

チェアマンエージェントから各エージェントへの接続は Tool as Agent として実装しています。（今回は技術検証が主目的であったため、A2Aを採用する必要はないと判断しました。）

今回は考慮しない部分

• 社内業務用システムのユーザーの権限によってエージェントにリクエストできる範囲の制限（Identityの部分）
• 各エージェントの細かいツール実装

各エージェントとMCPサーバーの技術スタック

エージェント

全てのエージェントをPython・Strands Agents・Bedrock AgentCore Runtimeで構築しています。

エージェント名	技術スタック	備考
チェアマンエージェント	Python・Strands Agents・Bedrock AgentCore Runtime	各エージェントへ接続するツールを定義
自社プロダクト マニュアル検索エージェント	Python・Strands Agents・Bedrock AgentCore Runtime	ナレッジベース検索するツールを定義
自社プロダクト API実行エージェント	Python・Strands Agents・Bedrock AgentCore Runtime	リモートMCPサーバーへ接続
社内業務用システム API実行エージェント	Python・Strands Agents・Bedrock AgentCore Runtime	リモートMCPサーバーへ接続
Amazon Connect 通話履歴分析エージェント	Python・Strands Agents・Bedrock AgentCore Runtime	Amazon Connect APIを実行するツールを定義
AWSコスト分析 エージェント	Python・Strands Agents・Bedrock AgentCore Runtime	agent.pyと同じ階層にローカルMCPサーバーを配置

MCPサーバー

MCPサーバー名	技術スタック	備考
自社プロダクトAPI実行MCPサーバー	Python・FastMCP・Bedrock AgentCore Runtime	FastMCPを使用したMCPサーバーの構築方法はこちら
社内業務用システムAPI実行MCPサーバー	Lambda・Bedrock AgentCore Gateway	LambdaとBedrock AgentCore GatewayでのMCPサーバーの構築方法はこちら
AWS Cost Explorer MCPサーバー	こちらから必要な部分のみ取得	agent.pyと同じ階層にローカルMCPサーバーを配置

↓ AWS Cost Explorer MCPはこちら ↓

なぜAWS CDKで構築したか

今回、社内業務用システムAPI実行MCPサーバーの構築を Lambda・Bedrock AgentCore Gateway で行うように設計したのですが、Lambdaを21個作る必要がありました。

AWSマネジメントコンソール上で21個のLambdaを1つずつ作成していくのは現実的ではないため、CDKでfor文を使ってまとめてデプロイするという結論に至りました。

エージェントについても同様で、Bedrock AgentCore Runtimeへのデプロイが7つ必要だったことから、こちらもCDKで一括デプロイする方針としました。

構築手順

0. 前提条件

• CDKはPythonで記述しています
• LambdaもPythonで記述しています
• エージェントもPythonで記述しています
• Lambdaで使用するLambdaレイヤーは事前にアップロード済みとします

1. 作業ディレクトリを用意して、CDKプロジェクトを初期化する

# 作業ディレクトリの用意
$ mkdir playground && cd $_
$ mkdir multi_agent_aws_cdk

# CDKプロジェクトの初期化
$ cdk init --language python
$ source .venv/bin/activate
$ python -m pip install -r requirements.txt

2. app.pyを編集する

app.pyを開き、以下をコピーします。

#!/usr/bin/env python3
import os

import aws_cdk as cdk

from multi_agent_aws_cdk.multi_agent_aws_cdk_stack import MultiAgentAwsCDKStack


app = cdk.App()
MultiAgentAwsCDKStack(
    app, "MultiAgentAwsCDKStack",
    env=cdk.Environment(
        account=os.getenv('CDK_DEFAULT_ACCOUNT'),
        region='ap-northeast-1'
    ),
    description="マルチエージェント構築スタック"
)

app.synth()

3. multi_agent_aws_cdk_stackを記述する

ボリュームが多いため、細かく区切って説明します。

3-1. 必要なライブラリをインポートする

from aws_cdk import (
    Stack,
    Duration,
    CfnOutput,
    RemovalPolicy,
    aws_lambda as _lambda,
    aws_ec2 as ec2,
    aws_iam as iam,
    aws_ecr_assets as ecr_assets,
    aws_bedrockagentcore as bedrockagentcore,
)
from constructs import Construct
import os

3-2. MultiAgentAwsCDKStackクラスを用意する

CDKプロジェクトを初期化した時点で以下のコードは生成されています。

class MultiAgentAwsCDKStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

3-3. 既存のVPC、サブネット、セキュリティグループを参照する

今回はVPC内にLambdaを作成するため、VPC、サブネット、セキュリティグループを参照する必要があります。（この要件がない場合は不要です。）

        # 既存のVPCを参照
        vpc = ec2.Vpc.from_lookup(
            self,
            "SampleVpc",
            vpc_id="vpc-xxxxxxxxxxxxxxx"
        )

        # サブネットの選択（private-A, private-C）
        subnet_private_a = ec2.Subnet.from_subnet_attributes(
            self,
            "SubnetPrivateA",
            subnet_id="subnet-xxxxxxxxxxxxxxxxx",
            availability_zone="ap-northeast-1a",
        )
        subnet_private_c = ec2.Subnet.from_subnet_attributes(
            self,
            "SubnetPrivateC",
            subnet_id="subnet-xxxxxxxxxxxxxxxxx",
            availability_zone="ap-northeast-1c",
        )

        subnet_selection = ec2.SubnetSelection(
            subnets=[subnet_private_a, subnet_private_c]
        )

        # 既存のセキュリティグループを参照
        security_group = ec2.SecurityGroup.from_security_group_id(
            self,
            "SampleDevSecurityGroup",
            security_group_id="sg-xxxxxxxxxxxxxxxxx"
        )

3-4. Lambda関数に設定する環境変数を用意する

        # 共通の環境変数
        common_environment = {
            "AWS_REGION_NAME": "ap-northeast-1",
            "RDS_SECRET": "sample-db-sm-01"
        }

3-5. Lambdaレイヤーを参照する

前述のとおり、既に登録されているLambdaレイヤーを参照します。

        # Lambdaレイヤーの参照
        sqlalchemy_layer_arn = f"arn:aws:lambda:ap-northeast-1:{self.account}:layer:sqlalchemy:1"
        sqlalchemy_layer = _lambda.LayerVersion.from_layer_version_arn(
            self,
            "SqlAlchemyLayer",
            layer_version_arn=sqlalchemy_layer_arn
        )

3-6. Lambda関数を作成する

n個のLambda関数を作成します。（ここでは3つのテーブルを例に説明します。）

注意: 以下のコードでは gateway、credential_provider_configurations、TABLE_PARAMS、COMMON_DATETIME_PARAMS を参照しています。これらはMCP GatewayおよびGateway Roleの定義と合わせて、このループの前に定義しておく必要があります。詳細は冒頭のGateway構築記事を参照してください。

        lambda_tables = ['users', 'makers', 'devices']

        # ループ開始（LambdaとAgentCore Gatewayターゲットを生成）
        for table in lambda_tables:
            fn_id = ''.join(word.capitalize() for word in table.split('_'))
            fn = _lambda.Function(
                self,
                f"SampleDev{fn_id}",
                function_name=f"sample-dev-{table.replace('_', '-')}",
                runtime=_lambda.Runtime.PYTHON_3_13,
                handler="lambda_function.lambda_handler",
                code=_lambda.Code.from_asset(
                    f"sample_agent_cdk/lib/sample_api_agent/lambda/{table}"
                ),
                vpc=vpc,
                vpc_subnets=subnet_selection,
                security_groups=[security_group],
                layers=[sqlalchemy_layer],
                environment=common_environment,
                timeout=Duration.seconds(30),
                memory_size=256,
                allow_public_subnet=True
            )
            fn.add_to_role_policy(iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                actions=["secretsmanager:GetSecretValue"],
                resources=["*"]
            ))

            schema_props = {
                "mode": bedrockagentcore.CfnGatewayTarget.SchemaDefinitionProperty(
                    type="string",
                    description="ここにツールのパラメータの詳細を記載"
                ),
                "id": bedrockagentcore.CfnGatewayTarget.SchemaDefinitionProperty(
                    type="string",
                    description="ここにツールのパラメータの詳細を記載"
                ),
            }
            # テーブル固有パラメータを生成
            for param_name, param_desc in TABLE_PARAMS.get(table, {}).items():
                schema_props[param_name] = bedrockagentcore.CfnGatewayTarget.SchemaDefinitionProperty(
                    type="string",
                    description=param_desc
                )
            # 共通パラメータを生成
            for param_name, param_desc in COMMON_DATETIME_PARAMS.items():
                schema_props[param_name] = bedrockagentcore.CfnGatewayTarget.SchemaDefinitionProperty(
                    type="string",
                    description=param_desc
                )

            # GatewayロールがLambdaを呼び出せるようにリソースベースポリシーを追加
            lambda_permission = _lambda.CfnPermission(
                self,
                f"SampleDev{fn_id}GatewayPermission",
                action="lambda:InvokeFunction",
                function_name=fn.function_name,
                principal="bedrock-agentcore.amazonaws.com",
                source_account=self.account,
            )

            # 各Lambda関数をGatewayのターゲットとして登録
            gateway_target = bedrockagentcore.CfnGatewayTarget(
                self,
                f"SampleDev{fn_id}Target",
                name=f"{table.replace('_', '-')}",
                gateway_identifier=gateway.attr_gateway_identifier,
                target_configuration=bedrockagentcore.CfnGatewayTarget.TargetConfigurationProperty(
                    mcp=bedrockagentcore.CfnGatewayTarget.McpTargetConfigurationProperty(
                        lambda_=bedrockagentcore.CfnGatewayTarget.McpLambdaTargetConfigurationProperty(
                            lambda_arn=fn.function_arn,
                            tool_schema=bedrockagentcore.CfnGatewayTarget.ToolSchemaProperty(
                                inline_payload=[
                                    bedrockagentcore.CfnGatewayTarget.ToolDefinitionProperty(
                                        name=f"sample_dev_{table}",
                                        description=f"ここにツールの詳細を記載する。",
                                        input_schema=bedrockagentcore.CfnGatewayTarget.SchemaDefinitionProperty(
                                            type="object",
                                            properties=schema_props
                                        )
                                    )
                                ]
                            )
                        )
                    )
                ),
                credential_provider_configurations=credential_provider_configurations,
                description=f"sample-dev-{table.replace('_', '-')} Lambda function target"
            )
            gateway_target.add_dependency(lambda_permission)

3-7. Dockerイメージのビルド & ECRへのプッシュ

エージェントごとにDockerイメージをビルドしてECRへプッシュします。DockerImageAssetを使うことで、cdk deploy実行時にビルドとプッシュが自動的に行われます。

        lib_dir = os.path.join(os.path.dirname(__file__), "lib")

        agent_image_dirs = {
            "sample-chairman-agent": os.path.join(lib_dir, "chairman_agent", "chairman_agent"),
            "sample-xxxxxxx-manual-search-agent": os.path.join(lib_dir, "xxxxxxx_manual_search_agent", "xxxxxxx_manual_search_agent"),
            "sample-xxxxxxx-api-agent": os.path.join(lib_dir, "xxxxxxx_api_agent", "xxxxxxx_api_agent"),
            "sample-yyyyy-api-agent": os.path.join(lib_dir, "yyyyy_api_agent", "yyyyy_api_agent"),
            "sample-amazon-connect-agent": os.path.join(lib_dir, "amazon_connect_agent", "amazon_connect_agent"),
            "sample-aws-cost-analysis-agent": os.path.join(lib_dir, "aws_cost_analysis_agent"),
            "sample-xxxxxxx-mcp": os.path.join(lib_dir, "xxxxxxx_api_agent", "xxxxxxx-mcp")
        }

        # construct_id を固定したいエージェントのマッピング（CDKリソースの置き換えを防ぐ）
        CONSTRUCT_ID_OVERRIDES = {
            "sample-chairman-agent": "ChairmanAgent",
        }

        image_assets = {}
        for agent_name, docker_dir in agent_image_dirs.items():
            asset_id = CONSTRUCT_ID_OVERRIDES.get(
                agent_name,
                ''.join(word.capitalize() for word in agent_name.split('-'))
            )
            # aws_cost_analysis_agent はビルドコンテキストが親ディレクトリのため file を明示
            docker_file = "aws_cost_analysis_agent/Dockerfile" if agent_name == "sample-aws-cost-analysis-agent" else None
            image_asset = ecr_assets.DockerImageAsset(
                self,
                f"{asset_id}Image",
                directory=docker_dir,
                **( {"file": docker_file} if docker_file else {} ),
            )
            image_assets[agent_name] = image_asset
            CfnOutput(
                self,
                f"{asset_id}ImageUri",
                value=image_asset.image_uri,
                description=f"ECR image URI for {agent_name}",
            )

3-8. Bedrock AgentCore Memoryを作成する

作成するMemoryは1つですが、将来複数のMemoryを管理しやすいようにループ構成にしています。

        MEMORY_AGENT_NAMES = ["sample-chairman-agent"]

        agent_memories: dict = {}
        for agent_name in MEMORY_AGENT_NAMES:
            asset_id = CONSTRUCT_ID_OVERRIDES.get(
                agent_name,
                ''.join(word.capitalize() for word in agent_name.split('-'))
            )
            memory = bedrockagentcore.CfnMemory(
                self,
                f"{asset_id}Memory",
                name=f"{agent_name.replace('-', '_')}_memory",
                event_expiry_duration=90,
            )
            memory.apply_removal_policy(RemovalPolicy.RETAIN)
            agent_memories[agent_name] = memory
            CfnOutput(
                self,
                f"{asset_id}MemoryId",
                value=memory.attr_memory_id,
                description=f"Memory ID for {agent_name}",
            )

3-9. エージェントをBedrock AgentCore Runtimeへデプロイする

        runtimes: dict = {}
        runtime_endpoints: dict = {}

        for agent_name in agent_image_dirs.keys():
            asset_id = CONSTRUCT_ID_OVERRIDES.get(
                agent_name,
                ''.join(word.capitalize() for word in agent_name.split('-'))
            )

            # AgentCore Runtime 実行ロール
            runtime_role = iam.Role(
                self,
                f"{asset_id}RuntimeRole",
                assumed_by=iam.ServicePrincipal("bedrock-agentcore.amazonaws.com"),
                managed_policies=[
                    iam.ManagedPolicy.from_aws_managed_policy_name("BedrockAgentCoreFullAccess"),
                ],
                inline_policies={
                    "EcrPullPolicy": iam.PolicyDocument(
                        statements=[
                            iam.PolicyStatement(
                                effect=iam.Effect.ALLOW,
                                actions=[
                                    "ecr:GetDownloadUrlForLayer",
                                    "ecr:BatchGetImage",
                                    "ecr:BatchCheckLayerAvailability",
                                    "ecr:GetAuthorizationToken",
                                ],
                                resources=["*"],
                            )
                        ]
                    )
                },
            )

            # xxxxxxx_manual_search_agent には Bedrock KnowledgeBase の Retrieve 権限を追加
            if agent_name == "sample-xxxxxxx-manual-search-agent":
                runtime_role.add_to_policy(iam.PolicyStatement(
                    effect=iam.Effect.ALLOW,
                    actions=["bedrock:Retrieve"],
                    resources=[f"arn:aws:bedrock:us-east-1:{self.account}:knowledge-base/ZZZZZZZZZZ"],
                ))

            # aws_cost_analysis_agent には Cost Explorer 権限を追加
            if agent_name == "sample-aws-cost-analysis-agent":
                runtime_role.add_to_policy(iam.PolicyStatement(
                    effect=iam.Effect.ALLOW,
                    actions=[
                        "ce:GetCostAndUsage",
                        "ce:GetCostForecast",
                        "ce:GetDimensionValues",
                        "ce:GetReservationCoverage",
                        "ce:GetSavingsPlansCoverage",
                        "ce:GetSavingsPlansUtilization",
                        "ce:ListCostAllocationTags",
                    ],
                    resources=["*"],  # Cost Explorer はリソースレベル制限不可
                ))

            # amazon_connect_agent には Amazon Connect 権限を追加
            if agent_name == "sample-amazon-connect-agent":
                runtime_role.add_to_policy(iam.PolicyStatement(
                    effect=iam.Effect.ALLOW,
                    actions=[
                        "connect:SearchContacts",
                        "connect:DescribeContact",
                        "connect:ListQueues",
                    ],
                    resources=[
                        f"arn:aws:connect:ap-northeast-1:{self.account}:instance/abcdef12-1234-abcd-1234-abcdef123456",
                        f"arn:aws:connect:ap-northeast-1:{self.account}:instance/abcdef12-1234-abcd-1234-abcdef123456/*",
                    ],
                ))

            # OTEL ログ書き込み用 CloudWatch Logs 権限（全エージェント共通）
            runtime_role.add_to_policy(iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                actions=[
                    "logs:CreateLogGroup",
                    "logs:CreateLogStream",
                    "logs:PutLogEvents",
                    "logs:DescribeLogStreams",
                    "logs:DescribeLogGroups",
                ],
                resources=["*"],
            ))

            # プロトコル設定: MCPサーバーは MCP、その他は HTTP (None)
            if agent_name == "sample-xxxxxxx-mcp":
                protocol_configuration = "MCP"
            else:
                protocol_configuration = None   # チェアマンも子エージェントも HTTP

            # MEMORY_ID を環境変数として渡す（chairman_agent のみ）
            env_vars = None
            if agent_name in agent_memories:
                env_vars = {"MEMORY_ID": agent_memories[agent_name].attr_memory_id}

            # AgentCore Runtime 定義
            runtime = bedrockagentcore.CfnRuntime(
                self,
                f"{asset_id}Runtime",
                agent_runtime_name=agent_name.replace('-', '_'),
                agent_runtime_artifact=bedrockagentcore.CfnRuntime.AgentRuntimeArtifactProperty(
                    container_configuration=bedrockagentcore.CfnRuntime.ContainerConfigurationProperty(
                        container_uri=image_assets[agent_name].image_uri,
                    )
                ),
                network_configuration=bedrockagentcore.CfnRuntime.NetworkConfigurationProperty(
                    network_mode="PUBLIC",
                ),
                role_arn=runtime_role.role_arn,
                protocol_configuration=protocol_configuration,
                environment_variables=env_vars,
            )

            runtimes[agent_name] = runtime

            # AgentCore Runtime エンドポイント定義
            runtime_endpoint = bedrockagentcore.CfnRuntimeEndpoint(
                self,
                f"{asset_id}RuntimeEndpoint",
                agent_runtime_id=runtime.attr_agent_runtime_id,
                name=f"{agent_name.replace('-', '_')}_endpoint",
            )

            runtime_endpoints[agent_name] = runtime_endpoint

            CfnOutput(
                self,
                f"{asset_id}RuntimeArn",
                value=runtime.attr_agent_runtime_arn,
                description=f"ARN of the {agent_name} AgentCore Runtime",
            )
            CfnOutput(
                self,
                f"{asset_id}RuntimeEndpointUrl",
                value=runtime_endpoint.attr_live_version,
                description=f"Live version endpoint URL for {agent_name}",
            )

3-10. ループ後にエージェントへ環境変数を注入する

チェアマンエージェントが子エージェントのARNを参照できるよう、ループ終了後にadd_property_overrideで注入します。ループ内では子エージェントのRuntimeがまだ作成されていないため、ループ後にまとめて処理する必要があります。

        chairman_runtime = runtimes["sample-chairman-agent"]
        child_arn_map = {
            "sample-xxxxxxx-manual-search-agent": "XXXXXXX_MANUAL_SEARCH_AGENT_ARN",
            "sample-xxxxxxx-api-agent":           "XXXXXXX_API_AGENT_ARN",
            "sample-yyyyy-api-agent":             "YYYYY_API_AGENT_ARN",
            "sample-amazon-connect-agent":        "AMAZON_CONNECT_AGENT_ARN",
            "sample-aws-cost-analysis-agent":     "AWS_COST_ANALYSIS_AGENT_ARN",
        }
        for child_name, env_key in child_arn_map.items():
            chairman_runtime.add_property_override(
                f"EnvironmentVariables.{env_key}",
                runtimes[child_name].attr_agent_runtime_arn,
            )

        # xxxxxxx_api_agent に sample-xxxxxxx-mcp の ARN を注入
        runtimes["sample-xxxxxxx-api-agent"].add_property_override(
            "EnvironmentVariables.XXXXXXX_MCP_ARN",
            runtimes["sample-xxxxxxx-mcp"].attr_agent_runtime_arn,
        )

        # yyyyy_api_agent に SAMPLE MCP Gateway URL を注入
        runtimes["sample-yyyyy-api-agent"].add_property_override(
            "EnvironmentVariables.YYYYY_MCP_GATEWAY_URL",
            gateway.attr_gateway_url,
        )

4. CloudFormationテンプレートを生成する

cdk synth

5. デプロイ実行

cdk deploy --require-approval never

まとめ

今回はAWS CDKを使用して、マルチエージェント構成を一括デプロイしました。

実装を通じて正直なところ、「ここまで多くのエージェントをまとめて管理するケースはどれくらいあるだろうか」という疑問は残っています。実用的な場面としては、以下のケースが考えられます。

• 大規模なマルチエージェント構成のAI機能が必要な場合
• エージェントの管理をすべてCDKプロジェクトに集約する方針のプロジェクト

一方で、エージェントとそれに紐づくMCPサーバーの数が増えるにつれて全体像が複雑になりやすいため、管理しやすい規模に抑えることを意識することが重要だと感じました。

CDK自体は非常に便利なツールであり、今後も活用していく予定です。

以上