A Token Bucket Rate Limiter in 30 Lines of Python (No Libraries)

Every API you call eventually tells you to slow down. Sometimes it’s a polite 429. Sometimes it’s a ban. Before you reach for a library, it’s worth knowing how little code a solid rate limiter takes.

The idea

A token bucket is a simple mental model:

• A bucket holds up to capacity tokens.
• Tokens drip in at a steady rate per second.
• Each request costs one token.
• No tokens? Wait until one arrives.

The bucket allows short bursts (up to capacity) while enforcing a long-term average rate, which is how most real APIs behave.

The code
python
import time
import threading

class TokenBucket:
def init(self, rate: float, capacity: int):
self.rate = rate # tokens added per second
self.capacity = capacity # max tokens in the bucket
self.tokens = capacity
self.updated = time.monotonic()
self.lock = threading.Lock()

def _refill(self):
    now = time.monotonic()
    elapsed = now - self.updated
    self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
    self.updated = now

def acquire(self, cost: int = 1):
    while True:
        with self.lock:
            self._refill()
            if self.tokens >= cost:
                self.tokens -= cost
                return
            wait = (cost - self.tokens) / self.rate
        time.sleep(wait)

Using it
python
import requests

bucket = TokenBucket(rate=5, capacity=10) # 5 req/sec, bursts up to 10

def fetch(url):
bucket.acquire()
return requests.get(url, timeout=10)

for i in range(30):
fetch("https://example.com/api/items")
print(f"request {i} sent")

The first ten requests fire immediately. After that, the loop settles into roughly five per second.

Why time.monotonic()?

time.time() follows the system clock, which can jump backward or forward when NTP adjusts it. monotonic() only moves forward, so your refill math can’t produce negative elapsed time.

Why sleep outside the lock?

If you sleep while holding the lock, every other thread blocks behind you, even ones that could proceed. Computing the wait inside the lock and sleeping outside keeps things concurrent.

Gotchas

• Multiple processes: this bucket lives in one process’s memory. For several workers, you need shared state, typically Redis with an atomic Lua script.
• Server-side limits: if the API sends Retry-After, respect it over your own estimate.
• Cost-based limits: some APIs charge more tokens for heavier endpoints. That’s what the cost parameter is for.
• Async code: swap time.sleep for await asyncio.sleep and the lock for an asyncio.Lock.
• AI agents as API clients: autonomous agents can call tools in loops nobody scripted, so a per-process limiter isn’t enough. Enforcement needs to sit centrally, which is what an AI agent governance layer does

Wrap-up

Thirty lines, no dependencies, and it handles bursts correctly. When you outgrow it, you’ll at least understand what the library is doing underneath.

What’s your go-to approach for rate limiting? Drop it in the comments.