アマゾン ウェブ サービス ジャパン (有志)Publicationへの投稿
🤖

AWS CDK の L2 Construct を使用し AgentCore Runtime w/VPCを作成する

mazyu36 / Yuki Matsuda
に公開
AWS
AWS CDK
AgentCore
tech

Amazon Bedrock AgentCore の L2 Construct (Runtime, Code Interpreter, Browser) が AWS CDK ver 2.221.0 でリリースされました 🥳

https://github.com/aws/aws-cdk/releases/tag/v2.221.0

今回は AgentCore Runtime の L2 Construct を試してみたいと思います。
過去記事の L1 Construct での実装例をベースします。

https://zenn.dev/aws_japan/articles/2025-09-25-agentcore-runtime-l1

上記を単純に L2 Construct に置き換えるだけではなく、 VPC 対応 も合わせて試してみます。

今回は AgentCore Runtime の VPC 対応サンプルとして以下を構築します。VPC 対応を行う場合は AgentCore Runtime が使用する ENI が VPC 内で作成され、そこ経由でアクセスが行われます。

  • VPC 内リソースとして Aurora Serverless v2 を作成しアクセス
  • ECR, CloudWatch Logs など、VPC 外リソースには VPC Endpoint 経由でアクセス

※実際はマルチ AZ 構成ですが、簡略化しています。

事前準備

alpha モジュールのインストール

Bedrock AgentCore の L2 Construct は alpha モジュールのため、個別に追加が必要です。

npm i @aws-cdk/aws-bedrock-agentcore-alpha

ディレクトリ構成

cdk init の後、lib/app_vpc 配下に Agent の実装と資材を格納しています。

.
├── bin
│   └── cdk-bedrock-agentcore.ts
├── cdk.json
├── lib
│   ├── app_vpc
│   │   ├── agent.py
│   │   ├── Dockerfile
│   │   ├── pyproject.toml
│   │   ├── README.md
│   │   └── uv.lock
│   └── cdk-bedrock-agentcore-stack.ts
├── package-lock.json
├── package.json
├── README.md
└── tsconfig.json

Agent の実装

Strands Agents のサンプル Option B: Custom Agent をベースにしています。

Secrets Manager から認証情報を取得するため boto3、Aurora Serverless v2 (MySQL) に書き込むため mysql-connector-python を追加しています。

pyproject.toml
lib/app_vpc/pyproject.toml
[project]
name = "app-vpc"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
+    "boto3>=1.40.40",
    "fastapi>=0.117.1",
    "httpx>=0.28.1",
    "pydantic>=2.11.9",
    "strands-agents>=1.9.1",
    "uvicorn>=0.37.0",
+    "mysql-connector-python>=8.0.33",
]

また Agent にはサンプルから DB に書き込みを行う処理を追加しています。以下の実装例では、Agent の Invoke ごとに ID を発行し結果を DB に書き込むようにしています。

本記事の主題ではないため、処理の最適化は割愛しています。

Agent の実装例
lib/app_vpc/agent.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any, Optional
from datetime import datetime, timezone
from strands import Agent
import os
import boto3
import mysql.connector
import json as json_lib
import uuid
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Strands Agent Server", version="1.0.0")

# Initialize Strands agent
strands_agent = Agent()

DB_ENDPOINT = os.environ.get("DB_ENDPOINT")
DB_PORT = int(os.environ.get("DB_PORT", 3306))
DB_NAME = os.environ.get("DB_NAME", "agentdb")
DB_SECRET_ARN = os.environ.get("DB_SECRET_ARN")
AWS_REGION = os.environ.get("AWS_REGION", "us-west-2")

REGION = AWS_REGION

# Secrets Manager client
secretsmanager_client = boto3.client("secretsmanager", region_name=REGION)

# Global cache for database credentials and connection
_db_credentials_cache = None
_db_connection = None


class InvocationRequest(BaseModel):
    input: Dict[str, Any]


class InvocationResponse(BaseModel):
    output: Dict[str, Any]


def get_db_credentials() -> Optional[Dict[str, str]]:
    """Retrieve database credentials from Secrets Manager (cached)"""
    global _db_credentials_cache

    if _db_credentials_cache is not None:
        logger.debug("Using cached database credentials")
        return _db_credentials_cache

    if not DB_SECRET_ARN:
        logger.warning("DB_SECRET_ARN not configured")
        return None

    try:
        response = secretsmanager_client.get_secret_value(SecretId=DB_SECRET_ARN)
        secret = json_lib.loads(response.get("SecretString", "{}"))
        _db_credentials_cache = {
            "user": secret.get("username", "admin"),
            "password": secret.get("password", ""),
            "host": DB_ENDPOINT,
            "port": DB_PORT,
            "database": DB_NAME,
        }
        logger.info("Retrieved and cached database credentials from Secrets Manager")
        return _db_credentials_cache
    except Exception as e:
        logger.error(f"Failed to get DB credentials: {str(e)}")
        return None


