diff --git a/kubernetes/aio/utils/__init__.py b/kubernetes/aio/utils/__init__.py
index b1244b9a0e..aabaa28673 100644
--- a/kubernetes/aio/utils/__init__.py
+++ b/kubernetes/aio/utils/__init__.py
@@ -18,3 +18,10 @@
     FailToCreateError, create_from_dict, create_from_yaml,
     create_from_yaml_single_item,
 )
+from .retry import (Backoff, DEFAULT_BACKOFF, DEFAULT_RETRY,
+                    DEFAULT_RETRY_AFTER_BACKOFF, async_on_error,
+                    async_on_retry_after_error, async_retry_on_conflict,
+                    is_conflict, is_retry_after_response,
+                    is_too_many_requests, on_error, on_retry_after_error,
+                    retry_after_backoff, retry_after_max_retries,
+                    retry_on_conflict, retry_after_seconds)
diff --git a/kubernetes/aio/utils/retry.py b/kubernetes/aio/utils/retry.py
new file mode 120000
index 0000000000..3a2eabf7d3
--- /dev/null
+++ b/kubernetes/aio/utils/retry.py
@@ -0,0 +1 @@
+../../utils/retry.py
\ No newline at end of file
diff --git a/kubernetes/aio/utils/retry_test.py b/kubernetes/aio/utils/retry_test.py
new file mode 100644
index 0000000000..e20b5d168f
--- /dev/null
+++ b/kubernetes/aio/utils/retry_test.py
@@ -0,0 +1,72 @@
+# Copyright 2026 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from kubernetes.aio.utils.retry import (
+    Backoff,
+    DEFAULT_RETRY,
+    async_retry_on_conflict,
+    retry_after_seconds,
+)
+
+
+class FakeError(Exception):
+
+    def __init__(self, status, headers=None):
+        super().__init__("status {0}".format(status))
+        self.status = status
+        self.headers = headers or {}
+
+
+class AioRetryTest(unittest.IsolatedAsyncioTestCase):
+
+    def test_default_retry_matches_client_go(self):
+        self.assertEqual(
+            DEFAULT_RETRY,
+            Backoff(steps=5, duration=0.01, factor=1.0, jitter=0.1),
+        )
+
+    def test_retry_after_seconds_parses_delay_seconds(self):
+        error = FakeError(429, {"Retry-After": "7"})
+
+        self.assertEqual(retry_after_seconds(error), 7.0)
+
+    async def test_async_retry_on_conflict_retries(self):
+        attempts = []
+        sleeps = []
+
+        async def fn():
+            attempts.append(1)
+            if len(attempts) < 3:
+                raise FakeError(409)
+            return "updated"
+
+        async def sleep(delay):
+            sleeps.append(delay)
+
+        result = await async_retry_on_conflict(
+            fn,
+            backoff=Backoff(steps=3, duration=1.0, factor=1.0),
+            sleep_func=sleep,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "updated")
+        self.assertEqual(len(attempts), 3)
+        self.assertEqual(sleeps, [1.0, 1.0])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/kubernetes/utils/__init__.py b/kubernetes/utils/__init__.py
index 92ab696782..f99a611424 100644
--- a/kubernetes/utils/__init__.py
+++ b/kubernetes/utils/__init__.py
@@ -19,3 +19,10 @@
 from .duration import parse_duration
 from .metrics import (get_nodes_metrics, get_pods_metrics,
                       get_pods_metrics_in_all_namespaces)
