AgentCore Code Interpreter を boto3 で直接操作して Excel レポートを生成してみた

uv init code-interpreter-demo
cd code-interpreter-demo
uv add bedrock-agentcore boto3

import boto3

# 実行するコード
code_to_execute = """
print("Hello World!!!")
"""

# クライアントの作成
client = boto3.client(
    "bedrock-agentcore",
    region_name="ap-northeast-1"
)

# セッションの開始
session_response = client.start_code_interpreter_session(
    codeInterpreterIdentifier="aws.codeinterpreter.v1",
    name="my-code-session",
    sessionTimeoutSeconds=900
)
session_id = session_response["sessionId"]

print(f"Started session: {session_id}\n\n")

try:
    # セッションIDを指定してコードの実行
    execute_response = client.invoke_code_interpreter(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id,
        name="executeCode",
        arguments={
            "language": "python",
            "code": code_to_execute
        }
    )

    # ストリームレスポンスから結果の取得
    for event in execute_response['stream']:
        if 'result' in event:
            result = event['result']
            if 'content' in result:
                for content_item in result['content']:
                    if content_item['type'] == 'text':
                        print(content_item['text'])

finally:
    # セッションの終了
    client.stop_code_interpreter_session(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id
    )
    print(f"\n\nStopped session: {session_id}")

import boto3

client = boto3.client(
    "bedrock-agentcore",
    region_name="ap-northeast-1"
)

session_response = client.start_code_interpreter_session(
    codeInterpreterIdentifier="aws.codeinterpreter.v1",
    name="analysis-session",
    sessionTimeoutSeconds=900
)
session_id = session_response["sessionId"]
print(f"Started session: {session_id}\n")

# CSVファイルの内容
csv_content = """商品,単価,数量
りんご,150,10
みかん,100,20
バナナ,200,5
ぶどう,350,8
もも,400,3"""

# 分析用コード
analysis_code = """
import pandas as pd

df = pd.read_csv('sales_data.csv')
df['金額'] = df['単価'] * df['数量']

print(df.to_string(index=False))
print(f'\\n合計金額: {df["金額"].sum():,}円')
print(f'平均単価: {df["単価"].mean():.0f}円')
"""

try:
    # 1. CSVファイルをアップロード
    client.invoke_code_interpreter(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id,
        name="writeFiles",
        arguments={
            "content": [
                {
                    "path": "sales_data.csv",
                    "text": csv_content
                }
            ]
        }
    )
    print("Uploaded: sales_data.csv")

    # 2. 分析コードを実行
    execute_response = client.invoke_code_interpreter(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id,
        name="executeCode",
        arguments={
            "language": "python",
            "code": analysis_code
        }
    )

    for event in execute_response['stream']:
        if 'result' in event:
            result = event['result']
            if 'content' in result:
                for content_item in result['content']:
                    if content_item['type'] == 'text':
                        print(content_item['text'])

finally:
    client.stop_code_interpreter_session(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id
    )
    print(f"\nStopped session: {session_id}")

import boto3

client = boto3.client(
    "bedrock-agentcore",
    region_name="ap-northeast-1"
)

session_response = client.start_code_interpreter_session(
    codeInterpreterIdentifier="aws.codeinterpreter.v1",
    name="excel-session",
    sessionTimeoutSeconds=900
)
session_id = session_response["sessionId"]
print(f"Started session: {session_id}\n")

# 売上CSVデータ
csv_content = """商品,単価,数量
りんご,150,10
みかん,100,20
バナナ,200,5
ぶどう,350,8
もも,400,3"""

