Retry Policy from HTTP Responses

import email.utils
import time
from datetime import timedelta
from temporalio.exceptions import ApplicationError
from temporalio import workflow
from typing import Optional, Tuple

with workflow.unsafe.imports_passed_through():
    from httpx import Response, Headers


# Adapted from the OpenAI Python client (https://github.com/openai/openai-python/blob/main/src/openai/_base_client.py)
# which is generated by the Stainless SDK Generator.
def _parse_retry_after_header(response_headers: Optional[Headers] = None) -> float | None:
    """Returns a float of the number of seconds (not milliseconds) to wait after retrying, or None if unspecified.

    About the Retry-After header: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After
    See also  https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After#syntax
    """
    if response_headers is None:
        return None

    # First, try the non-standard `retry-after-ms` header for milliseconds,
    # which is more precise than integer-seconds `retry-after`
    try:
        retry_ms_header = response_headers.get("retry-after-ms", None)
        return float(retry_ms_header) / 1000
    except (TypeError, ValueError):
        pass

    # Next, try parsing `retry-after` header as seconds (allowing nonstandard floats).
    retry_header = response_headers.get("retry-after")
    try:
        # note: the spec indicates that this should only ever be an integer
        # but if someone sends a float there's no reason for us to not respect it
        return float(retry_header)
    except (TypeError, ValueError):
        pass

    # Last, try parsing `retry-after` as a date.
    retry_date_tuple = email.utils.parsedate_tz(retry_header)
    if retry_date_tuple is None:
        return None

    retry_date = email.utils.mktime_tz(retry_date_tuple)
    return float(retry_date - time.time())

def _should_retry(response: Response) -> Tuple[bool, str]:
    # Note: this is not a standard header
    should_retry_header = response.headers.get("x-should-retry")

    # If the server explicitly says whether or not to retry, obey.
    if should_retry_header == "true":
        return True, f"Server requested retry via x-should-retry=true header (HTTP {response.status_code})"
    if should_retry_header == "false":
        return False, f"Server prevented retry via x-should-retry=false header (HTTP {response.status_code})"

    # Retry on request timeouts.
    if response.status_code == 408:
        return True, f"HTTP request timeout ({response.status_code}), will retry with backoff"

    # Retry on lock timeouts.
    if response.status_code == 409:
        return True, f"HTTP conflict/lock timeout ({response.status_code}), will retry with backoff"

    # Retry on rate limits.
    if response.status_code == 429:
        return True, f"HTTP rate limit exceeded ({response.status_code}), will retry with backoff"

    # Retry internal errors.
    if response.status_code >= 500:
        return True, f"HTTP server error ({response.status_code}), will retry with backoff"

    return False, f"HTTP client error ({response.status_code}), not retrying - check your request"


def http_response_to_application_error(response: Response) -> ApplicationError:
    """Transform HTTP response into Temporal ApplicationError for retry handling.

    This function implements generic HTTP retry logic based on status codes and headers.

    Args:
        response: The httpx.Response from a failed HTTP request

    Returns:
        ApplicationError: Always returns an ApplicationError configured for Temporal's retry system:
            - non_retryable: False for retryable errors, True for non-retryable
            - next_retry_delay: Server-provided delay hint (if valid)

    Note:
        Even when x-should-retry=true, this function returns an ApplicationError with
        non_retryable=False rather than raising an exception, for cleaner functional style.
    """
    should_retry, retry_message = _should_retry(response)
    if should_retry:
        # Calculate the retry delay only when retrying
        retry_after = _parse_retry_after_header(response.headers)
        # Make sure that the retry delay is in a reasonable range
        if retry_after is not None and 0 < retry_after <= 60:
            retry_after = timedelta(seconds=retry_after)
        else:
            retry_after = None

        # Add delay info for rate limits
        if response.status_code == 429 and retry_after is not None:
            retry_message = f"HTTP rate limit exceeded (429) (server requested {retry_after.total_seconds():.1f}s delay), will retry with backoff"

        return ApplicationError(
            retry_message,
            non_retryable=False,
            next_retry_delay=retry_after,
        )
    else:
        return ApplicationError(
            retry_message,
            non_retryable=True,
            next_retry_delay=None,
        )

from temporalio import activity
from openai import AsyncOpenAI
from openai.types.responses import Response
from dataclasses import dataclass
from util.http_retry import http_response_to_application_error
from openai import APIStatusError

# Temporal best practice: Create a data structure to hold the request parameters.
@dataclass
class OpenAIResponsesRequest:
    model: str
    instructions: str
    input: str

@activity.defn
async def create(request: OpenAIResponsesRequest) -> Response:
    # Temporal best practice: Disable retry logic in OpenAI API client library.
    client = AsyncOpenAI(max_retries=0)

    try:
        resp = await client.responses.create(
            model=request.model,
            instructions=request.instructions,
            input=request.input,
            timeout=15,
        )
        return resp
    except APIStatusError as e:
        raise http_response_to_application_error(e.response) from e