AI API Rate Limiting: Token Bucket, Redis, and SlowAPI for FastAPI LLM Endpoints in 2026

TL;DR — AI API rate limiting: token bucket, Redis, SlowAPI. KnowledgeLib: "Token bucket for most APIs (burst handling + simplicity). Sliding window log for precision. GCRA for memory-efficient production. Redis EVAL for atomic operations." Orq.ai: "Token bucket for LLM bursts. Maintain steady TPS, prevent degradation." MasteringBackend: "Multiple workers — in-memory counters don't sync. Redis for shared state. Global via middleware, per-route via dependencies." DasRoot: "Async Redis for non-blocking counter increments. aioredis for high-performance async operations." DigitalApplied: "RateLimit header: r=remaining, t=window. RateLimit-Policy: q=quota, w=window. Cloudflare adopted September 2025." Learn more with authentication, FastAPI tutorial, async backend, and Docker deployment.

KnowledgeLib provides the algorithm guidance: "Use token bucket for most APIs (best burst handling + simplicity); use sliding window log for precise accuracy; use GCRA for memory-efficient production systems. Key tool: Redis EVAL for atomic operations."

MasteringBackend adds the production caveat: "In production, most FastAPI apps often run with multiple workers, which may stop in-memory counters from syncing across workers. Use Redis for shared state."

Rate Limiting Architecture

Algorithm Comparison

Implementation

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class RateLimitAlgo(Enum):
    TOKEN_BUCKET = "token_bucket"
    FIXED_WINDOW = "fixed_window"
    SLIDING_WINDOW = "sliding_window"
    GCRA = "gcra"

