Amazon Bedrock AgentCore observability guide

pip install wandb boto3 strands-agents strands-agents-tools weave 

import wandb
import weave
import boto3
import json
from datetime import datetime
from typing import Dict, Any, Optional

# Login to W&B (you'll be prompted for your API key if not already logged in)
wandb.login()

# Initialize Weave with a project name
weave.init("bedrock-agent-observability")

mkdir my-custom-agent && cd my-custom-agent
uv init --python 3.11
uv add fastapi 'uvicorn[standard]' pydantic httpx strands-agents

export REGION=us-east-1
export ACCOUNT_ID=your_aws_account_id
export AGENT_NAME=myagent
export ECR_REPO=my-agent-repo
export ROLE_NAME=AgentCoreRuntime-ExecutionRole-${AGENT_NAME}
export IMAGE=${ACCOUNT_ID}.dkr.ecr.${REGION}.amazonaws.com/${ECR_REPO}:latest
export EXECUTION_ROLE_ARN=arn:aws:iam::${ACCOUNT_ID}:role/${ROLE_NAME}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AssumeRolePolicy",
      "Effect": "Allow",
      "Principal": {
        "Service": "bedrock-agentcore.amazonaws.com"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "aws:SourceAccount": "YOUR_AWS_ACCOUNT_ID"
        },
        "ArnLike": {
          "aws:SourceArn": "arn:aws:bedrock-agentcore:YOUR_REGION:YOUR_AWS_ACCOUNT_ID:*"
        }
      }
    }
  ]
}

aws iam create-role \
  --role-name $ROLE_NAME \
  --assume-role-policy-document file://trust-policy.json \
  --region $REGION

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ECRImageAccess",
      "Effect": "Allow",
      "Action": [
        "ecr:BatchGetImage",
        "ecr:GetDownloadUrlForLayer"
      ],
      "Resource": [
        "arn:aws:ecr:YOUR_REGION:YOUR_AWS_ACCOUNT_ID:repository/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:DescribeLogStreams",
        "logs:CreateLogGroup"
      ],
      "Resource": [
        "arn:aws:logs:YOUR_REGION:YOUR_AWS_ACCOUNT_ID:log-group:/aws/bedrock-agentcore/runtimes/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": [ "logs:DescribeLogGroups" ],
      "Resource": [ "arn:aws:logs:YOUR_REGION:YOUR_AWS_ACCOUNT_ID:log-group:*" ]
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": [
        "arn:aws:logs:YOUR_REGION:YOUR_AWS_ACCOUNT_ID:log-group:/aws/bedrock-agentcore/runtimes/*:log-stream:*"
      ]
    },
    {
      "Sid": "ECRTokenAccess",
      "Effect": "Allow",
      "Action": [ "ecr:GetAuthorizationToken" ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "xray:PutTraceSegments",
        "xray:PutTelemetryRecords",
        "xray:GetSamplingRules",
        "xray:GetSamplingTargets"
      ],
      "Resource": [ "*" ]
    },
    {
      "Effect": "Allow",
      "Resource": "*",
      "Action": "cloudwatch:PutMetricData",
      "Condition": {
        "StringEquals": {
          "cloudwatch:namespace": "bedrock-agentcore"
        }
      }
    },
    {
      "Sid": "GetAgentAccessToken",
      "Effect": "Allow",
      "Action": [
        "bedrock-agentcore:GetWorkloadAccessToken",
        "bedrock-agentcore:GetWorkloadAccessTokenForJWT",
        "bedrock-agentcore:GetWorkloadAccessTokenForUserId"
      ],
      "Resource": [
        "arn:aws:bedrock-agentcore:YOUR_REGION:YOUR_AWS_ACCOUNT_ID:workload-identity-directory/default",
        "arn:aws:bedrock-agentcore:YOUR_REGION:YOUR_AWS_ACCOUNT_ID:workload-identity-directory/default/workload-identity/${AGENT_NAME}-*"
      ]
    },
    {
      "Sid": "BedrockModelInvocation",
      "Effect": "Allow",
      "Action": [
        "bedrock:InvokeModel",
        "bedrock:InvokeModelWithResponseStream"
      ],
      "Resource": [
        "arn:aws:bedrock:*::foundation-model/*",
        "arn:aws:bedrock:YOUR_REGION:YOUR_AWS_ACCOUNT_ID:*"
      ]
    }
  ]
}

