AgentCore Code Interpreter × Strands Agents × GenU でファイル生成→ダウンロードまでを構築して得られた知見まとめ

agentcore create   # プロジェクト作成
agentcore dev      # ローカル開発サーバー起動
agentcore deploy   # AWS へデプロイ
agentcore invoke   # デプロイ済みエージェントの実行

npm install -g @aws/agentcore

agentcore create

Name:       CodeInterpreterAgent
Language:   Python
Build:      Container
Protocol:   HTTP
Framework:  Strands
Model:      Bedrock
Memory:     None
Network:    PUBLIC

cd CodeInterpreterAgent

CodeInterpreterAgent/
├── agentcore/
│   ├── agentcore.json          # リソース設定
│   ├── aws-targets.json        # デプロイ先設定
│   └── cdk/                    # CDK インフラ
│       └── lib/
│           └── cdk-stack.ts    # CDK スタック定義
└── app/
    └── MyAgent/
        ├── main.py             # エージェントのコード
        ├── model/
        │   └── load.py         # モデルロード
        ├── mcp_client/         # MCP クライアント
        ├── pyproject.toml      # Python 依存関係
        └── Dockerfile          # コンテナビルド

[project]
name = "MyAgent"
version = "0.1.0"
description = "Code Interpreter AgentCore Runtime"
requires-python = ">=3.10"
dependencies = [
    "aws-opentelemetry-distro",
    "bedrock-agentcore >= 1.0.3",
    "botocore[crt] >= 1.35.0",
    "strands-agents >= 1.13.0",
    "strands-agents-tools[agent_core_code_interpreter]",
    "boto3",
]

cd app/MyAgent
uv lock

import ast
import base64
import json
import os
import uuid

import boto3
from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from model.load import load_model
from strands_tools.code_interpreter import AgentCoreCodeInterpreter
from strands_tools.code_interpreter.models import ReadFilesAction

app = BedrockAgentCoreApp()
log = app.logger

# Code Interpreter の初期化
code_interpreter = AgentCoreCodeInterpreter(
    region=os.environ.get("AWS_REGION", "us-east-1")
)

WORKSPACE_DIR = "/tmp/ws"

def _is_under_dir(path: str, base_dir: str) -> bool:
    """パストラバーサル対策: path が base_dir 配下にあるか検証する。"""
    real_path = os.path.realpath(path)
    real_base = os.path.realpath(base_dir)
    return real_path == real_base or real_path.startswith(real_base + os.sep)

@tool
def download_file_from_sandbox(sandbox_filepath: str) -> str:
    """Download a file from the Code Interpreter sandbox to the local workspace.

    The Code Interpreter runs in an isolated sandbox. Files generated there
    are not directly accessible from this runtime. Use this tool to pull
    files from the sandbox before uploading to S3.

    Returns the local file path where the file was saved.

    Args:
        sandbox_filepath: The path to the file in the Code Interpreter sandbox.
                          Use a relative path from the sandbox working directory.
    """
    os.makedirs(WORKSPACE_DIR, exist_ok=True)

    result = code_interpreter.read_files(
        ReadFilesAction(type="readFiles", paths=[sandbox_filepath])
    )

    if result.get("status") != "success":
        raise RuntimeError(f"Failed to read sandbox file: {result}")

    # read_files() は _create_tool_result() で content を str() 化するため、
    # json.loads / ast.literal_eval でパースして blob / text を取り出す
    try:
        content_str = result["content"][0]["text"]
    except (KeyError, IndexError, TypeError) as e:
        raise RuntimeError(f"Unexpected read_files response: {result}") from e

    try:
        content_list = json.loads(content_str)
    except (json.JSONDecodeError, TypeError):
        content_list = ast.literal_eval(content_str)

    filename = os.path.basename(sandbox_filepath)
    local_filename = f"{uuid.uuid4()}_{filename}"
    local_path = os.path.join(WORKSPACE_DIR, local_filename)

    for item in content_list:
        if item.get("type") != "resource":
            continue

        resource = item.get("resource", {})
        if "blob" in resource:
            blob_data = resource["blob"]
            if isinstance(blob_data, bytes):
                # str() → ast.literal_eval() で bytes として復元された場合
                data = blob_data
            else:
                # base64 文字列として返された場合
                data = base64.b64decode(blob_data)
            with open(local_path, "wb") as f:
                f.write(data)
            return local_path
        elif "text" in resource:
            with open(local_path, "w", encoding="utf-8") as f:
                f.write(resource["text"])
            return local_path

    raise RuntimeError("No file content found in read_files response")

