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

Bedrock AgentCore Code Interpreter の CDK L1 Construct を試す

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

前回 AgentCore Runtime の CDK L1 Construct の使い方をご紹介しました。

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

今回は AgentCore Code Interpreter の CDK L1 Construct を試してみます。

AgentCore Code Interpreter とは

AgentCore Code Interpreter は、Agent が Sandbox 環境内で安全にコードやコマンドを記述・実行できる機能です。Agent が動作する環境とは 別環境で、任意のコード・コマンド実行を行い、データ漏洩やセキュリティリスクを防ぐために重要な役割を果たします。


（画像は Execute code and analyze data using Amazon Bedrock AgentCore Code Interpreter より引用）

※セキュリティ面以外にも Sandbox 環境を使用するメリットはあります。詳細は Why use Code Interpreter in agent development をご参照ください。

作るもの

AgentCore Runtime 上にデプロイした Agent から AgentCore Code Interpreter を使用して、S3 バケット上のスクリプトを実行し、結果を S3 バケットに保存する例を構築します。

これは AWS ドキュメントの例を使用したものになります。AgentCore Code Interpreter と AWS リソース連携を試したかったため、この例を使用しています。

https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/code-interpreter-s3-integration.html

シーケンスとしては以下です。

  • AgentCore Runtime にデプロイした Agent から AgentCore Code Interpreter の呼び出しを行う
  • AgentCore Code Interpreter 上でコマンドを実行し、S3 バケットからスクリプトの取得や結果の書き込みを行う。

事前準備

ディレクトリ構成

cdk init の後、lib/app-tools 配下に AgentCore Code Interpreter を使用する Agent を実装し、script 配下に AgentCore Code Interpreter で使用する python スクリプトを格納しています。

.
├── bin
│   └── cdk-bedrock-agentcore.ts
├── cdk.json
├── lib
│   ├── app_tools
│   └── cdk-buitin-tools-stack.ts
├── package-lock.json
├── package.json
├── README.md
├── script
│   └── generate_csv.py
└── tsconfig.json

Agent の実装

前回同様、Strands Agents のサンプル Option B: Custom Agent を使用しています。

また今回、AgentCore Code Interpreter 連携は、AWS のドキュメントの実装を参考に作成します。

AgentCore Code Interpreter リソースの作成/削除は IaC (CDK) の範疇となるため、Agent に実装する処理は、Code Interpreter session の開始からCode Interpreter session の終了までの一連の流れになります。

この例を参考に、Strands Agents のサンプルに対して、AgentCore Code Interpreter を使用する Tool を追加しています（以下実装例の execute_command を参照）。

Tool はドキュメントを参考に AgentCore Code Interpreter のセッションの開始から終了までを実装しています。

Agent の実装例
lib/app_tools/agent.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any
from datetime import datetime, timezone
from strands import Agent, tool
import os
import json
import boto3


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

CODE_INTERPRETER_ID = os.environ["CODE_INTERPRETER_ID"]
S3_BUCKET_NAME = os.environ["S3_BUCKET_NAME"]

REGION = "us-west-2"

# Global code interpreter client
bedrock_agentcore_client = boto3.client(
    "bedrock-agentcore",
    region_name=REGION,
    endpoint_url=f"https://bedrock-agentcore.{REGION}.amazonaws.com",
)

# Validation-focused system prompt
SYSTEM_PROMPT = """You are a helpful assistant"""


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


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


