Integrate · Agents Build · aws/agent-toolkit-for-aws · Skills Docs

Integrate · Agents Build · aws/agent-toolkit-for-aws · Skills Docs

Auth type	How the client authenticates	Best for
None (default)	IAM SigV4 signing on the request	Backend services with AWS credentials
AWS_IAM	IAM SigV4 signing on the request	Backend services, Lambda-to-agent calls
CUSTOM_JWT	Bearer token in Authorization header	Web/mobile apps with an identity provider

import boto3
import json
from botocore.exceptions import ClientError

client = boto3.client("bedrock-agentcore", region_name="<REGION>")

try:
    response = client.invoke_agent_runtime(
        agentRuntimeArn="<AGENT_RUNTIME_ARN>",
        qualifier="DEFAULT",  # or a specific version number
        payload=json.dumps({
            "prompt": "Hello, what can you do?"
        }).encode(),
        runtimeSessionId="session-123",  # reuse for multi-turn conversations
    )

    # Handle streaming response — response["response"] is a StreamingBody
    stream = response["response"]
    if hasattr(stream, "iter_lines"):
        for line in stream.iter_lines():
            if line:
                print(line.decode(), end="", flush=True)
    else:
        # Some SDK versions return raw bytes — read all at once
        content = stream.read()
        print(content.decode() if isinstance(content, bytes) else content)

except ClientError as e:
    code = e.response["Error"]["Code"]
    if code == "AccessDeniedException":
        # Missing bedrock-agentcore:InvokeAgentRuntime permission
        raise RuntimeError("Caller lacks InvokeAgentRuntime permission") from e
    elif code == "ValidationException":
        # Wrong ARN, bad payload format, invalid session ID
        raise RuntimeError(f"Invalid request: {e}") from e
    elif code == "ThrottlingException":
        # Retry with exponential backoff
        raise
    else:
        raise

import requests

AGENT_URL = "<ENDPOINT_URL>"
JWT_TOKEN = "<TOKEN_FROM_YOUR_IDP>"

response = requests.post(
    AGENT_URL,
    headers={
        "Authorization": f"Bearer {JWT_TOKEN}",
        "Content-Type": "application/json",
    },
    json={"prompt": "Hello, what can you do?"},
    stream=True,
)

for chunk in response.iter_content(chunk_size=None):
    print(chunk.decode(), end="", flush=True)

import {
  BedrockAgentCoreClient,
  InvokeAgentRuntimeCommand,
} from "@aws-sdk/client-bedrock-agentcore";

const client = new BedrockAgentCoreClient({ region: "<REGION>" });

const response = await client.send(
  new InvokeAgentRuntimeCommand({
    agentRuntimeArn: "<AGENT_RUNTIME_ARN>",
    qualifier: "DEFAULT",
    payload: new TextEncoder().encode(
      JSON.stringify({ prompt: "Hello, what can you do?" })
    ),
    runtimeSessionId: "session-123",
  })
);

// response.response is the streaming body
const decoder = new TextDecoder();
for await (const chunk of response.response) {
  process.stdout.write(decoder.decode(chunk));
}

const AGENT_URL = "<ENDPOINT_URL>";
const JWT_TOKEN = "<TOKEN_FROM_YOUR_IDP>";

const response = await fetch(AGENT_URL, {
  method: "POST",
  headers: {
    Authorization: `Bearer ${JWT_TOKEN}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({ prompt: "Hello, what can you do?" }),
});

const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  process.stdout.write(decoder.decode(value));
}

import uuid

# New conversation
session_id = str(uuid.uuid4())

# First turn
invoke(session_id, "What's the weather in Seattle?")

# Follow-up in same conversation
invoke(session_id, "What about tomorrow?")

# New conversation — new session
new_session_id = str(uuid.uuid4())
invoke(new_session_id, "Different topic entirely")

// Trust policy — who can assume this role
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"AWS": "arn:aws:iam::<CALLER_ACCOUNT_ID>:root"},
    "Action": "sts:AssumeRole",
    "Condition": {
      "StringEquals": {"sts:ExternalId": "<unique-external-id>"}
    }
  }]
}

// Permissions policy — what this role can do
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": "bedrock-agentcore:InvokeAgentRuntime",
    "Resource": "arn:aws:bedrock-agentcore:<REGION>:<AGENT_ACCOUNT_ID>:runtime/<RUNTIME_NAME>-*"
  }]
}

import boto3
import json

# Assume the role in Account B
sts = boto3.client("sts")
assumed = sts.assume_role(
    RoleArn="arn:aws:iam::<AGENT_ACCOUNT_ID>:role/<ROLE_NAME>",
    RoleSessionName="agent-invoker",
    ExternalId="<unique-external-id>",
)
creds = assumed["Credentials"]

# Use the temporary credentials to invoke the runtime
agentcore = boto3.client(
    "bedrock-agentcore",
    region_name="<REGION>",
    aws_access_key_id=creds["AccessKeyId"],
    aws_secret_access_key=creds["SecretAccessKey"],
    aws_session_token=creds["SessionToken"],
)

response = agentcore.invoke_agent_runtime(
    agentRuntimeArn="arn:aws:bedrock-agentcore:<REGION>:<AGENT_ACCOUNT_ID>:runtime/<RUNTIME_NAME>",
    qualifier="DEFAULT",
    payload=json.dumps({"prompt": "hello"}).encode(),
    runtimeSessionId="session-123",
)

import boto3

client = boto3.client("bedrock-agentcore", region_name="<REGION>")

response = client.invoke_agent_runtime_command(
    agentRuntimeArn="<AGENT_RUNTIME_ARN>",
    qualifier="DEFAULT",
    runtimeSessionId="session-123",   # must be an existing session
    command="ls -la /tmp && cat /tmp/status.json",
)

# Output streams back over HTTP/2 on response["response"]
for chunk in response["response"].iter_chunks():
    print(chunk.decode(), end="", flush=True)

aws events put-rule \
  --name AgentCoreCommandExecution \
  --event-pattern '{"source":["aws.bedrock-agentcore"],"detail-type":["AWS API Call via CloudTrail"],"detail":{"eventName":["InvokeAgentRuntimeCommand"]}}' \
  --state ENABLED