@tool
def upload_file_to_s3_and_retrieve_s3_url(filepath: str) -> str:
    """Upload a file from local workspace to S3 and return the S3 URL.

    GenU's frontend detects S3 URLs and generates its own presigned URL
    on demand, so we return the raw S3 URL instead of a presigned URL.

    Args:
        filepath: The local file path (under /tmp/ws) to upload
    """
    bucket = os.environ.get("FILE_BUCKET")
    if not bucket:
        return f"S3 upload skipped (FILE_BUCKET not set). Local path: {filepath}"

    if not _is_under_dir(filepath, WORKSPACE_DIR):
        raise ValueError(f"{filepath} is not under {WORKSPACE_DIR}.")

    _, ext = os.path.splitext(filepath)
    key = f"code-interpreter/{uuid.uuid4()}{ext}"

    s3 = boto3.client("s3")
    s3.upload_file(filepath, bucket, key)

    region = s3.meta.region_name
    s3_url = f"https://{bucket}.s3.{region}.amazonaws.com/{key}"
    return s3_url

SYSTEM_PROMPT = """あなたはデータ分析・レポート生成アシスタントです。

## 基本方針
- ユーザーの指示に基づいて、Code Interpreter でデータ分析やファイル生成を行います。
- Excel、CSV、グラフ画像など、さまざまな形式のファイルを生成できます。

## Code Interpreter サンドボックスに渡すコードの制約（厳守）
- `/tmp/` や `/var/` など、`outputs/` 以外のディレクトリにファイルを保存してはならない
- `os.makedirs('/tmp/ws', ...)` のような独自の作業ディレクトリを作成してはならない
- 日本語ファイル名を使用してはならない（英数字とアンダースコアのみ）
- 生成するすべてのファイルは **必ず `outputs/` ディレクトリにのみ** 保存すること
- コードの先頭で必ず `os.makedirs('outputs', exist_ok=True)` を実行すること
- ファイルパスは必ず `outputs/` で始まること（例: `outputs/report.xlsx`）

## ファイル提供の手順
Code Interpreter のサンドボックスと Runtime は完全に別のファイルシステムです。
サンドボックスで生成したファイルに Runtime から直接アクセスすることはできません。
ファイルをユーザーに提供するには、以下の 3 ステップを **必ずこの順序で** 実行してください。

1. **Code Interpreter でファイル生成**: `outputs/` ディレクトリに保存する
   ```python
   os.makedirs('outputs', exist_ok=True)
   df.to_excel('outputs/report.xlsx', index=False)
   ```
2. **`download_file_from_sandbox`**: サンドボックスから Runtime にダウンロードする
   - sandbox_filepath には `outputs/ファイル名` の形式で指定すること
   - 戻り値はローカルファイルパス。次のステップにそのまま渡すこと
   ```
   download_file_from_sandbox(sandbox_filepath="outputs/report.xlsx")
   ```
3. **`upload_file_to_s3_and_retrieve_s3_url`**: ステップ 2 の戻り値を渡して S3 にアップロードする
   - 戻り値はダウンロード URL
   ```
   upload_file_to_s3_and_retrieve_s3_url(filepath="<ステップ2の戻り値>")
   ```

## URL の出力形式
- 画像: `![ファイル名](URL)`
- Excel/CSV/PDF など: `[ファイル名](URL)`
"""

def process_prompt(prompt):
    """GenU が送る prompt（ContentBlock 配列）からテキストを抽出する。"""
    if isinstance(prompt, str):
        return prompt
    if isinstance(prompt, list):
        texts = [
            block["text"]
            for block in prompt
            if isinstance(block, dict) and "text" in block
        ]
        return "\n".join(texts) if texts else ""
    return str(prompt)

@app.entrypoint
async def invoke(payload):
    log.info("Invoking Code Interpreter Agent...")

    messages = payload.get("messages", [])
    prompt = process_prompt(payload.get("prompt", ""))

    agent = Agent(
        model=load_model(),
        system_prompt=SYSTEM_PROMPT,
        tools=[
            code_interpreter.code_interpreter,
            download_file_from_sandbox,
            upload_file_to_s3_and_retrieve_s3_url,
        ],
        messages=messages,
    )

    async for event in agent.stream_async(prompt):
        if "event" in event:
            yield event

if __name__ == "__main__":
    app.run()

from strands_tools.code_interpreter import AgentCoreCodeInterpreter

code_interpreter = AgentCoreCodeInterpreter(
    region=os.environ.get("AWS_REGION", "us-east-1")
)

@tool
def download_file_from_sandbox(sandbox_filepath: str) -> str:
    """Download a file from the Code Interpreter sandbox to the local workspace."""

@tool
def upload_file_to_s3_and_retrieve_s3_url(filepath: str) -> str:
    """Upload a file from local workspace to S3 and return the S3 URL."""

Code Interpreter のサンドボックスと Runtime は完全に別のファイルシステムです。
サンドボックスで生成したファイルに Runtime から直接アクセスすることはできません。

## Code Interpreter サンドボックスに渡すコードの制約（厳守）
- `/tmp/` や `/var/` など、`outputs/` 以外のディレクトリにファイルを保存してはならない
- `os.makedirs('/tmp/ws', ...)` のような独自の作業ディレクトリを作成してはならない

- 生成するすべてのファイルは **必ず `outputs/` ディレクトリにのみ** 保存すること

- 日本語ファイル名を使用してはならない（英数字とアンダースコアのみ）

async for event in agent.stream_async(prompt):
    if "event" in event:
        yield event

aws cloudformation describe-stacks \
  --stack-name AgentCoreStack \
  --query "Stacks[0].Outputs[?OutputKey=='FileBucketName'].OutputValue" \
  --output text

import {
  AgentCoreApplication,
  AgentCoreMcp,
  type AgentCoreProjectSpec,
  type AgentCoreMcpSpec,
} from '@aws/agentcore-cdk';
import { CfnOutput, Stack, type StackProps } from 'aws-cdk-lib';
import { Effect, PolicyStatement } from 'aws-cdk-lib/aws-iam';
import { Bucket } from 'aws-cdk-lib/aws-s3';
import { Construct } from 'constructs';

// GenU の AgentCore スタックが作成した S3 バケット名
const GENU_FILE_BUCKET_NAME =
  'agentcorestack-genericagentcoreagentcorefilebucket-xxxxxxxxxx';

export interface AgentCoreStackProps extends StackProps {
  spec: AgentCoreProjectSpec;
  mcpSpec?: AgentCoreMcpSpec;
  credentials?: Record<string, { credentialProviderArn: string; clientSecretArn?: string }>;
}

export class AgentCoreStack extends Stack {
  public readonly application: AgentCoreApplication;

  constructor(scope: Construct, id: string, props: AgentCoreStackProps) {
    super(scope, id, props);

    const { spec, mcpSpec, credentials } = props;

    this.application = new AgentCoreApplication(this, 'Application', {
      spec,
    });

    if (mcpSpec?.agentCoreGateways && mcpSpec.agentCoreGateways.length > 0) {
      new AgentCoreMcp(this, 'Mcp', {
        projectName: spec.name,
        mcpSpec,
        agentCoreApplication: this.application,
        credentials,
        projectTags: spec.tags,
      });
    }

    // GenU の AgentCore スタックが作成した S3 バケットを参照
    const fileBucket = Bucket.fromBucketName(
      this,
      'GenUFileBucket',
      GENU_FILE_BUCKET_NAME
    );

    // エージェント実行ロールに権限を付与
    const agentEnv = this.application.environments.values().next().value!;
    const { role } = agentEnv.runtime;

    // Code Interpreter 権限
    role.addToPrincipalPolicy(
      new PolicyStatement({
        effect: Effect.ALLOW,
        actions: [
          'bedrock-agentcore:CreateCodeInterpreter',
          'bedrock-agentcore:StartCodeInterpreterSession',
          'bedrock-agentcore:InvokeCodeInterpreter',
          'bedrock-agentcore:StopCodeInterpreterSession',
          'bedrock-agentcore:DeleteCodeInterpreter',
          'bedrock-agentcore:ListCodeInterpreters',
          'bedrock-agentcore:GetCodeInterpreter',
          'bedrock-agentcore:GetCodeInterpreterSession',
          'bedrock-agentcore:ListCodeInterpreterSessions',
        ],
        resources: ['*'],
      })
    );

    // GenU バケットへの書き込み権限
    fileBucket.grantWrite(role);

    // 環境変数の設定（GenU のバケット名を渡す）
    agentEnv.runtime.addEnvironmentVariable('FILE_BUCKET', GENU_FILE_BUCKET_NAME);

    new CfnOutput(this, 'StackNameOutput', {
      description: 'Name of the CloudFormation Stack',
      value: this.stackName,
    });
  }
}

agentcore deploy

agentcore status

arn:aws:bedrock-agentcore:ap-northeast-1:123456789012:runtime/xxxxxxxx

const envs: Record<string, Partial<StackInput>> = {
  myenv: {
    modelRegion: 'us-east-1',
    agentCoreRegion: 'ap-northeast-1',
    createGenericAgentCoreRuntime: true,
    agentBuilderEnabled: true,
    agentCoreExternalRuntimes: [
      {
        name: 'Code Interpreter Agent',
        arn: 'arn:aws:bedrock-agentcore:ap-northeast-1:123456789012:runtime/xxxxxxxx',
        description: 'Code Interpreter でデータ分析・レポート生成を行うエージェント',
      },
    ],
  },
};

npm run cdk:deploy