# 追加した AgentCore Code Interpreter 連携の Tool
@tool
def execute_command():
    """Create sample csv to Amazon S3 Bucket."""
    session_response = bedrock_agentcore_client.start_code_interpreter_session(
        codeInterpreterIdentifier=CODE_INTERPRETER_ID,
        name="combined-test-session",
        sessionTimeoutSeconds=1800,
    )
    session_id = session_response["sessionId"]
    print(f"Created session ID: {session_id}")

    print(f"Downloading CSV generation script from S3")
    command_to_execute = f"aws s3 cp s3://{S3_BUCKET_NAME}/generate_csv.py ."
    response = bedrock_agentcore_client.invoke_code_interpreter(
        codeInterpreterIdentifier=CODE_INTERPRETER_ID,
        sessionId=session_id,
        name="executeCommand",
        arguments={"command": command_to_execute},
    )

    for event in response["stream"]:
        print(json.dumps(event["result"], default=str, indent=2))

    print(f"Executing the CSV generation script")
    response = bedrock_agentcore_client.invoke_code_interpreter(
        codeInterpreterIdentifier=CODE_INTERPRETER_ID,
        sessionId=session_id,
        name="executeCommand",
        arguments={"command": "python generate_csv.py 5 10"},
    )

    for event in response["stream"]:
        print(json.dumps(event["result"], default=str, indent=2))

    print(f"Uploading generated artifact to S3")
    command_to_execute = (
        f"aws s3 cp generated_data.csv s3://{S3_BUCKET_NAME}/output_artifacts/"
    )
    response = bedrock_agentcore_client.invoke_code_interpreter(
        codeInterpreterIdentifier=CODE_INTERPRETER_ID,
        sessionId=session_id,
        name="executeCommand",
        arguments={"command": command_to_execute},
    )

    for event in response["stream"]:
        print(json.dumps(event["result"], default=str, indent=2))

    print(f"Stopping the code interpreter session")
    stop_response = bedrock_agentcore_client.stop_code_interpreter_session(
        codeInterpreterIdentifier=CODE_INTERPRETER_ID, sessionId=session_id
    )

    return stop_response


@app.post("/invocations", response_model=InvocationResponse)
async def invoke_agent(request: InvocationRequest):
    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.",
            )

        agent = Agent(
            tools=[execute_command],
            system_prompt=SYSTEM_PROMPT,
        )

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

        return InvocationResponse(output=response)

    except Exception as e:
        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)

python スクリプトの用意

上記の処理の中で使用する generate_csv.py は配布されていません。そのため LLM に作ってもらいました。引数を 2 つ渡すことで何かしら csv を出力するというだけで、特に深い意味はありません。

参考までに載せておきます。

python スクリプト
script/generate_csv.py
import sys
import csv
import random
from datetime import datetime, timedelta


def generate_random_data(rows, columns):
    # ヘッダーの作成
    headers = [f"Column_{i+1}" for i in range(columns)]

    # データの生成
    data = []
    start_date = datetime.now()

    for i in range(rows):
        row = []
        for j in range(columns):
            # 列ごとに異なるタイプのデータを生成
            if j % 3 == 0:
                # 数値データ
                value = random.uniform(0, 1000)
                row.append(round(value, 2))
            elif j % 3 == 1:
                # 日付データ
                date = start_date + timedelta(days=i)
                row.append(date.strftime("%Y-%m-%d"))
            else:
                # カテゴリデータ
                value = random.choice(["A", "B", "C", "D", "E"])
                row.append(value)
        data.append(row)

    return headers, data