@dataclass
class AIRateLimitGuide:
    """AI API rate limiting implementation guide."""

    def get_redis_limiter(self) -> str:
        """Redis-based async rate limiter."""
        return (
            "# === REDIS RATE LIMITER ===\n"
            "# pip install redis fastapi\n"
            "\n"
            "import redis.asyncio as redis\n"
            "from fastapi import FastAPI,\n"
            "    Request, HTTPException\n"
            "from fastapi.responses import (\n"
            "    JSONResponse)\n"
            "\n"
            "app = FastAPI()\n"
            "redis_client = redis.from_url(\n"
            "    'redis://localhost:6379')\n"
            "\n"
            "# Lua script for atomic\n"
            "#   check-and-increment\n"
            "RATE_SCRIPT = '''\n"
            "local key = KEYS[1]\n"
            "local limit = tonumber(ARGV[1])\n"
            "local window = tonumber(ARGV[2])\n"
            "local current = redis.call('INCR', key)\n"
            "if current == 1 then\n"
            "  redis.call('EXPIRE', key, window)\n"
            "end\n"
            "return current\n"
            "'''\n"
            "\n"
            "async def check_rate_limit(\n"
            "    key: str,\n"
            "    limit: int,\n"
            "    window: int) -> bool:\n"
            "    '''Check rate limit in Redis.''' \n"
            "    current = await redis_client.eval(\n"
            "        RATE_SCRIPT, 1,\n"
            "        key, limit, window)\n"
            "    return current <= limit\n"
            "\n"
            "def get_client_key(\n"
            "    request: Request) -> str:\n"
            "    '''Get rate limit key.''' \n"
            "    # Try user_id from JWT\n"
            "    user = getattr(\n"
            "      request.state, 'user', None)\n"
            "    if user:\n"
            "        return f'rate:{user.id}'\n"
            "    # Fall back to IP\n"
            "    ip = request.client.host\n"
            "    forwarded = (\n"
            "      request.headers.get(\n"
            "        'X-Forwarded-For'))\n"
            "    if forwarded:\n"
            "        ip = (\n"
            "          forwarded.split(',')[0])\n"
            "    return f'rate:{ip}'\n"
            "\n"
            "async def rate_limit_dependency(\n"
            "    request: Request,\n"
            "    limit: int = 10,\n"
            "    window: int = 60):\n"
            "    '''Per-route rate limit.''' \n"
            "    key = get_client_key(request)\n"
            "    ok = await check_rate_limit(\n"
            "        key, limit, window)\n"
            "    if not ok:\n"
            "        return JSONResponse(\n"
            "            status_code=429,\n"
            "            content={\n"
            "              'error':\n"
            "                'rate_limited'},\n"
            "            headers={\n"
            "              'Retry-After':\n"
            "                str(window)})\n"
            "    return None"
        )

    def get_slowapi(self) -> str:
        """SlowAPI rate limiting setup."""
        return (
            "# === SLOWAPI SETUP ===\n"
            "# pip install slowapi\n"
            "\n"
            "from slowapi import Limiter\n"
            "from slowapi.util import (\n"
            "    get_remote_address)\n"
            "from slowapi.errors import (\n"
            "    RateLimitExceeded)\n"
            "from slowapi.middleware import (\n"
            "    SlowAPIMiddleware)\n"
            "from fastapi import FastAPI,\n"
            "    Request\n"
            "from fastapi.responses import (\n"
            "    JSONResponse)\n"
            "\n"
            "app = FastAPI()\n"
            "\n"
            "# Redis backend for\n"
            "#   multi-worker sync\n"
            "limiter = Limiter(\n"
            "    key_func=get_remote_address,\n"
            "    storage_uri=\n"
            "      'redis://localhost:6379')\n"
            "\n"
            "app.state.limiter = limiter\n"
            "app.add_middleware(\n"
            "    SlowAPIMiddleware)\n"
            "\n"
            "# Custom 429 handler\n"
            "@app.exception_handler(\n"
            "    RateLimitExceeded)\n"
            "async def rate_limit_handler(\n"
            "    request, exc):\n"
            "    return JSONResponse(\n"
            "        status_code=429,\n"
            "        content={\n"
            "          'error':\n"
            "            'rate_limited',\n"
            "          'detail': str(exc)},\n"
            "        headers={\n"
            "          'Retry-After':\n"
            "            str(\n"
            "              exc.retry_after)})\n"
            "\n"
            "# Per-route limits\n"
            "@app.post('/v1/chat')\n"
            "@limiter.limit('10/minute')\n"
            "async def chat(\n"
            "    request: Request,\n"
            "    prompt: str):\n"
            "    return {'response': 'ok'}\n"
            "\n"
            "# Multiple limits\n"
            "@app.post('/v1/generate')\n"
            "@limiter.limit(\n"
            "    '10/minute', '100/hour')\n"
            "async def generate(\n"
            "    request: Request):\n"
            "    return {'ok': True}\n"
            "\n"
            "# Per-user key function\n"
            "def get_user_key(request: Request):\n"
            "    user = getattr(\n"
            "      request.state, 'user', None)\n"
            "    if user:\n"
            "        return f'user:{user.id}'\n"
            "    return (\n"
            "      get_remote_address(request))\n"
            "\n"
            "user_limiter = Limiter(\n"
            "    key_func=get_user_key,\n"
            "    storage_uri=\n"
            "      'redis://localhost:6379')\n"
            "\n"
            "# Exempt routes\n"
            "@app.get('/health')\n"
            "@limiter.exempt\n"
            "async def health():\n"
            "    return {'status': 'ok'}"
        )

    def get_token_bucket(self) -> str:
        """Token bucket algorithm in Redis."""
        return (
            "# === TOKEN BUCKET ===\n"
            "\n"
            "TOKEN_BUCKET_SCRIPT = '''\n"
            "local key = KEYS[1]\n"
            "local capacity = tonumber(ARGV[1])\n"
            "local refill_rate = tonumber(ARGV[2])\n"
            "local now = tonumber(ARGV[3])\n"
            "local tokens = tonumber(\n"
            "  redis.call('GET', key..':tokens')\n"
            "    or capacity)\n"
            "local last = tonumber(\n"
            "  redis.call('GET', key..':last')\n"
            "    or now)\n"
            "local elapsed = now - last\n"
            "tokens = math.min(\n"
            "  capacity,\n"
            "  tokens + elapsed * refill_rate)\n"
            "if tokens >= 1 then\n"
            "  tokens = tokens - 1\n"
            "  redis.call('SET',\n"
            "    key..':tokens', tokens)\n"
            "  redis.call('SET',\n"
            "    key..':last', now)\n"
            "  return 1\n"
            "else\n"
            "  redis.call('SET',\n"
            "    key..':tokens', tokens)\n"
            "  redis.call('SET',\n"
            "    key..':last', now)\n"
            "  return 0\n"
            "end\n"
            "'''\n"
            "\n"
            "async def token_bucket_check(\n"
            "    key: str,\n"
            "    capacity: int = 10,\n"
            "    refill_rate: float = 0.1):\n"
            "    '''Token bucket check.''' \n"
            "    import time\n"
            "    now = time.time()\n"
            "    result = await (\n"
            "      redis_client.eval(\n"
            "        TOKEN_BUCKET_SCRIPT, 1,\n"
            "        key, capacity,\n"
            "        refill_rate, now))\n"
            "    return result == 1"
        )

    def get_token_quota(self) -> str:
        """LLM token-based quota management."""
        return (
            "# === TOKEN QUOTA ===\n"
            "\n"
            "async def check_token_quota(\n"
            "    user_id: str,\n"
            "    estimated_tokens: int,\n"
            "    daily_limit: int = 100000):\n"
            "    '''Check token-based quota.''' \n"
            "    key = f'tokens:{user_id}'\n"
            "    used = await redis_client.incrby(\n"
            "        key, estimated_tokens)\n"
            "    if used == estimated_tokens:\n"
            "        # First use today\n"
            "        await redis_client.expire(\n"
            "            key, 86400)  # 24h\n"
            "    return used <= daily_limit\n"
            "\n"
            "# Tiered limits\n"
            "TIERS = {\n"
            "    'free': {\n"
            "        'rate': '10/minute',\n"
            "        'tokens': 100_000},\n"
            "    'pro': {\n"
            "        'rate': '100/minute',\n"
            "        'tokens': 1_000_000},\n"
            "    'enterprise': {\n"
            "        'rate': '1000/minute',\n"
            "        'tokens': 10_000_000},\n"
            "}\n"
            "\n"
            "async def get_tier_limit(\n"
            "    user_id: str) -> dict:\n"
            "    '''Get user tier limits.''' \n"
            "    tier = await get_user_tier(\n"
            "        user_id)\n"
            "    return TIERS.get(\n"
            "        tier, TIERS['free'])\n"
            "\n"
            "# GPU rate limiting\n"
            "import asyncio\n"
            "gpu_semaphore = (\n"
            "  asyncio.Semaphore(4))\n"
            "\n"
            "@app.post('/v1/chat')\n"
            "async def chat(\n"
            "    request: Request,\n"
            "    prompt: str):\n"
            "    user = request.state.user\n"
            "    tier = await (\n"
            "      get_tier_limit(user.id))\n"
            "    \n"
            "    # Rate limit check\n"
            "    key = f'rate:{user.id}'\n"
            "    ok = await (\n"
            "      check_rate_limit(\n"
            "        key, tier['tokens'], 60))\n"
            "    if not ok:\n"
            "        return JSONResponse(\n"
            "            status_code=429,\n"
            "            content={\n"
            "              'error':\n"
            "                'rate_limited'},\n"
            "            headers={\n"
            "              'Retry-After': '60'})\n"
            "    \n"
            "    # GPU concurrency\n"
            "    async with gpu_semaphore:\n"
            "        result = await (\n"
            "          call_ollama(prompt))\n"
            "    \n"
            "    # Track token usage\n"
            "    await check_token_quota(\n"
            "        user.id,\n"
            "        result['tokens'])\n"
            "    \n"
            "    return result"
        )

    def get_headers(self) -> str:
        """Rate limit response headers."""
        return (
            "# === RATE LIMIT HEADERS ===\n"
            "\n"
            "def rate_limit_headers(\n"
            "    remaining: int,\n"
            "    limit: int,\n"
            "    window: int,\n"
            "    reset: int) -> dict:\n"
            "    '''Build rate limit headers.''' \n"
            "    return {\n"
            "        # IETF draft headers\n"
            "        'RateLimit':\n"
            "          f'r={remaining}, t={window}',\n"
            "        'RateLimit-Policy':\n"
            "          f'q={limit};w={window}',\n"
            "        # Legacy headers\n"
            "        'X-RateLimit-Limit':\n"
            "          str(limit),\n"
            "        'X-RateLimit-Remaining':\n"
            "          str(remaining),\n"
            "        'X-RateLimit-Reset':\n"
            "          str(reset),\n"
            "    }\n"
            "\n"
            "# On 429 response\n"
            "return JSONResponse(\n"
            "    status_code=429,\n"
            "    content={\n"
            "        'error': 'rate_limited',\n"
            "        'limit': limit,\n"
            "        'remaining': 0,\n"
            "        'reset': reset,\n"
            "    },\n"
            "    headers={\n"
            "        **rate_limit_headers(\n"
            "            0, limit, window, reset),\n"
            "        'Retry-After':\n"
            "          str(window),\n"
            "    })"
        )

    def get_production_tips(self) -> dict:
        """Production tips."""
        return {
            "redis_backend": "Redis for multi-worker shared state. In-memory counters don't sync across uvicorn workers.",
            "token_bucket": "Token bucket for most APIs — best burst handling + simplicity. Refill at rate, consume per request.",
            "lua_atomic": "Lua script via Redis EVAL for atomic check-and-increment. Avoids race conditions.",
            "slowapi": "SlowAPI decorator-based with Redis backend. @limiter.limit('10/minute'). Per-route or global.",
            "per_user": "Rate limit by user_id from JWT, fall back to IP. Key: rate:{user_id} or rate:{ip}.",
            "token_quota": "Track tokens consumed, not just request count. 100K/day free, 1M/day pro. INCRBY in Redis.",
            "gpu_limit": "asyncio.Semaphore for concurrent GPU requests. RTX 4090: 2-4 streams. Queue excess with 503.",
            "tiers": "Tiered limits: free 10/min, pro 100/min, enterprise 1000/min. Different quotas per tier.",
            "headers": "RateLimit (r=remaining, t=window), RateLimit-Policy (q=quota, w=window), Retry-After on 429.",
            "429": "429 Too Many Requests with Retry-After. Include error body with limit, remaining, reset.",
        }

