Using the Claude API in Production Environments: Stability Guarantees and Best Practices

# ❌ What Not to Do – Never Do This
client = anthropic.Anthropic(api_key="sk-ant-xxx...")
# ✅ The correct approach - Use environment variables
import os
import anthropic
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

# Using AWS Secrets Manager
import boto3
import json
def get_api_key():
    """Retrieve the API key from AWS Secrets Manager"""
    client = boto3.client('secretsmanager', region_name='us-east-1')
    response = client.get_secret_value(SecretId='anthropic-api-key')
    secret = json.loads(response['SecretString'])
    return secret['api_key']
# Using Vault
import hvac
def get_api_key_from_vault():
    client = hvac.Client(url='https://vault.company.com')
    client.token = os.environ.get('VAULT_TOKEN')
    secret = client.secrets.kv.read_secret_version(path='ai-keys')
    return secret['data']['data']['anthropic_key']

import anthropic
import time
import random
from functools import wraps
def retry_with_exponential_backoff(
    max_retries=5,
    initial_wait=1.0,
    max_wait=60.0,
    jitter=True
):
    """Exponential Backoff Retry Decorator"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            wait_time = initial_wait
            
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except anthropic.RateLimitError as e:
                    if attempt == max_retries:
                        raise
                    
                    if jitter:
                        actual_wait = wait_time + random.uniform(0, wait_time * 0.1)
                    else:
                        actual_wait = wait_time
                    
                    print(f"Rate limit hit. Waiting {actual_wait:.1f}s (attempt {attempt + 1}/{max_retries})")
                    time.sleep(actual_wait)
                    wait_time = min(wait_time * 2, max_wait)
                    
                except anthropic.APIStatusError as e:
                    if e.status_code >= 500 and attempt < max_retries:
                        time.sleep(wait_time)
                        wait_time = min(wait_time * 2, max_wait)
                    else:
                        raise
        
        return wrapper
    return decorator
@retry_with_exponential_backoff(max_retries=5)
def call_claude(client, messages):
    return client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=messages
    )

import asyncio
import aiohttp
from asyncio import Semaphore
class ClaudeRequestQueue:
    def __init__(self, max_concurrent=10, requests_per_minute=60):
        self.semaphore = Semaphore(max_concurrent)
        self.rpm_limit = requests_per_minute
        self.request_times = []
    
    async def execute(self, client, messages):
        """Execute API requests via a queue"""
        async with self.semaphore:
            await self._rate_limit()
            return await self._make_request(client, messages)
    
    async def _rate_limit(self):
        """A Simple Speed Limit Implementation"""
        now = asyncio.get_event_loop().time()
        self.request_times = [t for t in self.request_times if now - t < 60]
        
        if len(self.request_times) >= self.rpm_limit:
            wait_time = 60 - (now - self.request_times[0])
            if wait_time > 0:
                await asyncio.sleep(wait_time)
        
        self.request_times.append(now)
    
    async def _make_request(self, client, messages):
        # Implementing Asynchronous API Calls
        pass

import logging
from dataclasses import dataclass
@dataclass
class APIUsageStats:
    input_tokens: int = 0
    output_tokens: int = 0
    total_requests: int = 0
    total_cost_usd: float = 0.0
# Claude Pricing (using claude-opus-4-5 as an example)
CLAUDE_PRICING = {
    "claude-opus-4-5": {
        "input": 3.0 / 1,000,000,   # $3 per million input tokens
        "output": 15.0 / 1,000,000  # $15 per million output tokens
    },
    "claude-3-5-sonnet-20241022": {
        "input": 3.0 / 1_000_000,
        "output": 15.0 / 1,000,000
    }
}
class CostTracker:
    def __init__(self):
        self.stats = APIUsageStats()
        self.logger = logging.getLogger(__name__)
    
    def record_usage(self, response, model: str):
        usage = response.usage
        pricing = CLAUDE_PRICING.get(model, CLAUDE_PRICING["claude-opus-4-5"])
        
        input_cost = usage.input_tokens * pricing["input"]
        output_cost = usage.output_tokens * pricing["output"]
        
        self.stats.input_tokens += usage.input_tokens
        self.stats.output_tokens += usage.output_tokens
        self.stats.total_requests += 1
        self.stats.total_cost_usd += (input_cost + output_cost)
        
        self.logger.info(
            f"Request cost: ${input_cost + output_cost:.6f} | "
            f"Tokens: {usage.input_tokens}in/{usage.output_tokens}out | "
            f"Total cost today: ${self.stats.total_cost_usd:.4f}"
        )
    
    def get_report(self) -> dict:
        return {
            "total_requests": self.stats.total_requests,
            "total_input_tokens": self.stats.input_tokens,
            "total_output_tokens": self.stats.output_tokens,
            "total_cost_usd": round(self.stats.total_cost_usd, 4)
        }

import hashlib
import redis
class ResponseCache:
    def __init__(self, redis_client, ttl=3600):
        self.redis = redis_client
        self.ttl = ttl
    
    def _cache_key(self, messages, model):
        content = str(messages) + model
        return "claude_cache:" + hashlib.md5(content.encode()).hexdigest()
    
    def get(self, messages, model):
        key = self._cache_key(messages, model)
        cached = self.redis.get(key)
        return cached.decode() if cached else None
    
    def set(self, messages, model, response_text):
        key = self._cache_key(messages, model)
        self.redis.setex(key, self.ttl, response_text)

import anthropic
def handle_api_error(error: Exception) -> dict:
    """Unified Error Handling"""
    if isinstance(error, anthropic.AuthenticationError):
        return {"code": "AUTH_ERROR", "message": "API Key无效，请检查配置", "retryable": False}
    
    elif isinstance(error, anthropic.RateLimitError):
        return {"code": "RATE_LIMIT", "message": "请求过于频繁", "retryable": True}
    
    elif isinstance(error, anthropic.APIStatusError):
        if error.status_code == 400:
            return {"code": "BAD_REQUEST", "message": "请求参数错误", "retryable": False}
        elif error.status_code >= 500:
            return {"code": "SERVER_ERROR", "message": "Anthropic服务异常", "retryable": True}
    
    elif isinstance(error, anthropic.APIConnectionError):
        return {"code": "CONNECTION_ERROR", "message": "网络连接失败", "retryable": True}
    
    return {"code": "UNKNOWN_ERROR", "message": str(error), "retryable": False}

# Integration with Sentry
import sentry_sdk
from sentry_sdk.integrations.anthropic import AnthropicIntegration
sentry_sdk.init(
    dsn="your-sentry-dsn",
    integrations=[AnthropicIntegration(include_prompts=False)],  # Note: Prompts (containing user data) are not logged
    traces_sample_rate=0.1
)

import anthropic
import os
# Production Environment Configuration
client = anthropic.Anthropic(
    api_key=os.environ.get("JIEKOU_API_KEY"),
    base_url="https://api.jiekou.ai/v1",
    timeout=30.0,
    max_retries=3  # Built-in retry
)
# No changes are required to the rest of the business logic; it is fully compatible with the official SDK