【AWS】Bedrock AgentCoreを使ってAIにPythonによるデータ分析ができるCode Interpreterを使わせてみる！

どうもこんにちは。今回は、Amazon Bedrock AgentCoreを使って、AIエージェントにCode Interpreterを実行させ、その実行証跡まで確認できる検証環境を作ってみました。

単に「答えが合っている」だけだと、本当にCode Interpreterを使ったのかは分かりません。そこで今回は、Strands Agentsのcallback handlerとmetricsを使って、code_interpreterの実行開始、結果、所要時間をCloudWatch LogsとRuntimeレスポンスの両方へ残す構成にしています。

この記事は誰向け？

この記事は以下のような方向けに執筆しています。

• Amazon Bedrock AgentCore を使ってみたい方
• Strands Agents から AgentCore Code Interpreter を使いたい方
• AWS CDK（Python）で Amazon Bedrock AgentCore Runtime へエージェントをデプロイしたい方
• 「Code Interpreterを使った証拠」をログとして残したい方

デモコードの場所

実際に動かしたい方は、以下を参照ください。

この検証でやりたいこと

今回のゴールは次の3点です。

• Strands Agentsで作成したPythonエージェントをDocker化する
• CDKでECRとBedrock AgentCore Runtimeへデプロイする
• Agentがcode_interpreterを実際に使ったことを、レスポンスとCloudWatch Logsで確認できるようにする

ポイントは最後の「証跡」です。モデルが回答文で「Pythonで計算しました」と言っても、それだけでは証拠として弱いです。そこで、Strandsが発行するcurrent_tool_useイベントとtoolResult、さらにmetrics.tool_metricsを突き合わせて、実際にツールが呼ばれたかどうかを判定します。

実装の流れ

0. 作業プロジェクトの用意

mkdir bedrock-agentcore-code-interpreter-demo && cd $_

1. CDKプロジェクトの雛形を整える

1-1. 依存パッケージ

まずはCDKの依存関係を定義します。

requirements.txt

aws-cdk-lib==2.255.0
constructs>=10.0.0,<11.0.0
cdk-ecr-deployment>=4.0.0,<5.0.0

requirements-dev.txt

-r requirements.txt
pytest>=8.0.0,<9.0.0

cdk-ecr-deploymentは、CDKがビルドしたDockerイメージを専用ECR Repositoryへコピーするために使います。

1-2. CDKエントリーポイント

app.pyは通常のCDKエントリーポイントです。

app.py

#!/usr/bin/env python3
import os

import aws_cdk as cdk

from infra.agentcore_code_interpreter_stack import AgentCoreCodeInterpreterDemoStack


app = cdk.App()

AgentCoreCodeInterpreterDemoStack(
    app,
    "AgentCoreCodeInterpreterDemoStack",
    env=cdk.Environment(
        account=os.environ.get("CDK_DEFAULT_ACCOUNT"),
        region=os.environ.get("CDK_DEFAULT_REGION", "ap-northeast-1"),
    ),
)

app.synth()

1-3. cdk.json

cdk.jsonは、常にこのプロジェクトの仮想環境を使うようにしておきます。

cdk.json

{
  "app": ".venv/bin/python app.py",
  "context": {
    "@aws-cdk/core:checkSecretUsage": true,
    "@aws-cdk/core:target-partitions": [
      "aws"
    ]
  }
}

.venv/bin/pythonを固定しておくと、仮想環境をactivateしていない状態でもCDKが同じPython環境を使ってくれます。

1-4. 仮想環境

python3.12 -m venv .venv
source .venv/bin/activate
pip install -r requirements-dev.txt -r agent/requirements.txt

AWS認証とリージョンを確認します。

aws sts get-caller-identity
export AWS_REGION=ap-northeast-1

必要であればbootstrapします。

cdk bootstrap

2. Strands AgentにCode Interpreterを組み込む

2-1. エージェント本体

エージェント実装はagent/code_interpreter_agent/agent.pyに置きます。

agent/code_interpreter_agent/agent.py

import os
import boto3
from strands import Agent
from strands.models import BedrockModel
from strands_tools.code_interpreter import AgentCoreCodeInterpreter


DEFAULT_REGION = "ap-northeast-1"
DEFAULT_MODEL_ID = "jp.anthropic.claude-sonnet-4-6"
DEFAULT_CODE_INTERPRETER_ID = "aws.codeinterpreter.v1"


