CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-anthropic

The official Python library for the anthropic API

—

Pending
Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

SecuritybySnyk

Pending

The risk profile of this skill

Overview
Eval results
Files

bedrock.mddocs/

AWS Bedrock Integration

Specialized client for accessing Claude models through Amazon Bedrock, with AWS authentication and Bedrock-specific configurations. This integration allows you to use Claude models within your AWS infrastructure.

Capabilities

Bedrock Client Classes

Synchronous and asynchronous clients for AWS Bedrock integration with Claude models.

class AnthropicBedrock:
    def __init__(
        self,
        *,
        aws_access_key: Optional[str] = None,
        aws_secret_key: Optional[str] = None,
        aws_session_token: Optional[str] = None,
        aws_region: Optional[str] = None,
        **kwargs
    ): ...
    
    messages: Messages
    completions: Completions

class AsyncAnthropicBedrock:
    def __init__(
        self,
        *,
        aws_access_key: Optional[str] = None,
        aws_secret_key: Optional[str] = None,
        aws_session_token: Optional[str] = None,
        aws_region: Optional[str] = None,
        **kwargs
    ): ...
    
    messages: AsyncMessages
    completions: AsyncCompletions

Usage Examples

Basic Bedrock Setup

from anthropic import AnthropicBedrock

# Basic configuration using AWS credentials
client = AnthropicBedrock(
    aws_region="us-east-1"
)

# Explicit credentials (not recommended for production)
client = AnthropicBedrock(
    aws_access_key="AKIA...",
    aws_secret_key="secret...",
    aws_region="us-east-1"
)

# Using session token for temporary credentials
client = AnthropicBedrock(
    aws_access_key="ASIA...",
    aws_secret_key="secret...",
    aws_session_token="token...",
    aws_region="us-east-1"
)

AWS Authentication Methods

import boto3
import os
from anthropic import AnthropicBedrock

# Method 1: Use default AWS credentials (recommended)
# This uses credentials from ~/.aws/credentials, environment variables, or IAM roles
client = AnthropicBedrock(
    aws_region="us-east-1"
)

# Method 2: Environment variables
os.environ["AWS_ACCESS_KEY_ID"] = "your-access-key"
os.environ["AWS_SECRET_ACCESS_KEY"] = "your-secret-key"
os.environ["AWS_DEFAULT_REGION"] = "us-east-1"

client = AnthropicBedrock()

# Method 3: AWS Profile
os.environ["AWS_PROFILE"] = "your-profile-name"

client = AnthropicBedrock(
    aws_region="us-west-2"
)

# Method 4: Using boto3 session
session = boto3.Session(
    aws_access_key_id="your-access-key",
    aws_secret_access_key="your-secret-key",
    region_name="us-east-1"
)

# Extract credentials from session
credentials = session.get_credentials()
client = AnthropicBedrock(
    aws_access_key=credentials.access_key,
    aws_secret_key=credentials.secret_key,
    aws_session_token=credentials.token,
    aws_region=session.region_name
)

Messages with Bedrock

# Create message using Bedrock
message = client.messages.create(
    model="anthropic.claude-sonnet-4-20250514-v1:0",  # Bedrock model ARN format
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Hello from AWS Bedrock!"}
    ]
)

print(message.content[0].text)

Bedrock Model Selection

# Available Claude models on Bedrock (example model IDs)
BEDROCK_MODELS = {
    "claude-sonnet-4": "anthropic.claude-sonnet-4-20250514-v1:0",
    "claude-haiku-3": "anthropic.claude-haiku-3-20241022-v1:0",
    "claude-opus-3": "anthropic.claude-opus-3-20240229-v1:0"
}

def create_bedrock_message(model_name: str, prompt: str) -> str:
    """Create message with Bedrock model"""
    
    if model_name not in BEDROCK_MODELS:
        raise ValueError(f"Unknown model: {model_name}")
    
    model_id = BEDROCK_MODELS[model_name]
    
    message = client.messages.create(
        model=model_id,
        max_tokens=1024,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )
    
    return message.content[0].text

# Usage
response = create_bedrock_message("claude-sonnet-4", "What is AWS Bedrock?")
print(response)

Cross-Region Configuration

class BedrockMultiRegion:
    """Manage Bedrock clients across multiple AWS regions"""
    
    def __init__(self, regions: List[str]):
        self.clients = {}
        for region in regions:
            self.clients[region] = AnthropicBedrock(aws_region=region)
    
    def create_message(self, region: str, **kwargs) -> Any:
        """Create message in specific region"""
        if region not in self.clients:
            raise ValueError(f"Region {region} not configured")
        
        return self.clients[region].messages.create(**kwargs)
    
    def find_best_region(self, model: str) -> str:
        """Find best region for a model (simplified example)"""
        # In practice, you'd check model availability per region
        if "opus" in model.lower():
            return "us-east-1"  # Opus might be primarily in us-east-1
        else:
            return "us-west-2"  # Other models in us-west-2

