Bedrock

# Invoke Claude
aws bedrock-runtime invoke-model \
  --model-id us.anthropic.claude-sonnet-5 \
  --content-type application/json \
  --accept application/json \
  --cli-binary-format raw-in-base64-out \
  --body '{
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 4096,
    "messages": [
      {"role": "user", "content": "Explain AWS Lambda in 3 sentences."}
    ]
  }' \
  response.json

# Claude Sonnet 5/Opus 5 think by default: content may start with a thinking block
cat response.json | jq -r '.content[] | select(.type=="text") | .text'

import boto3
import json

bedrock = boto3.client('bedrock-runtime')

def invoke_claude(prompt, max_tokens=4096):
    response = bedrock.invoke_model(
        modelId='us.anthropic.claude-sonnet-5',
        contentType='application/json',
        accept='application/json',
        body=json.dumps({
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': max_tokens,
            'messages': [
                {'role': 'user', 'content': prompt}
            ]
        })
    )

    result = json.loads(response['body'].read())
    # Skip thinking blocks (adaptive thinking is on by default for Sonnet 5).
    # max_tokens caps thinking + text, so a truncated response may have no text block.
    if result['stop_reason'] == 'max_tokens':
        print('Truncated at max_tokens: raise it or lower output_config.effort')
    return next((b['text'] for b in result['content'] if b['type'] == 'text'), '')

# Usage
response = invoke_claude('What is Amazon S3?')
print(response)

import boto3
import json

bedrock = boto3.client('bedrock-runtime')

def stream_claude(prompt):
    response = bedrock.invoke_model_with_response_stream(
        modelId='us.anthropic.claude-sonnet-5',
        contentType='application/json',
        accept='application/json',
        body=json.dumps({
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': 4096,
            'messages': [
                {'role': 'user', 'content': prompt}
            ]
        })
    )

    for event in response['body']:
        chunk = json.loads(event['chunk']['bytes'])
        if chunk['type'] == 'content_block_delta':
            yield chunk['delta'].get('text', '')

# Usage
for text in stream_claude('Write a haiku about cloud computing.'):
    print(text, end='', flush=True)

import boto3
import json

bedrock = boto3.client('bedrock-runtime')

def get_embedding(text):
    response = bedrock.invoke_model(
        modelId='amazon.titan-embed-text-v2:0',
        contentType='application/json',
        accept='application/json',
        body=json.dumps({
            'inputText': text,
            'dimensions': 1024,
            'normalize': True
        })
    )

    result = json.loads(response['body'].read())
    return result['embedding']

# Usage
embedding = get_embedding('AWS Lambda is a serverless compute service.')
print(f'Embedding dimension: {len(embedding)}')

import boto3
import json

bedrock = boto3.client('bedrock-runtime')

class Conversation:
    def __init__(self, system_prompt=None):
        self.messages = []
        self.system = system_prompt

    def chat(self, user_message):
        self.messages.append({
            'role': 'user',
            'content': user_message
        })

        body = {
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': 4096,
            'messages': self.messages
        }

        if self.system:
            body['system'] = self.system

        response = bedrock.invoke_model(
            modelId='us.anthropic.claude-sonnet-5',
            contentType='application/json',
            accept='application/json',
            body=json.dumps(body)
        )

        result = json.loads(response['body'].read())
        if result['stop_reason'] == 'max_tokens':
            # max_tokens caps thinking + text; don't store a truncated/empty turn
            self.messages.pop()
            raise RuntimeError('Truncated at max_tokens: raise it or lower output_config.effort')
        assistant_message = next(
            (b['text'] for b in result['content'] if b['type'] == 'text'), ''
        )

        self.messages.append({
            'role': 'assistant',
            'content': assistant_message
        })

        return assistant_message

# Usage
conv = Conversation(system_prompt='You are an AWS solutions architect.')
print(conv.chat('What database should I use for a chat application?'))
print(conv.chat('What about for time-series data?'))

# List all foundation models
aws bedrock list-foundation-models \
  --query 'modelSummaries[*].[modelId,modelName,providerName]' \
  --output table

# Filter by provider
aws bedrock list-foundation-models \
  --by-provider anthropic \
  --query 'modelSummaries[*].modelId'

# Get model details (includes modelLifecycle.status)
aws bedrock get-foundation-model \
  --model-identifier anthropic.claude-sonnet-5

# agreementAvailability.status AVAILABLE / NOT_AVAILABLE, authorizationStatus
aws bedrock get-foundation-model-availability \
  --model-id anthropic.claude-sonnet-5

# Anthropic one-time use case form (base64-encoded JSON:
# companyName, companyWebsite, intendedUsers, industryOption, otherIndustryOption, useCases)
aws bedrock put-use-case-for-model-access --form-data <base64-json>

# Programmatic agreement for third-party models
aws bedrock list-foundation-model-agreement-offers --model-id <model-id>
aws bedrock create-foundation-model-agreement --model-id <model-id> --offer-token <token>

# Free; returns inputTokens. Not supported for every model (e.g. CRIS-only Claude models)
aws bedrock-runtime count-tokens \
  --model-id anthropic.claude-3-5-haiku-20241022-v1:0 \
  --input '{"converse": {"messages": [{"role": "user", "content": [{"text": "Hello"}]}]}}'