AI API Rate Limiting Checklist

• [ ] Token bucket for most APIs — best burst handling + simplicity, tokens refill at rate
• [ ] Token bucket: allows bursts, consume token per request, refill at steady rate
• [ ] Fixed window: count per time window, reset at boundary — simple but allows burst at edge
• [ ] Sliding window log: timestamp each request, count in window — precise accuracy, high memory
• [ ] Sliding window counter: hybrid of fixed + sliding — good balance
• [ ] GCRA: virtual arrival time — memory-efficient for production at scale
• [ ] Leaky bucket: fixed output rate, queue requests — smooth output
• [ ] Redis for shared state across multiple uvicorn workers — in-memory counters don't sync
• [ ] Redis INCR for atomic counter, EXPIRE for window reset
• [ ] Lua script via Redis EVAL for atomic check-and-increment — avoids race conditions
• [ ] Async Redis operations (redis.asyncio) — non-blocking, no event loop stall
• [ ] Per-user rate limiting: key = f'rate:{user_id}' from JWT
• [ ] Per-IP rate limiting: key = f'rate:{ip}', check X-Forwarded-For behind proxy
• [ ] Per-API-key: key = f'rate:{api_key}' for service-to-service
• [ ] SlowAPI: pip install slowapi, Limiter with Redis storage_uri
• [ ] SlowAPI: @limiter.limit('10/minute') decorator per route
• [ ] SlowAPI: multiple limits @limiter.limit('10/minute', '100/hour')
• [ ] SlowAPI: @limiter.exempt for health check endpoints
• [ ] SlowAPI: custom key_func for per-user limiting
• [ ] SlowAPI: custom 429 handler with Retry-After header
• [ ] Custom middleware for global rate limiting — all routes pass through same logic
• [ ] Dependency injection for per-route limits — different limits per endpoint
• [ ] LLM token-based quota: track tokens consumed, not just request count
• [ ] Token quota: 100K/day free tier, 1M/day pro, 10M/day enterprise
• [ ] Redis INCRBY for token tracking, EXPIRE 86400 for daily reset
• [ ] GPU rate limiting: asyncio.Semaphore for concurrent LLM request cap
• [ ] GPU: RTX 4090 handles 2-4 concurrent 7B streams
• [ ] Queue depth threshold: reject with 503 + Retry-After when GPU full
• [ ] Tiered limits: free 10/min, pro 100/min, enterprise 1000/min
• [ ] Token bucket allows short bursts — useful for LLM applications with unpredictable bursts
• [ ] Maintain steady TPS while preventing service degradation
• [ ] RateLimit header: r=remaining quota, t=effective window (IETF draft)
• [ ] RateLimit-Policy header: q=total quota, w=window, pk=partition key
• [ ] Cloudflare adopted RateLimit headers September 2025, GitHub also uses them
• [ ] Retry-After header: seconds until retry allowed (on 429 response)
• [ ] X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset (legacy headers)
• [ ] 429 Too Many Requests status code with error body (limit, remaining, reset)
• [ ] Consistent headers across all endpoints — document in API docs
• [ ] Graceful degradation: queue or reject with informative error
• [ ] Cost tracking: estimate cost per request for billing integration
• [ ] Rate limit auth endpoints (login, refresh) to prevent brute force
• [ ] Read authentication for auth setup
• [ ] Read FastAPI tutorial for API setup
• [ ] Read async backend for async patterns
• [ ] Read Docker deployment for deployment
• [ ] Test: rate limit rejects after exceeding limit (429)
• [ ] Test: Redis backend syncs across multiple workers
• [ ] Test: token bucket allows bursts within capacity
• [ ] Test: per-user limits work with JWT identification
• [ ] Test: token quota tracks LLM token consumption
• [ ] Test: GPU semaphore limits concurrent requests
• [ ] Test: tiered limits apply correct rates per user tier
• [ ] Test: Retry-After header present on 429 responses
• [ ] Document algorithm choice, Redis config, key strategy, tier limits, header format

