Retry

ModelRetryBackoffSettings

Backoff configuration for runner-managed model retries.

Source code in src/agents/retry.py

@pydantic_dataclass
class ModelRetryBackoffSettings:
    """Backoff configuration for runner-managed model retries."""

    initial_delay: float | None = Field(default=None, ge=0)
    """Delay in seconds before the first retry attempt."""

    max_delay: float | None = Field(default=None, ge=0)
    """Maximum delay in seconds between retry attempts."""

    multiplier: float | None = Field(default=None, ge=0)
    """Multiplier applied after each retry attempt."""

    jitter: bool | None = None
    """Whether to apply random jitter to the computed delay."""

    def to_json_dict(self) -> dict[str, Any]:
        return dataclasses.asdict(self)

initial_delay class-attribute instance-attribute

initial_delay: float | None = Field(default=None, ge=0)

Delay in seconds before the first retry attempt.

max_delay class-attribute instance-attribute

max_delay: float | None = Field(default=None, ge=0)

Maximum delay in seconds between retry attempts.

multiplier class-attribute instance-attribute

multiplier: float | None = Field(default=None, ge=0)

Multiplier applied after each retry attempt.

jitter class-attribute instance-attribute

jitter: bool | None = None

Whether to apply random jitter to the computed delay.

ModelRetryNormalizedError dataclass

Normalized error facts exposed to retry policies.

Source code in src/agents/retry.py

@dataclass(init=False)
class ModelRetryNormalizedError:
    """Normalized error facts exposed to retry policies."""

    status_code: int | None = None
    error_code: str | None = None
    message: str | None = None
    request_id: str | None = None
    retry_after: float | None = None
    is_abort: bool = False
    is_network_error: bool = False
    is_timeout: bool = False

    def __init__(
        self,
        status_code: int | None = _UNSET,
        error_code: str | None = _UNSET,
        message: str | None = _UNSET,
        request_id: str | None = _UNSET,
        retry_after: float | None = _UNSET,
        is_abort: bool = _UNSET,
        is_network_error: bool = _UNSET,
        is_timeout: bool = _UNSET,
    ) -> None:
        explicit_fields: set[str] = set()

        def assign(name: str, value: Any, default: Any) -> Any:
            if value is _UNSET:
                return default
            explicit_fields.add(name)
            return value

        self.status_code = assign("status_code", status_code, None)
        self.error_code = assign("error_code", error_code, None)
        self.message = assign("message", message, None)
        self.request_id = assign("request_id", request_id, None)
        self.retry_after = assign("retry_after", retry_after, None)
        self.is_abort = assign("is_abort", is_abort, False)
        self.is_network_error = assign("is_network_error", is_network_error, False)
        self.is_timeout = assign("is_timeout", is_timeout, False)
        self._explicit_fields = frozenset(explicit_fields)

ModelRetryAdvice dataclass

Provider-specific retry guidance returned by model adapters.

Source code in src/agents/retry.py

@dataclass
class ModelRetryAdvice:
    """Provider-specific retry guidance returned by model adapters."""

    suggested: bool | None = None
    retry_after: float | None = None
    replay_safety: str | None = None
    reason: str | None = None
    normalized: ModelRetryNormalizedError | None = None
    response_started: bool = False
    """Whether the provider had begun emitting the response when the failure occurred."""

response_started class-attribute instance-attribute

response_started: bool = False

Whether the provider had begun emitting the response when the failure occurred.

ModelRetryAdviceRequest dataclass

Context passed to a model adapter when deriving retry advice.

Source code in src/agents/retry.py

@dataclass
class ModelRetryAdviceRequest:
    """Context passed to a model adapter when deriving retry advice."""

    error: Exception
    attempt: int
    stream: bool
    previous_response_id: str | None = None
    conversation_id: str | None = None

RetryDecision dataclass

Explicit retry decision returned by retry policies.

Source code in src/agents/retry.py

