Bedrock AgentCore RuntimeのInvokeAgentRuntimeCommand APIでエージェントセッション内からシェルコマンドを実行する

# main.py
import json
import sys
 
def handle_invoke(event):
    user_input = event.get("input", {}).get("text", "")
    return {
        "output": {
            "text": f"Received: {user_input}. This is a minimal test agent."
        }
    }
 
def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            request = json.loads(line)
            response = handle_invoke(request)
            print(json.dumps(response), flush=True)
        except json.JSONDecodeError:
            print(json.dumps({"error": "Invalid JSON"}), flush=True)
 
if __name__ == "__main__":
    main()

# 変数設定
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
BUCKET_NAME="agentcore-test-shell-cmd-${ACCOUNT_ID}"
REGION="us-west-2"
 
# S3バケット作成とコードアップロード
zip agent.zip main.py
aws s3 mb "s3://${BUCKET_NAME}" --region "$REGION"
aws s3 cp agent.zip "s3://${BUCKET_NAME}/agent.zip"
 
# IAMロール作成（AgentCoreサービスへの信頼ポリシー）
aws iam create-role \
  --role-name AgentCoreRuntimeTestRole \
  --assume-role-policy-document '{
    "Version": "2012-10-17",
    "Statement": [{
      "Effect": "Allow",
      "Principal": {"Service": "bedrock-agentcore.amazonaws.com"},
      "Action": "sts:AssumeRole"
    }]
  }'
 
# 必要なポリシーをアタッチ
aws iam attach-role-policy \
  --role-name AgentCoreRuntimeTestRole \
  --policy-arn arn:aws:iam::aws:policy/AmazonBedrockFullAccess
 
aws iam put-role-policy \
  --role-name AgentCoreRuntimeTestRole \
  --policy-name S3Access \
  --policy-document "{
    \"Version\": \"2012-10-17\",
    \"Statement\": [{
      \"Effect\": \"Allow\",
      \"Action\": [\"s3:GetObject\", \"s3:ListBucket\"],
      \"Resource\": [
        \"arn:aws:s3:::${BUCKET_NAME}\",
        \"arn:aws:s3:::${BUCKET_NAME}/*\"
      ]
    }]
  }"
 
# ランタイム作成
aws bedrock-agentcore-control create-agent-runtime \
  --region "$REGION" \
  --agent-runtime-name shell_cmd_test_agent \
  --role-arn "arn:aws:iam::${ACCOUNT_ID}:role/AgentCoreRuntimeTestRole" \
  --agent-runtime-artifact "{
    \"codeConfiguration\": {
      \"code\": {\"s3\": {\"bucket\": \"${BUCKET_NAME}\", \"prefix\": \"agent.zip\"}},
      \"runtime\": \"PYTHON_3_13\",
      \"entryPoint\": [\"main.py\"]
    }
  }" \
  --network-configuration '{"networkMode": "PUBLIC"}'
# → レスポンスの agentRuntimeId を控える
 
RUNTIME_ID="shell_cmd_test_agent-XXXXXXXXXX"  # 実際のIDに置き換え
 
# エンドポイント作成＋READYまでポーリング
aws bedrock-agentcore-control create-agent-runtime-endpoint \
  --region "$REGION" \
  --agent-runtime-id "$RUNTIME_ID" \
  --name shell_cmd_test_endpoint
 
while true; do
  STATUS=$(aws bedrock-agentcore-control get-agent-runtime-endpoint \
    --region "$REGION" \
    --agent-runtime-id "$RUNTIME_ID" \
    --endpoint-name shell_cmd_test_endpoint \
    --query 'status' --output text)
  echo "Endpoint status: $STATUS"
  [ "$STATUS" = "READY" ] && break
  sleep 10
done

import boto3, sys, uuid
 
client = boto3.client('bedrock-agentcore', region_name='us-west-2')
RUNTIME_ARN = "arn:aws:bedrock-agentcore:us-west-2:ACCOUNT_ID:runtime/RUNTIME_ID"
SESSION_ID = str(uuid.uuid4()) + "-" + str(uuid.uuid4())[:8]  # 45文字
 
def run_command(command, timeout=30, session_id=None):
    """コマンドを実行し、EventStreamを処理するヘルパー関数。"""
    response = client.invoke_agent_runtime_command(
        agentRuntimeArn=RUNTIME_ARN,
        runtimeSessionId=session_id or SESSION_ID,
        qualifier='DEFAULT',
        contentType='application/json',
        accept='application/vnd.amazon.eventstream',
        body={'command': command, 'timeout': timeout}
    )
    for event in response.get('stream', []):
        if 'chunk' in event:
            chunk = event['chunk']
            if 'contentStart' in chunk:
                print("[contentStart] Command execution started")
            if 'contentDelta' in chunk:
                delta = chunk['contentDelta']
                if delta.get('stdout'):
                    print(f"[stdout] {delta['stdout']}", end='')
                if delta.get('stderr'):
                    print(f"[stderr] {delta['stderr']}", end='', file=sys.stderr)
            if 'contentStop' in chunk:
                stop = chunk['contentStop']
                print(f"[contentStop] Exit code: {stop.get('exitCode')}, "
                      f"Status: {stop.get('status')}")

run_command('/bin/bash -c "echo Hello from AgentCore Runtime"')

[contentStart] Command execution started
[stdout] Hello from AgentCore Runtime
[contentStop] Exit code: 0, Status: COMPLETED

Linux localhost 6.1.158-15.288.amzn2023.aarch64 (Amazon Linux 2023ベース)
Python 3.13.9
User: root
Working directory: /

# APIコール1: ファイル作成
run_command('/bin/bash -c "mkdir -p /tmp/test_dir && echo \'file content\' > /tmp/test_dir/test.txt"')
 
# APIコール2（同一セッション）: ファイル確認
run_command('/bin/bash -c "cat /tmp/test_dir/test.txt"')
# → [stdout] file content

new_session = str(uuid.uuid4()) + "-" + str(uuid.uuid4())[:8]
run_command(
    '/bin/bash -c "cat /tmp/test_dir/test.txt 2>&1 || echo File NOT found"',
    session_id=new_session
)
# → [stdout] File NOT found

[stderr] ls: cannot access '/nonexistent_path': No such file or directory
[contentStop] Exit code: 2, Status: COMPLETED

run_command('/bin/bash -c "echo start && sleep 5 && echo end"', timeout=1)

[stdout] start
[contentStop] Exit code: -1, Status: TIMED_OUT

cmd_A: exit=0, time=3.51s, output=A_done
cmd_B: exit=0, time=3.51s, output=B_done
cmd_C: exit=0, time=3.51s, output=C_done
Total wall time: 3.51s