OAuth2 client credentials flow for server-to-server calls

The OAuth2 client credentials flow server-to-server pattern replaces long-lived API keys with short-lived bearer tokens minted by an authorization server. For backend services calling LLM platforms, it removes the scramble of rotating static secrets and gives you per-client audit trails without baking credentials into every deployment.

Step 1: Register your client with the authorization server

Before any code, you need a client registration. The auth server (Keycloak, Auth0, or a platform-native issuer) hands you a client_id and client_secret, plus the token endpoint URL and the scopes your service needs.

Request the narrowest scope possible. If the LLM platform exposes model:invoke or usage:read, take only what the calling service uses. Broad scopes inflate blast radius when a secret leaks.

Store the secret in your environment or a vault, never in source control:

export LLM_CLIENT_ID="svc-inference-prod"
export LLM_CLIENT_SECRET="********"
export TOKEN_URL="https://auth.example.com/oauth2/token"

If the issuer supports an audience parameter (common with Auth0-style servers), note the API identifier you must pass so the returned JWT carries the correct aud claim.

Step 2: Request a token from the token endpoint

The client credentials grant is a single POST. Send credentials via HTTP Basic auth and the grant parameters in the form body. I prefer Basic auth because it keeps the body clean and avoids leaking the secret in logs that capture request bodies.

import os
import requests

def fetch_token(audience: str | None = None) -> dict:
    payload = {"grant_type": "client_credentials", "scope": "model:invoke"}
    if audience:
        payload["audience"] = audience
    resp = requests.post(
        os.environ["TOKEN_URL"],
        auth=(os.environ["LLM_CLIENT_ID"], os.environ["LLM_CLIENT_SECRET"]),
        data=payload,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()

token = fetch_token()
print(token["access_token"], token["expires_in"])

A successful response looks like:

{
  "access_token": "eyJhbGciOi...",
  "token_type": "Bearer",
  "expires_in": 3600,
  "scope": "model:invoke"
}

Validate three things before using it: token_type is Bearer (case-insensitive), expires_in is a positive integer, and the scope string contains what you requested. If your auth server rejects Basic, move credentials into the body as client_id and client_secret. The spec allows both; pick what your issuer documents.

Step 3: Call the protected API with the bearer token

Strip the token type check to lowercase and prefix the header. Most LLM gateways expect Authorization: Bearer <token>. When you point a service at n4n.ai, the same OAuth2 client credentials flow server-to-server token works against its OpenAI-compatible endpoint that addresses 240+ models, and the gateway forwards your provider cache-control hints without extra config.

def chat_completion(token: str, prompt: str) -> str:
    resp = requests.post(
        "https://api.n4n.ai/v1/chat/completions",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": 128,
        },
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

For non-JWT opaque tokens, treat the string as opaque. For JWTs, decode locally only to inspect exp or scope; never trust a locally decoded JWT for authorization without signature verification against the issuer’s JWKS.

Step 4: Cache and refresh tokens correctly

Fetching a token on every request wastes round-trips and can trip rate limits on the auth server. Cache the token in memory with a safety buffer. I use a simple singleton with a lock; for multi-process workers, use a shared cache like Redis.

import time
import threading

class TokenCache:
    def __init__(self, buffer_seconds: int = 30):
        self._lock = threading.Lock()
        self._token = None
        self._expires_at = 0
        self._buffer = buffer_seconds

    def get(self) -> str:
        with self._lock:
            if time.time() < self._expires_at - self._buffer:
                return self._token
            data = fetch_token()
            self._token = data["access_token"]
            self._expires_at = time.time() + data["expires_in"]
            return self._token

    def force_expire(self) -> None:
        with self._lock:
            self._expires_at = 0

cache = TokenCache()

The buffer prevents using a token that expires mid-flight. If your auth server returns expires_in: 0 for very short tokens, skip caching and fetch per call. In a distributed setup, store the token under a key like oauth:client:svc-inference-prod with the TTL set to expires_in - buffer so all workers share one token and one refresh cycle.

Step 5: Handle errors and edge cases

A 401 from the resource server means the token is expired, revoked, or scoped wrong. Force a refresh and retry once:

def call_with_retry(prompt: str):
    for attempt in (0, 1):
        try:
            token = cache.get()
            return chat_completion(token, prompt)
        except requests.HTTPError as e:
            if e.response.status_code == 401 and attempt == 0:
                cache.force_expire()
                continue
            raise

Clock skew between your host and the auth server can make a token appear valid but get rejected. The 30-second buffer in Step 4 covers most cases. For stricter environments, query the auth server’s /.well-known/oauth-authorization-server for clock_skew if published.

Scope mismatches surface as invalid_scope at token endpoint or 403 at the API. Read error_description and adjust your registration, not your code. If the token endpoint returns 429, back off exponentially; auth servers are usually lower-capacity than the inference gateway.

Token revocation is out-of-band. If you suspect a leak, rotate the secret at the issuer; cached tokens remain valid until expires_in elapses, so keep lifetimes short (under an hour) in production.

Step 6: Verify the integration end to end

Verification should prove three things: the token endpoint returns a well-formed token, the bearer call succeeds, and refresh works after expiry.

Run a manual curl chain:

TOKEN=$(curl -s -u "$LLM_CLIENT_ID:$LLM_CLIENT_SECRET" \
  -d grant_type=client_credentials -d scope=model:invoke \
  $TOKEN_URL | jq -r .access_token)

curl -s https://api.n4n.ai/v1/chat/completions \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"ping"}]}' \
  | jq .choices[0].message.content

A returned string confirms the OAuth2 client credentials flow server-to-server call is wired correctly.

For automated confidence, write a pytest smoke test:

def test_token_and_call():
    data = fetch_token()
    assert data["token_type"].lower() == "bearer"
    assert data["expires_in"] > 0
    out = chat_completion(data["access_token"], "say ok")
    assert isinstance(out, str) and len(out) > 0

def test_refresh_path():
    c = TokenCache()
    t1 = c.get()
    c.force_expire()
    t2 = c.get()
    assert t1 != t2  # new token after forced expiry

Run these in CI against a staging issuer. Never point integration tests at production secrets.

Production notes

Rotate client secrets on a schedule. Most auth servers support two active secrets during overlap; swap them in your vault and redeploy without downtime.

Don’t log access_token values. Redact them in middleware:

import logging
logging.getLogger("requests").setLevel(logging.WARNING)

If your platform supports mutual TLS, bind the client certificate to the OAuth client for a second factor stronger than a shared secret.

The OAuth2 client credentials flow server-to-server model is boring in the best way: predictable tokens, clear scopes, and no key sprawl. Implement the cache once, test the refresh path, and your backend can talk to any compliant LLM gateway without touching credential files again.