FAQ

What are the best rate limiting algorithms for AI APIs?

Token bucket for most APIs (best burst handling), sliding window log for precision, GCRA for memory efficiency. KnowledgeLib: "Use token bucket for most APIs (best burst handling + simplicity); use sliding window log for precise accuracy; use GCRA for memory-efficient production systems. Key tool: Redis EVAL for atomic operations." Orq.ai: "Token bucket allows requests only if tokens available in virtual bucket. Particularly useful for LLM-based applications where bursts occur unpredictably. Helps maintain steady TPS while preventing service degradation." DigitalApplied: "RateLimit header reports remaining quota (r) and window (t). RateLimit-Policy reports total quota (q), window (w), partition key (pk). Cloudflare adopted this draft September 2025." Algorithms: (1) Token bucket: tokens refill at rate, consume per request, allows bursts. (2) Fixed window: count per time window, reset at boundary. (3) Sliding window log: timestamp each request, count in window. (4) Sliding window counter: hybrid of fixed + sliding. (5) GCRA: virtual arrival time, memory-efficient. (6) Leaky bucket: fixed output rate, queue requests.

How do you implement rate limiting in FastAPI with Redis?

Use async Redis operations for atomic counter increments and window checks. DasRoot: "FastAPI integration with Redis enhanced by async libraries like aioredis. Asynchronous operations such as incrementing counters and checking limits without blocking event loop." MasteringBackend: "In production, most FastAPI apps run with multiple workers — in-memory counters do not sync across workers. Use Redis for shared state. Global limit via middleware, per-route via dependencies." Implementation: (1) Redis INCR for atomic counter. (2) EXPIRE for window reset. (3) Lua script via EVAL for atomic check-and-increment. (4) async def rate_limit(key, limit, window): counter = await redis.incr(key), if counter == 1: await redis.expire(key, window), if counter > limit: return False. (5) Per-user: key = f'rate:{user_id}'. (6) Per-IP: key = f'rate:{ip}'. (7) Middleware for global, dependency for per-route. (8) Return 429 with Retry-After header.

