Bedrock AgentCore Code Interpreterでプリインストールにないライブラリをインストールして使う

Python 3.12.12
pip 26.0.1 from /opt/amazon/genesis1p-tools/venv/lib64/python3.12/site-packages/pip
aws-cli/2.32.22
genesis1ptools
Linux localhost 6.1.158-15.288.amzn2023.aarch64

ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
BUCKET_NAME="agentcore-ci-lib-test-${ACCOUNT_ID}"
REGION="us-east-1"
 
# S3 バケット作成
aws s3 mb "s3://${BUCKET_NAME}" --region "$REGION"
 
# IAM ロール作成（AgentCore 信頼ポリシー）
aws iam create-role \
  --role-name AgentCoreCITestRole \
  --assume-role-policy-document '{
    "Version": "2012-10-17",
    "Statement": [{
      "Effect": "Allow",
      "Principal": {"Service": "bedrock-agentcore.amazonaws.com"},
      "Action": "sts:AssumeRole"
    }]
  }'
 
# S3 アクセスポリシーを付与
aws iam put-role-policy \
  --role-name AgentCoreCITestRole \
  --policy-name S3Access \
  --policy-document "{
    \"Version\": \"2012-10-17\",
    \"Statement\": [{
      \"Effect\": \"Allow\",
      \"Action\": [\"s3:GetObject\", \"s3:PutObject\", \"s3:ListBucket\"],
      \"Resource\": [
        \"arn:aws:s3:::${BUCKET_NAME}\",
        \"arn:aws:s3:::${BUCKET_NAME}/*\"
      ]
    }]
  }"

import boto3, time
 
REGION = "us-east-1"
ACCOUNT_ID = "123456789012"  # 自分のアカウントIDに置き換え
ROLE_ARN = f"arn:aws:iam::{ACCOUNT_ID}:role/AgentCoreCITestRole"
 
control = boto3.client("bedrock-agentcore-control", region_name=REGION)
response = control.create_code_interpreter(
    name="ciLibInstallTest",
    executionRoleArn=ROLE_ARN,
    networkConfiguration={"networkMode": "PUBLIC"},
)
ci_id = response["codeInterpreterId"]
 
# READY になるまでポーリング
while True:
    status = control.get_code_interpreter(codeInterpreterId=ci_id)["status"]
    if status == "READY":
        break
    time.sleep(5)

data_client = boto3.client("bedrock-agentcore", region_name=REGION)
 
def start_session():
    return data_client.start_code_interpreter_session(
        codeInterpreterIdentifier=ci_id)["sessionId"]
 
def run_code(session_id, code):
    resp = data_client.invoke_code_interpreter(
        codeInterpreterIdentifier=ci_id, sessionId=session_id,
        name="executeCode", arguments={"code": code, "language": "python"})
    for event in resp.get("stream", []):
        if "result" in event:
            sc = event["result"].get("structuredContent", {})
            if sc.get("stdout"): print(sc["stdout"])
            if sc.get("stderr"): print(sc["stderr"])
 
def run_cmd(session_id, command):
    resp = data_client.invoke_code_interpreter(
        codeInterpreterIdentifier=ci_id, sessionId=session_id,
        name="executeCommand", arguments={"command": command})
    for event in resp.get("stream", []):
        if "result" in event:
            sc = event["result"].get("structuredContent", {})
            if sc.get("stdout"): print(sc["stdout"])
            if sc.get("stderr"): print(sc["stderr"])
 
def stop_session(session_id):
    data_client.stop_code_interpreter_session(
        codeInterpreterIdentifier=ci_id, sessionId=session_id)

session_id = start_session()
 
# cowsay がプリインストールにないことを確認
run_code(session_id, """
try:
    import cowsay
    print(f"cowsay installed: {cowsay.__version__}")
except ImportError:
    print("cowsay: NOT installed")
""")
# → cowsay: NOT installed
 
# PyPI から直接インストール
run_cmd(session_id, "pip install cowsay")

Collecting cowsay
  Downloading cowsay-6.1-py3-none-any.whl.metadata (5.6 kB)
