Uso de comandos de terminal con una función de ejecución

Puede crear una herramienta de interpretación de código personalizada con una función de ejecución para upload/download los archivos de Amazon S3. Esto permite que el código interactúe con los buckets de S3 para almacenar y recuperar datos.

Requisitos previos

Antes de crear un intérprete de código personalizado con acceso a S3, debe:

Código de muestra de python

Puede implementar la integración de S3 mediante boto3 (AWS SDK para Python). En el siguiente ejemplo, se utiliza boto3 para crear un intérprete de código personalizado con una función de ejecución que puede cargar o descargar archivos desde 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")

En este ejemplo puede ver cómo: