Refreshing Expired Tokens Automatically in a Python CLI

A CLI that logs in with OAuth gets two tokens: a short-lived access token, sent with every API request and typically valid for an hour, and a long-lived refresh token, used only to obtain new access tokens. Users should notice neither. They log in once with mytool login, and from then on every command just works — for weeks — until the refresh token itself expires or is revoked, at which point the CLI says clearly “your session has expired; run: mytool login”. Getting there means refreshing at the right moments: shortly before the access token expires, and once more if the server rejects a token that looked valid. It also means saving the new refresh token when the server rotates it, and not stampeding the token endpoint when several requests notice expiry at once. This guide implements all of that as an httpx.Auth class, so every request made through the client is covered, and tests it against a fake authorisation server. It belongs to the secrets and credentials topic.

Prerequisites

• A login flow that yields an access token, a refresh token and an expiry, such as OAuth device flow login for CLIs.
• uv add httpx (examples checked with httpx 0.28 and Python 3.13).

When to refresh

There are two triggers. Proactively, when the stored expiry is near: refreshing 60 seconds early avoids sending a token that expires in transit, and saves a wasted round trip. Reactively, when the API answers 401 although the token should still be valid — it was revoked, the server’s clock differs from yours, or the provider rotated signing keys. A reactive refresh happens once per request; a second 401 after a fresh token means something else is wrong, and retrying further would loop. When the refresh itself fails with invalid_grant (HTTP 400 or 401 from the token endpoint), the session is over: clear the stored tokens and ask the user to log in.

The recipe

# src/mytool/auth.py
from __future__ import annotations

import json
import threading
import time
from collections.abc import Generator
from dataclasses import asdict, dataclass
from pathlib import Path

import httpx

TOKEN_URL = "https://auth.example.com/oauth/token"
CLIENT_ID = "mytool-cli"
EARLY = 60                                         # refresh this many seconds before expiry


class LoginRequired(Exception):
    """The refresh token is missing, expired or revoked; the user must log in again."""


@dataclass
class Tokens:
    access_token: str
    refresh_token: str
    expires_at: float

    def expiring(self, now: float | None = None) -> bool:
        return (now or time.time()) >= self.expires_at - EARLY


class TokenStore:
    """File-backed store; swap in keyring for real credentials."""

    def __init__(self, path: Path) -> None:
        self.path = path

    def load(self) -> Tokens | None:
        try:
            return Tokens(**json.loads(self.path.read_text(encoding="utf-8")))
        except FileNotFoundError:
            return None

    def save(self, tokens: Tokens) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(asdict(tokens)), encoding="utf-8")
        tmp.chmod(0o600)
        tmp.replace(self.path)

    def clear(self) -> None:
        self.path.unlink(missing_ok=True)


class RefreshingAuth(httpx.Auth):
    """Attach the access token; refresh it before expiry and once more on a 401."""

    requires_response_body = True

    def __init__(self, store: TokenStore) -> None:
        self.store = store
        self._lock = threading.Lock()

    def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, None]:
        tokens = self.store.load()
        if tokens is None:
            raise LoginRequired("not logged in; run: mytool login")
        if tokens.expiring():
            tokens = yield from self._refresh(tokens)
        request.headers["Authorization"] = f"Bearer {tokens.access_token}"
        response = yield request
        if response.status_code == 401:            # revoked early, clock skew, server restart...
            tokens = yield from self._refresh(tokens, force=True)
            request.headers["Authorization"] = f"Bearer {tokens.access_token}"
            yield request

    def _refresh(self, stale: Tokens, *, force: bool = False):
        with self._lock:
            current = self.store.load() or stale
            # Another thread (or process) may have refreshed while we waited for the lock.
            if current.access_token != stale.access_token and not current.expiring():
                return current
            response = yield httpx.Request("POST", TOKEN_URL, data={
                "grant_type": "refresh_token",
                "refresh_token": current.refresh_token,
                "client_id": CLIENT_ID,
            })
            if response.status_code in (400, 401):  # invalid_grant: expired or revoked
                self.store.clear()
                raise LoginRequired("your session has expired; run: mytool login")
            response.raise_for_status()
            body = response.json()
            fresh = Tokens(
                access_token=body["access_token"],
                refresh_token=body.get("refresh_token", current.refresh_token),   # rotation
                expires_at=time.time() + int(body.get("expires_in", 3600)),
            )
            self.store.save(fresh)
            return fresh

httpx Auth flows

httpx.Auth.auth_flow is a generator: it modifies the request, yields it, receives the response, and may yield further requests — including requests to a different URL, such as the token endpoint. That makes it the ideal place for refresh logic, because it wraps every request the client sends, including retries and pagination, without any caller knowing. requires_response_body = True tells httpx to read response bodies before handing them to the flow, which _refresh needs to parse the token response. yield from self._refresh(...) lets the refresh helper issue its own request through the same flow and return the new tokens.

Using the client is then unremarkable:

store = TokenStore(user_state_path("mytool") / "tokens.json")
with httpx.Client(base_url="https://api.example.com", auth=RefreshingAuth(store)) as client:
    me = client.get("/me").json()

Rotation and storage

Many providers issue a new refresh token with every refresh and invalidate the old one — refresh token rotation, which limits the damage of a leaked token. The new value must be saved before anything else happens: lose it, and the next refresh fails with invalid_grant and the user has to log in again. save writes to a temporary file with private permissions and renames it into place, so a crash cannot leave a half-written token file; the reasoning is in handling file permissions and umask in CLIs. For real credentials, prefer the system keychain — storing tokens with keyring — behind the same load/save/clear interface.