def main():
    # コマンドライン引数の確認
    if len(sys.argv) != 3:
        print("Usage: python generate_csv.py <rows> <columns>")
        sys.exit(1)

    try:
        rows = int(sys.argv[1])
        columns = int(sys.argv[2])
    except ValueError:
        print("Error: rows and columns must be integers")
        sys.exit(1)

    # データの生成
    headers, data = generate_random_data(rows, columns)

    # CSVファイルの作成
    output_file = "generated_data.csv"
    with open(output_file, "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(headers)
        writer.writerows(data)

    print(f"Generated CSV file '{output_file}' with {rows} rows and {columns} columns")


if __name__ == "__main__":
    main()

CDK 実装内容

Stack（lib/cdk-buitin-tools-stack.ts）に実装しています。全量は以下です。

CDK 実装内容の全量
lib/cdk-buitin-tools-stack.ts
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as s3deploy from 'aws-cdk-lib/aws-s3-deployment';
import { DockerImageAsset, Platform } from 'aws-cdk-lib/aws-ecr-assets';
import * as path from 'path';
import * as agentcore from 'aws-cdk-lib/aws-bedrockagentcore';

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

    /**
     * S3 Bucket for script and output
     */
    const bucket = new s3.Bucket(this, 'CodeInterpreterBucket', {
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      autoDeleteObjects: true,
    });

    // Deploy script to S3 bucket
    new s3deploy.BucketDeployment(this, 'DeployCodeInterpreterAssets', {
      sources: [s3deploy.Source.asset('./script')],
      destinationBucket: bucket,
    });

    /**
     *  AgentCore interpreter
     */
    const codeInterpreterExecutionRole = new iam.Role(this, 'CodeInterpreterExecutionRole', {
      assumedBy: new iam.ServicePrincipal('bedrock-agentcore.amazonaws.com'),
      description: 'Execution role for AgentCore Code Interpreter',
    });

    bucket.grantReadWrite(codeInterpreterExecutionRole);

    const codeInterpreter = new agentcore.CfnCodeInterpreterCustom(this, 'CodeInterpreter', {
      name: 'MyCodeInterpreter',
      executionRoleArn: codeInterpreterExecutionRole.roleArn,
      networkConfiguration: {
        networkMode: 'SANDBOX', // 'PUBLIC' or 'SANDBOX' を設定
      },
    });

    /**
     * AgentCore Runtime
     */
    // Create Docker Image Asset
    const dockerImageAsset = new DockerImageAsset(this, 'DockerImageAsset', {
      directory: path.join(__dirname, 'app_tools'),
      platform: Platform.LINUX_ARM64,
      file: 'Dockerfile',
    });

    // Create IAM role for Bedrock AgentCore with required policies
    const agentCoreRole = new iam.Role(this, 'BedrockAgentCoreRole', {
      assumedBy: new iam.ServicePrincipal('bedrock-agentcore.amazonaws.com'),
      description: 'IAM role for Bedrock AgentCore Runtime',
    });

    const region = cdk.Stack.of(this).region;
    const accountId = cdk.Stack.of(this).account;

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        sid: 'ECRImageAccess',
        effect: iam.Effect.ALLOW,
        actions: ['ecr:BatchGetImage', 'ecr:GetDownloadUrlForLayer'],
        resources: [`arn:aws:ecr:${region}:${accountId}:repository/*`],
      }),
    );

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['logs:DescribeLogStreams', 'logs:CreateLogGroup'],
        resources: [`arn:aws:logs:${region}:${accountId}:log-group:/aws/bedrock-agentcore/runtimes/*`],
      }),
    );

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['logs:DescribeLogGroups'],
        resources: [`arn:aws:logs:${region}:${accountId}:log-group:*`],
      }),
    );

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['logs:CreateLogStream', 'logs:PutLogEvents'],
        resources: [`arn:aws:logs:${region}:${accountId}:log-group:/aws/bedrock-agentcore/runtimes/*:log-stream:*`],
      }),
    );

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        sid: 'ECRTokenAccess',
        effect: iam.Effect.ALLOW,
        actions: ['ecr:GetAuthorizationToken'],
        resources: ['*'],
      }),
    );

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: [
          'xray:PutTraceSegments',
          'xray:PutTelemetryRecords',
          'xray:GetSamplingRules',
          'xray:GetSamplingTargets',
        ],
        resources: ['*'],
      }),
    );

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['cloudwatch:PutMetricData'],
        resources: ['*'],
        conditions: {
          StringEquals: {
            'cloudwatch:namespace': 'bedrock-agentcore',
          },
        },
      }),
    );

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        sid: 'GetAgentAccessToken',
        effect: iam.Effect.ALLOW,
        actions: [
          'bedrock-agentcore:GetWorkloadAccessToken',
          'bedrock-agentcore:GetWorkloadAccessTokenForJWT',
          'bedrock-agentcore:GetWorkloadAccessTokenForUserId',
        ],
        resources: [
          `arn:aws:bedrock-agentcore:${region}:${accountId}:workload-identity-directory/default`,
          `arn:aws:bedrock-agentcore:${region}:${accountId}:workload-identity-directory/default/workload-identity/agentName-*`,
        ],
      }),
    );

    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        sid: 'BedrockModelInvocation',
        effect: iam.Effect.ALLOW,
        actions: ['bedrock:InvokeModel', 'bedrock:InvokeModelWithResponseStream'],
        resources: ['arn:aws:bedrock:*::foundation-model/*', `arn:aws:bedrock:${region}:${accountId}:*`],
      }),
    );

    // AgentCore Code Interpreter へのアクセス許可
    agentCoreRole.addToPolicy(
      new iam.PolicyStatement({
        sid: 'InvokeAgentCoreCodeInterpreter',
        effect: iam.Effect.ALLOW,
        actions: [
          'bedrock-agentcore:StartCodeInterpreterSession',
          'bedrock-agentcore:InvokeCodeInterpreter',
          'bedrock-agentcore:StopCodeInterpreterSession',
        ],
        resources: [codeInterpreter.attrCodeInterpreterArn],
      }),
    );

    // Bedrock AgentCore Runtime
    const runtime = new agentcore.CfnRuntime(this, 'AgentCoreRuntimeWithBuiltInTools', {
      agentRuntimeName: 'MyAgentRuntimWithBuiltInTools',
      agentRuntimeArtifact: {
        containerConfiguration: {
          containerUri: dockerImageAsset.imageUri,
        },
      },
      networkConfiguration: {
        networkMode: 'PUBLIC',
      },
      roleArn: agentCoreRole.roleArn,
      protocolConfiguration: 'HTTP',
      environmentVariables: {
        CODE_INTERPRETER_ID: codeInterpreter.attrCodeInterpreterId,
        S3_BUCKET_NAME: bucket.bucketName,
      },
    });

    // 依存関係を定義
    runtime.node.addDependency(agentCoreRole);

    // Runtime Endpoint
    new agentcore.CfnRuntimeEndpoint(this, 'RuntimeWithBuiltInToolsEndpoint', {
      agentRuntimeId: runtime.attrAgentRuntimeId,
      agentRuntimeVersion: runtime.attrAgentRuntimeVersion,
      name: 'MyAgentRuntimeEndpoint',
    });
  }
}

以下詳細を見ていきます。

S3 バケット

AgentCore Interpreter と連携する S3 バケットを作成しています。また BucketDeployment Construct を使用し、cdk deploy 時にリポジトリ内のスクリプトを S3 バケットにデプロイします。

lib/cdk-buitin-tools-stack.ts

/**
 * S3 Bucket for script and output
 */
const bucket = new s3.Bucket(this, 'CodeInterpreterBucket', {
    removalPolicy: cdk.RemovalPolicy.DESTROY,
    autoDeleteObjects: true,
});

// Deploy script to S3 bucket
new s3deploy.BucketDeployment(this, 'DeployCodeInterpreterAssets', {
    sources: [s3deploy.Source.asset('./script')],
    destinationBucket: bucket,
});

AgentCore Code Interpreter

AgentCore Code Interpreter を作成します。Execution Role を作成し、S3 バケットの読み取り/書き込み権限を付与した上で定義します。

lib/cdk-buitin-tools-stack.ts
/**
 *  AgentCore Code Interpreter
 */
const codeInterpreterExecutionRole = new iam.Role(this, 'CodeInterpreterExecutionRole', {
    assumedBy: new iam.ServicePrincipal('bedrock-agentcore.amazonaws.com'),
    description: 'Execution role for AgentCore Code Interpreter',
});

bucket.grantReadWrite(codeInterpreterExecutionRole);

const codeInterpreter = new agentcore.CfnCodeInterpreterCustom(this, 'CodeInterpreter', {
    name: 'MyCodeInterpreter',
    executionRoleArn: codeInterpreterExecutionRole.roleArn,
    networkConfiguration: {
        networkMode: 'SANDBOX', // 'PUBLIC' or 'SANDBOX' を設定
    },
});

