Menggunakan Perintah Terminal dengan peran eksekusi

Anda dapat membuat alat Penerjemah Kode khusus dengan peran eksekusi ke upload/download file dari Amazon S3. Ini memungkinkan kode Anda berinteraksi dengan bucket S3 untuk menyimpan dan mengambil data.

Prasyarat

Sebelum membuat Penerjemah Kode kustom dengan akses S3, Anda perlu:

Contoh kode Python

Anda dapat menerapkan integrasi S3 menggunakan boto3 (AWS SDK untuk Python). Contoh berikut menggunakan boto3 untuk membuat Penerjemah Kode kustom dengan peran eksekusi yang dapat mengunggah file ke atau mengunduh file dari Amazon S3.

import boto3
import json
import time

REGION = "<Region>"
CP_ENDPOINT_URL = f"https://bedrock-agentcore-control.{REGION}.amazonaws.com"
DP_ENDPOINT_URL = f"https://bedrock-agentcore.{REGION}.amazonaws.com"

# Update the accountId to reflect the correct S3 path.
S3_BUCKET_NAME = "DOC-EXAMPLE-BUCKET"

bedrock_agentcore_control_client = boto3.client(
    'bedrock-agentcore-control',
    region_name=REGION,
    endpoint_url=CP_ENDPOINT_URL
)
bedrock_agentcore_client = boto3.client(
    'bedrock-agentcore',
    region_name=REGION,
    endpoint_url=DP_ENDPOINT_URL
)

unique_name = f"s3InteractionEnv_{int(time.time())}"
create_response = bedrock_agentcore_control_client.create_code_interpreter(
    name=unique_name,
    description="Combined test code sandbox",
    executionRoleArn="arn:aws:iam::123456789012:role/S3InteractionRole",
    networkConfiguration={
        "networkMode": "SANDBOX"
    }
)
code_interpreter_id = create_response['codeInterpreterId']
print(f"Created custom interpreter ID: {code_interpreter_id}")

session_response = bedrock_agentcore_client.start_code_interpreter_session(
    codeInterpreterIdentifier=code_interpreter_id,
    name="combined-test-session",
    sessionTimeoutSeconds=1800
)
session_id = session_response['sessionId']
print(f"Created session ID: {session_id}")

print(f"Downloading CSV generation script from S3")
command_to_execute = f"aws s3 cp s3://{S3_BUCKET_NAME}/generate_csv.py ."
response = bedrock_agentcore_client.invoke_code_interpreter(
    codeInterpreterIdentifier=code_interpreter_id,
    sessionId=session_id,
    name="executeCommand",
    arguments={
        "command": command_to_execute
    }
)

for event in response["stream"]:
    print(json.dumps(event["result"], default=str, indent=2))

print(f"Executing the CSV generation script")
response = bedrock_agentcore_client.invoke_code_interpreter(
    codeInterpreterIdentifier=code_interpreter_id,
    sessionId=session_id,
    name="executeCommand",
    arguments={
        "command": "python generate_csv.py 5 10"
    }
)

for event in response["stream"]:
    print(json.dumps(event["result"], default=str, indent=2))

print(f"Uploading generated artifact to S3")
command_to_execute = f"aws s3 cp generated_data.csv s3://{S3_BUCKET_NAME}/output_artifacts/"
response = bedrock_agentcore_client.invoke_code_interpreter(
    codeInterpreterIdentifier=code_interpreter_id,
    sessionId=session_id,
    name="executeCommand",
    arguments={
        "command": command_to_execute
    }
)

for event in response["stream"]:
    print(json.dumps(event["result"], default=str, indent=2))

print(f"Stopping the code interpreter session")
stop_response = bedrock_agentcore_client.stop_code_interpreter_session(
    codeInterpreterIdentifier=code_interpreter_id,
    sessionId=session_id
)

print(f"Deleting the code interpreter")
delete_response = bedrock_agentcore_control_client.delete_code_interpreter(
    codeInterpreterId=code_interpreter_id
)
print(f"Code interpreter status from response: {delete_response['status']}")
print(f"Clean up completed, script run successful")

Contoh ini menunjukkan kepada Anda cara: