Retries and Backoff for CLI HTTP Calls

Your deploy script calls the release API, gets a 502 Bad Gateway because a load balancer was rotating an instance, and aborts — leaving a half-finished release and a user who reruns it by hand thirty seconds later, when it works. A short, polite retry would have hidden the blip completely. But retries have sharp edges: retrying a request that the server already processed can charge a customer twice, retrying in a tight loop turns a struggling service into a failed one, and retrying a 400 Bad Request just wastes a minute before showing the same error. This guide builds a small retry layer for an httpx-based CLI that retries the right failures, waits the right amount, respects the server, and is fully testable without real sleeps. It is part of the HTTP APIs topic.

Prerequisites

• Python 3.10+ and httpx.
• An API client module along the lines of building an API client CLI with httpx.
• Awareness of HTTP method semantics: GET, PUT and DELETE are idempotent; POST generally is not.

What to retry, and when it is safe

A retry policy answers two questions: is this failure transient, and is repeating the request harmless?

Transient failures are those where trying again later has a real chance of succeeding: connection failures, timeouts, 429 Too Many Requests, and the 5xx family — especially 502, 503 and 504, which usually mean a gateway could not reach a healthy backend. Everything in the 4xx range other than 429 (and occasionally 408) is permanent: the request itself is wrong and will be just as wrong in five seconds.

Harmlessness depends on the method and on where the failure happened:

A failed connect is always safe to retry: nothing reached the server. A read timeout is the dangerous case — the request was sent, and the server may have acted on it before the response was lost. For GET, PUT and DELETE that does not matter, since repeating them produces the same end state. For POST, retry only if the API supports an idempotency key: a unique header (commonly Idempotency-Key) that lets the server recognise the repeat and return the original result instead of acting twice.

How long to wait

Retrying immediately rarely helps — the condition that caused the failure is still there — and when many clients do it at once they amplify the overload. The standard answer is exponential backoff: wait a base delay, doubling on each attempt, up to a cap. Then add jitter, a random component that spreads clients out so they do not all retry in the same instant.

With a base of 0.5 seconds and a cap of 8, the first retry waits up to half a second and the fifth up to eight. "Full jitter" — choosing uniformly between zero and the computed delay — is simple and performs well in practice. For an interactive CLI, keep the total budget modest: a person will tolerate fifteen seconds of "retrying..." far better than two minutes.

Servers often know better than any formula. A 429 or 503 may carry a Retry-After header, either in seconds or as an HTTP date. Honour it when present, capped at a maximum so a misconfigured server cannot put your tool to sleep for an hour.

The recipe

The wrapper below sits in your API module and wraps client.send(). Sleep and randomness are injectable, which is what makes it testable without waiting.

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

import random
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone

import httpx

RETRY_STATUSES = frozenset({429, 502, 503, 504})
IDEMPOTENT = frozenset({"GET", "HEAD", "OPTIONS", "PUT", "DELETE"})


@dataclass
class RetryPolicy:
    attempts: int = 5                 # total tries, including the first
    base: float = 0.5
    cap: float = 8.0
    max_retry_after: float = 60.0
    sleep: Callable[[float], None] = time.sleep
    rand: Callable[[], float] = random.random
    on_retry: Callable[[int, float, str], None] = field(default=lambda n, d, why: None)

    def backoff(self, retry_number: int) -> float:
        ceiling = min(self.cap, self.base * 2 ** (retry_number - 1))
        return ceiling * self.rand()          # full jitter


def retry_after_seconds(response: httpx.Response) -> float | None:
    value = response.headers.get("Retry-After")
    if value is None:
        return None
    if value.strip().isdigit():
        return float(value)
    try:
        when = parsedate_to_datetime(value)
    except (TypeError, ValueError):
        return None
    return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())


def send_with_retries(client: httpx.Client, request: httpx.Request,
                      policy: RetryPolicy | None = None) -> httpx.Response:
    policy = policy or RetryPolicy()
    can_repeat = request.method in IDEMPOTENT or "Idempotency-Key" in request.headers
    for attempt in range(1, policy.attempts + 1):
        last = attempt == policy.attempts
        try:
            response = client.send(request)
        except httpx.ConnectError:
            if last:
                raise
            why = "connection failed"          # nothing was sent: always safe
        except (httpx.ReadTimeout, httpx.RemoteProtocolError):
            if last or not can_repeat:
                raise
            why = "no response"
        else:
            if response.status_code not in RETRY_STATUSES or last:
                return response
            if response.status_code != 429 and not can_repeat:
                return response
            why = f"HTTP {response.status_code}"
            server_delay = retry_after_seconds(response)
            response.close()
            if server_delay is not None:
                delay = min(server_delay, policy.max_retry_after)
                policy.on_retry(attempt, delay, why)
                policy.sleep(delay)
                continue
        delay = policy.backoff(attempt)
        policy.on_retry(attempt, delay, why)
        policy.sleep(delay)
    raise AssertionError("unreachable")

