AWS Bedrock

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "bedrock:ListAgents",
        "bedrock:GetAgent",
        "bedrock:ListTagsForResource"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:FilterLogEvents",
        "logs:GetLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:log-group:<your-cloudtrail-log-group>:*"
    }
  ]
}

AWS_REGION=ap-southeast-1
BEDROCK_CLOUDTRAIL_LOG_GROUP=/aws/cloudtrail/your-trail

import httpx, threading, uuid

def invoke_bedrock_agent(agent_id, session_id, input_text):
    job_id = str(uuid.uuid4())
    
    response = bedrock_agent_runtime.invoke_agent(
        agentId=agent_id,
        sessionId=session_id,
        inputText=input_text,
    )
    output = "".join(chunk["chunk"]["bytes"].decode()
                     for chunk in response["completion"]
                     if "chunk" in chunk)
    
    # Optional: add precise checks
    threading.Thread(target=lambda: httpx.post(
        f"{MUSTER_URL}/api/v1/jobs/{job_id}/quality",
        json={"agent_id": "bedrock-" + agent_id, "checks": [
            {"check_id": "output_not_empty", "severity": "HIGH", "passed": bool(output)},
        ]},
        timeout=2.0,
    ), daemon=True).start()
    
    return output