실행 역할과 함께 터미널 명령 사용

Amazon S3에서 파일을 업로드/다운로드하는 실행 역할로 사용자 지정 코드 해석기 도구를 생성할 수 있습니다. 이렇게 하면 코드가 S3 버킷과 상호 작용하여 데이터를 저장하고 검색할 수 있습니다.

사전 조건

S3 액세스 권한이 있는 사용자 지정 코드 인터프리터를 생성하기 전에 다음을 수행해야 합니다.

샘플 Python 코드

boto3(Python용 SDK)를 사용하여 S3 통합을 구현할 수 있습니다.AWS 다음 예제에서는 boto3를 사용하여 Amazon S3에 파일을 업로드하거나 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")

이 예제에서는 다음 작업을 하는 방법을 보여줍니다.