Downloading cowsay-6.1-py3-none-any.whl (25 kB)
Installing collected packages: cowsay
Successfully installed cowsay-6.1

run_code(session_id, 'import cowsay; cowsay.cow("Hello from AgentCore Code Interpreter!")')
stop_session(session_id)

______________________________________
| Hello from AgentCore Code Interpreter! |
  ======================================
                                      \
                                       \
                                         ^__^
                                         (oo)\_______
                                         (__)\       )\/\
                                             ||----w |
                                             ||     ||

# 事前準備（ローカルまたはCI）
pip download cowsay --dest ./wheels/ --only-binary=:all:
aws s3 sync ./wheels/ s3://${BUCKET_NAME}/cowsay_wheels/

session_id = start_session()
 
# S3 から wheel をダウンロード
run_cmd(session_id, f"mkdir -p /tmp/cowsay_s3 && aws s3 cp s3://{BUCKET_NAME}/cowsay_wheels/ /tmp/cowsay_s3/ --recursive")

download: s3://.../cowsay_wheels/cowsay-6.1-py3-none-any.whl
  to ../../../../tmp/cowsay_s3/cowsay-6.1-py3-none-any.whl

# オフラインインストール（PyPI へのアクセスなし）
run_cmd(session_id, "pip install --no-index --find-links /tmp/cowsay_s3/ cowsay")

Looking in links: /tmp/cowsay_s3/
Processing /tmp/cowsay_s3/cowsay-6.1-py3-none-any.whl
Installing collected packages: cowsay
Successfully installed cowsay-6.1

run_code(session_id, 'import cowsay; cowsay.tux("Installed from S3 wheels!")')
stop_session(session_id)

_________________________
| Installed from S3 wheels! |
  =========================
                              \
                               \
                                \
                                 .--.
                                |o_o |
                                |:_/ |
                               //   \ \
                              (|     | )
                             /'\_   _/`\
                             \___)=(___/

# 事前準備（ローカルまたはCI）
pip download httpx --dest ./httpx_wheels/ --only-binary=:all:
aws s3 sync ./httpx_wheels/ s3://${BUCKET_NAME}/httpx_wheels/

session_id = start_session()
 
# S3 から wheel をダウンロード
run_cmd(session_id, f"mkdir -p /tmp/wheels && aws s3 cp s3://{BUCKET_NAME}/httpx_wheels/ /tmp/wheels/ --recursive")
 
# pip を使わず wheel を展開して直接 import
run_code(session_id, """
import sys, os, zipfile, importlib
 
whl_path = "/tmp/wheels/h11-0.16.0-py3-none-any.whl"
extract_dir = "/tmp/lib_direct"
os.makedirs(extract_dir, exist_ok=True)
 
with zipfile.ZipFile(whl_path, 'r') as z:
    z.extractall(extract_dir)
 
sys.path.insert(0, extract_dir)
 
# プリインストール版ではなく展開版を確実にロード
if 'h11' in sys.modules:
    del sys.modules['h11']
h11 = importlib.import_module('h11')
print(f"h11 version: {h11.__version__}")
""")
stop_session(session_id)

h11 version: 0.16.0

# セッション1: cowsay をインストール → 動作確認
session1 = start_session()
run_cmd(session1, "pip install cowsay")
run_code(session1, "import cowsay; print('Session 1: OK')")
stop_session(session1)
 
# セッション2: 新規セッションで import を試みる
session2 = start_session()
run_code(session2, """
try:
    import cowsay
    print(f"cowsay available: YES")
except ImportError:
    print("cowsay available: NO (not persisted)")
""")
stop_session(session2)

cowsay in new session: NO (not persisted)

# Code Interpreter 削除
aws bedrock-agentcore-control delete-code-interpreter \
  --code-interpreter-id "$CI_ID" --region us-east-1
 
# S3 バケット削除
aws s3 rb "s3://${BUCKET_NAME}" --force --region us-east-1
 
# IAM ロール削除（ポリシーを先に外す）
aws iam delete-role-policy --role-name AgentCoreCITestRole --policy-name S3Access
aws iam delete-role --role-name AgentCoreCITestRole