aws iam put-role-policy \
  --role-name $ROLE_NAME \
  --policy-name AgentCoreRuntimePolicy \
  --policy-document file://policy.json \
  --region $REGION

aws ecr create-repository --repository-name $ECR_REPO --region $REGION

aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin $ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com 

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any
from datetime import datetime

from strands import Agent, tool
from strands.models import BedrockModel

@tool
def word_count(text: str) -> int:
    return len(text.split())

@tool
def reverse(text: str) -> str:
    return text[::-1]

bedrock_model = BedrockModel(
    model_id="anthropic.claude-3-5-sonnet-20240620-v1:0",
    temperature=0.3,
    streaming=False,
    region_name="us-east-1"  # IMPORTANT: set this to your exact Claude-enabled region
)

strands_agent = Agent(
    model=bedrock_model,
    tools=[word_count, reverse],
    system_prompt="You are a helpful assistant who uses tools when they help."
)

app = FastAPI(title="Strands Agent Server", version="1.0.0")

class InvocationRequest(BaseModel):
    input: Dict[str, Any]

class InvocationResponse(BaseModel):
    output: Dict[str, Any]

@app.post("/invocations", response_model=InvocationResponse)
async def invoke_agent(request: InvocationRequest):
    try:
        user_message = request.input.get("prompt", "")
        if not user_message:
            raise HTTPException(status_code=400, detail="No prompt found in input. Please provide a 'prompt' key in the input.")
        result = strands_agent(user_message)
        response = {
            "message": result.message,
            "timestamp": datetime.utcnow().isoformat(),
            "model": bedrock_model.config["model_id"],
        }
        return InvocationResponse(output=response)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent processing failed: {str(e)}")

@app.get("/ping")
async def ping():
    return {"status": "healthy"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)


FROM --platform=linux/arm64 ghcr.io/astral-sh/uv:python3.11-bookworm-slim
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-cache
COPY agent.py ./
EXPOSE 8080
CMD ["uv", "run", "uvicorn", "agent:app", "--host", "0.0.0.0", "--port", "8080"]

docker buildx build --platform linux/arm64 -t $IMAGE --push .

aws ecr describe-images --repository-name $ECR_REPO --region $REGION

import boto3
# Set your account info here
ACCOUNT_ID = “your”_id”  # <-- Replace with your AWS account ID
ROLE_ARN = “your agent ARN”  # <-- Replace with your role ARN

client = boto3.client('bedrock-agentcore-control', region_name='us-east-1')

response = client.create_agent_runtime(
    agentRuntimeName='strands_agent',
    agentRuntimeArtifact={
        'containerConfiguration': {
            'containerUri': f'{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/my-strands-agent:latest'
        }
    },
    networkConfiguration={
        "networkMode": "PUBLIC"
    },
    roleArn=ROLE_ARN
)

print("Agent Runtime created successfully!")
print(f"Agent Runtime ARN: {response['agentRuntimeArn']}")
print(f"Status: {response['status']}")

import boto3
import json
import random
import string
import weave

# 1. Initialize Weave Tracing (use your project name)
weave.init("bedrock-agentcore-demo")

# 2. Trace the core agent call as a "Call" with Weave.
@weave.op
def invoke_bedrock_agent(prompt: str):
    agent_core_client = boto3.client('bedrock-agentcore', region_name='us-east-1')
    agent_runtime_arn = "Your_Agent_Runtime_ARN_Here"  # <-- Replace with your actual agent runtime ARN
    

    # Build session id for Bedrock API (randomized, 40 chars)
    random_part = ''.join(random.choices(string.ascii_letters + string.digits, k=40))
    runtime_session_id = f"strandsagent-{random_part}"

    payload = json.dumps({
        "input": {"prompt": prompt}
    })

    response = agent_core_client.invoke_agent_runtime(
        agentRuntimeArn=agent_runtime_arn,
        runtimeSessionId=runtime_session_id,
        payload=payload,
        qualifier="DEFAULT"
    )

    response_body = response['response'].read()
    try:
        response_data = json.loads(response_body)
    except Exception:
        response_data = response_body  # fallback if not JSON

    return response_data

if __name__ == "__main__":
    # This call is tracked in Weave!
    resp = invoke_bedrock_agent(
        "how many words are in the string 'ML is cool' ?"
    )

    try:
        text = resp["output"]["message"]["content"][0]["text"]
        print(text)
    except Exception as e:
        print("Could not extract summary text:", e)