+from .retry import (Backoff, DEFAULT_BACKOFF, DEFAULT_RETRY,
+                    DEFAULT_RETRY_AFTER_BACKOFF, async_on_error,
+                    async_on_retry_after_error, async_retry_on_conflict,
+                    is_conflict, is_retry_after_response,
+                    is_too_many_requests, on_error, on_retry_after_error,
+                    retry_after_backoff, retry_after_max_retries,
+                    retry_on_conflict, retry_after_seconds)
diff --git a/kubernetes/utils/retry.py b/kubernetes/utils/retry.py
new file mode 100644
index 0000000000..34de35b113
--- /dev/null
+++ b/kubernetes/utils/retry.py
@@ -0,0 +1,353 @@
+# Copyright 2026 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+import random
+import re
+import time
+from dataclasses import dataclass
+from typing import Any, Awaitable, Callable, Optional, TypeVar
+
+
+T = TypeVar("T")
+
+
+# This module is a 1:1 Python implementation of the Kubernetes Go retry
+# algorithms used by client-go:
+# - https://github.com/kubernetes/client-go/blob/master/util/retry/util.go
+# - https://github.com/kubernetes/apimachinery/blob/master/pkg/util/wait/
+#   backoff.go
+# - https://github.com/kubernetes/client-go/blob/master/rest/with_retry.go
+# Backoff durations are represented as seconds instead of time.Duration.
+@dataclass(frozen=True)
+class Backoff:
+    """Retry backoff parameters.
+
+    This maps 1:1 to ``k8s.io/apimachinery/pkg/util/wait.Backoff`` for the
+    fields used by client-go retry helpers. ``steps`` is the maximum number of
+    times the operation is called, including the first call. ``duration`` is
+    the first delay in seconds, ``factor`` is multiplied into the next delay
+    after every failed attempt, and ``jitter`` adds up to that fraction of the
+    current delay. ``cap`` limits increased delay values.
+    """
+
+    steps: int
+    duration: float
+    factor: float
+    jitter: float = 0.0
+    cap: float = 0.0
+
+    def __post_init__(self):
+        if self.steps < 1:
+            raise ValueError("steps must be at least 1")
+        if self.duration < 0:
+            raise ValueError("duration must be non-negative")
+        if self.factor < 0:
+            raise ValueError("factor must be non-negative")
+        if self.jitter < 0:
+            raise ValueError("jitter must be non-negative")
+        if self.cap < 0:
+            raise ValueError("cap must be non-negative")
+
+
+# 1:1 with k8s.io/client-go/util/retry.DefaultBackoff.
+DEFAULT_BACKOFF = Backoff(steps=4, duration=0.01, factor=5.0, jitter=0.1)
+
+# 1:1 with k8s.io/client-go/util/retry.DefaultRetry.
+DEFAULT_RETRY = Backoff(steps=5, duration=0.01, factor=1.0, jitter=0.1)
+
+# Matches rest.Request's default response retry ceiling: one initial attempt
+# plus 10 retries after Retry-After responses.
+DEFAULT_RETRY_AFTER_BACKOFF = Backoff(
+    steps=11, duration=0.0, factor=1.0, jitter=0.0)
+
+
+def retry_after_backoff(
+    retries: Any = None,
+    backoff: Optional[Backoff] = None,
+) -> Backoff:
+    """Return the client-go REST Retry-After backoff for ``retries``.
+
+    ``backoff`` supplies the retry delay parameters. If ``retries`` is not
+    ``None``, it is interpreted as the retry ceiling and overrides
+    ``backoff.steps``. ``False`` and ``0`` disable retries by returning a
+    single-attempt backoff. Integer values and urllib3-style Retry objects with
+    a ``total`` value are treated as the retry ceiling.
+    """
+
+    if backoff is None:
+        backoff = DEFAULT_RETRY_AFTER_BACKOFF
+    if retries is None:
+        return backoff
+
+    return Backoff(
+        steps=retry_after_max_retries(retries) + 1,
+        duration=backoff.duration,
+        factor=backoff.factor,
+        jitter=backoff.jitter,
+        cap=backoff.cap,
+    )
+
+
+def retry_after_max_retries(retries: Any = None) -> int:
+    """Return the client-go REST Retry-After retry ceiling."""
+
+    if retries is None:
+        return 10
+    if retries is False:
+        return 0
+    if retries is True:
+        return 10
+    if isinstance(retries, int):
+        return max(0, retries)
+
+    total = getattr(retries, "total", None)
+    if total is False:
+        return 0
+    if total is True or total is None:
+        return 10
+    if isinstance(total, int):
+        return max(0, total)
+    return 10
+
+
+def is_conflict(error: Exception) -> bool:
+    """Return True if ``error`` represents an HTTP 409 Conflict response."""
+
+    return getattr(error, "status", None) == 409
+
+
+def is_too_many_requests(error: Exception) -> bool:
+    """Return True if ``error`` represents HTTP 429 Too Many Requests."""
+
+    return getattr(error, "status", None) == 429
+
+
+def is_retry_after_response(error: Exception) -> bool:
+    """Return True if ``error`` should be retried after Retry-After."""
+
+    status = getattr(error, "status", None)
+    if status != 429 and (status is None or status < 500):
+        return False
+    return retry_after_seconds(error) is not None
+
+
+def retry_after_seconds(
+    error: Exception,
+    now: Optional[Callable[[], float]] = None,
+) -> Optional[float]:
+    """Return the Retry-After delay in seconds, if ``error`` provides one.
+
+    client-go accepts integer seconds. Invalid, missing, or negative values
+    return ``None`` so callers can fall back to their normal backoff.
+    """
+
+    headers = getattr(error, "headers", None) or {}
+    value = None
+    for key, header_value in headers.items():
+        if key.lower() == "retry-after":
+            value = header_value
+            break
+    if value is None:
+        return None
+
+    value = str(value).strip()
+    if not re.match(r"^[0-9]+$", value):
+        return None
+    return float(int(value))
+
+
+def on_error(
+    backoff: Backoff,
+    retriable: Callable[[Exception], bool],
+    fn: Callable[[], T],
+    sleep_func: Callable[[float], None] = time.sleep,
+    random_func: Callable[[], float] = random.random,
+) -> T:
+    """Run ``fn`` and retry while ``retriable`` returns True.
+
+    This is a 1:1 implementation of
+    ``k8s.io/client-go/util/retry.OnError``.
+    """
+
+    steps = backoff.steps
+    duration = backoff.duration
+    last_error = None
+    while steps > 0:
+        try:
+            return fn()
+        except Exception as error:
+            if not retriable(error):
+                raise
+            last_error = error
+
+            if steps == 1:
+                break
+
+            delay, duration, steps = _delay(
+                steps, duration, backoff, random_func)
+            sleep_func(delay)
+
+    raise last_error
+
+
+def retry_on_conflict(
+    fn: Callable[[], T],
+    backoff: Backoff = DEFAULT_RETRY,
+    sleep_func: Callable[[float], None] = time.sleep,
+    random_func: Callable[[], float] = random.random,
+) -> T:
+    """Run ``fn`` and retry on HTTP 409 Conflict responses.
+
+    This is a 1:1 implementation of
+    ``k8s.io/client-go/util/retry.RetryOnConflict``. Callers should re-read the
+    object inside ``fn`` before each write attempt, so every retry uses the
+    latest resource version.
+    """
+
+    return on_error(backoff, is_conflict, fn, sleep_func, random_func)
+
+
+async def async_on_error(
+    backoff: Backoff,
+    retriable: Callable[[Exception], bool],
+    fn: Callable[[], Awaitable[T]],
+    sleep_func: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    random_func: Callable[[], float] = random.random,
+) -> T:
+    """Async 1:1 implementation of client-go ``retry.OnError``."""
+
+    steps = backoff.steps
+    duration = backoff.duration
+    last_error = None
+    while steps > 0:
+        try:
+            return await fn()
+        except Exception as error:
+            if not retriable(error):
+                raise
+            last_error = error
+
+            if steps == 1:
+                break
+
+            delay, duration, steps = _delay(
+                steps, duration, backoff, random_func)
+            await sleep_func(delay)
+
+    raise last_error
+
+
+async def async_retry_on_conflict(
+    fn: Callable[[], Awaitable[T]],
+    backoff: Backoff = DEFAULT_RETRY,
+    sleep_func: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    random_func: Callable[[], float] = random.random,
+) -> T:
+    """Async 1:1 implementation of client-go ``retry.RetryOnConflict``."""
+
+    return await async_on_error(
+        backoff, is_conflict, fn, sleep_func, random_func)
+
+
+def on_retry_after_error(
+    backoff: Backoff,
+    retriable: Callable[[Exception], bool],
+    fn: Callable[[], T],
+    sleep_func: Callable[[float], None] = time.sleep,
+    random_func: Callable[[], float] = random.random,
+) -> T:
+    """Run ``fn`` with client-go REST Retry-After sleep semantics.
+
+    This is a 1:1 implementation of the wait selection in
+    ``k8s.io/client-go/rest.WithRetry``: if the retriable response carries a
+    valid ``Retry-After`` header, that delay is used; otherwise the supplied
+    backoff delay is used.
+    """
+
+    steps = backoff.steps
+    duration = backoff.duration
+    last_error = None
+    while steps > 0:
+        try:
+            return fn()
+        except Exception as error:
+            if not retriable(error):
+                raise
+            last_error = error
+
+            if steps == 1:
+                break
+
+            delay, duration, steps = _delay(
+                steps, duration, backoff, random_func)
+            retry_after = retry_after_seconds(error)
+            if retry_after is not None and retry_after > delay:
+                delay = retry_after
+            sleep_func(delay)
+
+    raise last_error
+
+
+async def async_on_retry_after_error(
+    backoff: Backoff,
+    retriable: Callable[[Exception], bool],
+    fn: Callable[[], Awaitable[T]],
+    sleep_func: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    random_func: Callable[[], float] = random.random,
+) -> T:
+    """Async implementation of client-go REST Retry-After sleep semantics."""
+
+    steps = backoff.steps
+    duration = backoff.duration
+    last_error = None
+    while steps > 0:
+        try:
+            return await fn()
+        except Exception as error:
+            if not retriable(error):
+                raise
+            last_error = error
+
+            if steps == 1:
+                break
+
+            delay, duration, steps = _delay(
+                steps, duration, backoff, random_func)
+            retry_after = retry_after_seconds(error)
+            if retry_after is not None and retry_after > delay:
+                delay = retry_after
+            await sleep_func(delay)
+
+    raise last_error
+
+
+def _delay(
+    steps: int,
+    duration: float,
+    backoff: Backoff,
+    random_func: Callable[[], float],
+) -> tuple[float, float, int]:
+    steps -= 1
+    if backoff.factor == 0:
+        next_duration = duration
+    else:
+        next_duration = duration * backoff.factor
+        if backoff.cap > 0 and next_duration > backoff.cap:
+            next_duration = backoff.cap
+            steps = 0
+    delay = duration
+    if backoff.jitter:
+        delay += random_func() * backoff.jitter * delay
+    return delay, next_duration, steps
diff --git a/kubernetes/utils/retry_test.py b/kubernetes/utils/retry_test.py
new file mode 100644
index 0000000000..b37082c892
--- /dev/null
+++ b/kubernetes/utils/retry_test.py
@@ -0,0 +1,550 @@
+# Copyright 2026 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from kubernetes.client.exceptions import ApiException
+from kubernetes.utils.retry import (
+    Backoff,
+    DEFAULT_BACKOFF,
+    DEFAULT_RETRY,
+    DEFAULT_RETRY_AFTER_BACKOFF,
+    async_on_error,
+    async_on_retry_after_error,
+    async_retry_on_conflict,
+    is_conflict,
+    is_retry_after_response,
+    is_too_many_requests,
+    on_error,
+    on_retry_after_error,
+    retry_after_backoff,
+    retry_after_max_retries,
+    retry_on_conflict,
+    retry_after_seconds,
+)
+from urllib3.util.retry import Retry
+
+
+def api_exception(status, headers=None):
+    error = ApiException(status=status)
+    error.headers = headers or {}
+    return error
+
+
+class RetryTest(unittest.TestCase):
+
+    def test_default_backoffs_match_client_go(self):
+        self.assertEqual(
+            DEFAULT_BACKOFF,
+            Backoff(steps=4, duration=0.01, factor=5.0, jitter=0.1),
+        )
+        self.assertEqual(
+            DEFAULT_RETRY,
+            Backoff(steps=5, duration=0.01, factor=1.0, jitter=0.1),
+        )
+        self.assertEqual(
+            DEFAULT_RETRY_AFTER_BACKOFF,
+            Backoff(steps=11, duration=0.0, factor=1.0, jitter=0.0),
+        )
+
+    def test_retry_after_backoff_uses_configuration_retries(self):
+        self.assertEqual(retry_after_max_retries(None), 10)
+        self.assertEqual(retry_after_max_retries(False), 0)
+        self.assertEqual(retry_after_max_retries(0), 0)
+        self.assertEqual(retry_after_max_retries(3), 3)
+        self.assertEqual(retry_after_max_retries(Retry(total=2)), 2)
+        self.assertEqual(retry_after_backoff(3).steps, 4)
+
+    def test_retry_after_backoff_can_be_configured(self):
+        configured = Backoff(
+            steps=4, duration=1.0, factor=2.0, jitter=0.3, cap=5.0)
+
+        self.assertEqual(retry_after_backoff(None, configured), configured)
+        self.assertEqual(
+            retry_after_backoff(1, configured),
+            Backoff(
+                steps=2, duration=1.0, factor=2.0, jitter=0.3, cap=5.0),
+        )
+
+    def test_on_error_returns_without_retry(self):
+        attempts = []
+
+        def fn():
+            attempts.append(1)
+            return "ok"
+
+        result = on_error(
+            Backoff(steps=3, duration=1.0, factor=2.0),
+            lambda e: True,
+            fn,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(len(attempts), 1)
+
+    def test_on_error_retries_retriable_errors(self):
+        attempts = []
+        sleeps = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 3:
+                raise api_exception(500)
+            return "ok"
+
+        result = on_error(
+            Backoff(steps=4, duration=1.0, factor=2.0),
+            lambda e: getattr(e, "status", None) == 500,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(len(attempts), 3)
+        self.assertEqual(sleeps, [1.0, 2.0])
+
+    def test_on_error_exhausts_retries(self):
+        test_error = api_exception(500)
+        attempts = []
+        sleeps = []
+
+        def fn():
+            attempts.append(1)
+            raise test_error
+
+        with self.assertRaises(ApiException) as ctx:
+            on_error(
+                Backoff(steps=3, duration=1.0, factor=2.0),
+                lambda e: getattr(e, "status", None) == 500,
+                fn,
+                sleep_func=sleeps.append,
+                random_func=lambda: 0.0,
+            )
+
+        self.assertIs(ctx.exception, test_error)
+        self.assertEqual(len(attempts), 3)
+        self.assertEqual(sleeps, [1.0, 2.0])
+
+    def test_on_error_raises_non_retriable_error(self):
+        sleeps = []
+
+        def fn():
+            raise api_exception(400)
+
+        with self.assertRaises(ApiException) as ctx:
+            on_error(
+                Backoff(steps=3, duration=1.0, factor=2.0),
+                lambda e: getattr(e, "status", None) == 500,
+                fn,
+                sleep_func=sleeps.append,
+            )
+
+        self.assertEqual(ctx.exception.status, 400)
+        self.assertEqual(sleeps, [])
+
+    def test_on_error_applies_jitter(self):
+        sleeps = []
+        attempts = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(500)
+            return "ok"
+
+        result = on_error(
+            Backoff(steps=2, duration=10.0, factor=1.0, jitter=0.5),
+            lambda e: getattr(e, "status", None) == 500,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.25,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [11.25])
+
+    def test_on_error_keeps_duration_when_factor_is_zero(self):
+        sleeps = []
+        attempts = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 3:
+                raise api_exception(500)
+            return "ok"
+
+        result = on_error(
+            Backoff(steps=3, duration=1.0, factor=0.0),
+            lambda e: getattr(e, "status", None) == 500,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [1.0, 1.0])
+
+    def test_backoff_rejects_invalid_values(self):
+        with self.assertRaises(ValueError):
+            Backoff(steps=0, duration=1.0, factor=1.0)
+        with self.assertRaises(ValueError):
+            Backoff(steps=1, duration=-1.0, factor=1.0)
+        with self.assertRaises(ValueError):
+            Backoff(steps=1, duration=1.0, factor=-1.0)
+        with self.assertRaises(ValueError):
+            Backoff(steps=1, duration=1.0, factor=1.0, jitter=-1.0)
+        with self.assertRaises(ValueError):
+            Backoff(steps=1, duration=1.0, factor=1.0, cap=-1.0)
+
+    def test_on_error_stops_after_capped_sleep(self):
+        test_error = api_exception(500)
+        attempts = []
+        sleeps = []
+
+        def fn():
+            attempts.append(1)
+            raise test_error
+
+        with self.assertRaises(ApiException) as ctx:
+            on_error(
+                Backoff(steps=4, duration=1.0, factor=5.0, cap=3.0),
+                lambda e: getattr(e, "status", None) == 500,
+                fn,
+                sleep_func=sleeps.append,
+                random_func=lambda: 0.0,
+            )
+
+        self.assertIs(ctx.exception, test_error)
+        self.assertEqual(len(attempts), 1)
+        self.assertEqual(sleeps, [1.0])
+
+    def test_retry_on_conflict_never_returns(self):
+        conflict_error = api_exception(409)
+
+        def fn():
+            raise conflict_error
+
+        with self.assertRaises(ApiException) as ctx:
+            retry_on_conflict(
+                fn,
+                backoff=Backoff(steps=3, duration=0.0, factor=1.0),
+                sleep_func=lambda delay: None,
+                random_func=lambda: 0.0,
+            )
+
+        self.assertIs(ctx.exception, conflict_error)
+
+    def test_retry_on_conflict_returns_immediately(self):
+        attempts = []
+
+        def fn():
+            attempts.append(1)
+            return "updated"
+
+        result = retry_on_conflict(
+            fn,
+            backoff=Backoff(steps=3, duration=0.0, factor=1.0),
+        )
+
+        self.assertEqual(result, "updated")
+        self.assertEqual(len(attempts), 1)
+
+    def test_retry_on_conflict_returns_immediately_on_non_conflict(self):
+        test_error = api_exception(500)
+
+        def fn():
+            raise test_error
+
+        with self.assertRaises(ApiException) as ctx:
+            retry_on_conflict(
+                fn,
+                backoff=Backoff(steps=3, duration=0.0, factor=1.0),
+            )
+
+        self.assertIs(ctx.exception, test_error)
+
+    def test_retry_on_conflict_keeps_retrying(self):
+        attempts = []
+        sleeps = []
+
+        def fn():
+            if len(attempts) < 2:
+                attempts.append(1)
+                raise api_exception(409)
+            return "updated"
+
+        result = retry_on_conflict(
+            fn,
+            backoff=Backoff(steps=3, duration=1.0, factor=1.0),
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "updated")
+        self.assertEqual(len(attempts), 2)
+        self.assertEqual(sleeps, [1.0, 1.0])
+
+    def test_retry_after_seconds_parses_delay_seconds(self):
+        error = api_exception(429, {"Retry-After": "7"})
+
+        self.assertEqual(retry_after_seconds(error), 7.0)
+
+    def test_retry_after_seconds_rejects_http_date(self):
+        error = api_exception(
+            429,
+            {"retry-after": "Wed, 21 Oct 2015 07:28:00 GMT"},
+        )
+
+        self.assertIsNone(retry_after_seconds(error))
+
+    def test_retry_after_seconds_returns_none_without_header(self):
+        self.assertIsNone(retry_after_seconds(api_exception(429)))
+
+    def test_retry_after_seconds_returns_none_for_invalid_header(self):
+        error = api_exception(429, {"Retry-After": "invalid"})
+
+        self.assertIsNone(retry_after_seconds(error))
+
+    def test_retry_after_seconds_rejects_float_values(self):
+        error = api_exception(429, {"Retry-After": "0.5"})
+
+        self.assertIsNone(retry_after_seconds(error))
+
+    def test_on_error_uses_backoff_with_retry_after(self):
+        attempts = []
+        sleeps = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(429, {"Retry-After": "3"})
+            return "ok"
+
+        result = on_error(
+            Backoff(steps=3, duration=1.0, factor=2.0),
+            is_too_many_requests,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [1.0])
+
+    def test_on_retry_after_error_honors_retry_after_for_429(self):
+        attempts = []
+        sleeps = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(429, {"Retry-After": "3"})
+            return "ok"
+
+        result = on_retry_after_error(
+            Backoff(steps=3, duration=1.0, factor=2.0),
+            is_too_many_requests,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [3.0])
+
+    def test_on_retry_after_error_uses_backoff_when_longer(self):
+        attempts = []
+        sleeps = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(429, {"Retry-After": "1"})
+            return "ok"
+
+        result = on_retry_after_error(
+            Backoff(steps=3, duration=3.0, factor=2.0),
+            is_too_many_requests,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [3.0])
+
+    def test_on_retry_after_error_honors_retry_after_for_5xx(self):
+        attempts = []
+        sleeps = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(500, {"Retry-After": "3"})
+            return "ok"
+
+        result = on_retry_after_error(
+            Backoff(steps=3, duration=1.0, factor=2.0),
+            lambda e: getattr(e, "status", None) == 500,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [3.0])
+
+    def test_on_retry_after_error_falls_back_without_retry_after(self):
+        sleeps = []
+        attempts = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(429)
+            return "ok"
+
+        result = on_retry_after_error(
+            Backoff(steps=3, duration=1.0, factor=2.0),
+            is_too_many_requests,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [1.0])
+
+    def test_on_retry_after_error_falls_back_with_invalid_retry_after(self):
+        sleeps = []
+        attempts = []
+
+        def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(429, {"Retry-After": "invalid"})
+            return "ok"
+
+        result = on_retry_after_error(
+            Backoff(steps=3, duration=1.0, factor=2.0),
+            is_too_many_requests,
+            fn,
+            sleep_func=sleeps.append,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [1.0])
+
+    def test_retry_after_seconds_rejects_negative_values(self):
+        error = api_exception(429, {"Retry-After": "-1"})
+
+        self.assertIsNone(retry_after_seconds(error))
+
+    def test_status_helpers(self):
+        self.assertTrue(is_conflict(api_exception(409)))
+        self.assertFalse(is_conflict(api_exception(500)))
+        self.assertTrue(is_too_many_requests(api_exception(429)))
+        self.assertFalse(is_too_many_requests(api_exception(409)))
+        self.assertTrue(
+            is_retry_after_response(
+                api_exception(429, {"Retry-After": "1"})))
+        self.assertTrue(
+            is_retry_after_response(
+                api_exception(500, {"Retry-After": "1"})))
+        self.assertFalse(is_retry_after_response(api_exception(429)))
+        self.assertFalse(
+            is_retry_after_response(
+                api_exception(500, {"Retry-After": "invalid"})))
+        self.assertFalse(
+            is_retry_after_response(
+                api_exception(409, {"Retry-After": "1"})))
+
+
+class AsyncRetryTest(unittest.IsolatedAsyncioTestCase):
+
+    async def test_async_on_error_retries_retriable_errors(self):
+        attempts = []
+        sleeps = []
+
+        async def fn():
+            attempts.append(1)
+            if len(attempts) < 3:
+                raise api_exception(500)
+            return "ok"
+
+        async def sleep(delay):
+            sleeps.append(delay)
+
+        result = await async_on_error(
+            Backoff(steps=4, duration=1.0, factor=2.0),
+            lambda e: getattr(e, "status", None) == 500,
+            fn,
+            sleep_func=sleep,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(len(attempts), 3)
+        self.assertEqual(sleeps, [1.0, 2.0])
+
+    async def test_async_on_retry_after_error_honors_retry_after_for_429(self):
+        attempts = []
+        sleeps = []
+
+        async def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(429, {"Retry-After": "3"})
+            return "ok"
+
+        async def sleep(delay):
+            sleeps.append(delay)
+
+        result = await async_on_retry_after_error(
+            Backoff(steps=3, duration=1.0, factor=2.0),
+            is_too_many_requests,
+            fn,
+            sleep_func=sleep,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "ok")
+        self.assertEqual(sleeps, [3.0])
+
+    async def test_async_retry_on_conflict_retries(self):
+        attempts = []
+
+        async def fn():
+            attempts.append(1)
+            if len(attempts) < 2:
+                raise api_exception(409)
+            return "updated"
+
+        async def sleep(delay):
+            pass
+
+        result = await async_retry_on_conflict(
+            fn,
+            backoff=Backoff(steps=3, duration=0.0, factor=1.0),
+            sleep_func=sleep,
+        )
+
+        self.assertEqual(result, "updated")
+        self.assertEqual(len(attempts), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/setup-asyncio.py b/setup-asyncio.py
index ecad9d520a..8928ec8161 100644
--- a/setup-asyncio.py
+++ b/setup-asyncio.py
@@ -48,6 +48,7 @@
     packages=[
         'kubernetes.aio',
         'kubernetes.aio.config',
+        'kubernetes.aio.utils',
         'kubernetes.aio.client',
         'kubernetes.aio.client.api',
         'kubernetes.aio.client.models'],