How it works
The Bedrock connector:- Discovers all Bedrock agents via
boto3.list_agents() - Streams invocations by reading
InvokeAgentevents from CloudTrail
Connectors
AWS Bedrock
Automatic monitoring of AWS Bedrock agents via the muster connector.
Documentation Index
Fetch the complete documentation index at: /llms.txt
Use this file to discover all available pages before exploring further.
Automatic monitoring of AWS Bedrock agents via the muster connector.
boto3.list_agents()InvokeAgent events from CloudTrail{
"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