# Excel生成コード（openpyxl + グラフ付き）
excel_code = """
import pandas as pd
from openpyxl import Workbook
from openpyxl.chart import BarChart, Reference
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side

# CSVを読み込んで集計
df = pd.read_csv('sales_data.csv')
df['金額'] = df['単価'] * df['数量']

# ワークブック作成
wb = Workbook()
ws = wb.active
ws.title = '売上レポート'

# ヘッダースタイル
header_font = Font(bold=True, color='FFFFFF', size=12)
header_fill = PatternFill(start_color='4472C4', end_color='4472C4', fill_type='solid')
header_alignment = Alignment(horizontal='center')
thin_border = Border(
    left=Side(style='thin'),
    right=Side(style='thin'),
    top=Side(style='thin'),
    bottom=Side(style='thin')
)

# ヘッダー行
headers = ['商品', '単価', '数量', '金額']
for col, header in enumerate(headers, 1):
    cell = ws.cell(row=1, column=col, value=header)
    cell.font = header_font
    cell.fill = header_fill
    cell.alignment = header_alignment
    cell.border = thin_border

# データ行
for row_idx, row in df.iterrows():
    for col_idx, value in enumerate([row['商品'], row['単価'], row['数量'], row['金額']], 1):
        cell = ws.cell(row=row_idx + 2, column=col_idx, value=value)
        cell.border = thin_border
        if col_idx >= 2:
            cell.number_format = '#,##0'

# 合計行
total_row = len(df) + 2
ws.cell(row=total_row, column=1, value='合計').font = Font(bold=True)
ws.cell(row=total_row, column=4, value=df['金額'].sum()).font = Font(bold=True)
ws.cell(row=total_row, column=4).number_format = '#,##0'
for col in range(1, 5):
    ws.cell(row=total_row, column=col).border = thin_border

# 列幅調整
ws.column_dimensions['A'].width = 12
ws.column_dimensions['B'].width = 10
ws.column_dimensions['C'].width = 10
ws.column_dimensions['D'].width = 12

# 棒グラフ追加
chart = BarChart()
chart.type = 'col'
chart.title = '商品別売上金額'
chart.y_axis.title = '金額（円）'
chart.x_axis.title = '商品'

data = Reference(ws, min_col=4, min_row=1, max_row=len(df) + 1)
categories = Reference(ws, min_col=1, min_row=2, max_row=len(df) + 1)
chart.add_data(data, titles_from_data=True)
chart.set_categories(categories)
chart.shape = 4
chart.width = 15
chart.height = 10

ws.add_chart(chart, 'F2')

# 保存
output_path = 'sales_report.xlsx'
wb.save(output_path)
print(f'Excel ファイルを生成しました: {output_path}')
print(f'\\nレポート内容:')
print(df.to_string(index=False))
print(f'\\n合計金額: {df["金額"].sum():,}円')
"""

try:
    # 1. CSVファイルをアップロード
    client.invoke_code_interpreter(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id,
        name="writeFiles",
        arguments={
            "content": [
                {
                    "path": "sales_data.csv",
                    "text": csv_content
                }
            ]
        }
    )
    print("Uploaded: sales_data.csv")

    # 2. Excel生成コードを実行
    execute_response = client.invoke_code_interpreter(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id,
        name="executeCode",
        arguments={
            "language": "python",
            "code": excel_code
        }
    )

    for event in execute_response['stream']:
        if 'result' in event:
            result = event['result']
            if 'content' in result:
                for content_item in result['content']:
                    if content_item['type'] == 'text':
                        print(content_item['text'])

    # 3. 生成したExcelファイルを取り出す
    read_response = client.invoke_code_interpreter(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id,
        name="readFiles",
        arguments={
            "paths": ["sales_report.xlsx"]
        }
    )

    for event in read_response['stream']:
        if 'result' in event:
            result = event['result']
            if 'content' in result:
                for content_item in result['content']:
                    if content_item['type'] == 'resource':
                        # バイナリファイルは resource.blob に bytes で直接格納されている
                        file_data = content_item['resource']['blob']
                        with open('sales_report.xlsx', 'wb') as f:
                            f.write(file_data)
                        print(f"\nダウンロード完了: sales_report.xlsx ({len(file_data):,} bytes)")

finally:
    client.stop_code_interpreter_session(
        codeInterpreterIdentifier="aws.codeinterpreter.v1",
        sessionId=session_id
    )
    print(f"\nStopped session: {session_id}")