def create_agent(callback_handler=None) -> Agent:
    region = os.environ.get("AWS_REGION", DEFAULT_REGION)
    model = BedrockModel(
        model_id=os.environ.get("BEDROCK_MODEL_ID", DEFAULT_MODEL_ID),
        boto_session=boto3.Session(region_name=region),
    )
    code_interpreter = AgentCoreCodeInterpreter(
        region=region,
        identifier=os.environ.get(
            "AGENTCORE_CODE_INTERPRETER_ID",
            DEFAULT_CODE_INTERPRETER_ID,
        ),
        session_timeout_seconds=900,
    )
    return Agent(
        model=model,
        tools=[code_interpreter.code_interpreter],
        system_prompt="計算やデータ処理は必ず code_interpreter を使う",
        callback_handler=callback_handler,
    )

ここで使っているaws.codeinterpreter.v1はAWS管理版のCode Interpreterです。そのため、今回はAWS::BedrockAgentCore::CodeInterpreterCustomは作りません。

2-2. システムプロンプト

今回は「本当にツールを使ったか」を見たいため、システムプロンプトでも明示的に、計算やデータ処理ではcode_interpreterを使うよう指示しています。

SYSTEM_PROMPT = """あなたは計算、データ分析、コード検証を行う日本語アシスタントです。

計算、統計、データ処理、アルゴリズム、またはコードの検証を求められた場合は、
推測だけで回答せず、必ず code_interpreter ツールで実際にコードを実行してください。
ツールが失敗した場合は成功したように装わず、失敗したことを明示してください。
最終回答は実行結果に基づいて簡潔に説明してください。
"""

3. Code Interpreterの実行証跡を集める

3-1. なぜ証跡収集が必要か

単に最終回答を見るだけでは、モデルが頭の中で計算したのか、Code Interpreterを使ったのかが区別できません。

そこで今回は、次の情報を記録します。

• toolUseId
• phase: started / completed / failed
• action
• language
• 実行コード
• 実行結果
• durationMs

3-2. callback handlerでcurrent_tool_useを捕まえる

Strandsでは、ツール実行中にcurrent_tool_useがcallbackへ流れてきます。これを使って、Code Interpreterの開始時点を取ります。

agent/code_interpreter_agent/evidence.py

def callback(self, **kwargs: Any) -> None:
    current_tool_use = kwargs.get("current_tool_use")
    if isinstance(current_tool_use, dict):
        self._capture_tool_use(current_tool_use)

    message = kwargs.get("message")
    if isinstance(message, dict):
        self._capture_message(message)

Code Interpreterの入力はストリーミング途中ではJSON文字列の断片として届くことがあるため、入力が完成したタイミングでactionやcodeを解釈するようにしています。

3-3. toolResultで成功・失敗を判定する

ツールの実行後はmessage.content[].toolResultが届くため、これでcompletedかfailedかを判定します。

def _result_status(tool_result: dict[str, Any]) -> bool:
    status = str(tool_result.get("status", "success")).lower()
    return status not in {"error", "failed", "failure"}

3-4. metricsもあわせて見る

callbackだけでなく、最終的なAgentResult.metrics.tool_metricsも参照します。これで、成功回数や総実行時間も確認できます。

metric = result.metrics.tool_metrics["code_interpreter"]
metric.call_count
metric.success_count
metric.error_count
metric.total_time

この値をcallbackで記録した情報と付き合わせることで、レスポンスへ返すinvocationCountやdurationMsの精度を上げています。

3-5. CloudWatch Logsへ構造化ログを出す

ログは1行JSONで出力します。

event = {
    "event": "code_interpreter_evidence",
    "requestId": self.request_id,
    "toolUseId": invocation.tool_use_id,
    "phase": phase,
    "action": action,
    "language": language,
    "code": _truncate(code, self.max_log_chars),
    "result": _truncate(invocation.result, self.max_log_chars),
    "durationMs": invocation.duration_ms,
}
logger.info(json.dumps(event, ensure_ascii=False, default=str))

コードや実行結果が長すぎるとログが読みにくくなるため、4000文字上限で切り詰め、truncatedとoriginalCharsも残しています。

4. Runtimeレスポンスに証跡要約を含める

runtime_app.pyでは、各リクエストごとにrequestIdを発行し、エージェント実行後に証跡サマリをレスポンスへ埋め込みます。

agent/code_interpreter_agent/runtime_app.py

def invoke_agent(payload: dict) -> dict:
    prompt = payload.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip():
        raise ValueError("payload.prompt must be a non-empty string")

    request_id = str(uuid.uuid4())
    collector = CodeInterpreterEvidenceCollector(request_id=request_id)
    result = create_agent(callback_handler=collector.callback)(prompt.strip())
    evidence = collector.finalize(result)

    return {
        "response": extract_text(result),
        "evidence": {
            "requestId": request_id,
            "codeInterpreter": evidence,
        },
    }

応答例は次のようになります。

{
  "response": "最初の10個のフィボナッチ数は 0, 1, 1, 2, 3, 5, 8, 13, 21, 34 です。",
  "evidence": {
    "requestId": "8bb13bec-9e20-4d39-b877-e1e9d932f991",
    "codeInterpreter": {
      "used": true,
      "successful": true,
      "invocationCount": 1,
      "durationMs": 1234
    }
  }
}

ここで大事なのは、usedはモデルの自己申告ではなく、実際にcurrent_tool_useが発生した場合だけtrueになる点です。

5. Dockerイメージを作る

AgentCore Runtimeへ載せるため、エージェントをDocker化します。

agent/Dockerfile

FROM public.ecr.aws/docker/library/python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
ENV PORT=8080

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY code_interpreter_agent ./code_interpreter_agent

EXPOSE 8080

CMD ["python", "-m", "code_interpreter_agent.runtime_app"]

依存関係はAgent用に分けています。

agent/requirements.txt

bedrock-agentcore>=1.3.0,<2.0.0
strands-agents>=1.0.0,<2.0.0
strands-agents-tools>=0.2.0,<1.0.0
boto3>=1.42.0

6. CDKでECRとAgentCore Runtimeを作る

6-1. ECR Repository

まずは専用ECR Repositoryを作成します。

infra/agentcore_code_interpreter_stack.py

image_repository = ecr.Repository(
    self,
    "AgentImageRepository",
    repository_name="agentcore-code-interpreter-demo",
    image_scan_on_push=True,
    image_tag_mutability=ecr.TagMutability.IMMUTABLE,
    lifecycle_rules=[
        ecr.LifecycleRule(
            description="Keep the ten most recent agent images",
            max_image_count=10,
        )
    ],
    removal_policy=RemovalPolicy.DESTROY,
    empty_on_delete=True,
)

6-2. DockerImageAssetをECRへコピー

CDK標準のDockerImageAssetだけだと専用Repositoryを使いづらいため、cdk-ecr-deploymentでコピーします。

image_asset = ecr_assets.DockerImageAsset(
    self,
    "AgentImageAsset",
    directory=str(Path(__file__).resolve().parents[1] / "agent"),
    platform=ecr_assets.Platform.LINUX_ARM64,
)

image_tag = image_asset.asset_hash
runtime_image_uri = image_repository.repository_uri_for_tag(image_tag)

ecr_deployment.ECRDeployment(
    self,
    "DeployAgentImage",
    src=ecr_deployment.DockerImageName(image_asset.image_uri),
    dest=ecr_deployment.DockerImageName(runtime_image_uri),
)

6-3. Runtime実行ロール

AgentCore Runtimeが必要とする権限を付与します。

runtime_role.add_to_policy(
    iam.PolicyStatement(
        actions=[
            "bedrock-agentcore:StartCodeInterpreterSession",
            "bedrock-agentcore:InvokeCodeInterpreter",
            "bedrock-agentcore:StopCodeInterpreterSession",
        ],
        resources=[
            f"arn:{Aws.PARTITION}:bedrock-agentcore:{Aws.REGION}:"
            "aws:code-interpreter/aws.codeinterpreter.v1"
        ],
    )
)

これに加えて、ECR pull、CloudWatch Logs、bedrock:InvokeModelも付与しています。

6-4. AgentCore Runtime

最後にRuntime本体です。

runtime = bedrock_agentcore.CfnRuntime(
    self,
    "AgentRuntime",
    agent_runtime_name="AgentCoreCodeInterpreterDemoRuntime",
    description="Strands agent using managed AgentCore Code Interpreter.",
    agent_runtime_artifact=...,
    network_configuration=bedrock_agentcore.CfnRuntime.NetworkConfigurationProperty(
        network_mode="PUBLIC",
    ),
    protocol_configuration="HTTP",
    role_arn=runtime_role.role_arn,
    environment_variables={
        "AWS_REGION": self.region,
        "BEDROCK_MODEL_ID": model_id,
        "AGENTCORE_CODE_INTERPRETER_ID": "aws.codeinterpreter.v1",
        "EVIDENCE_LOG_MAX_CHARS": "4000",
    },
    lifecycle_configuration=bedrock_agentcore.CfnRuntime.LifecycleConfigurationProperty(
        idle_runtime_session_timeout=900,
        max_lifetime=3600,
    ),
)

7. テストを書く

今回は「証跡が本当に残るか」が重要なので、そこを厚めにテストしています。

たとえば次のような観点です。

• startedとcompletedログが出るか
• Code Interpreter未使用時にused: falseになるか
• 複数回呼び出し時にinvocationCountが正しく増えるか
• 長いコードや結果が切り詰められるか
• 分割されたJSON入力でも最終的にactionやcodeを正しく取れるか

例:

tests/test_evidence.py

def test_collects_successful_code_interpreter_evidence(caplog) -> None:
    collector = CodeInterpreterEvidenceCollector(
        request_id="request-1",
        clock=FakeClock(10.0, 11.25),
        max_log_chars=100,
    )
    ...
    evidence = collector.finalize(result)

    assert evidence == {
        "used": True,
        "successful": True,
        "invocationCount": 1,
        "durationMs": 1250,
    }

CDK側も、ECR、Runtime、環境変数、IAM権限をassertionsで検証しています。

8. デプロイして確認する

pytest
cdk synth
cdk deploy

今回は、フィボナッチ数列の計算をしてもらおうと思います。以下の操作をして、 ランタイムプレイグラウンド の画面へ遷移してください。

1. Bedrock AgentCore コンソールへ遷移
2. 左メニューの テスト セクション以下の ランタイムプレイグラウンド をクリック
3. ランタイムエージェントで、今回作成したランタイム名を選択し、エンドポイントが DEFAULT となっていることを確認

では、入力欄に以下を入力します。証跡確保のために、Pythonコードも明示させます。

{"prompt": "Pythonで最初の10個のフィボナッチ数を計算してください。計算に使用したPythonコードを明示してください。"}

実行 ボタンをクリックしたら、以下のようにレスポンスが返ってくるはずです。（ちょっと見づらいけど、整形したら以下のようになります！）

## 実行結果

---

### 📝 使用したPythonコード

``python
def fibonacci(n):
    fibs = []
    a, b = 0, 1
    for _ in range(n):
        fibs.append(a)
        a, b = b, a + b
        return fibs
        
        result = fibonacci(10)
        for i, val in enumerate(result, 1):
            print(f'F({i:2d}) = {val}')
``

---

### ✅ 計算結果（最初の10個のフィボナッチ数）

| 番号 | フィボナッチ数 |
|------|--------------|
| F(1)  | 0  |
| F(2)  | 1  |
| F(3)  | 1  |
| F(4)  | 2  |
| F(5)  | 3  |
| F(6)  | 5  |
| F(7)  | 8  |
| F(8)  | 13 |
| F(9)  | 21 |
| F(10) | 34 |

---

### 💡 アルゴリズムの説明

- 変数 `a=0`, `b=1` から開始します。
- 各ステップで `a` をリストに追加し、`a, b = b, a + b` で次の値を計算します。
- これを10回繰り返すことで、**0から始まる最初の10個のフィボナッチ数**が得られます。
- 計算量は **O(n)** と非常に効率的です。

続いてCloudWatch Logsで同じrequestIdを検索し、次のようなcode_interpreter_evidenceログを確認します。

{
  "event": "code_interpreter_evidence",
  "requestId": "xxxxxxxxxx-xxxx-xxxx-xxxxx-xxxxxxxxxx",
  "toolUseId": "tooluse_...",
  "phase": "completed",
  "action": "executeCode",
  "language": "python",
  "code": {
    "value": "print(...)",
    "truncated": false,
    "originalChars": 10
  },
  "result": {
    "value": "[{\"text\": \"...\"}]",
    "truncated": false,
    "originalChars": 20
  },
  "durationMs": 1234
}

これで、「最終回答が合っている」だけでなく、「実際にCode Interpreterが呼ばれ、何を実行して、成功したのか」まで追えるようになります。

まとめ

今回は、Amazon Bedrock AgentCore RuntimeへStrands Agentをデプロイし、AWS管理版Code Interpreterを使う構成を作ってみました。

構成自体は比較的シンプルですが、実運用や検証では「本当にツールを使ったのか」を見たい場面が多いと思います。そこを、Strandsのcallback handlerとmetrics、CloudWatch Logs、Runtimeレスポンスを組み合わせて確認できるようにしたのが今回のポイントです。

次の段階としては、CSV分析やグラフ生成など、Code Interpreterらしいユースケースへ広げていくと面白そうです。