セッションにファイルを書き込む

Code Interpreter を使用して、サンドボックス環境でファイルを読み書きできます。これにより、データファイルをアップロードし、コードで処理して、結果を取得できます。

依存関係のインストール

次のコマンドを実行して、必要なパッケージをインストールします。

pip install bedrock-agentcore

ファイルツールを使用してコードとデータをアップロードする

次の Python コードは、Code Interpreter セッションにファイルをアップロードし、それらのファイルを処理するコードを実行する方法を示しています。必要なファイルは data.csvであり、このパッケージでstats.py利用できます。

# file_mgmt_ci_agent.py

from bedrock_agentcore.tools.code_interpreter_client import CodeInterpreter
import json
from typing import Dict, Any, List

#Configure and Start the code interpreter session
code_client = CodeInterpreter('<Region>')
code_client.start()

#read the content of the sample data file
data_file = "data.csv"

try:
    with open(data_file, 'r', encoding='utf-8') as data_file_content:
        data_file_content = data_file_content.read()
    #print(data_file_content)
except FileNotFoundError:
    print(f"Error: The file '{data_file}' was not found.")
except Exception as e:
    print(f"An error occurred: {e}")

#read the content of the python script to analyze the sample file
code_file = "stats.py"
try:
    with open(code_file, 'r', encoding='utf-8') as code_file_content:
        code_file_content = code_file_content.read()
    #print(code_file_content)
except FileNotFoundError:
    print(f"Error: The file '{code_file}' was not found.")
except Exception as e:
    print(f"An error occurred: {e}")

files_to_create = [
                {
                    "path": "data.csv",
                    "text": data_file_content
                },
                {
                    "path": "stats.py",
                    "text": code_file_content
                }]

#define the method to call the invoke API
def call_tool(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:

    response = code_client.invoke(tool_name, arguments)
    for event in response["stream"]:
        return json.dumps(event["result"], indent=2)

#write the sample data and analysis script into the code interpreter session
writing_files = call_tool("writeFiles", {"content": files_to_create})
print(f"writing files: {writing_files}")

#List and validate that the files were written successfully
listing_files = call_tool("listFiles", {"path": ""})
print(f"listing files: {listing_files}")

#Run the python script to analyze the sample data file
execute_code = call_tool("executeCode", {
                "code": files_to_create[1]['text'],
                "language": "python",
                "clearContext": True})
print(f"code execution result: {execute_code}")

#Clean up and stop the code interpreter session
code_client.stop()