def get_db_connection():
    """Get or create MySQL database connection (cached)"""
    global _db_connection

    if _db_connection is not None:
        try:
            # Check if connection is still valid
            if _db_connection.is_connected():
                logger.debug("Using existing cached database connection")
                return _db_connection
        except Exception as e:
            logger.warning(f"Cached connection is no longer valid: {str(e)}")
            _db_connection = None

    try:
        credentials = get_db_credentials()
        if not credentials:
            return None

        _db_connection = mysql.connector.connect(**credentials)
        logger.info("Created new database connection")
        return _db_connection
    except Exception as e:
        logger.error(f"Failed to connect to database: {str(e)}")
        return None


def save_execution_result(
    execution_id: str,
    agent_name: str,
    input_data: str,
    output_data: str,
    status: str = "completed"
) -> bool:
    """Save execution result to the database"""
    connection = None
    try:
        connection = get_db_connection()
        if not connection:
            logger.warning("Could not connect to database for saving results")
            return False

        cursor = connection.cursor()

        insert_query = """
            INSERT INTO agent_execution_results
            (execution_id, agent_name, input, output, status)
            VALUES (%s, %s, %s, %s, %s)
        """

        cursor.execute(insert_query, (
            execution_id,
            agent_name,
            input_data[:65535],  # TEXT limit
            output_data[:16777215],  # LONGTEXT limit
            status
        ))

        connection.commit()
        logger.info(f"Saved execution result for {execution_id}")
        return True

    except Exception as e:
        logger.error(f"Failed to save execution result: {str(e)}")
        return False
    finally:
        if connection:
            connection.close()

@app.post("/invocations", response_model=InvocationResponse)
async def invoke_agent(request: InvocationRequest):
    execution_id = str(uuid.uuid4())
    try:
        user_message = request.input.get("prompt", "")
        if not user_message:
            raise HTTPException(
                status_code=400,
                detail="No prompt found in input. Please provide a 'prompt' key in the input.",
            )

        result = strands_agent(user_message)
        response = {
            "message": result.message,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "model": "strands-agent",
            "execution_id": execution_id,
        }

        # Save execution result to database
        input_str = json_lib.dumps(request.input)
        output_str = json_lib.dumps(response)
        save_execution_result(
            execution_id=execution_id,
            agent_name="strands-agent",
            input_data=input_str,
            output_data=output_str,
            status="completed"
        )

        return InvocationResponse(output=response)

    except HTTPException:
        # Log failed execution to database
        save_execution_result(
            execution_id=execution_id,
            agent_name="strands-agent",
            input_data=json_lib.dumps(request.input),
            output_data=json_lib.dumps({"error": "HTTP Exception"}),
            status="failed"
        )
        raise
    except Exception as e:
        # Log error execution to database
        save_execution_result(
            execution_id=execution_id,
            agent_name="strands-agent",
            input_data=json_lib.dumps(request.input),
            output_data=json_lib.dumps({"error": str(e)}),
            status="error"
        )
        raise HTTPException(
            status_code=500, detail=f"Agent processing failed: {str(e)}"
        )


@app.get("/ping")
async def ping():
    return {"status": "healthy"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8080)

CDK 実装内容

本題の Stack （lib/cdk-bedrock-agentcore-stack.ts）の実装内容です。全量は以下になります。

Stack の実装内容
lib/cdk-bedrock-agentcore-stack.ts
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as path from 'path';
import * as ec2 from 'aws-cdk-lib/aws-ec2';
import * as rds from 'aws-cdk-lib/aws-rds';
import * as agentcore from '@aws-cdk/aws-bedrock-agentcore-alpha';

export class CdkBedrockAgentCoreStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    /**
     * VPC for Aurora and AgentCore Runtime
     */
    const vpc = new ec2.Vpc(this, 'AgentCoreVpc', {
      maxAzs: 2,
      natGateways: 1,
    });

    /**
     * Aurora Serverless v2 Database
     */
    const auroraCluster = new rds.DatabaseCluster(this, 'AuroraCluster', {
      engine: rds.DatabaseClusterEngine.auroraMysql({
        version: rds.AuroraMysqlEngineVersion.VER_3_10_0,
      }),
      writer: rds.ClusterInstance.serverlessV2('writer'),
      vpc,
      vpcSubnets: {
        subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
      },
      serverlessV2MaxCapacity: 2,
      serverlessV2MinCapacity: 0.5,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      defaultDatabaseName: 'agentdb',
      enableDataApi: true,
    });

    /**
     * AgentCore Runtime
     */
    const runtime = new agentcore.Runtime(this, 'Runtime', {
      runtimeName: 'MyAgentRuntime',
      agentRuntimeArtifact: agentcore.AgentRuntimeArtifact.fromAsset(path.join(__dirname, 'app_vpc')),
      networkConfiguration: agentcore.RuntimeNetworkConfiguration.usingVpc(this, {
        vpc,
        vpcSubnets: { subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS },
      }),
      environmentVariables: {
        DB_ENDPOINT: auroraCluster.clusterEndpoint.hostname,
        DB_PORT: auroraCluster.clusterEndpoint.port.toString(),
        DB_NAME: 'agentdb',
        DB_SECRET_ARN: auroraCluster.secret!.secretArn,
        AWS_REGION: cdk.Stack.of(this).region,
      },
    })

    // Runtime Endpoint を追加
    runtime.addEndpoint('MyEndpoint', {})

    // Bedrock の Invoke の権限付与
    runtime.role.addToPrincipalPolicy(
      new iam.PolicyStatement({
        sid: 'BedrockModelInvocation',
        effect: iam.Effect.ALLOW,
        actions: ['bedrock:InvokeModel', 'bedrock:InvokeModelWithResponseStream'],
        resources: ['arn:aws:bedrock:*::foundation-model/*', `arn:aws:bedrock:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:*`],
      }),
    );

    // Aurora の Secrets へのアクセス許可
    auroraCluster.secret?.grantRead(runtime.role);

    // Runtime -> Aurora の SG 設定
    auroraCluster.connections.allowDefaultPortFrom(runtime);

    // CloudWatch Alarm の作成
    runtime.metricSystemErrors().createAlarm(
      this, 'SystemErrosAlarm', {
      threshold: 10,
      evaluationPeriods: 1,
    });

    /**
     * VPC Endpoints for Private Subnet Runtime
     */
    // ECR Docker Endpoint
    vpc.addInterfaceEndpoint('EcrDockerEndpoint', {
      service: ec2.InterfaceVpcEndpointAwsService.ECR_DOCKER,
      subnets: {
        subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
      },
      privateDnsEnabled: true,
    });

    // ECR API Endpoint
    vpc.addInterfaceEndpoint('EcrApiEndpoint', {
      service: ec2.InterfaceVpcEndpointAwsService.ECR,
      subnets: {
        subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
      },
      privateDnsEnabled: true,
    });

    // CloudWatch Logs Endpoint
    vpc.addInterfaceEndpoint('LogsEndpoint', {
      service: ec2.InterfaceVpcEndpointAwsService.CLOUDWATCH_LOGS,
      subnets: {
        subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
      },
      privateDnsEnabled: true,
    });

    // S3 Gateway Endpoint
    vpc.addGatewayEndpoint('S3Endpoint', {
      service: ec2.GatewayVpcEndpointAwsService.S3,
      subnets: [
        {
          subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
        },
      ],
    });

    // Bedrock Runtime Endpoint
    vpc.addInterfaceEndpoint('BedrockRuntimeEndpoint', {
      service: ec2.InterfaceVpcEndpointAwsService.BEDROCK_RUNTIME,
      subnets: {
        subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
      },
      privateDnsEnabled: true,
    });

    // Secrets Manager Endpoint
    vpc.addInterfaceEndpoint('SecretsManagerEndpoint', {
      service: ec2.InterfaceVpcEndpointAwsService.SECRETS_MANAGER,
      subnets: {
        subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
      },
      privateDnsEnabled: true,
    });
  };
};

AgentCore Runtime の設定

L2 Construct を使用することで、アプリケーションパスを指定するだけでコンテナイメージのビルドとデプロイが自動化されます。VPC 設定や Runtime Endpoint の追加も簡潔に記述できます。

/**
 * AgentCore Runtime
 */
