pypi-anthropic

1
# AWS Bedrock Integration
2

3
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.
4

5
## Capabilities
6

7
### Bedrock Client Classes
8

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

11
```python { .api }
12
class AnthropicBedrock:
13
    def __init__(
14
        self,
15
        *,
16
        aws_access_key: Optional[str] = None,
17
        aws_secret_key: Optional[str] = None,
18
        aws_session_token: Optional[str] = None,
19
        aws_region: Optional[str] = None,
20
        **kwargs
21
    ): ...
22
    
23
    messages: Messages
24
    completions: Completions
25

26
class AsyncAnthropicBedrock:
27
    def __init__(
28
        self,
29
        *,
30
        aws_access_key: Optional[str] = None,
31
        aws_secret_key: Optional[str] = None,
32
        aws_session_token: Optional[str] = None,
33
        aws_region: Optional[str] = None,
34
        **kwargs
35
    ): ...
36
    
37
    messages: AsyncMessages
38
    completions: AsyncCompletions
39
```
40

41
## Usage Examples
42

43
### Basic Bedrock Setup
44

45
```python
46
from anthropic import AnthropicBedrock
47

48
# Basic configuration using AWS credentials
49
client = AnthropicBedrock(
50
    aws_region="us-east-1"
51
)
52

53
# Explicit credentials (not recommended for production)
54
client = AnthropicBedrock(
55
    aws_access_key="AKIA...",
56
    aws_secret_key="secret...",
57
    aws_region="us-east-1"
58
)
59

60
# Using session token for temporary credentials
61
client = AnthropicBedrock(
62
    aws_access_key="ASIA...",
63
    aws_secret_key="secret...",
64
    aws_session_token="token...",
65
    aws_region="us-east-1"
66
)
67
```
68

69
### AWS Authentication Methods
70

71
```python
72
import boto3
73
import os
74
from anthropic import AnthropicBedrock
75

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

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

87
client = AnthropicBedrock()
88

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

92
client = AnthropicBedrock(
93
    aws_region="us-west-2"
94
)
95

96
# Method 4: Using boto3 session
97
session = boto3.Session(
98
    aws_access_key_id="your-access-key",
99
    aws_secret_access_key="your-secret-key",
100
    region_name="us-east-1"
101
)
102

103
# Extract credentials from session
104
credentials = session.get_credentials()
105
client = AnthropicBedrock(
106
    aws_access_key=credentials.access_key,
107
    aws_secret_key=credentials.secret_key,
108
    aws_session_token=credentials.token,
109
    aws_region=session.region_name
110
)
111
```
112

113
### Messages with Bedrock
114

115
```python
116
# Create message using Bedrock
117
message = client.messages.create(
118
    model="anthropic.claude-sonnet-4-20250514-v1:0",  # Bedrock model ARN format
119
    max_tokens=1024,
120
    messages=[
121
        {"role": "user", "content": "Hello from AWS Bedrock!"}
122
    ]
123
)
124

125
print(message.content[0].text)
126
```
127

128
### Bedrock Model Selection
129

130
```python
131
# Available Claude models on Bedrock (example model IDs)
132
BEDROCK_MODELS = {
133
    "claude-sonnet-4": "anthropic.claude-sonnet-4-20250514-v1:0",
134
    "claude-haiku-3": "anthropic.claude-haiku-3-20241022-v1:0",
135
    "claude-opus-3": "anthropic.claude-opus-3-20240229-v1:0"
136
}
137

138
def create_bedrock_message(model_name: str, prompt: str) -> str:
139
    """Create message with Bedrock model"""
140
    
141
    if model_name not in BEDROCK_MODELS:
142
        raise ValueError(f"Unknown model: {model_name}")
143
    
144
    model_id = BEDROCK_MODELS[model_name]
145
    
146
    message = client.messages.create(
147
        model=model_id,
148
        max_tokens=1024,
149
        messages=[
150
            {"role": "user", "content": prompt}
151
        ]
152
    )
153
    
154
    return message.content[0].text
155

156
# Usage
157
response = create_bedrock_message("claude-sonnet-4", "What is AWS Bedrock?")
158
print(response)
159
```
160

