openai-compatible-errors 0.1.0

Normalize OpenAI-compatible API failures, plan conservative retries, redact logs, and inspect SSE streams

openai-compatible-errors

openai-compatible-errors is a zero-runtime-dependency Python library for the failure boundary around OpenAI-compatible HTTP APIs. It turns varied HTTPX, OpenAI Python SDK and JSON error shapes into a small immutable snapshot; plans retries only when the caller supplies enough replay evidence; redacts bounded log data; and incrementally inspects Chat Completions or Responses SSE streams.

It does not send requests, sleep, retry automatically, buffer a complete stream, or retain raw provider payloads.

python -m pip install openai-compatible-errors

Python 3.10 or newer is required. The installed wheel has no dependencies and ships a py.typed marker.

For a no-network, no-credential path through a real OpenAI client exception, install the development extra and run examples/mock_transport_retry.py:

python -m pip install -e '.[dev]'
python examples/mock_transport_retry.py

Normalize failures without retaining the response

import httpx

from openai_compatible_errors import normalize_error

response = httpx.Response(
    429,
    request=httpx.Request("POST", "https://api.example/v1/responses"),
    headers={"retry-after": "2", "x-request-id": "req_01"},
    json={
        "error": {
            "type": "requests",
            "code": "rate_limit_exceeded",
            "message": "provider-controlled detail",
        }
    },
)

error = normalize_error(response)
assert error.category.value == "rate_limit"
assert error.status == 429
assert error.retry_after_ms == 2_000
assert error.request_id == "req_01"
assert error.provider_message is None

The same function recognizes OpenAI Python SDK exception structures without making the SDK a runtime dependency:

try:
    client.responses.create(model="example-model", input="hello")
except Exception as exc:
    safe_error = normalize_error(exc)
    logger.warning(
        "API call failed",
        extra={"api_error": safe_error.to_log_dict()},
    )

Keep the snapshot under its own extra key. Python logging reserves names such as message, so expanding the snapshot directly into a LogRecord would raise instead of recording the failure.

Compatibility tests use real httpx==0.28.1 objects and exceptions emitted by a real openai==2.49.0 client through MockTransport across 400, 401, 403, 404, 409, 422, 429 and 5xx responses. Targeted SDK exception constructors cover timeout and response-validation paths. Duck-typed objects with instance fields such as status_code, headers, body, or response are also supported. Unknown object properties are not invoked.

Safe-by-default data contract

ApiError is a frozen, slotted dataclass. It has no fields for raw bodies, headers, exception causes, tracebacks, prompt text, or model output.

Field	Default behavior
message	Stable library-owned text based on category
status, code, type	Retained only after validation
request_id	Extracted from a small allowlist of correlation headers
retry_after_ms	Parsed from bounded Retry-After seconds or HTTP date
provider_message	Omitted unless include_provider_message=True

Even after the provider-message opt-in, common credentials are redacted and the string is bounded. This is risk reduction, not a guarantee that arbitrary provider text is appropriate for your logs.

Plan conservative retries

HTTP method alone cannot prove replay safety. The caller must state both the request phase and whether the particular operation is safe to replay:

from openai_compatible_errors import (
    ReplaySafety,
    RequestPhase,
    RetryContext,
    decide_retry,
)

plan = decide_retry(
    error,
    RetryContext(
        method="POST",
        phase=RequestPhase.HTTP_ERROR,
        replay_safety=ReplaySafety.SAFE,  # caller-owned operation contract
        attempt=1,  # one-based failed attempt
        elapsed_ms=350,
    ),
)

if plan.action.value == "retry":
    # Your application owns cancellation, sleep and request replay.
    schedule_retry_after(plan.delay_ms or 0)

The decision is deliberately three-state:

• retry only for a transient category, known replay-safe operation, known phase, no observed stream output, and remaining attempt/time/delay budgets;
• do_not_retry for permanent failures, unsafe replay, partial output, cancellation, completion, or an exhausted budget;
• manual_decision when evidence is missing, unclassified, or invalid.

Server Retry-After/Retry-After-Ms evidence is rounded up; duplicate hints use the longest delay. Extremely large values saturate to a bounded sentinel, and a present but malformed hint fails closed instead of silently falling back to a short local delay. A server delay is used only when it fits both max_delay_ms and the remaining elapsed-time budget. Local backoff supports full jitter and an injectable random function for deterministic tests.

Inspect streaming replay boundaries

SSEInspector incrementally decodes UTF-8, including a code point split across network chunks. It understands SSE framing, [DONE], Chat Completions chunks, Responses event types, error events and unexpected EOF. It records only state; it never retains generated output.

from openai_compatible_errors import SSEInspector, SyncSSEAdapter

inspector = SSEInspector()
for chunk in SyncSSEAdapter(response.iter_bytes(), inspector):
    consume(chunk)  # each source chunk is yielded unchanged

state = inspector.state
if state.unexpected_eof and state.has_output:
    # Replaying could duplicate already-observed output.
    raise RuntimeError("stream ended after partial output")

AsyncSSEAdapter provides the same contract for an AsyncIterable of bytes-like chunks. Both adapters are one-shot and preserve backpressure: they request one source chunk, inspect it, and yield it before requesting another. They never prefetch, access the network, sleep, or retry. If a consumer stops early, the adapter does not falsely claim that the source reached EOF.

Terminal stream states are:

• done: [DONE] or response.completed was observed;
• incomplete: response.incomplete ended the stream without claiming a successful completion;
• error: a provider error event, malformed UTF-8/JSON, oversized event, or source iteration failure occurred;
• unexpected_eof: the source exhausted without a completion or error event.

has_output is intentionally conservative and includes Responses partial-image events. Unknown data-bearing vendor events also set it because their payload may already be user-visible. A false positive only prevents an automatic replay; a false negative could duplicate output. Non-standard JSON constants, non-finite numbers and duplicate object keys are protocol errors rather than ambiguous inputs.

Sanitize arbitrary log context

from openai_compatible_errors import SanitizerLimits, sanitize_for_log

safe_context = sanitize_for_log(
    diagnostic_context,
    limits=SanitizerLimits(max_depth=4, max_nodes=250),
)

The sanitizer redacts sensitive key names and common credential formats, limits depth, keys, sequence items, nodes, scanned characters and replacements, and breaks cycles. It uses instance data rather than walking arbitrary properties. Exception causes, tracebacks and arguments are never traversed; exception messages require a separate explicit opt-in. Exhausting a replacement budget returns a generic redaction marker instead of partially sanitized text.

Never use real credentials, prompts, model output or customer payloads as test fixtures. Redaction is a last defensive layer, not permission to log sensitive data.

Categories and boundaries

Normalized categories include authentication, permission, rate limit, quota, conflict, validation, not found, payload too large, timeout, network, upstream, server, schema, endpoint, aborted, stream and unknown. HTTP 409 maps to the explicit conflict category and remains a manual_decision: the library cannot infer whether resolving or replaying a state conflict is safe. Classification prefers HTTP status, exception class and structured code/type evidence. Free-form exception messages are not inspected or retained by default.

This package targets broadly used OpenAI-compatible shapes; compatibility is not a claim of partnership, certification, or complete behavior parity with any API provider. Validate retry safety against your own operation contract.

When to use something else

Keep your existing client library's native exceptions if one client and one provider already give you a stable error contract. Use an application-level resilience library when you need circuit breaking, cancellation-aware sleep or request execution: this package only returns a plan. Use a provider-specific stream parser when you must preserve every proprietary event field; this inspector deliberately retains only replay-boundary state.

Development and security

See CONTRIBUTING.md for the reproducible validation commands, SECURITY.md for private reporting guidance, and RELEASING.md for token-free PyPI Trusted Publishing setup.

MIT licensed.