Using it from the API layer is a one-line change per request: build the request, then send it through the wrapper. The on_retry hook is how the command layer tells the user what is happening:

import typer

from mytool.retry import RetryPolicy, send_with_retries


def announce(n: int, delay: float, why: str) -> None:
    typer.echo(f"  {why}; retrying in {delay:.1f}s (attempt {n + 1})", err=True)


def get_json(client, path: str):
    request = client.build_request("GET", path)
    response = send_with_retries(client, request, RetryPolicy(on_retry=announce))
    response.raise_for_status()
    return response.json()

httpx.HTTPTransport(retries=N) is still worth setting: it retries failed connection attempts at the transport level, below this wrapper. The two layers are complementary — the transport handles "could not connect", the wrapper handles everything that needs a policy decision.

UX considerations

• Say that you are retrying. One stderr line per retry — the reason, the delay, the attempt — turns a mysterious pause into visible resilience. Silence for fifteen seconds reads as a hang.
• Expose the budget. A --retries N option (with 0 to disable) helps in CI, where a fast failure may be preferable, and in flaky networks, where users want more patience.
• Report the final failure with its history. "the service returned 503 after 5 attempts over 14s" tells the user that retrying by hand right now is pointless.
• Never retry user errors. A typo in a project name producing a 404 should fail on the first attempt, instantly.
• Respect Ctrl+C during waits. time.sleep is interruptible by KeyboardInterrupt, so a user can always abandon a retry loop; make sure your top-level handler exits cleanly, as in handling KeyboardInterrupt cleanly.

Testing the behaviour

Inject a fake sleep that records delays and a fixed rand, then drive responses from a MockTransport that replays a scripted sequence. The whole suite runs instantly:

# tests/test_retry.py
import httpx
import pytest

from mytool.retry import RetryPolicy, send_with_retries


def scripted(*outcomes):
    """A transport that returns (or raises) each outcome in turn."""
    calls = []

    def handler(request):
        calls.append(request)
        outcome = outcomes[len(calls) - 1]
        if isinstance(outcome, Exception):
            raise outcome
        return outcome

    return httpx.Client(transport=httpx.MockTransport(handler), base_url="https://api.test"), calls


def policy(slept):
    return RetryPolicy(sleep=slept.append, rand=lambda: 1.0)


def test_retries_503_then_succeeds():
    client, calls = scripted(httpx.Response(503), httpx.Response(503), httpx.Response(200))
    slept = []
    r = send_with_retries(client, client.build_request("GET", "/x"), policy(slept))
    assert r.status_code == 200
    assert len(calls) == 3
    assert slept == [0.5, 1.0]


def test_does_not_retry_404():
    client, calls = scripted(httpx.Response(404))
    r = send_with_retries(client, client.build_request("GET", "/x"), policy([]))
    assert r.status_code == 404 and len(calls) == 1


def test_honours_retry_after():
    client, _ = scripted(httpx.Response(429, headers={"Retry-After": "3"}), httpx.Response(200))
    slept = []
    send_with_retries(client, client.build_request("GET", "/x"), policy(slept))
    assert slept == [3.0]


def test_post_is_not_retried_on_503_without_key():
    client, calls = scripted(httpx.Response(503), httpx.Response(200))
    r = send_with_retries(client, client.build_request("POST", "/x"), policy([]))
    assert r.status_code == 503 and len(calls) == 1


def test_post_with_idempotency_key_is_retried():
    client, calls = scripted(httpx.Response(503), httpx.Response(201))
    req = client.build_request("POST", "/x", headers={"Idempotency-Key": "abc"})
    assert send_with_retries(client, req, policy([])).status_code == 201


def test_gives_up_after_budget():
    client, calls = scripted(*[httpx.ConnectError("down")] * 5)
    with pytest.raises(httpx.ConnectError):
        send_with_retries(client, client.build_request("GET", "/x"), policy([]))
    assert len(calls) == 5


def test_backoff_is_capped():
    p = RetryPolicy(rand=lambda: 1.0)
    assert [p.backoff(n) for n in range(1, 8)] == [0.5, 1, 2, 4, 8, 8, 8]

With rand fixed at 1.0 the jitter disappears and delays are exact, which makes the backoff schedule itself a tested contract. The same scripted-transport idea extends to testing whole commands, as in mocking filesystem and network in CLI tests.

Conclusion

Good retries are narrow and polite: only transient failures, only requests that are safe to repeat, capped exponential backoff with jitter, the server's Retry-After when it gives one, and a visible message each time. Keep the policy in one small function with injectable sleep and randomness, and it becomes one of the best-tested parts of your CLI rather than an untested loop nobody dares touch.

Frequently asked questions

Related

• Up: Calling HTTP APIs from Python CLIs
• Building an API client CLI with httpx
• Downloading files with progress in Python
• Rate-limiting concurrent requests in CLIs
• Friendly error messages and tracebacks