Amazon Bedrock AgentCore harness をゼロコードで実機検証する

# GA 直後の API は boto3 1.43.x 以降が必要（システムの 1.42.x は create-harness 未対応）
python3 -m venv venv && . venv/bin/activate
pip install -q --upgrade boto3   # => 1.43.33

REGION=us-west-2; ACCOUNT=$(aws sts get-caller-identity --query Account --output text)
ROLE=agentcore-harness-verify-role
cat > trust.json <<JSON
{ "Version":"2012-10-17","Statement":[{"Effect":"Allow",
  "Principal":{"Service":"bedrock-agentcore.amazonaws.com"},"Action":"sts:AssumeRole",
  "Condition":{"StringEquals":{"aws:SourceAccount":"$ACCOUNT"},
    "ArnLike":{"aws:SourceArn":"arn:aws:bedrock-agentcore:$REGION:$ACCOUNT:*"}}}]}
JSON
aws iam create-role --role-name $ROLE --assume-role-policy-document file://trust.json
 
cat > perms.json <<JSON
{ "Version":"2012-10-17","Statement":[
  {"Sid":"Bedrock","Effect":"Allow",
   "Action":["bedrock:InvokeModel","bedrock:InvokeModelWithResponseStream"],
   "Resource":["arn:aws:bedrock:*::foundation-model/*","arn:aws:bedrock:$REGION:$ACCOUNT:*"]},
  {"Sid":"XRay","Effect":"Allow",
   "Action":["xray:PutTraceSegments","xray:PutTelemetryRecords","xray:GetSamplingRules","xray:GetSamplingTargets"],"Resource":"*"},
  {"Sid":"Logs","Effect":"Allow",
   "Action":["logs:CreateLogGroup","logs:CreateLogStream","logs:PutLogEvents","logs:DescribeLogStreams","logs:DescribeLogGroups"],
   "Resource":"arn:aws:logs:$REGION:$ACCOUNT:log-group:*"},
  {"Sid":"Metrics","Effect":"Allow","Action":"cloudwatch:PutMetricData","Resource":"*",
   "Condition":{"StringEquals":{"cloudwatch:namespace":"bedrock-agentcore"}}},
  {"Sid":"WorkloadIdentity","Effect":"Allow",
   "Action":["bedrock-agentcore:GetWorkloadAccessToken","bedrock-agentcore:GetWorkloadAccessTokenForJWT"],
   "Resource":["arn:aws:bedrock-agentcore:$REGION:$ACCOUNT:workload-identity-directory/default",
     "arn:aws:bedrock-agentcore:$REGION:$ACCOUNT:workload-identity-directory/default/workload-identity/*"]},
  {"Sid":"Memory","Effect":"Allow",
   "Action":["bedrock-agentcore:CreateEvent","bedrock-agentcore:ListEvents","bedrock-agentcore:RetrieveMemoryRecords",
     "bedrock-agentcore:ListMemoryRecords","bedrock-agentcore:GetMemoryRecord","bedrock-agentcore:GetMemory"],
   "Resource":"arn:aws:bedrock-agentcore:$REGION:$ACCOUNT:memory/*"}]}
JSON
aws iam put-role-policy --role-name $ROLE --policy-name harness-verify-perms \
  --policy-document file://perms.json

import boto3, time, uuid
 
REGION = "us-west-2"
ACCOUNT = boto3.client("sts").get_caller_identity()["Account"]
ROLE_ARN = f"arn:aws:iam::{ACCOUNT}:role/agentcore-harness-verify-role"
ctrl = boto3.client("bedrock-agentcore-control", REGION)
data = boto3.client("bedrock-agentcore", REGION)
 
# CreateHarness: harnessName と executionRoleArn だけ（モデル/メモリ/ツールは指定しない）
h = ctrl.create_harness(harnessName="verifyZeroCode", executionRoleArn=ROLE_ARN)["harness"]
hid, harness_arn = h["harnessId"], h["arn"]
 