Avoiding refresh stampedes

When a command makes several concurrent requests — threads uploading files, say — they may all notice expiry at the same moment. Without coordination each would refresh, and with rotation all but the first would fail because the refresh token they sent is already spent. The lock serialises refreshes within the process, and the check at the top of _refresh makes latecomers reuse the token the first one obtained. Two processes — two commands in different terminals — share the token file but not the lock; re-reading the store inside _refresh handles the common case, and a file lock around the refresh, as in file locking for concurrent CLI runs, closes the gap for tools that are often run in parallel.

UX considerations

• Ask for login only when necessary, and say exactly what to run. LoginRequired carries the message; the entry point prints it and exits with a distinct code (for example 77, EX_NOPERM).
• Do not prompt mid-command. Starting an interactive login in the middle of a scripted command hangs automation. Fail with instructions instead; offer mytool login separately.
• Support non-interactive credentials. CI should use a token from an environment variable or a client-credentials grant rather than a user’s refresh token; see reading secrets from env and files.
• Show session status. mytool auth status printing who is logged in and when the session ends saves support questions; never print the tokens themselves.
• Log refreshes at debug level — “refreshed access token, expires 15:42” — without token values, as in redacting secrets from CLI output and logs.

Testing the behaviour

A fake server implemented as an httpx.MockTransport handler plays both roles: a token endpoint that rotates refresh tokens and rejects stale ones, and an API that accepts only the current access token. Each test sets up stored tokens in a particular state and checks what the client does:

# tests/test_auth.py
import time

import httpx
import pytest

from mytool.auth import TOKEN_URL, LoginRequired, RefreshingAuth, TokenStore, Tokens


class FakeServer:
    """Token endpoint plus an API that accepts only the current access token."""

    def __init__(self):
        self.valid_access = "access-1"
        self.valid_refresh = "refresh-1"
        self.refreshes = 0

    def __call__(self, request: httpx.Request) -> httpx.Response:
        if str(request.url) == TOKEN_URL:
            form = dict(httpx.QueryParams(request.content.decode()))
            if form["refresh_token"] != self.valid_refresh:
                return httpx.Response(400, json={"error": "invalid_grant"})
            self.refreshes += 1
            self.valid_access = f"access-{self.refreshes + 1}"
            self.valid_refresh = f"refresh-{self.refreshes + 1}"        # rotating refresh tokens
            return httpx.Response(200, json={"access_token": self.valid_access,
                                             "refresh_token": self.valid_refresh,
                                             "expires_in": 3600})
        if request.headers.get("authorization") != f"Bearer {self.valid_access}":
            return httpx.Response(401)
        return httpx.Response(200, json={"me": "ann"})


@pytest.fixture
def setup(tmp_path):
    server = FakeServer()
    store = TokenStore(tmp_path / "tokens.json")
    client = httpx.Client(transport=httpx.MockTransport(server), auth=RefreshingAuth(store))
    yield server, store, client
    client.close()


def test_valid_token_is_used_without_refresh(setup):
    server, store, client = setup
    store.save(Tokens("access-1", "refresh-1", time.time() + 3600))
    assert client.get("https://api.example.com/me").json() == {"me": "ann"}
    assert server.refreshes == 0


def test_expiring_token_is_refreshed_first_and_rotation_saved(setup):
    server, store, client = setup
    store.save(Tokens("access-1", "refresh-1", time.time() + 10))   # inside the early window
    assert client.get("https://api.example.com/me").status_code == 200
    assert server.refreshes == 1
    assert store.load().refresh_token == "refresh-2"


def test_401_triggers_one_refresh_and_retry(setup):
    server, store, client = setup
    store.save(Tokens("access-1", "refresh-1", time.time() + 3600))
    server.valid_access = "revoked-elsewhere"
    assert client.get("https://api.example.com/me").status_code == 200
    assert server.refreshes == 1


def test_revoked_refresh_token_requires_login(setup):
    server, store, client = setup
    store.save(Tokens("access-1", "stolen-and-rotated", time.time() - 1))
    with pytest.raises(LoginRequired, match="mytool login"):
        client.get("https://api.example.com/me")
    assert store.load() is None


def test_not_logged_in(setup):
    _, _, client = setup
    with pytest.raises(LoginRequired, match="not logged in"):
        client.get("https://api.example.com/me")

The tests cover the four situations users meet — a valid token, an expiring token, a token revoked server-side, and a dead refresh token — plus the not-logged-in case. Expiry is controlled through expires_at rather than by patching time, which keeps the tests simple. The rotation assertion (refresh-2 saved) is the one that catches the most damaging bug: forgetting to persist the new refresh token.

Conclusion

Users should log in once and then forget about tokens. Wrap the refresh logic in an httpx.Auth flow so every request is covered; refresh proactively a minute before expiry and reactively once after a 401; save rotated refresh tokens atomically and privately before continuing; serialise refreshes with a lock and re-read the store so concurrent requests reuse one new token; clear the store and raise a clear “run: mytool login” error on invalid_grant; never prompt mid-command; and test every state against a fake token server.

Frequently asked questions

Related

• Up: Secrets and credentials in Python CLIs
• OAuth device flow login for CLIs
• Storing tokens with keyring
• Supporting multiple profiles and accounts
• Retries and backoff for CLI HTTP calls