AI API Authentication: JWT, API Keys, RBAC, and Refresh Token Rotation for FastAPI in 2026

TL;DR — AI API authentication: JWT, API keys, RBAC, refresh rotation. Safeguard: "Layered dependency chain: token extraction, user loading, permission enforcement. JWT only if stateless needed — opaque tokens simpler. Refresh rotation with reuse detection. HttpOnly cookies for browser, bearer for server/mobile." TheLinuxCode: "OAuth2 password flow + JWT. User table, token endpoint, dependency validates JWT, role checks. Access TTL 15 min, refresh TTL 7-30 days." OneUptime: "API Keys for service-to-service, JWT for user auth, OAuth2 for third-party. APIKeyHeader, require_permission factory." FastAPI FullAuth: "JWT access/refresh with rotation, Argon2 hashing, RBAC, HttpOnly cookies, CSRF middleware, rate limiting." Learn more with FastAPI tutorial, async backend, Docker deployment, and rate limiting.

Safeguard frames the architecture: "The FastAPI dependency system is powerful and it is also where most auth bugs originate. The pattern we recommend is a layered chain: a low-level dependency that extracts and validates the token, a middle dependency that loads the user from the validated claims, and high-level dependencies that enforce specific permissions."

TheLinuxCode adds the implementation: "OAuth2 password flow + JWT access tokens, because it teaches the mechanics and works for service-to-service and mobile clients. User table with hashed password and role, token endpoint for login, dependency that validates JWT and loads current user, authorization dependencies for role checks."

AI API Authentication Architecture

Auth Method Comparison

Implementation

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

class AuthMethod(Enum):
    JWT = "jwt"
    API_KEY = "api_key"
    RBAC = "rbac"
    REFRESH = "refresh"