How do you use SlowAPI for rate limiting FastAPI AI endpoints?

SlowAPI provides decorator-based rate limiting with Redis backend for multi-worker support. MasteringBackend: "SlowAPI for FastAPI rate limiting. Per-route dependencies or global middleware. In production with multiple workers, use Redis backend — in-memory counters do not sync." SlowAPI: (1) pip install slowapi. (2) from slowapi import Limiter. (3) limiter = Limiter(key_func=get_remote_address, storage_uri='redis://localhost:6379'). (4) app.state.limiter = limiter. (5) @app.post('/v1/chat') @limiter.limit('10/minute') async def chat(request: Request, ...). (6) Per-user: key_func=lambda request: get_user_id(request). (7) Multiple limits: @limiter.limit('10/minute', '100/hour'). (8) Custom response: 429 with Retry-After. (9) Exempt routes: @limiter.exempt. (10) Redis backend for multi-worker shared state.

How do you manage LLM API quotas and GPU rate limiting?

Track token usage per user, set token-based quotas, and limit concurrent GPU requests. Orq.ai: "Token bucket for LLM applications where bursts occur unpredictably. Maintain steady TPS while preventing service degradation." Quota management: (1) Token-based quota: track tokens consumed per user, not just request count. (2) Per-user quota: e.g., 100K tokens/day for free tier, 1M for pro. (3) Redis: INCRBY tokens used, compare against quota. (4) GPU rate limiting: asyncio.Semaphore for concurrent LLM requests. (5) Queue depth threshold: reject with 503 + Retry-After when full. (6) Tiered limits: different rates for different user tiers. (7) Burst allowance: token bucket allows short bursts. (8) Cost tracking: estimate cost per request for billing. (9) Rate limit headers: RateLimit, RateLimit-Policy, Retry-After. (10) Graceful degradation: queue or reject with informative error.

What HTTP headers should you return for rate-limited AI APIs?

Return RateLimit, RateLimit-Policy, and Retry-After headers per the IETF draft. DigitalApplied: "RateLimit reports remaining service limit — r for available quota, t for effective window. RateLimit-Policy reports quota policy — q for total quota, w for window, pk for partition key. Cloudflare adopted this draft September 2025; GitHub also uses these headers." Headers: (1) RateLimit: r=remaining, t=window in seconds. (2) RateLimit-Policy: q=total quota, w=window, pk=partition key. (3) Retry-After: seconds until retry allowed (on 429). (4) X-RateLimit-Limit: total requests per window (legacy). (5) X-RateLimit-Remaining: remaining requests. (6) X-RateLimit-Reset: epoch time of window reset. (7) 429 Too Many Requests status code. (8) Include error body with limit, remaining, reset. (9) Consistent headers across all endpoints. (10) Document headers in API docs.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →