Claude Code Research

Rate Limiting & Token Management

6 min read

Rate Limiting & Token Management

Master the art of managing API rate limits, optimizing token usage, and controlling costs when using Claude Code and Anthropic APIs.

Overview

Effective rate limit management and token optimization are crucial for building reliable, cost-effective applications with Claude Code. This guide covers everything from understanding rate limits to implementing advanced optimization strategies.

1. Understanding Rate Limits

API Usage Tiers (2025)

Free Tier

  • Monthly Budget: Up to $10 of API usage
  • Rate Limits:
    • 5 requests per minute (RPM)
    • 20,000 input tokens per minute (ITPM)
    • 300,000 tokens per day
  • Best For: Experimentation and small projects

Build Tiers

TierDepositWait TimeMonthly UsageRate Limits
Build 1$5NoneUp to $10050 RPM, 40K ITPM
Build 2$407 daysUp to $5001,000 RPM, 400K ITPM
Build 3$2007 daysUp to $1,0002,000 RPM, 400K ITPM
Build 4$40014 daysUp to $5,0004,000 RPM, 800K ITPM

Claude.ai Subscription Plans

  • Free Plan: Variable daily cap based on demand
  • Pro Plan ($20/month):
    • At least 5x free usage
    • ~45 messages every 5 hours
    • ~10-40 Claude Code prompts every 5 hours
  • Max Plan ($100/month):
    • 5x Pro usage (~225 messages/5 hours)
  • Max Plan ($200/month):
    • 20x Pro usage (~900 messages/5 hours)
    • Soft limit of 50 sessions/month

Understanding the 5-Hour Window

Claude.ai uses a rolling 5-hour window for rate limiting:

  • Limits reset every 5 hours
  • Usage is measured within each window
  • Plan your intensive work around these windows

2. Token Counting and Measurement

Using the Token Count API

import anthropic
 
client = anthropic.Anthropic()
 
# Count tokens before sending
response = client.messages.count_tokens(
    model="claude-3-5-sonnet-20241022",
    messages=[
        {"role": "user", "content": "Your prompt here"}
    ]
)
 
print(f"Input tokens: {response.input_tokens}")

Token Counting Best Practices

  1. Pre-flight Checks: Count tokens before sending expensive requests
  2. Batch Analysis: Count tokens for multiple prompts to optimize batching
  3. Cost Estimation: Calculate costs before executing large operations

Getting Token Counts from Responses

message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    messages=[{"role": "user", "content": "Hello"}],
    max_tokens=1000
)
 
# Access usage data
print(f"Input tokens: {message.usage.input_tokens}")
print(f"Output tokens: {message.usage.output_tokens}")

3. Rate Limit Handling Strategies

Client-Side Rate Limiting

Token Bucket Algorithm

import time
from collections import deque
 
class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate
        self.last_refill = time.time()
    
    def consume(self, tokens=1):
        self.refill()
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False
    
    def refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, 
                         self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

Exponential Backoff

import time
import random
 
def exponential_backoff(attempt, base_delay=1, max_delay=60):
    """Calculate exponential backoff with jitter"""
    delay = min(base_delay * (2 ** attempt), max_delay)
    jitter = random.uniform(0, delay * 0.1)
    return delay + jitter
 
# Usage
for attempt in range(5):
    try:
        response = client.messages.create(...)
        break
    except anthropic.RateLimitError:
        delay = exponential_backoff(attempt)
        print(f"Rate limited. Waiting {delay:.2f}s...")
        time.sleep(delay)

Request Queue Management

import asyncio
from queue import PriorityQueue
 
class RequestQueue:
    def __init__(self, rate_limiter):
        self.queue = PriorityQueue()
        self.rate_limiter = rate_limiter
    
    async def add_request(self, priority, request_func):
        await self.queue.put((-priority, request_func))
    
    async def process_queue(self):
        while True:
            if not self.queue.empty() and self.rate_limiter.consume():
                _, request_func = await self.queue.get()
                try:
                    await request_func()
                except Exception as e:
                    print(f"Request failed: {e}")
            else:
                await asyncio.sleep(0.1)

4. Cost Optimization Techniques

Model Selection Strategy

ModelInput CostOutput CostBest Use Case
Claude 3 Haiku$0.25/M$1.25/MSimple tasks, classification
Claude 3.5 Sonnet$3/M$15/MBalanced performance/cost
Claude 4 Opus$15/M$75/MComplex reasoning, analysis

Prompt Caching

Prompt caching can reduce costs by 90%+ for repetitive prompts:

# Enable caching for frequently used prompts
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    messages=[
        {
            "role": "user", 
            "content": "Your system prompt here",
            "cache_control": {"type": "ephemeral"}
        }
    ],
    max_tokens=1000
)

Caching Benefits

  • Cache writes: Slightly higher initial cost
  • Cache reads: 3.00/M for input tokens
  • Ideal for: System prompts, templates, repeated contexts

Batch Processing

Take advantage of 50% discount for non-urgent tasks:

# Use batch API for bulk operations
batch_response = client.batches.create(
    requests=[
        {
            "custom_id": "req-1",
            "params": {
                "model": "claude-3-haiku-20240307",
                "messages": [{"role": "user", "content": "Task 1"}]
            }
        },
        # ... more requests
    ]
)

Token-Efficient Prompting

  1. Concise Instructions: Use clear, brief prompts
  2. Structured Output: Request specific formats to reduce tokens
  3. Context Pruning: Remove unnecessary context between turns
  4. Smart Truncation: Intelligently truncate long inputs

5. Monitoring and Analytics

Built-in Claude Code Commands

# Check current usage
/cost
 
# Monitor session tokens
/status
 
# View rate limit status
/info

Open Source Monitoring Tools

Claude Code Usage Monitor

Terminal-based real-time monitoring with:

  • Token consumption tracking
  • Burn rate calculations
  • ML-based predictions
  • Support for all plan types
# Install
npm install -g claude-code-usage-monitor
 
# Run
ccum --plan pro --dashboard

Prometheus/Grafana Integration

# docker-compose.yml
services:
  prometheus:
    image: prom/prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
  
  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"

Custom Monitoring Implementation

import time
from dataclasses import dataclass
from typing import Dict, List
 
@dataclass
class UsageMetrics:
    timestamp: float
    input_tokens: int
    output_tokens: int
    model: str
    cost: float
 
class UsageMonitor:
    def __init__(self):
        self.metrics: List[UsageMetrics] = []
        
    def track_usage(self, response):
        usage = response.usage
        cost = self.calculate_cost(
            usage.input_tokens, 
            usage.output_tokens,
            response.model
        )
        
        metric = UsageMetrics(
            timestamp=time.time(),
            input_tokens=usage.input_tokens,
            output_tokens=usage.output_tokens,
            model=response.model,
            cost=cost
        )
        self.metrics.append(metric)
    
    def get_hourly_stats(self) -> Dict:
        # Calculate hourly statistics
        pass

6. Error Handling and Recovery

Common Rate Limit Errors

429 Too Many Requests

try:
    response = client.messages.create(...)
except anthropic.RateLimitError as e:
    print(f"Rate limit hit: {e}")
    # Implement backoff strategy

Session Limits

  • Understand 5-hour reset windows
  • Plan intensive work accordingly
  • Use multiple accounts for critical operations

Robust Error Handling

import logging
from tenacity import retry, stop_after_attempt, wait_exponential
 
logger = logging.getLogger(__name__)
 
@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=60)
)
async def resilient_api_call(client, **kwargs):
    try:
        return await client.messages.create(**kwargs)
    except anthropic.RateLimitError as e:
        logger.warning(f"Rate limit: {e}")
        raise
    except anthropic.APIConnectionError as e:
        logger.error(f"Connection error: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

7. Advanced Optimization Strategies

Multi-Account Management

class MultiAccountManager:
    def __init__(self, api_keys: List[str]):
        self.clients = [
            anthropic.Anthropic(api_key=key) 
            for key in api_keys
        ]
        self.current_index = 0
    
    def get_next_client(self):
        client = self.clients[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.clients)
        return client

Intelligent Request Routing

class IntelligentRouter:
    def __init__(self):
        self.model_costs = {
            "haiku": {"input": 0.25, "output": 1.25},
            "sonnet": {"input": 3, "output": 15},
            "opus": {"input": 15, "output": 75}
        }
    
    def select_model(self, task_complexity, budget_remaining):
        if task_complexity < 0.3 and budget_remaining > 10:
            return "claude-3-haiku-20240307"
        elif task_complexity < 0.7:
            return "claude-3-5-sonnet-20241022"
        else:
            return "claude-4-opus-20250514"

Context Window Management

class ContextManager:
    def __init__(self, max_tokens=150000):
        self.max_tokens = max_tokens
        self.conversation_history = []
    
    def add_turn(self, role, content, tokens):
        self.conversation_history.append({
            "role": role,
            "content": content,
            "tokens": tokens
        })
        self._prune_if_needed()
    
    def _prune_if_needed(self):
        total_tokens = sum(turn["tokens"] for turn in self.conversation_history)
        while total_tokens > self.max_tokens and len(self.conversation_history) > 2:
            removed = self.conversation_history.pop(0)
            total_tokens -= removed["tokens"]

Best Practices Summary

  1. Pre-flight Checks: Always count tokens before expensive operations
  2. Implement Client-Side Limiting: Don’t rely solely on server rate limits
  3. Use Appropriate Models: Match model complexity to task requirements
  4. Cache Aggressively: Use prompt caching for repeated contexts
  5. Monitor Continuously: Deploy comprehensive monitoring solutions
  6. Plan for Failures: Implement robust retry logic with backoff
  7. Optimize Context: Prune unnecessary conversation history
  8. Batch When Possible: Take advantage of batch processing discounts

Quick Reference

Environment Variables

# Rate limit configuration
export ANTHROPIC_RATE_LIMIT_STRATEGY="token_bucket"
export ANTHROPIC_MAX_RETRIES=5
export ANTHROPIC_BACKOFF_BASE=2
 
# Monitoring
export CLAUDE_USAGE_TRACKING=true
export CLAUDE_COST_ALERTS_THRESHOLD=100

CLI Commands

# Check usage
claude /cost
claude /status
 
# Manage rate limits
claude /compact  # Reduce context
claude /clear   # Reset conversation

See Also

0 items under this folder.