Super Search — Rate Limiting & Caching Reference

Full implementation of ratelimit.py used in the Super Search MCP server (/root/docker/super-search/ratelimit.py).

Architecture

                  ┌─ Cache HIT? ──→ return cached result
                  │
Caller ──→ rate_limited_search(provider)
                  │
                  ├─ Cache MISS ──→ acquire rate-limit token (TokenBucket)
                  │                      │
                  │               ┌──────┘
                  │               ├─ 200 OK ──→ cache.set(TTL) ──→ return
                  │               ├─ 429/5xx ──→ _retry_with_backoff (max 3)
                  │               └─ permanent failure ──→ raise

TokenBucket

Async-aware token bucket with acquire() that blocks until a token is available:

class TokenBucket:
    def __init__(self, tokens: float, interval: float) -> None:
        self._capacity = float(tokens)
        self._tokens = float(tokens)
        self._interval = float(interval)
        self._refill_rate = self._capacity / self._interval
        self._last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(self._capacity, self._tokens + elapsed * self._refill_rate)
        self._last_refill = now

    async def acquire(self) -> None:
        async with self._lock:
            self._refill()
            if self._tokens >= 1.0:
                self._tokens -= 1.0
                return
            wait = (1.0 - self._tokens) / self._refill_rate
        await asyncio.sleep(wait)
        async with self._lock:
            self._refill()
            self._tokens -= 1.0

Provider Rate Limits

Configured per provider based on API tier and pricing:

Provider	Tokens	Interval	Effective Rate	Rationale
searxng	30	1.0s	30/sec	Local instance, virtually free
exa	10	1.0s	10/sec	Paid tier, moderate allowance
firecrawl	5	60.0s	~0.08/sec	1k/month free tier, conserve
opencorporates	1	1.0s	~1/sec	Free tier, no API key
courtlistener	10	6.0s	~1.67/sec	Free tier, ~100/min max

TTLCache

Dict-based with async lock and per-key expiration:

class TTLCache:
    def __init__(self) -> None:
        self._cache: dict[str, tuple[Any, float]] = {}  # key → (value, expires_at)
        self._lock = asyncio.Lock()

    async def get(self, key: str) -> Optional[Any]:
        async with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                return None
            value, expires_at = entry
            if time.monotonic() >= expires_at:
                del self._cache[key]
                return None
            return value

    async def set(self, key: str, value: Any, ttl: int) -> None:
        async with self._lock:
            self._cache[key] = (value, time.monotonic() + ttl)

Cache TTLs

Provider	TTL	Rationale
searxng	120s	Web results change frequently
exa	300s	Neural search — moderate freshness
opencorporates	3600s	Company records don't change often
courtlistener	3600s	Court opinions are infrequent additions

Cache key format: {provider}:{query.strip().lower()}:{limit}

Retry with Backoff

Handles transient failures (429, 503, 502, 504) with exponential backoff:

RETRYABLE_STATUSES: set[int] = {429, 503, 502, 504}

async def _retry_with_backoff(
    coro_factory: Callable[[], Any],
    provider: str,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> Any:
    for attempt in range(max_retries + 1):
        try:
            result = await coro_factory()
            return result
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            if status not in RETRYABLE_STATUSES or attempt == max_retries:
                raise
            retry_after = e.response.headers.get("Retry-After", "")
            if retry_after and retry_after.isdigit():
                delay = float(retry_after)
            else:
                delay = base_delay * (2 ** attempt)
            logger.warning(
                "Provider %s returned %d (attempt %d/%d). Retrying in %.1fs…",
                provider, status, attempt + 1, max_retries, delay,
            )
            await asyncio.sleep(delay)

Delays: 1s → 2s → 4s → 8s. Respects Retry-After header when present.

rate_limited_search Decorator

Composite decorator: cache-check → rate-limit → retry → cache-store:

def rate_limited_search(provider: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(query: str, limit: int = 5, **kwargs):
            # 1. Check cache
            key = cache_key(provider, query, limit)
            cached = await _cache.get(key)
            if cached is not None:
                return cached

            # 2. Acquire rate-limit token
            bucket = get_bucket(provider)
            await bucket.acquire()

            # 3. Execute with retry
            ttl = CACHE_TTL.get(provider, 300)

            async def _call():
                return await func(query, limit=limit, **kwargs)

            result = await _retry_with_backoff(_call, provider)

            # 4. Cache result (only on success)
            if result:
                await _cache.set(key, result, ttl)

            return result
        return wrapper
    return decorator

Usage in server.py

from ratelimit import rate_limited_search

@rate_limited_search("opencorporates")
async def _opencorporates_search(query: str, limit: int = 5) -> list[dict]:
    """Search company records via OpenCorporates API."""
    # ... implementation — rate limiting + caching handled by decorator

@rate_limited_search("courtlistener")
async def _courtlistener_search(query: str, limit: int = 5) -> list[dict]:
    """Search US court opinions via CourtListener API."""
    # ... implementation

Dependencies

asyncio (stdlib)
time (stdlib)
logging (stdlib)
functools.wraps (stdlib)
httpx (for HTTPStatusError type in retry handler only)

No external dependencies required for the rate-limiting module itself.

6.7 KiB Raw Permalink Blame History