@dataclass
class AIAPIAuthGuide:
    """AI API authentication implementation guide."""

    def get_jwt_setup(self) -> str:
        """JWT authentication setup."""
        return (
            "# === JWT AUTH SETUP ===\n"
            "# pip install python-jose\n"
            "#   passlib[bcrypt] pydantic\n"
            "\n"
            "from fastapi import FastAPI,\n"
            "    Depends, HTTPException, status\n"
            "from fastapi.security import (\n"
            "    OAuth2PasswordBearer,\n"
            "    OAuth2PasswordRequestForm)\n"
            "from jose import jwt, JWTError\n"
            "from passlib.context import (\n"
            "    CryptContext)\n"
            "from pydantic import BaseModel\n"
            "from datetime import (\n"
            "    datetime, timedelta)\n"
            "from typing import Annotated\n"
            "\n"
            "app = FastAPI()\n"
            "\n"
            "# Config (from env in production)\n"
            "SECRET_KEY = 'change-in-prod'\n"
            "ALGORITHM = 'HS256'\n"
            "ACCESS_TOKEN_EXPIRE = 15  # min\n"
            "REFRESH_TOKEN_EXPIRE = 7  # days\n"
            "\n"
            "pwd_context = CryptContext(\n"
            "    schemes=['bcrypt'],\n"
            "    deprecated='auto')\n"
            "\n"
            "oauth2_scheme = (\n"
            "    OAuth2PasswordBearer(\n"
            "        tokenUrl='/token'))\n"
            "\n"
            "# Models\n"
            "class Token(BaseModel):\n"
            "    access_token: str\n"
            "    refresh_token: str\n"
            "    token_type: str = 'bearer'\n"
            "\n"
            "class User(BaseModel):\n"
            "    username: str\n"
            "    role: str = 'user'\n"
            "    permissions: list = []\n"
            "\n"
            "# Token creation\n"
            "def create_access_token(\n"
            "    data: dict):\n"
            "    expire = datetime.utcnow(\n"
            "      ) + timedelta(\n"
            "        minutes=\n"
            "          ACCESS_TOKEN_EXPIRE)\n"
            "    return jwt.encode(\n"
            "        {**data, 'exp': expire},\n"
            "        SECRET_KEY,\n"
            "        algorithm=ALGORITHM)\n"
            "\n"
            "def create_refresh_token(\n"
            "    data: dict):\n"
            "    expire = datetime.utcnow(\n"
            "      ) + timedelta(\n"
            "        days=\n"
            "          REFRESH_TOKEN_EXPIRE)\n"
            "    return jwt.encode(\n"
            "        {**data, 'exp': expire,\n"
            "         'type': 'refresh'},\n"
            "        SECRET_KEY,\n"
            "        algorithm=ALGORITHM)"
        )

    def get_dependency_chain(self) -> str:
        """Layered dependency chain."""
        return (
            "# === DEPENDENCY CHAIN ===\n"
            "\n"
            "# Layer 1: Extract and validate token\n"
            "async def get_token_payload(\n"
            "    token: str = Depends(\n"
            "      oauth2_scheme)):\n"
            "    '''Extract and validate\n"
            "    JWT token.''' \n"
            "    try:\n"
            "        payload = jwt.decode(\n"
            "            token, SECRET_KEY,\n"
            "            algorithms=[ALGORITHM])\n"
            "        return payload\n"
            "    except JWTError:\n"
            "        raise HTTPException(\n"
            "            status_code=401,\n"
            "            detail='Invalid token')\n"
            "\n"
            "# Layer 2: Load user from claims\n"
            "async def get_current_user(\n"
            "    payload: dict = Depends(\n"
            "      get_token_payload)):\n"
            "    '''Load user from token.''' \n"
            "    username = payload.get('sub')\n"
            "    if not username:\n"
            "        raise HTTPException(\n"
            "            status_code=401,\n"
            "            detail='Invalid token')\n"
            "    # Lookup user in DB\n"
            "    user = await lookup_user(\n"
            "        username)\n"
            "    if not user:\n"
            "        raise HTTPException(\n"
            "            status_code=401,\n"
            "            detail='User not found')\n"
            "    return user\n"
            "\n"
            "# Layer 3: Permission enforcement\n"
            "def require_role(\n"
            "    allowed_roles: list[str]):\n"
            "    '''Factory for role check.''' \n"
            "    async def check_role(\n"
            "        user: User = Depends(\n"
            "          get_current_user)):\n"
            "        if user.role not in (\n"
            "          allowed_roles):\n"
            "            raise HTTPException(\n"
            "                status_code=403,\n"
            "                detail=\n"
            "                  'Permission denied')\n"
            "        return user\n"
            "    return check_role\n"
            "\n"
            "def require_permission(\n"
            "    permission: str):\n"
            "    '''Factory for permission.''' \n"
            "    async def check_perm(\n"
            "        user: User = Depends(\n"
            "          get_current_user)):\n"
            "        if permission not in (\n"
            "          user.permissions):\n"
            "            raise HTTPException(\n"
            "                status_code=403,\n"
            "                detail=\n"
            "                  'Permission denied')\n"
            "        return user\n"
            "    return check_perm\n"
            "\n"
            "# Usage in endpoints\n"
            "@app.post('/v1/chat')\n"
            "async def chat(\n"
            "    user: User = Depends(\n"
            "      get_current_user)):\n"
            "    return {'user': user.username}\n"
            "\n"
            "@app.post('/admin/models')\n"
            "async def admin_models(\n"
            "    user: User = Depends(\n"
            "      require_role(['admin']))):\n"
            "    return {'admin': True}"
        )

    def get_api_key_auth(self) -> str:
        """API key authentication for services."""
        return (
            "# === API KEY AUTH ===\n"
            "from fastapi.security import (\n"
            "    APIKeyHeader)\n"
            "from enum import Enum\n"
            "\n"
            "class ServicePermission(Enum):\n"
            "    READ = 'read'\n"
            "    WRITE = 'write'\n"
            "    ADMIN = 'admin'\n"
            "\n"
            "api_key_header = APIKeyHeader(\n"
            "    name='X-API-Key',\n"
            "    auto_error=False)\n"
            "\n"
            "# Store in secrets manager\n"
            "API_KEYS = {\n"
            "    'service-a-key': {\n"
            "        'name': 'service-a',\n"
            "        'permissions': [\n"
            "          ServicePermission.READ]},\n"
            "    'service-b-key': {\n"
            "        'name': 'service-b',\n"
            "        'permissions': [\n"
            "          ServicePermission.READ,\n"
            "          ServicePermission.WRITE]},\n"
            "}\n"
            "\n"
            "async def get_api_key_client(\n"
            "    api_key: str = Depends(\n"
            "      api_key_header)):\n"
            "    '''Validate API key.''' \n"
            "    if not api_key or (\n"
            "      api_key not in API_KEYS):\n"
            "        raise HTTPException(\n"
            "            status_code=401,\n"
            "            detail='Invalid API key')\n"
            "    return API_KEYS[api_key]\n"
            "\n"
            "def require_api_permission(\n"
            "    perm: ServicePermission):\n"
            "    '''Permission factory.''' \n"
            "    async def check(\n"
            "        client = Depends(\n"
            "          get_api_key_client)):\n"
            "        if perm not in (\n"
            "          client['permissions']):\n"
            "            raise HTTPException(\n"
            "                status_code=403,\n"
            "                detail=\n"
            "                  'Permission denied')\n"
            "        return client\n"
            "    return check\n"
            "\n"
            "# Usage\n"
            "@app.post('/internal/embed')\n"
            "async def embed(\n"
            "    client = Depends(\n"
            "      require_api_permission(\n"
            "        ServicePermission.WRITE))):\n"
            "    return {'service':\n"
            "      client['name']}"
        )

    def get_refresh_rotation(self) -> str:
        """Refresh token rotation with reuse detection."""
        return (
            "# === REFRESH TOKEN ROTATION ===\n"
            "import hashlib\n"
            "\n"
            "# DB table: refresh_tokens\n"
            "# user_id, token_hash, family_id,\n"
            "#   expires_at, revoked_at,\n"
            "#   rotation_count\n"
            "\n"
            "def hash_token(token: str):\n"
            "    return hashlib.sha256(\n"
            "        token.encode()).hexdigest()\n"
            "\n"
            "@app.post('/token')\n"
            "async def login(\n"
            "    form: OAuth2Password\n"
            "      RequestForm):\n"
            "    '''Login and issue tokens.''' \n"
            "    user = await (\n"
            "      authenticate_user(\n"
            "        form.username,\n"
            "        form.password))\n"
            "    if not user:\n"
            "        raise HTTPException(\n"
            "            status_code=401,\n"
            "            detail=\n"
            "              'Invalid credentials')\n"
            "    \n"
            "    access = create_access_token(\n"
            "        {'sub': user.username,\n"
            "         'role': user.role})\n"
            "    refresh = (\n"
            "      create_refresh_token(\n"
            "        {'sub': user.username}))\n"
            "    \n"
            "    # Store refresh hash\n"
            "    family_id = str(uuid.uuid4())\n"
            "    await db.execute(\n"
            "        'INSERT INTO '\n"
            "        'refresh_tokens '\n"
            "        '(user_id, token_hash, '\n"
            "        'family_id, expires_at) '\n"
            "        'VALUES (?,?,?,?)',\n"
            "        user.id, hash_token(\n"
            "          refresh),\n"
            "        family_id,\n"
            "        datetime.utcnow() +\n"
            "          timedelta(days=7))\n"
            "    \n"
            "    return Token(\n"
            "        access_token=access,\n"
            "        refresh_token=refresh)\n"
            "\n"
            "@app.post('/refresh')\n"
            "async def refresh_token(\n"
            "    refresh: str):\n"
            "    '''Rotate refresh token.''' \n"
            "    token_hash = hash_token(\n"
            "        refresh)\n"
            "    \n"
            "    # Check if valid\n"
            "    row = await db.fetch_one(\n"
            "        'SELECT * FROM '\n"
            "        'refresh_tokens WHERE '\n"
            "        'token_hash = ? AND '\n"
            "        'revoked_at IS NULL',\n"
            "        token_hash)\n"
            "    \n"
            "    if not row:\n"
            "        # Check if revoked\n"
            "        #   (reuse detection)\n"
            "        revoked = (\n"
            "          await db.fetch_one(\n"
            "            'SELECT family_id '\n"
            "            'FROM refresh_tokens '\n"
            "            'WHERE token_hash = ?',\n"
            "            token_hash))\n"
            "        if revoked:\n"
            "            # COMPROMISE!\n"
            "            #   Revoke family\n"
            "            await db.execute(\n"
            "                'UPDATE '\n"
            "                'refresh_tokens '\n"
            "                'SET revoked_at = ? '\n"
            "                'WHERE family_id = ?',\n"
            "                datetime.utcnow(),\n"
            "                revoked[\n"
            "                  'family_id'])\n"
            "        raise HTTPException(\n"
            "            status_code=401,\n"
            "            detail='Invalid token')\n"
            "    \n"
            "    # Revoke old token\n"
            "    await db.execute(\n"
            "        'UPDATE refresh_tokens '\n"
            "        'SET revoked_at = ? '\n"
            "        'WHERE id = ?',\n"
            "        datetime.utcnow(),\n"
            "        row['id'])\n"
            "    \n"
            "    # Issue new refresh\n"
            "    new_refresh = (\n"
            "      create_refresh_token(\n"
            "        {'sub': row['user_id']}))\n"
            "    await db.execute(\n"
            "        'INSERT INTO '\n"
            "        'refresh_tokens '\n"
            "        '(user_id, token_hash, '\n"
            "        'family_id, expires_at) '\n"
            "        'VALUES (?,?,?,?)',\n"
            "        row['user_id'],\n"
            "        hash_token(new_refresh),\n"
            "        row['family_id'],\n"
            "        datetime.utcnow() +\n"
            "          timedelta(days=7))\n"
            "    \n"
            "    new_access = (\n"
            "      create_access_token(\n"
            "        {'sub': row['user_id']}))\n"
            "    \n"
            "    return Token(\n"
            "        access_token=new_access,\n"
            "        refresh_token=\n"
            "          new_refresh)"
        )

    def get_production_tips(self) -> dict:
        """Production tips."""
        return {
            "layered_deps": "Layered chain: extract token, load user, enforce permissions. Each layer one responsibility, raises 401/403.",
            "jwt_vs_opaque": "JWT for stateless. Opaque tokens if Redis/DB hit per request — simpler, easier revoke, no crypto missteps.",
            "access_ttl": "Access token TTL: 15-60 min. Short-lived. Include only needed claims — clients can read JWT.",
            "refresh_ttl": "Refresh token TTL: 7-30 days. Rotation with reuse detection. Reuse = compromise, revoke family.",
            "api_keys": "APIKeyHeader for service-to-service. Store in secrets manager. Permission-based access. Rotate regularly.",
            "cookies": "HttpOnly + Secure + SameSite=Lax for browser. Never localStorage (XSS). CSRF middleware with cookies.",
            "password": "Argon2id (default) or bcrypt. Never store plaintext. passlib.context.CryptContext.",
            "rbac": "require_role and require_permission factory functions. Per-route enforcement. Readable dependencies.",
            "revocation": "Plan revocation path: token blacklist in Redis, session revoke by device, family revoke on reuse.",
            "audit": "Structured audit logging for all privileged endpoints. Rate limiting on auth endpoints.",
        }