const runtime = new agentcore.Runtime(this, "Runtime", {
  runtimeName: "MyAgentRuntime",

  // アプリケーションの実装があるパスを指定
  agentRuntimeArtifact: agentcore.AgentRuntimeArtifact.fromAsset(
    path.join(__dirname, "app_vpc")
  ),

  // VPC関連の設定（セキュリティグループは自動生成）
  networkConfiguration: agentcore.RuntimeNetworkConfiguration.usingVpc(this, {
    vpc,
    vpcSubnets: { subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS }, // Private Subnet を指定
  }),

  // 環境変数
  environmentVariables: {
    DB_ENDPOINT: auroraCluster.clusterEndpoint.hostname,
    DB_PORT: auroraCluster.clusterEndpoint.port.toString(),
    DB_NAME: "agentdb",
    DB_SECRET_ARN: auroraCluster.secret!.secretArn,
    AWS_REGION: cdk.Stack.of(this).region,
  },
});

agentRuntimeArtifact は ECR のレポジトリを参照することも可能です。CI/CD を組む場合などはこちらの方が良いかと思います。

declare const repository: ecr.Repository;

const runtime = new agentcore.Runtime(this, "Runtime", {
  runtimeName: "MyAgentRuntime",
  agentRuntimeArtifact:
    agentcore.AgentRuntimeArtifact.fromEcrRepository(repository), // ECR のレポジトリを指定。
});

ネットワーク設定では、VPC 指定時にセキュリティグループが自動生成されます。手動で指定することも可能です。

declare const mySecurityGroup: ec2.SecurityGroup;

const runtime = new agentcore.Runtime(this, "Runtime", {
  runtimeName: "MyAgentRuntime",
  agentRuntimeArtifact: agentcore.AgentRuntimeArtifact.fromAsset(
    path.join(__dirname, "app_vpc")
  ),
  networkConfiguration: agentcore.RuntimeNetworkConfiguration.usingVpc(this, {
    vpc,
    vpcSubnets: { subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS },
    securityGroups: [mySecurityGroup], // SG を指定
  }),
});

Runtime Endpoint の設定

Runtime Construct の addEndpoint で作成可能です。以下の例の場合 version は未指定でこの場合は 1 になります。

// Runtime Endpoint を追加
runtime.addEndpoint("MyEndpoint", {});

以下のようにバージョンを指定することも可能です。

runtime.addEndpoint("MyEndpoint", {
  version: "2", // version を指定
});

AgentCore Runtime は DEFAULT のバージョンを保持しており、いわゆる LATEST に該当します。これはアーティファクトの更新など上がっていく形になります。

一方で Production リリースしている場合は特定のバージョンを参照させておき挙動は変えず、テスト/リリースのタイミングで切り替えるといったことが必要になってきます。その場合に Endpoint によるバージョニングが活用できます。

利用例は CDK のドキュメントにも記載されているので併せてご参照ください。

https://docs.aws.amazon.com/cdk/api/v2/docs/aws-bedrock-agentcore-alpha-readme.html#runtime-versioning

AgentCore Runtime の Executiron Role の権限、SG の設定

L2 Construct の抽象化が活かせる点です。

L2 Constuct では Execution Role が自動生成され必須の権限も付与されます。そのため追加で必要な権限を付与していきます。

まず Model を Invoke するための権限は追加が必要です。

Inference Profile の Construct を使用する方法もありますが、執筆時点では Claude Sonnet 4 以降が追加されていません。そのため今回はポリシーを直接作成し付与しています。

// Bedrock の Invoke の権限付与
runtime.role.addToPrincipalPolicy(
  new iam.PolicyStatement({
    sid: "BedrockModelInvocation",
    effect: iam.Effect.ALLOW,
    actions: ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream"],
    resources: [
      "arn:aws:bedrock:*::foundation-model/*",
      `arn:aws:bedrock:${cdk.Stack.of(this).region}:${
        cdk.Stack.of(this).account
      }:*`,
    ],
  })
);

また Aurora にアクセスするために必要な Secrets Manager へのアクセス権限を付与します。grant で簡単にできます。

// Aurora の Secrets へのアクセス許可
auroraCluster.secret?.grantRead(runtime.role);

また Aurora Serverless v2 にアクセスするための SG 設定も行います。これも connections を活用することで簡単にできます。

// Runtime -> Aurora の SG 設定
auroraCluster.connections.allowDefaultPortFrom(runtime);

Metric の使用

Metric の抽象化メソッドにより、CloudWatch Alarm やダッシュボードの作成が容易になります。

// CloudWatch Alarm の作成
runtime.metricSystemErrors().createAlarm(
  this, 'SystemErrosAlarm', {
  threshold: 10,
  evaluationPeriods: 1,
});

