Managing OpenAI Timeouts in Python for Request Limits, Retries and Long-Running Responses

Configure the transport and the operation separately

An OpenAI timeout should not be one arbitrary number wrapped around every model call.

The official Python SDK lets you configure a timeout for the client or override it for one request. It also retries several failures twice by default. For production code, decide whether the SDK owns retries or your application does, then add a separate application deadline around the complete operation.

import httpx
from openai import AsyncOpenAI

client = AsyncOpenAI(
    timeout=httpx.Timeout(
        120.0,
        connect=5.0,
        read=60.0,
        write=10.0,
        pool=5.0,
    ),
    max_retries=0,
)

This separates connection setup from waiting for response data, but it is still not a strict 120-second deadline for model generation, retries, backoff, validation, and fallback together.

A timeout also does not prove that OpenAI rejected the request. The API may have accepted and processed it before the client stopped waiting. That creates uncertainty around both the result and billed usage.

The examples target Python 3.13 and openai 2.49.0. SDK defaults and APIs were verified against the official documentation on August 4, 2026.

What each timeout controls

The SDK accepts either a float or an httpx.Timeout object. A float is convenient, but phase-specific values make the policy visible.

The SDK currently uses a 10-minute default timeout. It also automatically retries connection errors, HTTP 408, 409, 429, and server errors at least twice by default.

That means one application call can contain several network attempts. The wall-clock duration can therefore exceed one configured timeout.

The lower-level distinction between connection setup, read inactivity, and a true end-to-end deadline is covered in What Python Requests Timeouts Mean for Connect, Read and Failure Modes. The OpenAI SDK uses a different client, but the transport-level failure model is closely related.

You can override a client policy for a single request:

response = await client.with_options(
    timeout=180.0,
).responses.create(
    model="gpt-5.5",
    input="Review this architecture and identify failure modes.",
)

Long reasoning, tool use, and large outputs may need a larger read budget than a short classification request. The value should follow the operation, model, and latency distribution rather than one global constant.

A manual policy with one retry budget

The example below disables SDK retries so the application owns the complete retry budget. The policy makes four decisions explicit:

• how long the user-facing operation may run;
• how long one model attempt may consume;
• which failures are transient enough to retry;
• how many provider calls may cross the boundary.

from __future__ import annotations

import asyncio
import random
from dataclasses import dataclass

import openai
from openai import AsyncOpenAI


class InvalidModelResponse(RuntimeError):
    pass


class GenerationDeadlineExceeded(TimeoutError):
    pass


@dataclass(frozen=True)
class GenerationPolicy:
    max_attempts: int = 3
    operation_timeout: float = 150.0
    attempt_timeout: float = 90.0
    base_backoff: float = 0.5
    max_backoff: float = 4.0


@dataclass(frozen=True)
class Generation:
    text: str
    request_id: str | None
    input_tokens: int | None
    output_tokens: int | None
    attempts: int
    elapsed: float


RETRYABLE = (
    TimeoutError,
    openai.APITimeoutError,
    openai.APIConnectionError,
    openai.RateLimitError,
    openai.InternalServerError,
)


def _backoff(policy: GenerationPolicy, attempt: int) -> float:
    ceiling = min(
        policy.max_backoff,
        policy.base_backoff * 2 ** (attempt - 1),
    )
    return random.uniform(0.0, ceiling)