161
### Cross-Region Configuration
162

163
```python
164
class BedrockMultiRegion:
165
    """Manage Bedrock clients across multiple AWS regions"""
166
    
167
    def __init__(self, regions: List[str]):
168
        self.clients = {}
169
        for region in regions:
170
            self.clients[region] = AnthropicBedrock(aws_region=region)
171
    
172
    def create_message(self, region: str, **kwargs) -> Any:
173
        """Create message in specific region"""
174
        if region not in self.clients:
175
            raise ValueError(f"Region {region} not configured")
176
        
177
        return self.clients[region].messages.create(**kwargs)
178
    
179
    def find_best_region(self, model: str) -> str:
180
        """Find best region for a model (simplified example)"""
181
        # In practice, you'd check model availability per region
182
        if "opus" in model.lower():
183
            return "us-east-1"  # Opus might be primarily in us-east-1
184
        else:
185
            return "us-west-2"  # Other models in us-west-2
186

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

190
best_region = multi_region.find_best_region("claude-sonnet-4")
191
message = multi_region.create_message(
192
    region=best_region,
193
    model="anthropic.claude-sonnet-4-20250514-v1:0",
194
    max_tokens=1024,
195
    messages=[{"role": "user", "content": "Hello!"}]
196
)
197
```
198

199
### Bedrock with IAM Roles
200

201
```python
202
import boto3
203
from anthropic import AnthropicBedrock
204

205
def create_bedrock_client_with_role(role_arn: str, session_name: str) -> AnthropicBedrock:
206
    """Create Bedrock client using IAM role assumption"""
207
    
208
    # Create STS client
209
    sts_client = boto3.client('sts')
210
    
211
    # Assume role
212
    response = sts_client.assume_role(
213
        RoleArn=role_arn,
214
        RoleSessionName=session_name
215
    )
216
    
217
    # Extract temporary credentials
218
    credentials = response['Credentials']
219
    
220
    # Create Bedrock client with temporary credentials
221
    return AnthropicBedrock(
222
        aws_access_key=credentials['AccessKeyId'],
223
        aws_secret_key=credentials['SecretAccessKey'],
224
        aws_session_token=credentials['SessionToken'],
225
        aws_region="us-east-1"
226
    )
227

228
# Usage
229
bedrock_client = create_bedrock_client_with_role(
230
    role_arn="arn:aws:iam::123456789012:role/BedrockAccessRole",
231
    session_name="anthropic-session"
232
)
233

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

241
### Async Bedrock Usage
242

243
```python
244
import asyncio
245
from anthropic import AsyncAnthropicBedrock
246

247
async def bedrock_async_example():
248
    # Create async Bedrock client
249
    async_client = AsyncAnthropicBedrock(
250
        aws_region="us-east-1"
251
    )
252
    
253
    # Async message creation
254
    message = await async_client.messages.create(
255
        model="anthropic.claude-sonnet-4-20250514-v1:0",
256
        max_tokens=1024,
257
        messages=[
258
            {"role": "user", "content": "Async Bedrock request"}
259
        ]
260
    )
261
    
262
    return message.content[0].text
263

264
# Run async
265
result = asyncio.run(bedrock_async_example())
266
print(f"Async Bedrock result: {result}")
267
```
268

269
### Bedrock Error Handling
270

271
```python
272
import boto3.exceptions
273
from anthropic import AnthropicBedrock, APIError
274

275
def robust_bedrock_request(prompt: str, max_retries: int = 3) -> Optional[str]:
276
    """Make Bedrock request with robust error handling"""
277
    
278
    for attempt in range(max_retries):
279
        try:
280
            client = AnthropicBedrock(aws_region="us-east-1")
281
            
282
            message = client.messages.create(
283
                model="anthropic.claude-sonnet-4-20250514-v1:0",
284
                max_tokens=1024,
285
                messages=[
286
                    {"role": "user", "content": prompt}
287
                ]
288
            )
289
            
290
            return message.content[0].text
291
            