# Usage
multi_region = BedrockMultiRegion(["us-east-1", "us-west-2", "eu-west-1"])

best_region = multi_region.find_best_region("claude-sonnet-4")
message = multi_region.create_message(
    region=best_region,
    model="anthropic.claude-sonnet-4-20250514-v1:0",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)

Bedrock with IAM Roles

import boto3
from anthropic import AnthropicBedrock

def create_bedrock_client_with_role(role_arn: str, session_name: str) -> AnthropicBedrock:
    """Create Bedrock client using IAM role assumption"""
    
    # Create STS client
    sts_client = boto3.client('sts')
    
    # Assume role
    response = sts_client.assume_role(
        RoleArn=role_arn,
        RoleSessionName=session_name
    )
    
    # Extract temporary credentials
    credentials = response['Credentials']
    
    # Create Bedrock client with temporary credentials
    return AnthropicBedrock(
        aws_access_key=credentials['AccessKeyId'],
        aws_secret_key=credentials['SecretAccessKey'],
        aws_session_token=credentials['SessionToken'],
        aws_region="us-east-1"
    )

# Usage
bedrock_client = create_bedrock_client_with_role(
    role_arn="arn:aws:iam::123456789012:role/BedrockAccessRole",
    session_name="anthropic-session"
)

message = bedrock_client.messages.create(
    model="anthropic.claude-sonnet-4-20250514-v1:0",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello with IAM role!"}]
)

Async Bedrock Usage

import asyncio
from anthropic import AsyncAnthropicBedrock

async def bedrock_async_example():
    # Create async Bedrock client
    async_client = AsyncAnthropicBedrock(
        aws_region="us-east-1"
    )
    
    # Async message creation
    message = await async_client.messages.create(
        model="anthropic.claude-sonnet-4-20250514-v1:0",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "Async Bedrock request"}
        ]
    )
    
    return message.content[0].text

# Run async
result = asyncio.run(bedrock_async_example())
print(f"Async Bedrock result: {result}")

Bedrock Error Handling

import boto3.exceptions
from anthropic import AnthropicBedrock, APIError

def robust_bedrock_request(prompt: str, max_retries: int = 3) -> Optional[str]:
    """Make Bedrock request with robust error handling"""
    
    for attempt in range(max_retries):
        try:
            client = AnthropicBedrock(aws_region="us-east-1")
            
            message = client.messages.create(
                model="anthropic.claude-sonnet-4-20250514-v1:0",
                max_tokens=1024,
                messages=[
                    {"role": "user", "content": prompt}
                ]
            )
            
            return message.content[0].text
            
        except boto3.exceptions.NoCredentialsError:
            print("❌ AWS credentials not found")
            print("Configure credentials using AWS CLI or environment variables")
            return None
            
        except boto3.exceptions.ClientError as e:
            error_code = e.response['Error']['Code']
            
            if error_code == 'AccessDeniedException':
                print("❌ Access denied to Bedrock")
                print("Check IAM permissions for Bedrock access")
                return None
                
            elif error_code == 'ThrottlingException':
                print(f"⏳ Throttled (attempt {attempt + 1})")
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                    continue
                    
            elif error_code == 'ModelNotReadyException':
                print("⏳ Model not ready, retrying...")
                if attempt < max_retries - 1:
                    time.sleep(5)
                    continue
                    
            print(f"❌ AWS error: {error_code}")
            return None
            
        except APIError as e:
            print(f"❌ Anthropic API error: {e}")
            return None
            
        except Exception as e:
            print(f"❌ Unexpected error: {e}")
            return None
    
    print("❌ Max retries reached")
    return None

# Usage
result = robust_bedrock_request("What are the benefits of using AWS Bedrock?")
if result:
    print(f"Success: {result}")

Bedrock Configuration Management

import json
import boto3
from typing import Dict, Any