async def generate(
    client: AsyncOpenAI,
    prompt: str,
    *,
    policy: GenerationPolicy = GenerationPolicy(),
) -> Generation:
    loop = asyncio.get_running_loop()
    started = loop.time()
    deadline = started + policy.operation_timeout
    last_error: BaseException | None = None

    for attempt in range(1, policy.max_attempts + 1):
        remaining = deadline - loop.time()
        if remaining <= 0:
            break

        try:
            async with asyncio.timeout(
                min(policy.attempt_timeout, remaining)
            ):
                response = await client.responses.create(
                    model="gpt-5.5",
                    input=prompt,
                )

            text = response.output_text.strip()
            if not text:
                raise InvalidModelResponse("Model returned no text")

            usage = response.usage
            return Generation(
                text=text,
                request_id=response._request_id,
                input_tokens=usage.input_tokens if usage else None,
                output_tokens=usage.output_tokens if usage else None,
                attempts=attempt,
                elapsed=loop.time() - started,
            )
        except asyncio.CancelledError:
            raise
        except InvalidModelResponse:
            raise
        except RETRYABLE as exc:
            last_error = exc

            if attempt == policy.max_attempts:
                raise

            remaining = deadline - loop.time()
            delay = min(_backoff(policy, attempt), max(0.0, remaining))

            if delay <= 0:
                break

            await asyncio.sleep(delay)

    raise GenerationDeadlineExceeded(
        "OpenAI generation exceeded its operation deadline"
    ) from last_error

The HTTPX configuration on the client still protects connect, read, write, and pool phases. attempt_timeout limits one complete SDK call, while operation_timeout is shared by all attempts and backoff.

Empty model output is treated as a contract failure rather than silently retried.

A production policy should also honor provider retry guidance and attach a stable operation ID to every attempt.

Streaming needs two clocks

For streaming responses, a read timeout usually limits silence while waiting for more data. It is not the same as a time-to-first-token objective or a total stream deadline.

This is the same reason a read timeout in a conventional HTTP client is not a total request deadline. The Python Requests timeout guide explains that behavior in more detail; LLM streaming adds first-token latency and model generation time on top of it.

A useful streaming policy distinguishes:

1. maximum time until the first text delta;
2. maximum silence between later deltas;
3. maximum duration of the complete generation.

from collections.abc import AsyncIterator
import asyncio

from openai import AsyncOpenAI


async def _next_text_delta(iterator: AsyncIterator[object]) -> str:
    while True:
        event = await anext(iterator)

        if event.type == "response.output_text.delta":
            return event.delta


async def stream_text(
    client: AsyncOpenAI,
    prompt: str,
    *,
    first_token_timeout: float = 30.0,
    chunk_timeout: float = 20.0,
    total_timeout: float = 180.0,
) -> AsyncIterator[str]:
    async with asyncio.timeout(total_timeout):
        stream = await client.responses.create(
            model="gpt-5.5",
            input=prompt,
            stream=True,
        )
        iterator = stream.__aiter__()

        try:
            async with asyncio.timeout(first_token_timeout):
                yield await _next_text_delta(iterator)

            while True:
                try:
                    async with asyncio.timeout(chunk_timeout):
                        yield await _next_text_delta(iterator)
                except StopAsyncIteration:
                    break
        finally:
            await stream.close()

The finally block matters when the user disconnects, the deadline expires, or the application stops reading early. It releases the underlying connection.

Cancellation still does not guarantee that remote computation or billing stopped at the same instant. It only guarantees that this application stopped waiting and cleaned up its local resources.

Classify failures before choosing a reaction

Not every failure should enter the same retry path.

Authentication errors, malformed requests, and unsupported parameters are not temporary failures. Retrying them adds latency without changing the outcome.

For successful API responses, log response._request_id. For APIStatusError, the exception exposes a request ID that can be logged even when the request failed.

The result and the bill can become uncertain

A model request may be accepted before the connection drops or the caller deadline expires. The generated result may never reach the application, while usage may still have occurred.

This is the AI equivalent of an ambiguous outcome.

A retry can therefore create:

• another billed generation;
• a different answer for the same user action;
• duplicate downstream tool calls;
• two workers persisting competing results.

Carry a stable operation ID through your own workflow. Deduplicate jobs and writes in the application, and make downstream tool calls independently idempotent.

Do not assume that retry support in an SDK provides exactly-once generation or billing semantics unless the API explicitly documents that guarantee.

See Idempotency in REST APIs: Keys, Concurrency and Safe Retries for the server-side state and recovery model behind safe repeated operations.