AgentCore Runtime におけるメトリクスの詳細は以下をご参照ください。

https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/observability-runtime-metrics.html

VPC Endpoint の設定

VPC Endpoint を設定しない場合、AgentCore Runtime から VPC 外リソースへのアクセスは NAT Gateway を経由します。

これにより NAT Gateway のコストが発生し、インターネット接続のない VPC では動作しません。本構成では VPC Endpoint 経由でアクセスするよう設定しています。

まず AgentCore Runtime において必須となる VPC Endpoint は以下の通りドキュメントに記載されています。

  • Amazon ECR Requirements:
    • Docker endpoint: com.amazonaws.region.ecr.dkr
    • ECR API endpoint: com.amazonaws.region.ecr.api
  • Amazon S3 Requirements:
    • Gateway endpoint for ECR docker layer storage: com.amazonaws.region.s3
  • CloudWatch Requirements:
    • Logs endpoint: com.amazonaws.region.logs

https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/agentcore-vpc.html#agentcore-vpc-endpoints

上記に加え LLM を Invoke する際に使用する Bedrock Runtime, Aurora にアクセスする際に使用する Secrets Manager の VPC Endpoint も追加しています。

lib/cdk-bedrock-agentcore-stack.ts
**
 * VPC Endpoints for Private Subnet Runtime
 */
// ECR Docker Endpoint
vpc.addInterfaceEndpoint('EcrDockerEndpoint', {
  service: ec2.InterfaceVpcEndpointAwsService.ECR_DOCKER,
  subnets: {
    subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
  },
  privateDnsEnabled: true,
});

// ECR API Endpoint
vpc.addInterfaceEndpoint('EcrApiEndpoint', {
  service: ec2.InterfaceVpcEndpointAwsService.ECR,
  subnets: {
    subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
  },
  privateDnsEnabled: true,
});

// CloudWatch Logs Endpoint
vpc.addInterfaceEndpoint('LogsEndpoint', {
  service: ec2.InterfaceVpcEndpointAwsService.CLOUDWATCH_LOGS,
  subnets: {
    subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
  },
  privateDnsEnabled: true,
});

// S3 Gateway Endpoint
vpc.addGatewayEndpoint('S3Endpoint', {
  service: ec2.GatewayVpcEndpointAwsService.S3,
  subnets: [
    {
      subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
    },
  ],
});

// Bedrock Runtime Endpoint
vpc.addInterfaceEndpoint('BedrockRuntimeEndpoint', {
  service: ec2.InterfaceVpcEndpointAwsService.BEDROCK_RUNTIME,
  subnets: {
    subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
  },
  privateDnsEnabled: true,
});

// Secrets Manager Endpoint
vpc.addInterfaceEndpoint('SecretsManagerEndpoint', {
  service: ec2.InterfaceVpcEndpointAwsService.SECRETS_MANAGER,
  subnets: {
    subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
  },
  privateDnsEnabled: true,
});

動作確認

cdk deploy し動作確認を行います。

今回 Aurora Serverless v2 を使用しているためクエリエディタが使用可能です。

https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/query-editor.html

対象 DB に接続し、Agent が書き込みを行うためのテーブル作成の SQL を流します。

CREATE TABLE IF NOT EXISTS agent_execution_results (
  id INT AUTO_INCREMENT PRIMARY KEY,
  execution_id VARCHAR(255) NOT NULL UNIQUE,
  agent_name VARCHAR(255) NOT NULL,
  input TEXT,
  output LONGTEXT,
  status VARCHAR(50),
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  INDEX idx_execution_id (execution_id),
  INDEX idx_agent_name (agent_name),
  INDEX idx_created_at (created_at)
);

Agent を Invoke します。問題なく動作しました。

クエリエディタで SELECT するとデータを取得することができました。AgentCore Runtime から Aurora への書き込みもできています。

終わりに

AgentCore Runtime の L2 Construct により、Agent のデプロイが簡潔になり、権限設定、セキュリティグループ設定、メトリクス作成が容易に行えます。ぜひ Agent 構築にご活用ください。

アマゾン ウェブ サービス ジャパン (有志)
アマゾン ウェブ サービス ジャパン (有志)Publication

この Publication に投稿している記事は、アマゾン ウェブ サービス ジャパン合同会社または Amazon Web Services, Inc. 所属社員による個人の見解であり、所属する組織の公式見解ではありません。参加したい従業員の方は、Sugiyama Suguru までお知らせください。

Discussion