class BedrockConfig:
    """Configuration management for Bedrock deployment"""
    
    def __init__(self, config_file: str = "bedrock-config.json"):
        self.config_file = config_file
        self.config = self.load_config()
    
    def load_config(self) -> Dict[str, Any]:
        """Load configuration from file"""
        try:
            with open(self.config_file, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return self.default_config()
    
    def default_config(self) -> Dict[str, Any]:
        """Default configuration"""
        return {
            "regions": ["us-east-1", "us-west-2"],
            "models": {
                "fast": "anthropic.claude-haiku-3-20241022-v1:0",
                "balanced": "anthropic.claude-sonnet-4-20250514-v1:0",
                "powerful": "anthropic.claude-opus-3-20240229-v1:0"
            },
            "max_tokens": 1024,
            "timeout": 30,
            "max_retries": 3
        }
    
    def create_client(self, region: str = None) -> AnthropicBedrock:
        """Create configured Bedrock client"""
        region = region or self.config["regions"][0]
        
        return AnthropicBedrock(
            aws_region=region,
            timeout=self.config["timeout"],
            max_retries=self.config["max_retries"]
        )
    
    def create_message(self, prompt: str, model_type: str = "balanced") -> str:
        """Create message with configured defaults"""
        client = self.create_client()
        model = self.config["models"].get(model_type, self.config["models"]["balanced"])
        
        message = client.messages.create(
            model=model,
            max_tokens=self.config["max_tokens"],
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        
        return message.content[0].text

# Usage
config = BedrockConfig()

# Quick message with defaults
response = config.create_message("Explain machine learning", model_type="fast")
print(response)

# Create client for custom usage
client = config.create_client(region="us-west-2")

Bedrock with VPC Configuration

import boto3
from anthropic import AnthropicBedrock

def create_vpc_bedrock_client(vpc_config: Dict[str, Any]) -> AnthropicBedrock:
    """Create Bedrock client configured for VPC access"""
    
    # Note: VPC configuration typically happens at the AWS service level
    # This is an example of how you might handle VPC-specific setups
    
    session = boto3.Session()
    
    # Configure session for VPC if needed
    if vpc_config.get("endpoint_url"):
        # Custom endpoint for VPC endpoint
        client = AnthropicBedrock(
            aws_region=vpc_config["region"],
            # Custom endpoint configuration would go here
        )
    else:
        client = AnthropicBedrock(
            aws_region=vpc_config["region"]
        )
    
    return client

# VPC configuration example
vpc_config = {
    "region": "us-east-1",
    "vpc_id": "vpc-12345678",
    "subnet_ids": ["subnet-12345678", "subnet-87654321"],
    "security_group_ids": ["sg-12345678"],
    "endpoint_url": "https://bedrock.us-east-1.amazonaws.com"  # VPC endpoint
}

vpc_client = create_vpc_bedrock_client(vpc_config)

Bedrock Cost Optimization

import time
from typing import Dict, List

class BedrockCostOptimizer:
    """Optimize Bedrock usage for cost efficiency"""
    
    def __init__(self):
        self.usage_stats = {}
        self.model_costs = {
            # Example costs per 1K tokens (input/output)
            "anthropic.claude-haiku-3-20241022-v1:0": (0.00025, 0.00125),
            "anthropic.claude-sonnet-4-20250514-v1:0": (0.003, 0.015),
            "anthropic.claude-opus-3-20240229-v1:0": (0.015, 0.075)
        }
    
    def select_optimal_model(self, prompt: str, priority: str = "cost") -> str:
        """Select model based on optimization priority"""
        
        if priority == "cost" and len(prompt) < 1000:
            return "anthropic.claude-haiku-3-20241022-v1:0"
        elif priority == "quality":
            return "anthropic.claude-opus-3-20240229-v1:0"
        else:
            return "anthropic.claude-sonnet-4-20250514-v1:0"
    
    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """Estimate cost for request"""
        if model not in self.model_costs:
            return 0.0
        
        input_cost, output_cost = self.model_costs[model]
        return (input_tokens * input_cost / 1000) + (output_tokens * output_cost / 1000)
    
    def create_cost_optimized_message(self, client: AnthropicBedrock, prompt: str, **kwargs) -> tuple:
        """Create message with cost tracking"""
        
        # Select optimal model
        model = self.select_optimal_model(prompt, kwargs.get("priority", "cost"))
        
        # Create message
        start_time = time.time()
        message = client.messages.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            **{k: v for k, v in kwargs.items() if k != "priority"}
        )
        duration = time.time() - start_time
        
        # Track usage
        usage = message.usage
        cost = self.estimate_cost(model, usage.input_tokens, usage.output_tokens)
        
        self.usage_stats[model] = self.usage_stats.get(model, {
            "requests": 0,
            "total_cost": 0.0,
            "total_tokens": 0
        })
        
        self.usage_stats[model]["requests"] += 1
        self.usage_stats[model]["total_cost"] += cost
        self.usage_stats[model]["total_tokens"] += usage.input_tokens + usage.output_tokens
        
        return message, {
            "model": model,
            "cost": cost,
            "duration": duration,
            "tokens": usage.input_tokens + usage.output_tokens
        }

# Usage
optimizer = BedrockCostOptimizer()
client = AnthropicBedrock(aws_region="us-east-1")

message, stats = optimizer.create_cost_optimized_message(
    client,
    "Write a short summary of cloud computing",
    max_tokens=200,
    priority="cost"
)

print(f"Model: {stats['model']}")
print(f"Cost: ${stats['cost']:.6f}")
print(f"Tokens: {stats['tokens']}")
print(f"Response: {message.content[0].text}")

docs

batching.md

bedrock.md

beta.md

completions.md

configuration.md

errors.md

index.md

messages.md

models.md

streaming.md

tools.md

vertex.md

tile.json