# GetHarness を READY までポーリング（作成→READY は createdAt→updatedAt から算出）
while True:
    g = ctrl.get_harness(harnessId=hid)["harness"]
    if g["status"] in ("READY", "CREATE_FAILED"):
        break
    time.sleep(3)
print("create->ready:", (g["updatedAt"] - g["createdAt"]).total_seconds(), "s")
print("model :", g["model"])         # 自動既定: global.anthropic.claude-sonnet-4-6
print("memory:", g["memory"])        # 自動生成された managedMemoryConfiguration の ARN
print("limits:", g["maxIterations"], g["timeoutSeconds"], g["truncation"])
 
# 初回 InvokeHarness（既定モデル）。runtimeSessionId は 33 文字以上が必須
sid = (uuid.uuid4().hex * 2)[:40]
t = time.time(); first = None; chunks = []
r = data.invoke_harness(harnessArn=harness_arn, runtimeSessionId=sid,
    messages=[{"role": "user", "content": [{"text": "In one sentence, what model are you and who built you?"}]}])
for ev in r["stream"]:
    if "contentBlockDelta" in ev:
        if first is None:
            first = time.time() - t
        chunks.append(ev["contentBlockDelta"]["delta"].get("text", ""))
    elif "metadata" in ev:
        print("usage:", ev["metadata"]["usage"])
print(f"first token {first:.2f}s / full {time.time()-t:.2f}s")
print("".join(chunks))
 
# 自動生成された Memory リソースを直接確認
mem_id = g["memory"]["managedMemoryConfiguration"]["arn"].split("/")[-1]
m = ctrl.get_memory(memoryId=mem_id)["memory"]
print("memory status:", m["status"], "expiry:", m["eventExpiryDuration"],
      "strategies:", [s["type"] for s in m["strategies"]])

import boto3, uuid
 
REGION = "us-west-2"
ctrl = boto3.client("bedrock-agentcore-control", REGION)
data = boto3.client("bedrock-agentcore", REGION)
harness_arn = [x for x in ctrl.list_harnesses()["harnesses"]
               if x["harnessName"] == "verifyZeroCode"][0]["arn"]
 
def invoke(session_id, text, model=None):
    kw = dict(harnessArn=harness_arn, runtimeSessionId=session_id,
              messages=[{"role": "user", "content": [{"text": text}]}])
    if model:
        kw["model"] = model
    out = []
    for ev in data.invoke_harness(**kw)["stream"]:
        if "contentBlockDelta" in ev:
            out.append(ev["contentBlockDelta"]["delta"].get("text", ""))
    return "".join(out)
 
# --- 同一セッションでプロバイダー切替 ---
sid = (uuid.uuid4().hex * 2)[:40]
print("T1:", invoke(sid, "Remember: codename is BLUEFALCON, launch number is 47. Confirm you noted them."))
print("T2:", invoke(sid, "Repeat my codename and launch number from earlier, and say which company built you.",
                    model={"bedrockModelConfig": {"modelId": "openai.gpt-oss-120b-1:0"}}))
 
# --- 対照実験（毎回クリーンな新規セッション）---
Q = "What model and company made you? One short sentence."
print("default:", invoke((uuid.uuid4().hex * 2)[:40], Q))                  # => Claude / Anthropic
print("gpt-oss:", invoke((uuid.uuid4().hex * 2)[:40], Q,
                         {"bedrockModelConfig": {"modelId": "openai.gpt-oss-120b-1:0"}}))  # => OpenAI
# 存在しない modelId を渡すと、ストリーム反復中に ConverseStream の
# ValidationException（EventStreamError）が送出される

ctrl = boto3.client("bedrock-agentcore-control", "us-west-2")
iam = boto3.client("iam")
hid = [h["harnessId"] for h in ctrl.list_harnesses()["harnesses"]
       if h["harnessName"] == "verifyZeroCode"][0]
ctrl.delete_harness(harnessId=hid)  # 自動生成 Memory は既定でカスケード削除
iam.delete_role_policy(RoleName="agentcore-harness-verify-role", PolicyName="harness-verify-perms")
iam.delete_role(RoleName="agentcore-harness-verify-role")