AI API Authentication Checklist

• [ ] Layered dependency chain: Layer 1 extract/validate token, Layer 2 load user, Layer 3 enforce permissions
• [ ] Each layer has one responsibility, each raises 401 or 403 with clear error
• [ ] OAuth2PasswordBearer(tokenUrl='/token') for token extraction
• [ ] JWT for stateless validation — opaque tokens if Redis/DB hit per request (simpler, easier revoke)
• [ ] Stripe and GitHub use opaque tokens — JWT not always the right choice
• [ ] JWT claims readable by clients — include only what needed (sub, role, exp)
• [ ] Access token TTL: 15-60 minutes — short-lived
• [ ] Refresh token TTL: 7-30 days — longer lived, stored securely
• [ ] Refresh token rotation: each refresh produces new token, invalidates old one
• [ ] Refresh token reuse detection: if revoked token presented, invalidate entire token family
• [ ] OAuth 2.1 recommendation: rotation with reuse detection requires server-side state
• [ ] Store refresh token hash in DB: user_id, token_hash, family_id, expires_at, revoked_at
• [ ] One row per active session — cheap, enables revocation
• [ ] Auth0 and Okta ship refresh rotation by default in 2026
• [ ] API keys for service-to-service communication (low complexity)
• [ ] APIKeyHeader(name='X-API-Key', auto_error=False) for API key extraction
• [ ] Store API keys with service_name and permissions in secrets manager
• [ ] require_permission factory function for permission-checking dependencies
• [ ] Multiple API keys with different permission levels (read, write, admin)
• [ ] Rotate API keys regularly — never hardcode in source
• [ ] RBAC: RoleChecker with allowed_roles list, require_role factory
• [ ] RBAC: require_permission factory for fine-grained per-route enforcement
• [ ] RBAC: current_user, require_role(), require_permission() dependencies
• [ ] Password hashing: Argon2id (default) or bcrypt via passlib.context.CryptContext
• [ ] Never store plaintext passwords — always hash
• [ ] Browser apps: HttpOnly + Secure + SameSite=Lax cookies — right default in 2026
• [ ] Never store tokens in localStorage — XSS exposure materially worse
• [ ] CSRF middleware when using cookie transport
• [ ] Server-to-server and mobile: bearer tokens appropriate, short access TTL
• [ ] JWT SECRET_KEY must never be hardcoded — load from environment variables
• [ ] Consider asymmetric keys (RS256) if multiple services verify tokens
• [ ] Token blacklist in Redis for JWT revocation
• [ ] Session management: list active sessions (device, IP, last used), revoke one device
• [ ] Rate limiting on auth endpoints (login, refresh) — prevent brute force
• [ ] Structured audit logging for all privileged endpoints
• [ ] Email verification for user registration
• [ ] OAuth2 social login (Google, GitHub) as optional
• [ ] Passkey/WebAuthn support as modern auth option
• [ ] Security headers middleware
• [ ] Write tests: one test per endpoint category (public, authenticated, admin)
• [ ] Read FastAPI tutorial for API setup
• [ ] Read async backend for async patterns
• [ ] Read Docker deployment for deployment
• [ ] Read rate limiting for API protection
• [ ] Test: login returns valid access and refresh tokens
• [ ] Test: protected endpoints reject invalid tokens (401)
• [ ] Test: role-based endpoints reject insufficient roles (403)
• [ ] Test: API key auth works for service-to-service
• [ ] Test: refresh token rotation issues new tokens
• [ ] Test: refresh token reuse detection revokes family
• [ ] Test: permission factory enforces per-route access
• [ ] Document auth methods, dependency chain, token lifecycle, rotation flow, RBAC design

FAQ

How do you implement authentication in FastAPI for AI APIs?

Use layered dependency chain: token extraction, user loading, permission enforcement. Safeguard: "FastAPI dependency system is powerful and where most auth bugs originate. Layered chain: low-level dependency extracts and validates token, middle dependency loads user from validated claims, high-level dependencies enforce specific permissions. Each layer one responsibility, each raises 401 or 403 with clear error." TheLinuxCode: "OAuth2 password flow + JWT access tokens for service-to-service and mobile. User table with hashed password and role, token endpoint for login, dependency that validates JWT and loads current user, authorization dependencies for role checks." OneUptime: "API Keys for service-to-service (low complexity), JWT for user auth (medium), OAuth2 for third-party (high). OAuth2PasswordBearer for token extraction." Implementation: (1) OAuth2PasswordBearer(tokenUrl='/token'). (2) Layered deps: get_token -> get_current_user -> require_role. (3) JWT or opaque tokens. (4) API keys for service-to-service. (5) RBAC with require_role/require_permission. (6) Each layer raises 401/403.

Should you use JWT or opaque tokens for AI API authentication?

Use JWT for stateless validation; opaque tokens with session lookup if you hit Redis/DB every request. Safeguard: "Use JWTs only if you genuinely need stateless validation. Reflexive choice of JWT for every API is most common over-engineering. If auth flow involves hit to Redis or Postgres on every request, opaque tokens with session lookup are simpler, easier to revoke, avoid entire family of JWT cryptographic missteps. Stripe and GitHub both ship API tokens as opaque strings." Decision: (1) JWT: stateless, no DB lookup, short-lived (15-60 min), harder to revoke. (2) Opaque tokens: session lookup per request, easy revocation, simpler, no crypto missteps. (3) Stripe, GitHub use opaque tokens. (4) JWT for distributed services where multiple services verify tokens. (5) Opaque for single-service APIs with Redis/DB hit anyway. (6) JWT claims readable by clients — include only what needed.