networkMode ですが、SANDBOX の場合は S3 バケットへのアクセスのみ可能、かつ Public Internet から隔離された Sandbox 環境でコードやコマンドが実行されます。

AgentCore Code Interpreter が Public Internet にアクセスする必要がある場合は PUBLIC を指定してください。

以下は Document の説明の抜粋です。

Configure network access - Choose PUBLIC mode if your Code Interpreter needs to connect to the public internet. If your Code Interpreter supports connection to Amazon S3, and if you want your Code Interpreter session to remain isolated from the public internet, choose SANDBOX mode.

なおネットワーク設定として、VPC サポートがアップデートで追加されているため、こちらの対応も時間の問題かと思います。

マネコン上では VPC を選択できますが、執筆時点では CloudFomartion (CFn) では設定できませんでした。

5:24:08 PM | UPDATE_FAILED        | AWS::BedrockA
gentCore::CodeInterpreterCustom | CodeInterpreter
Properties validation failed for resource CodeInt
erpreter with message:
[#/NetworkConfiguration/NetworkMode: VPC is not a
valid enum value]

AgentCore Runtime

基本的には前回の AgentCore Runtime の記事と同様なので、そちらをご参照ください。

以下前回記事との差分のみを記載します。

AgentCore Runtime -> AgentCore Code Interpreter へのアクセスが発生するため、追加のポリシーを付与します。今回はセッションの start/stop, invoke のみ使用するためそれのみ許可しています。

lib/cdk-buitin-tools-stack.ts
// AgentCore Code Interpreter へのアクセス許可
agentCoreRole.addToPolicy(
    new iam.PolicyStatement({
    sid: 'InvokeAgentCoreCodeInterpreter',
    effect: iam.Effect.ALLOW,
    actions: [
        'bedrock-agentcore:StartCodeInterpreterSession',
        'bedrock-agentcore:InvokeCodeInterpreter',
        'bedrock-agentcore:StopCodeInterpreterSession',
    ],
    resources: [codeInterpreter.attrCodeInterpreterArn],
    }),
);

なお AgentCore Code Interpreter の ARN として以下の 2 パターンが存在します。

  • code-interpreter: arn:${Partition}:bedrock-agentcore:${Region}:aws:code-interpreter/${CodeInterpreterId}
  • code-interpreter-custom: arn:${Partition}:bedrock-agentcore:${Region}:${Account}:code-interpreter-custom/${CodeInterpreterId}

今回のように自分で AgentCore Code Interpreter のリソースを作成した場合は後者になります。前者は デフォルトで AWS 側で作成されているマネージドなリソースになります（アカウント ID が ARN に含まれていない）。

あとは AgentCore Runtime の定義を行います。前回の差分としては Agent の実装上で使用する、AgentCore Code Interpreter の ID と、S3 バケット名を環境変数で設定しているのみです。

lib/cdk-buitin-tools-stack.ts
// Bedrock AgentCore Runtime
const runtime = new agentcore.CfnRuntime(this, 'AgentCoreRuntimeWithBuiltInTools', {

    //...中略

    // 環境変数を設定
    environmentVariables: {
        CODE_INTERPRETER_ID: codeInterpreter.attrCodeInterpreterId,
        S3_BUCKET_NAME: bucket.bucketName,
    },
});

動作確認

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

マネコンから AgentCore Runtime 上にデプロイした Agent に対して CSV の作成を指示します。応答は正常なものが返ってきました。

ログを見たところ AgentCore Code Interpreter 連携用の Tool (execute_command) が呼び出されていることがわかります。

S3 バケットを確認すると CSV ファイルが出力されています。問題ないことが確認できました。

おわりに

AWS CDK で AgentCore Code Intepreter を実装すると、今回の例のように権限の設定の抽象化が活用できる点がメリットかと思います。

L2 Construct 化も進行中のため、今後さらに使いやすくなるでしょう。

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

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

Discussion