bedrock_test.py

#%%
import boto3
import json
import logging
aws_access_key_id='AKIA2UC26WDCAFOF7X64'
aws_secret_access_key='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'
aws_region = "eu-central-1"
#%%
from anthropic import AnthropicBedrock

client = AnthropicBedrock(
    # Authenticate by either providing the keys below or use the default AWS credential providers, such as
    # using ~/.aws/credentials or the "AWS_SECRET_ACCESS_KEY" and "AWS_ACCESS_KEY_ID" environment variables.
    aws_access_key=aws_access_key_id,
    aws_secret_key=aws_secret_access_key,
    # Temporary credentials can be used with aws_session_token.
    # Read more at https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp.html.
    aws_session_token=None,
    # aws_region changes the aws region to which the request is made. By default, we read AWS_REGION,
    # and if that's not present, we default to us-east-1. Note that we do not read ~/.aws/config for the region.
    aws_region=aws_region,
)
modelid='anthropic.claude-v2:1'
message = client.messages.create(
    model=modelid,
    max_tokens=256,
    messages=[{"role": "user", "content": "Hello, world"}]
)
print(message.content)
#%%

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Shows how to generate a message with Anthropic Claude (on demand).
"""
import boto3
import json
import logging

from botocore.exceptions import ClientError


logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

def generate_message(bedrock_runtime, model_id, system_prompt, messages, max_tokens):

    body=json.dumps(
        {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": max_tokens,
            "system": system_prompt,
            "messages": messages
        }  
    )  

    
    response = bedrock_runtime.invoke_model(body=body, modelId=model_id)
    response_body = json.loads(response.get('body').read())
   
    return response_body

def main():
    """
    Entrypoint for Anthropic Claude message example.
    """

    try:
        # Define your credentials here (temporary for dev/testing only)

        aws_region = "eu-central-1"  # or the correct region

        # Use boto3 session to inject credentials
        session = boto3.Session(
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
            region_name=aws_region
        )

        bedrock_runtime = session.client(service_name='bedrock-runtime')

        model_id = 'anthropic.claude-v2:1'
        system_prompt = "Please respond only with emoji."
        max_tokens = 1000

        # Prompt with user turn only.
        user_message = {"role": "user", "content": "Hello World"}
        messages = [user_message]

        response = generate_message(bedrock_runtime, model_id, system_prompt, messages, max_tokens)
        print("User turn only.")
        print(json.dumps(response, indent=4))

        # Prompt with both user and assistant messages
        assistant_message = {"role": "assistant", "content": "<emoji>"}
        messages = [user_message, assistant_message]
        response = generate_message(bedrock_runtime, model_id, system_prompt, messages, max_tokens)
        print("User turn and prefilled assistant response.")
        print(json.dumps(response, indent=4))

    except ClientError as err:
        message = err.response["Error"]["Message"]
        logger.error("A client error occurred: %s", message)
        print("A client error occurred: " + format(message))


main()

#%%
from botocore.exceptions import ClientError
session = boto3.Session(
    aws_access_key_id=aws_access_key_id,
    aws_secret_access_key=aws_secret_access_key,
    region_name=aws_region
)

bedrock_runtime = session.client("bedrock-runtime")

model_id = "anthropic.claude-v2:1"

payload = {
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 100,
    "messages": [{"role": "user", "content": "Hello!"}]
}

try:
    response = bedrock_runtime.invoke_model_with_response_stream(
        modelId=model_id,
        body=json.dumps(payload),
        contentType="application/json",
        accept="application/json"
    )

    print("✅ Streaming appears to be enabled for Claude v2.1.")
    for event in response['body']:
        chunk = event['chunk']['bytes']
        print(chunk.decode(), end="")

except ClientError as e:
    code = e.response['Error']['Code']
    if code == "AccessDeniedException":
        print("❌ Streaming is NOT enabled for Claude v2.1: Access denied.")
    elif code == "ValidationException":
        print("⚠️ Model does not support streaming or bad payload.")
    else:
        print(f"❌ Unexpected error: {e}")
except Exception as e:
    print(f"❌ General error: {e}")
#%%
messages = [
    {"role": "user", "content": "Can you tell me a fun fact about llamas?"}
]

payload = {
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 256,
    "messages": messages
}

# ✅ 1. Test NON-streaming (invoke_model)
print("🧪 Testing invoke_model (non-streaming)...")
try:
    response = client.invoke_model(
        modelId=model_id,
        body=json.dumps(payload),
        contentType="application/json",
        accept="application/json"
    )
    result = json.loads(response["body"].read().decode("utf-8"))
    print("✅ invoke_model succeeded.")
    print("🧠 Claude's reply:", result["content"][0]["text"])
except ClientError as e:
    print("❌ invoke_model failed:", e)

# ❌ 2. Test Streaming (invoke_model_with_response_stream)
print("\n🧪 Testing invoke_model_with_response_stream (streaming)...")
try:
    stream_response = client.invoke_model_with_response_stream(
        modelId=model_id,
        body=json.dumps(payload),
        contentType="application/json",
        accept="application/json"
    )

    print("✅ Streaming supported. Response:")
    for event in stream_response["body"]:
        chunk = event.get("chunk", {}).get("bytes", b"")
        if chunk:
            decoded = json.loads(chunk.decode("utf-8"))
            delta = decoded.get("delta", {}).get("content", "")
            print(delta, end="", flush=True)

except ClientError as e:
    code = e.response["Error"]["Code"]
    if code == "AccessDeniedException":
        print("❌ AccessDeniedException: Streaming is not enabled for your role.")
    elif code == "ValidationException":
        print("⚠️ ValidationException: Model might not support streaming or payload is malformed.")
    else:
        print(f"❌ Unexpected error: {e}")
except Exception as e:
    print(f"❌ General error: {e}")
#%%