292
        except boto3.exceptions.NoCredentialsError:
293
            print("❌ AWS credentials not found")
294
            print("Configure credentials using AWS CLI or environment variables")
295
            return None
296
            
297
        except boto3.exceptions.ClientError as e:
298
            error_code = e.response['Error']['Code']
299
            
300
            if error_code == 'AccessDeniedException':
301
                print("❌ Access denied to Bedrock")
302
                print("Check IAM permissions for Bedrock access")
303
                return None
304
                
305
            elif error_code == 'ThrottlingException':
306
                print(f"⏳ Throttled (attempt {attempt + 1})")
307
                if attempt < max_retries - 1:
308
                    time.sleep(2 ** attempt)
309
                    continue
310
                    
311
            elif error_code == 'ModelNotReadyException':
312
                print("⏳ Model not ready, retrying...")
313
                if attempt < max_retries - 1:
314
                    time.sleep(5)
315
                    continue
316
                    
317
            print(f"❌ AWS error: {error_code}")
318
            return None
319
            
320
        except APIError as e:
321
            print(f"❌ Anthropic API error: {e}")
322
            return None
323
            
324
        except Exception as e:
325
            print(f"❌ Unexpected error: {e}")
326
            return None
327
    
328
    print("❌ Max retries reached")
329
    return None
330

331
# Usage
332
result = robust_bedrock_request("What are the benefits of using AWS Bedrock?")
333
if result:
334
    print(f"Success: {result}")
335
```
336

337
### Bedrock Configuration Management
338

339
```python
340
import json
341
import boto3
342
from typing import Dict, Any
343

344
class BedrockConfig:
345
    """Configuration management for Bedrock deployment"""
346
    
347
    def __init__(self, config_file: str = "bedrock-config.json"):
348
        self.config_file = config_file
349
        self.config = self.load_config()
350
    
351
    def load_config(self) -> Dict[str, Any]:
352
        """Load configuration from file"""
353
        try:
354
            with open(self.config_file, 'r') as f:
355
                return json.load(f)
356
        except FileNotFoundError:
357
            return self.default_config()
358
    
359
    def default_config(self) -> Dict[str, Any]:
360
        """Default configuration"""
361
        return {
362
            "regions": ["us-east-1", "us-west-2"],
363
            "models": {
364
                "fast": "anthropic.claude-haiku-3-20241022-v1:0",
365
                "balanced": "anthropic.claude-sonnet-4-20250514-v1:0",
366
                "powerful": "anthropic.claude-opus-3-20240229-v1:0"
367
            },
368
            "max_tokens": 1024,
369
            "timeout": 30,
370
            "max_retries": 3
371
        }
372
    
373
    def create_client(self, region: str = None) -> AnthropicBedrock:
374
        """Create configured Bedrock client"""
375
        region = region or self.config["regions"][0]
376
        
377
        return AnthropicBedrock(
378
            aws_region=region,
379
            timeout=self.config["timeout"],
380
            max_retries=self.config["max_retries"]
381
        )
382
    
383
    def create_message(self, prompt: str, model_type: str = "balanced") -> str:
384
        """Create message with configured defaults"""
385
        client = self.create_client()
386
        model = self.config["models"].get(model_type, self.config["models"]["balanced"])
387
        
388
        message = client.messages.create(
389
            model=model,
390
            max_tokens=self.config["max_tokens"],
391
            messages=[
392
                {"role": "user", "content": prompt}
393
            ]
394
        )
395
        
396
        return message.content[0].text
397

398
# Usage
399
config = BedrockConfig()
400

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

405
# Create client for custom usage
406
client = config.create_client(region="us-west-2")
407
```
408

409
### Bedrock with VPC Configuration
410

411
```python
412
import boto3
413
from anthropic import AnthropicBedrock
414

415
def create_vpc_bedrock_client(vpc_config: Dict[str, Any]) -> AnthropicBedrock:
416
    """Create Bedrock client configured for VPC access"""
417
    
418
    # Note: VPC configuration typically happens at the AWS service level
419
    # This is an example of how you might handle VPC-specific setups
420
    
