HTTPX: test timeouts and retries before connecting an AI workflow

Give the lookup a failure policy before adding AI

An AI worker that checks a job status still depends on ordinary network code. A failed connection, slow response or rejected credential should produce a predictable outcome. Otherwise, the surrounding agent may wait too long, repeat work or mistake an unavailable service for a successful lookup.

This AI-assisted editorial guide implements a deliberately narrow policy: a fixed read-only GET can make two attempts when connection establishment fails. Other errors stop immediately. Six local scenarios test that behavior. This is an original mocked example, not a live-service reliability benchmark or a complete agent runtime.

1. Install a pinned client in a separate environment

The example was executed on Windows with Python 3.13.15 and HTTPX 0.28.1 on October 14, 2026. The package supports Python 3.8 or newer. Use the PowerShell commands below, save both Python blocks in order as retry_demo.py, then run the last command. Installation needs internet access; the test itself does not. [1]

No model key, CRM account or customer record is required. The address ending in example.invalid is a dummy destination intercepted by the test transport. Keep it unchanged while running this demonstration.

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install "httpx==0.28.1"
# Save both Python blocks below as retry_demo.py first.
.\.venv\Scripts\python.exe retry_demo.py

2. Separate timeout settings from retry decisions

HTTPX exposes connect, read, write and pool timeouts. Connect concerns establishing the connection; read and write concern waiting for chunks of data; pool concerns obtaining a connection. These settings are not a single elapsed-time limit for the entire business task. A response that keeps delivering chunks can take longer than the read timeout in total. [2]

Here the illustrative limits are 3, 10, 5 and 2 seconds respectively. Choose real values from the dependency’s behavior and your workflow’s time budget. The helper retries only ConnectTimeout or ConnectError, waits a quarter second once and re-raises a second failure. Its default pause is real; the test replaces it with a recorder.

Calling raise_for_status before decoding JSON makes an unsuccessful HTTP response raise HTTPStatusError. Timeout errors and HTTP status errors are different exception categories, so a broad catch-and-retry rule would erase useful distinctions. [3][4]

import time
import httpx

TIMEOUT = httpx.Timeout(connect=3.0, read=10.0, write=5.0, pool=2.0)

def fetch_status(client, pause=time.sleep):
    # Only for this fixed, read-only GET. Two attempts total.
    for attempt in range(2):
        try:
            response = client.get("https://status.example.invalid/job/demo")
            response.raise_for_status()
            return response.json()
        except (httpx.ConnectTimeout, httpx.ConnectError):
            if attempt == 1:
                raise
            pause(0.25)

3. Exercise success and failure without a live API

HTTPX MockTransport accepts a handler that supplies responses or raises errors locally. Append this block to the same file. Each scenario receives a fresh client and sequence, so request counts are deterministic. The handler also checks that the four configured timeout values reached the request. [5]

The tests record the requested pause rather than sleeping. They verify our control flow and configuration, not real socket timing, DNS behavior or the wall-clock enforcement of network timeouts. No HTTP request leaves this test transport.

# TESTS: MockTransport intercepts all requests; no network is used.
def check(name, events, expected_calls, expected_error=None):
    calls, pauses = [], []
    sequence = iter(events)
    def handler(request):
        calls.append(request)
        assert request.extensions["timeout"] == {
            "connect": 3.0, "read": 10.0, "write": 5.0, "pool": 2.0,
        }
        event = next(sequence)
        if isinstance(event, type) and issubclass(event, Exception):
            raise event("Synthetic failure", request=request)
        return httpx.Response(event, json={"state": "ready"}, request=request)

    with httpx.Client(transport=httpx.MockTransport(handler), timeout=TIMEOUT) as client:
        if expected_error is None:
            assert fetch_status(client, pauses.append) == {"state": "ready"}
        else:
            try:
                fetch_status(client, pauses.append)
            except expected_error:
                pass
            else:
                raise AssertionError("Expected failure was not raised")
    assert len(calls) == expected_calls
    assert pauses == ([0.25] if expected_calls == 2 else [])
    print(f"{name}: {len(calls)} attempt(s), passed")

check("success", [200], 1)
check("connect then success", [httpx.ConnectTimeout, 200], 2)
check("connect exhausted", [httpx.ConnectError, httpx.ConnectError], 2, httpx.ConnectError)
check("read timeout", [httpx.ReadTimeout], 1, httpx.ReadTimeout)
check("HTTP 401", [401], 1, httpx.HTTPStatusError)
check("HTTP 503", [503], 1, httpx.HTTPStatusError)

4. Read the six results as a small contract

All six scenarios passed locally. Immediate success used one attempt. A connection timeout followed by success used two. Repeated connection errors stopped after two. A read timeout, HTTP 401 and HTTP 503 each stopped after one. Only the two connection-retry scenarios requested a pause.

Leaving 503 unretried is intentional here. If your service needs a rate-limit or temporary-server-error policy, design it separately around that API’s guidance, including its retry hints, a cap and a job deadline. A credential rejection should route to a configuration check, not an endless loop. A JSON decoding error also remains visible rather than becoming a fabricated status.

5. Connect the tested helper to a real workflow

Before replacing the mock, verify that the real endpoint is genuinely read-only and approved for this worker. Keep the destination configured by the application rather than selected freely by a model. Return a clear unavailable or review-needed outcome when attempts are exhausted; never report that a job is ready merely because a lookup failed.

HTTPX also offers connection retries through HTTPTransport. Its documented retry option covers ConnectError and ConnectTimeout; handling read/write errors or status codes requires another policy. Choose one retry layer. Combining an SDK retry, a transport retry and a workflow retry can multiply requests beyond the budget you intended. [5]

Test real timing in a controlled staging service, then measure the overall job deadline independently. Log attempt counts and error categories without copying credentials or customer response bodies. Finally, evaluate any write operation separately: creating an invoice or sending a message requires duplicate-prevention and reconciliation decisions that this read-only example does not implement.

Sources & further reading

1. PyPI: HTTPX 0.28.1 installation and Python requirements
2. HTTPX documentation: timeouts
3. HTTPX documentation: exceptions
4. HTTPX QuickStart: response status codes
5. HTTPX documentation: HTTP and mock transports

Make your AI worker’s integrations dependable

Sociologix can help define API permissions, timeout budgets, failure handling and human escalation for AI agents and business automation. Bring us one workflow and its external dependencies.