@dataclass
class RetryDecision:
    """Explicit retry decision returned by retry policies."""

    retry: bool
    delay: float | None = None
    reason: str | None = None
    approve_unsafe_replay: bool = False
    """Explicit application approval to replay a request the provider marked replay-unsafe.

    This is deliberately separate from ``retry``: an ordinary ``RetryDecision(retry=True)``
    never bypasses replay protection. Set this only for workloads where repeating
    provider-side work that may already have happened is acceptable.
    """
    _hard_veto: bool = field(default=False, init=False, repr=False, compare=False)
    _delegable_replay_veto: bool = field(default=False, init=False, repr=False, compare=False)
    _approves_replay: bool = field(default=False, init=False, repr=False, compare=False)

approve_unsafe_replay class-attribute instance-attribute

approve_unsafe_replay: bool = False

Explicit application approval to replay a request the provider marked replay-unsafe.

This is deliberately separate from retry: an ordinary RetryDecision(retry=True) never bypasses replay protection. Set this only for workloads where repeating provider-side work that may already have happened is acceptable.

RetryPolicyContext dataclass

Context passed to runtime retry policy callbacks.

Source code in src/agents/retry.py

@dataclass
class RetryPolicyContext:
    """Context passed to runtime retry policy callbacks."""

    error: Exception
    attempt: int
    max_retries: int
    stream: bool
    normalized: ModelRetryNormalizedError
    provider_advice: ModelRetryAdvice | None = None
    previous_response_id: str | None = None
    conversation_id: str | None = None
    _provider_authority: _ProviderRetryAuthority = field(
        init=False,
        repr=False,
        compare=False,
    )

    def __post_init__(self) -> None:
        advice = self.provider_advice
        replay_safety = advice.replay_safety if advice is not None else None
        self._provider_authority = _ProviderRetryAuthority(
            suggested=advice.suggested if advice is not None else None,
            replay_safety=(replay_safety if replay_safety in {"safe", "unsafe"} else "unknown"),
            response_started=advice.response_started if advice is not None else False,
        )

    @property
    def response_started(self) -> bool:
        """Whether the provider had begun emitting the response when the failure occurred."""
        return self._provider_authority.response_started

    @property
    def replay_safety(self) -> str:
        """Provider replay classification: ``"safe"``, ``"unsafe"`` or ``"unknown"``."""
        return self._provider_authority.replay_safety

    @property
    def stateful_request(self) -> bool:
        """Whether the request carried ``previous_response_id`` or ``conversation_id``."""
        return bool(self.previous_response_id or self.conversation_id)

response_started property

response_started: bool

Whether the provider had begun emitting the response when the failure occurred.

replay_safety property

replay_safety: str

Provider replay classification: "safe", "unsafe" or "unknown".

stateful_request property

stateful_request: bool

Whether the request carried previous_response_id or conversation_id.

ModelRetrySettings

Opt-in runner-managed retry settings for model calls.

Source code in src/agents/retry.py

@pydantic_dataclass
class ModelRetrySettings:
    """Opt-in runner-managed retry settings for model calls."""

    max_retries: int | None = None
    """Retries allowed after the initial model request."""

    backoff: ModelRetryBackoffInput | None = None
    """Backoff settings applied when the policy retries without an explicit delay."""

    policy: Callable[..., Any] | None = Field(default=None, exclude=True, repr=False)
    """Runtime-only retry policy callback. This field is not serialized."""

    def __post_init__(self) -> None:
        self.backoff = _coerce_backoff_settings(self.backoff)

    def to_json_dict(self) -> dict[str, Any]:
        backoff = _coerce_backoff_settings(self.backoff)
        return {
            "max_retries": self.max_retries,
            "backoff": backoff.to_json_dict() if backoff is not None else None,
        }

max_retries class-attribute instance-attribute

max_retries: int | None = None

Retries allowed after the initial model request.

backoff class-attribute instance-attribute

backoff: ModelRetryBackoffInput | None = None

Backoff settings applied when the policy retries without an explicit delay.

policy class-attribute instance-attribute

policy: Callable[..., Any] | None = Field(
    default=None, exclude=True, repr=False
)

Runtime-only retry policy callback. This field is not serialized.