How do you implement refresh token rotation with reuse detection?

Each refresh produces new token and invalidates old one; reuse triggers session family revocation. Safeguard: "Refresh-token rotation with reuse detection. Each refresh produces new refresh token and invalidates old one. If previously-used refresh token presented again, treat as compromise signal and invalidate entire token family for user. OAuth 2.1 recommendation, requires server-side state. One row per active session: token family ID, current token hash, user ID, rotation counter. Auth0 and Okta ship this by default in 2026." TheLinuxCode: "Refresh tokens table with user_id, token_hash, expires_at, revoked_at. Store hash(token). On refresh, verify hash exists and not revoked, revoke old and issue new." Rotation: (1) Store refresh token hash in DB. (2) On refresh: verify hash, revoke old, issue new. (3) Reuse detection: if revoked token presented, invalidate entire family. (4) Token family ID for tracking. (5) Rotation counter. (6) Access TTL: 15-60 min, refresh TTL: 7-30 days.

How do you implement API key authentication for service-to-service AI APIs?

Use APIKeyHeader dependency with permission-based access control. OneUptime: "API Keys for service-to-service communication (low complexity). Store API keys with permissions. APIKeyHeader(name=X-API-Key). require_permission factory function for permission-checking dependencies." API keys: (1) APIKeyHeader(name='X-API-Key', auto_error=False). (2) Store keys with service_name and permissions in DB/secrets. (3) async def get_api_key_client(api_key=Depends(api_key_header)): lookup, raise 401 if invalid. (4) require_permission(permission) factory: check permission, raise 403 if missing. (5) @app.post('/data', dependencies=[Depends(require_permission(ServicePermission.WRITE))]). (6) Multiple keys with different permission levels. (7) Rotate keys regularly. (8) Store in secrets manager, not hardcoded.

How do you secure browser-facing AI applications with cookies vs bearer tokens?

Use HttpOnly cookies for browser apps; bearer tokens for server-to-server and mobile. Safeguard: "For browser-facing applications, session cookies with HttpOnly, Secure, SameSite=Lax flags remain right default in 2026. Bearer tokens in localStorage still recommended in tutorials and should not be. XSS exposure materially worse. For server-to-server and mobile, bearer tokens appropriate: short access TTL 15-60 min, refresh rotation, clear revocation path." FastAPI FullAuth: "Bearer or cookie transport: opt into HttpOnly cookies that carry both access and refresh tokens, out of JavaScript reach. Wire CSRFMiddleware when using cookie transport." Security: (1) Browser: HttpOnly + Secure + SameSite=Lax cookies. (2) Server/mobile: bearer tokens, short TTL. (3) Never store tokens in localStorage (XSS). (4) CSRF middleware with cookie transport. (5) Access TTL: 15-60 min. (6) Refresh TTL: 7-30 days. (7) Revocation path for both patterns.

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