421
    session = boto3.Session()
422
    
423
    # Configure session for VPC if needed
424
    if vpc_config.get("endpoint_url"):
425
        # Custom endpoint for VPC endpoint
426
        client = AnthropicBedrock(
427
            aws_region=vpc_config["region"],
428
            # Custom endpoint configuration would go here
429
        )
430
    else:
431
        client = AnthropicBedrock(
432
            aws_region=vpc_config["region"]
433
        )
434
    
435
    return client
436

437
# VPC configuration example
438
vpc_config = {
439
    "region": "us-east-1",
440
    "vpc_id": "vpc-12345678",
441
    "subnet_ids": ["subnet-12345678", "subnet-87654321"],
442
    "security_group_ids": ["sg-12345678"],
443
    "endpoint_url": "https://bedrock.us-east-1.amazonaws.com"  # VPC endpoint
444
}
445

446
vpc_client = create_vpc_bedrock_client(vpc_config)
447
```
448

449
### Bedrock Cost Optimization
450

451
```python
452
import time
453
from typing import Dict, List
454

455
class BedrockCostOptimizer:
456
    """Optimize Bedrock usage for cost efficiency"""
457
    
458
    def __init__(self):
459
        self.usage_stats = {}
460
        self.model_costs = {
461
            # Example costs per 1K tokens (input/output)
462
            "anthropic.claude-haiku-3-20241022-v1:0": (0.00025, 0.00125),
463
            "anthropic.claude-sonnet-4-20250514-v1:0": (0.003, 0.015),
464
            "anthropic.claude-opus-3-20240229-v1:0": (0.015, 0.075)
465
        }
466
    
467
    def select_optimal_model(self, prompt: str, priority: str = "cost") -> str:
468
        """Select model based on optimization priority"""
469
        
470
        if priority == "cost" and len(prompt) < 1000:
471
            return "anthropic.claude-haiku-3-20241022-v1:0"
472
        elif priority == "quality":
473
            return "anthropic.claude-opus-3-20240229-v1:0"
474
        else:
475
            return "anthropic.claude-sonnet-4-20250514-v1:0"
476
    
477
    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
478
        """Estimate cost for request"""
479
        if model not in self.model_costs:
480
            return 0.0
481
        
482
        input_cost, output_cost = self.model_costs[model]
483
        return (input_tokens * input_cost / 1000) + (output_tokens * output_cost / 1000)
484
    
485
    def create_cost_optimized_message(self, client: AnthropicBedrock, prompt: str, **kwargs) -> tuple:
486
        """Create message with cost tracking"""
487
        
488
        # Select optimal model
489
        model = self.select_optimal_model(prompt, kwargs.get("priority", "cost"))
490
        
491
        # Create message
492
        start_time = time.time()
493
        message = client.messages.create(
494
            model=model,
495
            messages=[{"role": "user", "content": prompt}],
496
            **{k: v for k, v in kwargs.items() if k != "priority"}
497
        )
498
        duration = time.time() - start_time
499
        
500
        # Track usage
501
        usage = message.usage
502
        cost = self.estimate_cost(model, usage.input_tokens, usage.output_tokens)
503
        
504
        self.usage_stats[model] = self.usage_stats.get(model, {
505
            "requests": 0,
506
            "total_cost": 0.0,
507
            "total_tokens": 0
508
        })
509
        
510
        self.usage_stats[model]["requests"] += 1
511
        self.usage_stats[model]["total_cost"] += cost
512
        self.usage_stats[model]["total_tokens"] += usage.input_tokens + usage.output_tokens
513
        
514
        return message, {
515
            "model": model,
516
            "cost": cost,
517
            "duration": duration,
518
            "tokens": usage.input_tokens + usage.output_tokens
519
        }
520

521
# Usage
522
optimizer = BedrockCostOptimizer()
523
client = AnthropicBedrock(aws_region="us-east-1")
524

525
message, stats = optimizer.create_cost_optimized_message(
526
    client,
527
    "Write a short summary of cloud computing",
528
    max_tokens=200,
529
    priority="cost"
530
)
531

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