Fallback changes more than availability

Switching from one model or provider to another can restore availability, but it can also change output quality, latency, safety behavior, token accounting, tool support, and structured-output reliability.

A fallback should therefore be named and observable. Validate its output against the same business contract, but do not pretend the providers are semantically identical.

In Ghostviber, AI calls sit inside longer workflows that include credit reservation, asynchronous execution, validation, and provider routing. A fallback is only successful when the returned output is acceptable and the billing workflow can settle one user operation correctly.

Composing the policy with RelPrim

RelPrim can make the complete async operation explicit while leaving the official OpenAI SDK responsible for transport and API access.

The example disables SDK retries, then composes timeout, bounded retry, validation, fallback, and a structured result in RelPrim.

import httpx
from openai import AsyncOpenAI
from relprim import (
    RetryPolicy,
    TimeoutPolicy,
    async_operation,
    fallback_chain,
    validation_policy,
    validator,
)


client = AsyncOpenAI(
    max_retries=0,
    timeout=httpx.Timeout(
        90.0,
        connect=5.0,
        read=60.0,
        write=10.0,
        pool=5.0,
    ),
)


async def call_primary(prompt: str) -> str:
    response = await client.responses.create(
        model="gpt-5.5",
        input=prompt,
    )
    return response.output_text


async def call_backup(prompt: str) -> str:
    response = await client.responses.create(
        model="gpt-5.5-mini",
        input=prompt,
    )
    return response.output_text


non_empty = validation_policy(
    validator(
        "non_empty_output",
        lambda value: bool(value.strip()),
        message="Model output must not be empty.",
    )
)


result = await (
    async_operation("generate_ai_response", call_primary)
    .with_retry(RetryPolicy(max_attempts=3))
    .with_timeout(TimeoutPolicy(seconds=120))
    .with_validation(non_empty)
    .with_fallbacks(
        fallback_chain(("backup_model", call_backup))
    )
    .run("Summarize the incident in three concrete actions.")
)

print(result.value)
print(result.report.to_dict())

The execution report preserves how the result was produced instead of returning only the final text. Fallback names also appear in reports and structured events.

RelPrim does not replace the OpenAI SDK, provider telemetry, durable workflow state, or idempotency around downstream effects. Its role is to make the operation policy composable and observable.

See the RelPrim advanced usage guide for the current builder API.

Test the policy without calling a model

A fake provider can deterministically simulate a timeout, an invalid result, or a slow stream. Tests should assert the final outcome and the number of provider attempts, not only that one exception occurred.

import asyncio

import pytest
from relprim import RetryPolicy, TimeoutPolicy, async_operation


@pytest.mark.asyncio
async def test_timeout_is_retried_within_the_operation() -> None:
    calls = 0

    async def fake_provider(prompt: str) -> str:
        nonlocal calls
        calls += 1

        if calls == 1:
            await asyncio.sleep(0.05)

        return f"completed: {prompt}"

    result = await (
        async_operation("test_generation", fake_provider)
        .with_retry(RetryPolicy(max_attempts=2))
        .with_timeout(TimeoutPolicy(seconds=0.01))
        .run("incident")
    )

    assert result.value == "completed: incident"
    assert calls == 2

Also test a rate limit followed by success, validation failure across every attempt, fallback activation, cancellation, and a stream that stalls after several chunks.

Before you ship

Make transport timeouts explicit. Keep one owner for retries. Bound the complete operation, including backoff and fallback. Treat streaming first-token latency, inter-chunk silence, and total duration as separate concerns.

Record provider, model, attempt number, timeout phase, latency, request ID, validation result, fallback reason, and token usage when available. Preserve an unknown outcome when the request may have completed remotely.

A timeout limits how long your application waits. It does not prove that model execution or billing never happened.

Which part of an LLM timeout policy has caused the most trouble in your system: first-token latency, a stalled stream, or uncertainty after a non-streaming request fails?