MCP Server Rate Limiting + Caching Pattern

When an MCP server wraps multiple external APIs (public records, web search, etc.), many free-tier APIs enforce rate limits. The Super Search MCP server at /root/docker/super-search/server.py uses this pattern.

Architecture

Query → TTL Cache (in-memory, per-provider TTL)
          ↓ miss
       Token Bucket Rate Limiter (per-provider)
          ↓ allow
       Provider API → success? → return + cache
          ↓ 429 / rate limited
       Fallback to next provider in chain
          ↓ all depleted
       Return cached or error

Token Bucket Implementation

from collections import defaultdict
import time

RATE_LIMITS = {
    "searxng":       {"tokens": 30, "interval": 1.0},   # local, high
    "exa":           {"tokens": 10, "interval": 1.0},   # paid tier
    "firecrawl":     {"tokens": 5,  "interval": 60.0},  # 1k/mo
    "opencorporates": {"tokens": 1, "interval": 1.0},   # 1/sec free
    "courtlistener": {"tokens": 10, "interval": 6.0},   # ~100/min free
}

CACHE_TTL = {
    "searxng": 120,          # 2 min
    "exa": 300,              # 5 min
    "opencorporates": 3600,  # 1 hour — company records
    "courtlistener": 3600,   # 1 hour — court cases
}

_buckets = {}
_cache = {}

async def _rate_limited_call(provider, fn, cache_key, *args):
    """Rate-limited, cached provider call with auto-fallback."""
    # Check TTL cache
    if cache_key in _cache:
        age = time.time() - _cache[cache_key]["ts"]
        if age < CACHE_TTL.get(provider, 300):
            return _cache[cache_key]["data"]

    # Token bucket
    now = time.time()
    rl = RATE_LIMITS.get(provider, {"tokens": 5, "interval": 1.0})
    if provider not in _buckets:
        _buckets[provider] = {"tokens": rl["tokens"], "last": now}
    
    b = _buckets[provider]
    elapsed = now - b["last"]
    b["tokens"] = min(rl["tokens"], b["tokens"] + int(elapsed / rl["interval"]) * rl["tokens"])
    b["last"] = now
    
    if b["tokens"] <= 0:
        raise RateLimitedError(f"{provider} rate limited")
    b["tokens"] -= 1

    # Call provider
    result = await fn(*args)
    _cache[cache_key] = {"data": result, "ts": time.time()}
    return result

Fallback Chain Pattern

In each tool function, try providers in priority order, catching rate-limit errors:

async def web_search(query, limit=5):
    for provider, fn in [
        ("searxng", _searxng_search),
        ("exa", _exa_search),
        ("opencorporates", _opencorporates_search),
        ("courtlistener", _courtlistener_search),
        ("firecrawl", _firecrawl_search),
    ]:
        try:
            results = await _rate_limited_call(provider, fn, f"{provider}:{query}", query, limit)
            if results:
                return {"provider": provider, "results": results}
        except RateLimitedError:
            continue  # try next provider
        except Exception:
            continue  # try next provider
    return {"error": "All search providers exhausted"}

Free Public Records APIs for Skip Tracing

API	Endpoint	Free Tier	Rate Limit	Best For
OpenCorporates	`api.opencorporates.com/v0.4/companies/search`	Yes	~1 req/sec	Company registration lookup
CourtListener	`www.courtlistener.com/api/rest/v3/opinions/`	Yes	~100 req/min	Federal/state court opinions
USPTO	`data.uspto.gov/`	Yes	Generous	Trademark/patent search
PACER (paid)	`pacer.uscourts.gov/`	$0.10/page	Billing-based	Federal court cases, bankruptcies
SEC EDGAR	`sec.gov/cgi-bin/cik_lookup`	Yes	Generous	Corporate filings

Key Design Decisions

TTL cache prevents duplicate API calls for identical queries within the TTL window
Token bucket per provider prevents any single provider from being overwhelmed
Rate-limited providers fall through silently to the next provider in the chain
The chain always tries free options before paid ones
No provider failure blocks the user — they just get results from a different source

4.0 KiB Raw Permalink Blame History