diff --git a/aws_lambda_powertools/utilities/auth_alpha/__init__.py b/aws_lambda_powertools/utilities/auth_alpha/__init__.py
index 5809a0dec7e..2090e16bd89 100644
--- a/aws_lambda_powertools/utilities/auth_alpha/__init__.py
+++ b/aws_lambda_powertools/utilities/auth_alpha/__init__.py
@@ -6,15 +6,21 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.auth_alpha.exceptions import AuthFailureReason as AuthFailureReason
     from aws_lambda_powertools.utilities.auth_alpha.jwt import AuthErrorContext as AuthErrorContext
-    from aws_lambda_powertools.utilities.auth_alpha.jwt import AuthFailureReason as AuthFailureReason
     from aws_lambda_powertools.utilities.auth_alpha.jwt import JWTVerifier as JWTVerifier
+    from aws_lambda_powertools.utilities.auth_alpha.oauth2 import OAuth2Client as OAuth2Client
 
-__all__ = ["AuthErrorContext", "AuthFailureReason", "JWTVerifier"]
+__all__ = ["AuthErrorContext", "AuthFailureReason", "JWTVerifier", "OAuth2Client"]
 
 
 def __getattr__(name: str) -> object:
-    modules = {"AuthErrorContext": "jwt", "AuthFailureReason": "jwt", "JWTVerifier": "jwt"}
+    modules = {
+        "AuthErrorContext": "jwt",
+        "AuthFailureReason": "exceptions",
+        "JWTVerifier": "jwt",
+        "OAuth2Client": "oauth2",
+    }
     if name in modules:
         value = getattr(importlib.import_module(f"{__name__}.{modules[name]}"), name)
         globals()[name] = value
diff --git a/aws_lambda_powertools/utilities/auth_alpha/_internal/errors.py b/aws_lambda_powertools/utilities/auth_alpha/_internal/errors.py
new file mode 100644
index 00000000000..89e0f2de078
--- /dev/null
+++ b/aws_lambda_powertools/utilities/auth_alpha/_internal/errors.py
@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from functools import wraps
+from typing import TYPE_CHECKING, ParamSpec, TypeVar
+
+from aws_lambda_powertools.utilities.auth_alpha.exceptions import AuthError
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+_P = ParamSpec("_P")
+_T = TypeVar("_T")
+
+
+def sanitize_errors(operation: Callable[_P, _T]) -> Callable[_P, _T]:
+    """Detach provider exceptions before an Auth error leaves a public operation."""
+
+    @wraps(operation)
+    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
+        try:
+            return operation(*args, **kwargs)
+        except AuthError as error:
+            # `raise ... from None` only suppresses display of the context.
+            # Clear both references and use a bare re-raise so Python does not
+            # attach the active exception again.
+            error.__context__ = None
+            error.__cause__ = None
+            raise
+
+    return wrapper
diff --git a/aws_lambda_powertools/utilities/auth_alpha/_internal/http.py b/aws_lambda_powertools/utilities/auth_alpha/_internal/http.py
index 109ebca4836..322372b2445 100644
--- a/aws_lambda_powertools/utilities/auth_alpha/_internal/http.py
+++ b/aws_lambda_powertools/utilities/auth_alpha/_internal/http.py
@@ -1,16 +1,22 @@
 from __future__ import annotations
 
 import json
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 import urllib3
-from urllib3.connection import HTTPConnection
+from urllib3.poolmanager import pool_classes_by_scheme
 
 from aws_lambda_powertools.utilities.auth_alpha._internal.deadline import Deadline, RequestError
+from aws_lambda_powertools.utilities.auth_alpha._internal.transport import (
+    DeadlineHTTPSConnectionPool,
+    response_deadline,
+)
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
 
+    from urllib3.connectionpool import HTTPConnectionPool
+
 _MAX_JSON_BYTES = 1024 * 1024
 
 
@@ -19,6 +25,27 @@ class HTTPClient:
 
     def __init__(self) -> None:
         self.pool = urllib3.PoolManager(cert_reqs="CERT_REQUIRED")
+        self.pool.pool_classes_by_scheme = pool_classes_by_scheme.copy()
+        pool_classes = cast("dict[str, type[HTTPConnectionPool]]", self.pool.pool_classes_by_scheme)
+        pool_classes["https"] = DeadlineHTTPSConnectionPool
+
+    def _open_response(
+        self,
+        method: str,
+        url: str,
+        deadline: Deadline,
+        **options: Any,
+    ) -> urllib3.response.BaseHTTPResponse:
+        with response_deadline(deadline):
+            return self.pool.request(
+                method,
+                url,
+                timeout=urllib3.Timeout(total=deadline.remaining()),
+                retries=False,
+                redirect=False,
+                preload_content=False,
+                **options,
+            )
 
     def json_request(
         self,
@@ -31,15 +58,12 @@ def json_request(
     ) -> tuple[int, dict[str, Any]]:
         response = None
         try:
-            response = self.pool.request(
+            response = self._open_response(
                 method,
                 url,
+                deadline,
                 body=body,
                 headers=headers,
-                timeout=urllib3.Timeout(total=deadline.remaining()),
-                retries=False,
-                redirect=False,
-                preload_content=False,
             )
             if response.status != 200:
                 deadline.remaining()
@@ -55,23 +79,67 @@ def json_request(
 
     @staticmethod
     def _read_json(response: urllib3.response.BaseHTTPResponse, deadline: Deadline) -> dict[str, Any]:
+        content = HTTPClient._read_body(response, deadline, limit=_MAX_JSON_BYTES, decode_content=False)
+        try:
+            data = json.loads(content)
+        except (ValueError, UnicodeError, RecursionError):
+            raise RequestError() from None
+        if not isinstance(data, dict):
+            raise RequestError()
+        return data
+
+    def request(
+        self,
+        method: str,
+        url: str,
+        deadline: Deadline,
+        *,
+        headers: Mapping[str, str],
+        **options: Any,
+    ) -> urllib3.response.HTTPResponse:
+        """Buffer an authenticated response within one network time budget."""
+        response = None
+        try:
+            response = self._open_response(
+                method,
+                url,
+                deadline,
+                headers=headers,
+                **options,
+            )
+            content = self._read_body(response, deadline)
+            return urllib3.HTTPResponse(
+                body=content,
+                status=response.status,
+                headers=response.headers,
+                reason=response.reason,
+                version=response.version,
+                request_method=method,
+                request_url=url,
+                decode_content=False,
+            )
+        finally:
+            if response is not None:
+                response.close()
+                response.release_conn()
+
+    @staticmethod
+    def _read_body(
+        response: urllib3.response.BaseHTTPResponse,
+        deadline: Deadline,
+        *,
+        limit: int | None = None,
+        decode_content: bool = True,
+    ) -> bytes:
         chunks = bytearray()
         while True:
-            remaining = deadline.remaining()
-            connection = response.connection
-            if isinstance(connection, HTTPConnection) and connection.sock is not None:
-                connection.sock.settimeout(remaining)
-            chunk = response.read1(min(65536, _MAX_JSON_BYTES + 1 - len(chunks)), decode_content=False)
+            deadline.remaining()
+            size = 65536 if limit is None else min(65536, limit + 1 - len(chunks))
+            chunk = response.read1(size, decode_content=decode_content)
             deadline.remaining()
             if not chunk:
                 break
             chunks.extend(chunk)
-            if len(chunks) > _MAX_JSON_BYTES:
+            if limit is not None and len(chunks) > limit:
                 raise RequestError()
-        try:
-            data = json.loads(chunks)
-        except (ValueError, UnicodeError, RecursionError):
-            raise RequestError() from None
-        if not isinstance(data, dict):
-            raise RequestError()
-        return data
+        return bytes(chunks)
diff --git a/aws_lambda_powertools/utilities/auth_alpha/_internal/scopes.py b/aws_lambda_powertools/utilities/auth_alpha/_internal/scopes.py
new file mode 100644
index 00000000000..d80e0901f29
--- /dev/null
+++ b/aws_lambda_powertools/utilities/auth_alpha/_internal/scopes.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from aws_lambda_powertools.utilities.auth_alpha._internal.validation import string_list
+
+
+def valid_scope(value: str) -> bool:
+    return bool(value) and all(33 <= ord(character) <= 126 and character not in {'"', "\\"} for character in value)
+
+
+def required_scopes(scopes: list[str] | None) -> tuple[str, ...]:
+    values = string_list(scopes if scopes is not None else [])
+    if not all(valid_scope(value) for value in values):
+        raise ValueError("Scopes must be valid OAuth scope tokens")
+    return values
diff --git a/aws_lambda_powertools/utilities/auth_alpha/_internal/transport.py b/aws_lambda_powertools/utilities/auth_alpha/_internal/transport.py
new file mode 100644
index 00000000000..bf2b496a7a6
--- /dev/null
+++ b/aws_lambda_powertools/utilities/auth_alpha/_internal/transport.py
@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+from contextvars import ContextVar
+from http.client import HTTPResponse
+from io import BufferedReader, RawIOBase
+from typing import TYPE_CHECKING
+
+from urllib3.connection import HTTPSConnection
+from urllib3.connectionpool import HTTPSConnectionPool
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from socket import socket
+    from typing import Protocol
+
+    from typing_extensions import Buffer
+
+    from aws_lambda_powertools.utilities.auth_alpha._internal.deadline import Deadline
+
+    class _ResponseStream(Protocol):
+        def readinto1(self, buffer: Buffer, /) -> int: ...
+        def close(self) -> None: ...
+
+
+_current_deadline: ContextVar[Deadline] = ContextVar("auth_response_deadline")
+
+
+@contextmanager
+def response_deadline(deadline: Deadline) -> Iterator[None]:
+    """Pass the operation's deadline to responses created by this synchronous call."""
+    token = _current_deadline.set(deadline)
+    try:
+        yield
+    finally:
+        _current_deadline.reset(token)
+
+
+class _DeadlineReader(RawIOBase):
+    """Check the original budget on every refill, including HTTP framing reads."""
+
+    def __init__(self, stream: _ResponseStream, sock: socket, deadline: Deadline) -> None:
+        super().__init__()
+        self._stream = stream
+        self._socket = sock
+        self._deadline = deadline
+
+    def readable(self) -> bool:
+        return True
+
+    def readinto(self, buffer: Buffer, /) -> int:
+        self._socket.settimeout(self._deadline.remaining())
+        # Unlike readinto(), readinto1() performs at most one underlying read.
+        count = self._stream.readinto1(buffer)
+        self._deadline.remaining()
+        return count
+
+    def close(self) -> None:
+        try:
+            self._stream.close()
+        finally:
+            super().close()
+
+
+class _DeadlineResponse(HTTPResponse):
+    def __init__(
+        self,
+        sock: socket,
+        debuglevel: int = 0,
+        method: str | None = None,
+        url: str | None = None,
+    ) -> None:
+        deadline = _current_deadline.get()
+        super().__init__(sock, debuglevel=debuglevel, method=method, url=url)
+        # Retain the wrapper through body consumption: read1() can also parse
+        # chunk-size lines, delimiters and trailers before returning to our loop.
+        # The stream owns the socket reference even for Connection: close.
+        self.fp = BufferedReader(_DeadlineReader(self.fp, sock, deadline), buffer_size=8192)
+
+
+class _DeadlineHTTPSConnection(HTTPSConnection):
+    response_class = _DeadlineResponse
+
+
+class DeadlineHTTPSConnectionPool(HTTPSConnectionPool):
+    """Use the stdlib response hook without overriding urllib3's request machinery."""
+
+    ConnectionCls = _DeadlineHTTPSConnection
diff --git a/aws_lambda_powertools/utilities/auth_alpha/exceptions.py b/aws_lambda_powertools/utilities/auth_alpha/exceptions.py
new file mode 100644
index 00000000000..e6f598183e4
--- /dev/null
+++ b/aws_lambda_powertools/utilities/auth_alpha/exceptions.py
@@ -0,0 +1,29 @@
+"""Credential-free errors raised by the Auth utility."""
+
+from enum import Enum
+
+
+class AuthFailureReason(str, Enum):
+    """Stable, credential-free reasons suitable for application logs and metrics."""
+
+    MISSING_TOKEN = "missing_token"  # nosec B105
+    INVALID_TOKEN = "invalid_token"  # nosec B105
+    INVALID_CLAIMS = "invalid_claims"
+    TOKEN_EXPIRED = "token_expired"  # nosec B105
+    INVALID_SIGNATURE = "invalid_signature"
+    INSUFFICIENT_SCOPE = "insufficient_scope"
+    FORBIDDEN = "forbidden"
+    JWKS_UNAVAILABLE = "jwks_unavailable"
+    TOKEN_EXCHANGE_FAILED = "token_exchange_failed"  # nosec B105
+    DOWNSTREAM_REQUEST_FAILED = "downstream_request_failed"
+
+
+class AuthError(Exception):
+    """Base error with a fixed message that never includes credential material."""
+
+    message = "Authentication failed"
+    reason = AuthFailureReason.INVALID_TOKEN
+    retryable = False
+
+    def __init__(self) -> None:
+        super().__init__(self.message)
diff --git a/aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/authorization.py b/aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/authorization.py
index b035991901e..6732104425e 100644
--- a/aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/authorization.py
+++ b/aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/authorization.py
@@ -3,7 +3,7 @@
 from collections.abc import Mapping
 from typing import Any
 
-from aws_lambda_powertools.utilities.auth_alpha._internal.validation import string_list
+from aws_lambda_powertools.utilities.auth_alpha._internal.scopes import required_scopes, valid_scope
 from aws_lambda_powertools.utilities.auth_alpha.jwt.exceptions import (
     AuthError,
     AuthFailureReason,
@@ -11,6 +11,8 @@
     InvalidTokenError,
 )
 
+__all__ = ["required_scopes"]
+
 
 class MissingTokenError(InvalidTokenError):
     """No authorization header was supplied."""
@@ -65,17 +67,6 @@ def _authorization_values(headers: Any) -> list[Any]:
     return values
 
 
-def valid_scope(value: str) -> bool:
-    return bool(value) and all(33 <= ord(character) <= 126 and character not in {'"', "\\"} for character in value)
-
-
-def required_scopes(scopes: list[str] | None) -> tuple[str, ...]:
-    values = string_list(scopes if scopes is not None else [])
-    if not all(valid_scope(value) for value in values):
-        raise ValueError("Scopes must be valid OAuth scope tokens")
-    return values
-
-
 def enforce_scopes(claims: dict[str, Any], expected: tuple[str, ...]) -> None:
     value: Any = next((claims[name] for name in ("scope", "scp", "scopes") if name in claims), [])
     if isinstance(value, str):
diff --git a/aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/errors.py b/aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/errors.py
index c99f88e54d2..0ac7598fe60 100644
--- a/aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/errors.py
+++ b/aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/errors.py
@@ -1,30 +1,3 @@
-from __future__ import annotations
+from aws_lambda_powertools.utilities.auth_alpha._internal.errors import sanitize_errors
 
-from functools import wraps
-from typing import TYPE_CHECKING, ParamSpec, TypeVar
-
-from aws_lambda_powertools.utilities.auth_alpha.jwt.exceptions import AuthError
-
-if TYPE_CHECKING:
-    from collections.abc import Callable
-
-_P = ParamSpec("_P")
-_T = TypeVar("_T")
-
-
-def sanitize_errors(operation: Callable[_P, _T]) -> Callable[_P, _T]:
-    """Detach provider exceptions before an Auth error leaves a public operation."""
-
-    @wraps(operation)
-    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
-        try:
-            return operation(*args, **kwargs)
-        except AuthError as error:
-            # `raise ... from None` only suppresses display of the context.
-            # Clear both references and use a bare re-raise so Python does not
-            # attach the active exception again.
-            error.__context__ = None
-            error.__cause__ = None
-            raise
-
-    return wrapper
+__all__ = ["sanitize_errors"]
diff --git a/aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py b/aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py
index 6c460b34790..03f2044f302 100644
--- a/aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py
+++ b/aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py
@@ -1,30 +1,16 @@
-"""Credential-free errors raised by the Auth utility."""
-
-from enum import Enum
-
-
-class AuthFailureReason(str, Enum):
-    """Stable, credential-free reasons suitable for application logs and metrics."""
-
-    MISSING_TOKEN = "missing_token"  # nosec B105
-    INVALID_TOKEN = "invalid_token"  # nosec B105
-    INVALID_CLAIMS = "invalid_claims"
-    TOKEN_EXPIRED = "token_expired"  # nosec B105
-    INVALID_SIGNATURE = "invalid_signature"
-    INSUFFICIENT_SCOPE = "insufficient_scope"
-    FORBIDDEN = "forbidden"
-    JWKS_UNAVAILABLE = "jwks_unavailable"
-
-
-class AuthError(Exception):
-    """Base error with a fixed message that never includes credential material."""
-
-    message = "Authentication failed"
-    reason = AuthFailureReason.INVALID_TOKEN
-    retryable = False
-
-    def __init__(self) -> None:
-        super().__init__(self.message)
+"""Credential-free errors raised by JWT verification."""
+
+from aws_lambda_powertools.utilities.auth_alpha.exceptions import AuthError, AuthFailureReason
+
+__all__ = [
+    "AuthError",
+    "AuthFailureReason",
+    "InvalidTokenError",
+    "InvalidClaimsError",
+    "TokenExpiredError",
+    "InvalidSignatureError",
+    "JWKSFetchError",
+]
 
 
 class InvalidTokenError(AuthError):
diff --git a/aws_lambda_powertools/utilities/auth_alpha/oauth2/__init__.py b/aws_lambda_powertools/utilities/auth_alpha/oauth2/__init__.py
new file mode 100644
index 00000000000..51c2bfea71e
--- /dev/null
+++ b/aws_lambda_powertools/utilities/auth_alpha/oauth2/__init__.py
@@ -0,0 +1,23 @@
+"""OAuth 2.0 client-credentials token acquisition."""
+
+from __future__ import annotations
+
+import importlib
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.auth_alpha.oauth2.client import OAuth2Client as OAuth2Client
+
+__all__ = ["OAuth2Client"]
+
+
+def __getattr__(name: str) -> object:
+    if name == "OAuth2Client":
+        value = importlib.import_module(f"{__name__}.client").OAuth2Client
+        globals()[name] = value
+        return value
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(__all__))
diff --git a/aws_lambda_powertools/utilities/auth_alpha/oauth2/client.py b/aws_lambda_powertools/utilities/auth_alpha/oauth2/client.py
new file mode 100644
index 00000000000..9c4cf66faa2
--- /dev/null
+++ b/aws_lambda_powertools/utilities/auth_alpha/oauth2/client.py
@@ -0,0 +1,376 @@
+from __future__ import annotations
+
+import base64
+import re
+import threading
+import time
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from string import ascii_letters, digits, hexdigits
+from typing import TYPE_CHECKING, Any
+from urllib.parse import quote_plus, urlencode, urlsplit
+
+import urllib3
+
+from aws_lambda_powertools.utilities.auth_alpha._internal.deadline import Deadline, RequestError
+from aws_lambda_powertools.utilities.auth_alpha._internal.errors import sanitize_errors
+from aws_lambda_powertools.utilities.auth_alpha._internal.http import HTTPClient
+from aws_lambda_powertools.utilities.auth_alpha._internal.scopes import required_scopes
+from aws_lambda_powertools.utilities.auth_alpha._internal.validation import (
+    finite_seconds,
+    https_url,
+    is_nonempty_string,
+)
+from aws_lambda_powertools.utilities.auth_alpha.oauth2.exceptions import DownstreamRequestError, TokenExchangeError
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+_BEARER_TOKEN = re.compile(r"[-A-Za-z0-9._~+/]+=*")
+_HEADER_NAME = re.compile(r"[-!#$%&'*+.^_`|~0-9A-Za-z]+")
+_RESOURCE_CHARACTERS = frozenset(ascii_letters + digits + "-._~:/?[]@!$&'()*+,;=")
+_HEX_DIGITS = frozenset(hexdigits)
+
+
+def _valid_resource_characters(resource: str) -> bool:
+    """Scan once, accepting URI characters and complete percent escapes, but no fragment."""
+    characters = iter(resource)
+    for character in characters:
+        if character == "%":
+            if next(characters, "") not in _HEX_DIGITS or next(characters, "") not in _HEX_DIGITS:
+                return False
+        elif character not in _RESOURCE_CHARACTERS:
+            return False
+    return True
+
+
+@dataclass(frozen=True)
+class _AccessToken:
+    value: str = field(repr=False)
+    expires_at: float | None
+
+    def cacheable(self) -> bool:
+        return self.expires_at is not None and time.monotonic() < self.expires_at - 30
+
+    def usable(self) -> bool:
+        return self.expires_at is None or time.monotonic() < self.expires_at
+
+
+@dataclass
+class _Exchange:
+    done: threading.Event = field(default_factory=threading.Event, repr=False)
+    token: _AccessToken | None = field(default=None, repr=False)
+    retryable: bool = False
+
+
+class OAuth2Client:
+    """Acquire bearer tokens using client credentials for one configured resource.
+
+    Parameters
+    ----------
+    token_url : str
+        Trusted HTTPS OAuth token endpoint.
+    client_id : str
+        Identifier for a client supporting ``client_secret_basic``.
+    client_secret : str | Callable[[], str]
+        Secret or loader invoked for each exchange attempt.
+    scopes : list[str], optional
+        Scopes requested on every exchange.
+    audience : str, optional
+        Provider-specific audience request field, mutually exclusive with resource.
+    resource : str, optional
+        RFC 8707 resource request field, mutually exclusive with audience.
+        Must be an absolute URI without a fragment; URNs are supported.
+    timeout_seconds : float
+        Positive acquisition budget including retries, by default 3.
+
+    Notes
+    -----
+    Instances do not share tokens. Tokens are reacquired 30 seconds before
+    expiration. Short-lived tokens and tokens without a lifetime are not cached.
+    Configure timeouts on application-provided secret loaders.
+
+    Examples
+    --------
+    ```python
+    client = OAuth2Client(
+        token_url="https://idp.example.com/token",
+        client_id="orders",
+        client_secret=load_secret,
+        resource="https://inventory.example.com",
+        scopes=["inventory:read"],
+    )
+    headers = client.auth_headers()
+    ```
+    """
+
+    def __init__(
+        self,
+        *,
+        token_url: str,
+        client_id: str,
+        client_secret: str | Callable[[], str],
+        scopes: list[str] | None = None,
+        audience: str | None = None,
+        resource: str | None = None,
+        timeout_seconds: float = 3,
+    ) -> None:
+        self._token_url = https_url(token_url)
+        if not is_nonempty_string(client_id):
+            raise ValueError("A nonempty OAuth client ID is required")
+        if not callable(client_secret) and (not isinstance(client_secret, str) or not client_secret):
+            raise ValueError("client_secret must be a nonempty string or a callable")
+        if audience is not None and resource is not None:
+            raise ValueError("audience and resource are mutually exclusive")
+        self._client_id = client_id
+        self._client_secret = client_secret
+        self._scopes = required_scopes(scopes)
+        self._timeout = finite_seconds(timeout_seconds, positive=True)
+        self._fields = {"grant_type": "client_credentials"}
+        if self._scopes:
+            self._fields["scope"] = " ".join(self._scopes)
+        for name, value in (("audience", audience), ("resource", resource)):
+            if value is not None:
+                if not is_nonempty_string(value):
+                    raise ValueError("Resource selection must be a nonempty string")
+                self._fields[name] = value
+        try:
+            self._client_id.encode("utf-8")
+            self._body = urlencode(self._fields).encode()
+        except UnicodeError:
+            raise ValueError("OAuth configuration must contain valid UTF-8 strings") from None
+        if resource is not None:
+            self._validate_resource(resource)
+        self._http = HTTPClient()
+        self._cached_token: _AccessToken | None = None
+        self._flight: _Exchange | None = None
+        self._lock = threading.Lock()
+
+    def __repr__(self) -> str:
+        return "<OAuth2Client>"
+
+    @staticmethod
+    def _validate_resource(resource: str) -> None:
+        try:
+            scheme = urlsplit(resource).scheme
+            # Older urlsplit versions also recognize schemes starting with a digit.
+            valid = scheme[:1].isalpha() and _valid_resource_characters(resource)
+        except ValueError:
+            valid = False
+        if not valid:
+            raise ValueError("resource must be an absolute URI without a fragment")
+
+    @sanitize_errors
+    def auth_headers(self) -> dict[str, str]:
+        """Return an Authorization header for this client's configured resource.
+
+        Raises
+        ------
+        TokenExchangeError
+            A usable bearer token could not be obtained within the budget.
+
+        Examples
+        --------
+        ```python
+        headers = client.auth_headers()
+        response = http.request("GET", trusted_inventory_url, headers=headers)
+        ```
+        """
+        try:
+            token = self._get_token(Deadline(self._timeout))
+        except RequestError as error:
+            raise TokenExchangeError(retryable=error.retryable) from None
+        return {"Authorization": f"Bearer {token.value}"}
+
+    @sanitize_errors
+    def request(
+        self,
+        method: str,
+        url: str,
+        *,
+        timeout: float = 5,
+        headers: Mapping[str, str] | None = None,
+        **options: Any,
+    ) -> urllib3.response.BaseHTTPResponse:
+        """Send a synchronous HTTPS request using this resource's bearer token.
+
+        Only trusted destination URLs should be supplied. Redirects and retries
+        are disabled, and an existing Authorization header is rejected.
+        ``body``, ``fields``, ``json``, ``encode_multipart`` and
+        ``multipart_boundary`` are forwarded to urllib3.
+
+        Parameters
+        ----------
+        method : str
+            HTTP method.
+        url : str
+            Trusted HTTPS destination for this resource's credentials.
+        timeout : float
+            Positive downstream timeout, separate from acquisition, by default 5.
+        headers : Mapping[str, str], optional
+            Additional headers with HTTP token names, excluding Authorization.
+
+        Returns
+        -------
+        urllib3.response.BaseHTTPResponse
+            Downstream response; inspect its status before consuming its body.
+
+        Raises
+        ------
+        TokenExchangeError
+            Token acquisition failed.
+        DownstreamRequestError
+            Downstream transport failed; the request is never replayed automatically.
+        ValueError
+            Request configuration is invalid.
+
+        Examples
+        --------
+        ```python
+        response = client.request("GET", "https://inventory.example.com/items")
+        if response.status == 200:
+            items = response.json()
+        ```
+        """
+        target = https_url(url)
+        duration = finite_seconds(timeout, positive=True)
+        allowed = {"body", "fields", "json", "encode_multipart", "multipart_boundary"}
+        if not options.keys() <= allowed:
+            raise ValueError("Unsupported authenticated request option")
+        if not isinstance(method, str) or not re.fullmatch(r"[A-Za-z]+", method):
+            raise ValueError("A valid HTTP method is required")
+        request_headers = self._request_headers(headers)
+        request_headers.update(self.auth_headers())
+        deadline = Deadline(duration)
+        try:
+            return self._http.request(
+                method.upper(),
+                target,
+                deadline,
+                headers=request_headers,
+                **options,
+            )
+        except (urllib3.exceptions.HTTPError, OSError, ValueError, TypeError, RequestError):
+            raise DownstreamRequestError() from None
+
+    @staticmethod
+    def _request_headers(headers: Mapping[str, str] | None) -> dict[str, str]:
+        if headers is None:
+            return {}
+        if not isinstance(headers, Mapping):
+            raise ValueError("Request headers must be a mapping of strings")
+        for name, value in headers.items():
+            if (
+                not isinstance(name, str)
+                or not isinstance(value, str)
+                or not _HEADER_NAME.fullmatch(name)
+                or name.lower() == "authorization"
+                or any(character in value for character in ("\r", "\n"))
+            ):
+                raise ValueError("Request headers must be valid and must not include Authorization")
+        return dict(headers)
+
+    def _get_token(self, deadline: Deadline) -> _AccessToken:
+        with self._lock:
+            if self._cached_token is not None and self._cached_token.cacheable():
+                return self._cached_token
+            self._cached_token = None
+            owner = self._flight is None
+            if self._flight is None:
+                self._flight = _Exchange()
+            flight = self._flight
+        if owner:
+            self._run_exchange(flight, deadline)
+        elif not flight.done.wait(timeout=deadline.remaining()):
+            raise TokenExchangeError(retryable=True)
+        deadline.remaining()
+        if flight.token is None:
+            raise TokenExchangeError(retryable=flight.retryable)
+        if not flight.token.usable():
+            raise TokenExchangeError()
+        return flight.token
+
+    def _run_exchange(self, flight: _Exchange, deadline: Deadline) -> None:
+        try:
+            token = self._exchange(deadline)
+            with self._lock:
+                if token.cacheable():
+                    self._cached_token = token
+                flight.token = token
+        except (TokenExchangeError, RequestError) as error:
+            flight.retryable = error.retryable
+            raise
+        finally:
+            # Waiters keep this flight's result, including uncacheable short
+            # tokens. Calls starting after completion must acquire their own.
+            with self._lock:
+                self._flight = None
+                flight.done.set()
+
+    def _exchange(self, deadline: Deadline) -> _AccessToken:
+        for attempt in range(3):
+            try:
+                return self._exchange_once(deadline)
+            except RequestError as error:
+                if not error.retryable or attempt == 2:
+                    raise TokenExchangeError(retryable=error.retryable) from None
+                delay = 0.1 * 2**attempt
+                if deadline.remaining() <= delay:
+                    raise TokenExchangeError(retryable=error.retryable) from None
+                time.sleep(delay)
+        raise TokenExchangeError()
+
+    def _credentials(self) -> str:
+        try:
+            secret = self._client_secret if isinstance(self._client_secret, str) else self._client_secret()
+        except Exception:
+            # Secret providers can raise arbitrary exceptions containing their
+            # configuration or response data. None of it crosses this boundary.
+            raise TokenExchangeError() from None
+        if not isinstance(secret, str) or not secret:
+            raise TokenExchangeError()
+        try:
+            credentials = f"{quote_plus(self._client_id)}:{quote_plus(secret)}"
+        except UnicodeError:
+            raise TokenExchangeError() from None
+        return base64.b64encode(credentials.encode()).decode()
+
+    def _exchange_once(self, deadline: Deadline) -> _AccessToken:
+        authorization = self._credentials()
+        started = time.monotonic()
+        status, payload = self._http.json_request(
+            "POST",
+            self._token_url,
+            deadline,
+            body=self._body,
+            headers={
+                "Authorization": f"Basic {authorization}",
+                "Content-Type": "application/x-www-form-urlencoded",
+            },
+        )
+        if status != 200:
+            raise RequestError(retryable=status == 429 or 500 <= status <= 599)
+        return self._parse_token(payload, started)
+
+    @staticmethod
+    def _parse_token(payload: dict[str, Any], started: float) -> _AccessToken:
+        value = payload.get("access_token")
+        token_type = payload.get("token_type")
+        if (
+            not isinstance(value, str)
+            or not _BEARER_TOKEN.fullmatch(value)
+            or not isinstance(token_type, str)
+            or token_type.lower() != "bearer"
+        ):
+            raise TokenExchangeError()
+        expires_at = None
+        if "expires_in" in payload:
+            try:
+                lifetime = finite_seconds(payload["expires_in"], positive=True)
+            except ValueError:
+                raise TokenExchangeError() from None
+            expires_at = started + lifetime
+        token = _AccessToken(value, expires_at)
+        if not token.usable():
+            raise TokenExchangeError()
+        return token
diff --git a/aws_lambda_powertools/utilities/auth_alpha/oauth2/exceptions.py b/aws_lambda_powertools/utilities/auth_alpha/oauth2/exceptions.py
new file mode 100644
index 00000000000..f4cdd7f060e
--- /dev/null
+++ b/aws_lambda_powertools/utilities/auth_alpha/oauth2/exceptions.py
@@ -0,0 +1,32 @@
+"""Credential-free errors raised by outbound OAuth requests."""
+
+from aws_lambda_powertools.utilities.auth_alpha.exceptions import AuthError, AuthFailureReason
+
+__all__ = ["AuthError", "AuthFailureReason", "TokenExchangeError", "DownstreamRequestError"]
+
+
+class TokenExchangeError(AuthError):
+    """A usable bearer token could not be obtained.
+
+    ``retryable`` indicates a transient endpoint failure or acquisition timeout.
+    Invalid responses, rejected credentials, and secret-loader failures are not
+    retried automatically. Exception messages never include provider details.
+    """
+
+    message = "Unable to acquire an access token"
+    reason = AuthFailureReason.TOKEN_EXCHANGE_FAILED
+
+    def __init__(self, *, retryable: bool = False) -> None:
+        self.retryable = retryable
+        super().__init__()
+
+
+class DownstreamRequestError(AuthError):
+    """The authenticated HTTP operation could not complete.
+
+    The server may already have performed the operation. Callers must decide
+    whether replay is safe; this error does not advise automatic retries.
+    """
+
+    message = "Authenticated request failed"
+    reason = AuthFailureReason.DOWNSTREAM_REQUEST_FAILED
diff --git a/docs/api_doc/auth_alpha.md b/docs/api_doc/auth_alpha.md
index 52409094848..30854875ab5 100644
--- a/docs/api_doc/auth_alpha.md
+++ b/docs/api_doc/auth_alpha.md
@@ -5,3 +5,5 @@
 ::: aws_lambda_powertools.utilities.auth_alpha.AuthErrorContext
 ::: aws_lambda_powertools.utilities.auth_alpha.jwt.exceptions
 ::: aws_lambda_powertools.utilities.auth_alpha.jwt.testing
+::: aws_lambda_powertools.utilities.auth_alpha.oauth2.client
+::: aws_lambda_powertools.utilities.auth_alpha.oauth2.exceptions
diff --git a/docs/getting-started/install.md b/docs/getting-started/install.md
index e744b9e35e0..f17a68ee92b 100644
--- a/docs/getting-started/install.md
+++ b/docs/getting-started/install.md
@@ -43,6 +43,7 @@ Some features require additional dependencies. Install them as needed:
 | [Validation](../utilities/validation.md) | `pip install "aws-lambda-powertools[validation]"` | `fastjsonschema` |
 | [Parser](../utilities/parser.md) | `pip install "aws-lambda-powertools[parser]"` | `pydantic` |
 | [JWT verification (alpha)](../utilities/auth.md) | `pip install "aws-lambda-powertools[jwt]"` | `PyJWT`, `cryptography`, `urllib3` |
+| [OAuth2 client (alpha)](../utilities/oauth2.md) | `pip install "aws-lambda-powertools[oauth2]"` | `urllib3` |
 | [Data Masking](../utilities/data_masking.md) | `pip install "aws-lambda-powertools[datamasking]"` | `aws-encryption-sdk`, `jsonpath-ng` |
 | [Datadog Metrics](../core/metrics/datadog.md) | `pip install "aws-lambda-powertools[datadog]"` | `datadog-lambda` |
 | [Kafka (Avro)](../utilities/kafka.md) | `pip install "aws-lambda-powertools[kafka-consumer-avro]"` | `avro` |
diff --git a/docs/index.md b/docs/index.md
index 475eda10a74..6a4c1317b0f 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -54,7 +54,7 @@ Powertools for AWS Lambda (Python) is a developer toolkit to implement Serverles
 | [Metrics](./core/metrics.md) | Custom Metrics created asynchronously via CloudWatch Embedded Metric Format (EMF) |
 | [Event Handler](./core/event_handler/api_gateway.md) | Event handler for API Gateway, ALB, Lambda Function URL, VPC Lattice, AppSync, and Bedrock Agents |
 | [Parameters](./utilities/parameters.md) | Retrieve and cache parameter values from Parameter Store, Secrets Manager, AppConfig, or DynamoDB |
-| [Auth (alpha)](./utilities/auth.md) | Verify JWT access tokens and use verified claims in Lambda workloads |
+| [Auth (alpha)](./utilities/auth.md) | Verify JWT access tokens and [acquire client-credentials tokens](./utilities/oauth2.md) for downstream APIs |
 | [Parser](./utilities/parser.md) | Data parsing and deep validation using Pydantic |
 | [Batch Processing](./utilities/batch.md) | Handle partial failures for SQS, Kinesis Data Streams, and DynamoDB Streams |
 | [Idempotency](./utilities/idempotency.md) | Make your Lambda functions idempotent and prevent duplicate execution |
diff --git a/docs/utilities/auth.md b/docs/utilities/auth.md
index 849f4ef3abd..f84879cc4c2 100644
--- a/docs/utilities/auth.md
+++ b/docs/utilities/auth.md
@@ -9,6 +9,8 @@ status: new
 
 Auth verifies JWT access tokens in any Lambda workload. Use `verify()` directly, create Event Handler middleware with `require()`, or build a Lambda authorizer response with `authorize()`.
 
+For client-credentials tokens used to call a downstream API, see [OAuth2 client (alpha)](oauth2.md).
+
 ```mermaid
 flowchart LR
     Token["JWT access token"] --> Verify["JWTVerifier.verify()"]
@@ -168,7 +170,8 @@ The example template disables API Gateway authorizer-result caching so every req
 --8<-- "examples/auth_alpha/jwt/templates/sam.yaml"
 ```
 
-If you enable Gateway caching, include all request attributes used by authorization in its identity sources to prevent decisions from being reused across different authorization inputs. Even with a complete cache key, a cached allow can outlive the JWT expiration until the cache TTL expires. Keep result caching disabled when every request must respect token expiration. This cache is independent of the verifier JWKS cache.
+If you enable Gateway caching, include all request attributes used by authorization in its identity sources to prevent decisions from being reused across different authorization inputs.
+Even with a complete cache key, a cached allow can outlive the JWT expiration until the cache TTL expires. Keep result caching disabled when every request must respect token expiration. This cache is independent of the verifier JWKS cache.
 
 ### Errors and diagnostics
 
diff --git a/docs/utilities/oauth2.md b/docs/utilities/oauth2.md
new file mode 100644
index 00000000000..2938852bb5e
--- /dev/null
+++ b/docs/utilities/oauth2.md
@@ -0,0 +1,163 @@
+---
+title: OAuth2 client (alpha)
+description: Acquire and cache client-credentials tokens for downstream APIs
+status: new
+---
+
+!!! warning "Alpha / experimental"
+    This utility ships under the `auth_alpha` namespace while we collect feedback. Its public API may change before GA. Pin your Powertools version before using it in production.
+
+`OAuth2Client` obtains bearer tokens for a Lambda function calling an OAuth2-protected API. Each client owns its resource configuration and token cache. It supports the client-credentials grant with `client_secret_basic` authentication.
+
+Use [JWT verification](auth.md) to authenticate incoming requests. The OAuth client obtains separate credentials for outgoing requests; it does not forward an incoming caller's token.
+
+## Key features
+
+* Cache access tokens across warm Lambda invocations and reacquire them before expiration.
+* Coordinate concurrent token requests within one client.
+* Resolve a client secret for each exchange attempt.
+* Select a downstream API using a provider-specific audience or an RFC 8707 resource indicator.
+* Obtain headers for your HTTP client or send a synchronous authenticated request.
+* Report fixed failure reasons without exposing credentials or provider responses.
+
+## Getting started
+
+### Install
+
+```shell
+pip install "aws-lambda-powertools[oauth2]"
+```
+
+The `oauth2` extra installs urllib3. It does not require PyJWT or cryptography. The client is available from both `aws_lambda_powertools.utilities.auth_alpha` and `aws_lambda_powertools.utilities.auth_alpha.oauth2`.
+
+### Call a downstream API
+
+Create the client outside the Lambda handler so warm invocations reuse its token cache. This complete Lambda loads its client secret through Parameters and calls an inventory API:
+
+```python title="client_credentials.py"
+--8<-- "examples/auth_alpha/oauth2/src/client_credentials.py"
+```
+
+Configure `TOKEN_URL`, `CLIENT_ID`, `CLIENT_SECRET_NAME`, and `INVENTORY_URL` as deployment settings. The secret must be a plain string. The function needs permission to retrieve that secret and outbound HTTPS connectivity to both endpoints. Token exchange itself requires no additional IAM permissions.
+
+`OAuth2Client` provides two operations:
+
+| Method | Use when |
+| ------ | -------- |
+| `auth_headers()` | You want an Authorization header for an application-owned HTTP client |
+| `request(method, url, ...)` | You want the utility to send an authenticated HTTPS request |
+
+Both methods acquire a token only when one is needed. Construction performs no network requests.
+
+### Choose the resource
+
+| Parameter | Token-request field | Purpose |
+| --------- | ------------------- | ------- |
+| `audience` | `audience=<value>` | Provider-specific API selection, such as an Auth0 API identifier |
+| `resource` | `resource=<value>` | One resource indicator for providers supporting RFC 8707 |
+
+These parameters are mutually exclusive and are not interchangeable. Configure the parameter supported by your provider. If neither is supplied, the provider must select the intended API through its client configuration or scope conventions; scopes alone do not universally identify a resource.
+
+`resource` must be an absolute URI without a fragment, such as `https://inventory.example.com` or `urn:example:inventory`. Query parameters and percent-encoded characters are preserved. Relative paths and malformed URI characters are rejected during construction. `audience` remains a provider-specific, nonempty string.
+
+Use a separate client for each API. The eventual request URL does not change the token's audience, and clients do not share token caches. Changing the requested resource requires creating a new client.
+
+### Use your own HTTP client
+
+`auth_headers()` returns a new dictionary containing `Authorization: Bearer <token>`. You can pass it to urllib3, requests, httpx, or another HTTP client:
+
+```python title="headers.py"
+--8<-- "examples/auth_alpha/oauth2/src/headers.py"
+```
+
+The example validates that `INVENTORY_URL` uses HTTPS before obtaining any credentials or sending requests. It uses an environment-provided secret; the Parameters loader from the first example also works here. With your own HTTP client, enforce HTTPS and configure its timeouts, redirects, and retries yourself. Never log the returned headers or forward them to an untrusted destination.
+
+## Advanced
+
+### Token lifetimes and concurrency
+
+Tokens are cached while more than 30 seconds of their positive `expires_in` remain. On demand, the client reacquires a token when 30 seconds or less remain. This performs a new client-credentials exchange; it does not use an OAuth refresh token.
+
+Tokens with an advertised lifetime of 30 seconds or less, or without `expires_in`, are returned without caching. A call does not loop trying to obtain a longer-lived token. Invalid lifetimes and tokens that expire during acquisition are rejected.
+Lifetime accounting uses a monotonic clock starting immediately before the token request, after secret lookup. Secret lookup consumes the acquisition budget but does not shorten the newly issued token's lifetime.
+
+Concurrent callers share one in-progress exchange, including short-lived tokens and failures. A waiting caller has its own acquisition deadline. Separate clients and Lambda execution environments have separate caches.
+
+### Secret rotation and client authentication
+
+`client_secret` accepts a nonempty string or a callable returning one. A callable runs for each exchange attempt, including retries. The client does not cache the callable's returned secret separately.
+
+An already cached access token can remain usable after a secret changes. Parameters also has its own cache: the first example's `max_age=300` can delay observation of a changed secret by five minutes. Configure secret-provider timeouts independently; the client cannot interrupt an application-supplied callable.
+
+The token endpoint receives form-encoded client identifiers and secrets through HTTP Basic authentication. They are not included in the form body. Providers requiring `client_secret_post`, private-key JWT, mTLS, or interactive grants need a different client.
+
+### Timeouts, retries, and destination safety
+
+`timeout_seconds`, defaulting to three seconds, is the token-acquisition budget, including waiting, secret lookup, token requests, and retry backoff. Configure the Lambda timeout to leave time for token acquisition, the downstream request, and your error handling.
+
+Transport failures, HTTP 429, and HTTP 5xx responses allow at most two retries within the acquisition budget. Backoff starts at 100 milliseconds, then 200 milliseconds. Other HTTP failures, malformed token responses, and secret-loader failures are not retried.
+
+`request()` uses a separate `timeout`, defaulting to five seconds, for connecting to the downstream API and buffering its response. It returns an urllib3 HTTP response with `.status`, `.headers`, `.data`, and `.json()`. Non-success HTTP responses are returned for your application to interpret.
+
+The remaining budget is enforced while reading response headers and bodies, including chunked response framing.
+This is not a universal wall-clock limit: synchronous DNS resolution, application-provided secret loaders, and upload producers cannot be interrupted. Their elapsed time still consumes the budget. Configure their timeouts separately where supported, and leave room in the Lambda invocation timeout.
+
+The helper requires HTTPS, rejects an existing Authorization header, and never follows redirects or automatically retries downstream requests. It forwards only `body`, `fields`, `json`, `encode_multipart`, and `multipart_boundary` options to urllib3. Use `auth_headers()` with your own client for streaming responses or other transport options.
+
+Header names must use HTTP token syntax: letters, digits, and the permitted token punctuation. Empty names, whitespace (including trailing spaces or tabs), and delimiters such as colons are rejected before token acquisition. Authorization is rejected regardless of casing.
+
+!!! warning "Use trusted destination URLs"
+    `request()` does not derive or restrict destinations from the configured audience or resource. Supply trusted URLs from application configuration; never pass a caller-controlled destination. A token intended for one API must not be sent to another.
+
+### Errors and diagnostics
+
+OAuth errors inherit from the common `AuthError` in `auth_alpha.exceptions`. Existing JWT exception imports continue to work.
+
+| Exception | Reason | Retryable |
+| --------- | ------ | --------- |
+| `TokenExchangeError` | `token_exchange_failed` | True for transient endpoint failures or acquisition timeouts; otherwise false |
+| `DownstreamRequestError` | `downstream_request_failed` | False: the server may already have performed the operation |
+
+Use the fixed `reason.value` and `retryable` fields for logs and metrics:
+
+```python title="diagnostics.py"
+--8<-- "examples/auth_alpha/oauth2/src/diagnostics.py"
+```
+
+The utility performs no automatic logging. It removes provider exception chains before exposing an auth error. Never log client secrets, access tokens, Authorization headers, or full provider responses.
+
+### Calling downstream APIs from an MCP tool
+
+An MCP server can use the same client after authorizing the incoming caller. Obtain a separate token for the downstream API instead of forwarding the caller's bearer token. In an async tool, offload this synchronous client to a worker thread:
+
+```python
+import asyncio
+from urllib.parse import quote
+
+from mcp.server.auth.middleware.auth_context import get_access_token
+
+# inventory_api is the configured OAuth2Client from client_credentials.py.
+async def check_stock(sku: str) -> dict:
+    caller = get_access_token()
+    if caller is None or "inventory:read" not in caller.scopes:
+        raise PermissionError("Inventory read permission is required")
+    response = await asyncio.to_thread(
+        inventory_api.request,
+        "GET",
+        f"{INVENTORY_URL}/stock/{quote(sku, safe='')}",
+        timeout=5,
+    )
+    if response.status != 200:
+        raise RuntimeError("Inventory lookup failed")
+    return response.json()
+```
+
+The MCP SDK owns transport authentication and protocol error responses; adapt the permission error to your SDK's handling. Cancelling the awaiting task does not stop an in-progress worker thread, so network timeouts still apply. No MCP dependency is added to Powertools.
+
+## Testing your code
+
+Mock the client operation when testing application behavior, and test your provider configuration separately:
+
+```python title="test_client_credentials.py"
+--8<-- "examples/auth_alpha/oauth2/tests/test_client_credentials.py"
+```
diff --git a/examples/auth_alpha/oauth2/src/client_credentials.py b/examples/auth_alpha/oauth2/src/client_credentials.py
new file mode 100644
index 00000000000..866a2820436
--- /dev/null
+++ b/examples/auth_alpha/oauth2/src/client_credentials.py
@@ -0,0 +1,32 @@
+import os
+from urllib.parse import quote
+
+from aws_lambda_powertools.utilities import parameters
+from aws_lambda_powertools.utilities.auth_alpha import OAuth2Client
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+def load_secret() -> str:
+    secret = parameters.get_secret(os.environ["CLIENT_SECRET_NAME"], max_age=300)
+    if not isinstance(secret, str):
+        raise ValueError("Expected a string client secret")
+    return secret
+
+
+INVENTORY_URL = os.environ["INVENTORY_URL"]
+
+inventory_api = OAuth2Client(
+    token_url=os.environ["TOKEN_URL"],
+    client_id=os.environ["CLIENT_ID"],
+    client_secret=load_secret,
+    scopes=["inventory:read"],
+    audience=INVENTORY_URL,
+)
+
+
+def lambda_handler(event: dict, context: LambdaContext):
+    sku = quote(event["sku"], safe="")
+    response = inventory_api.request("GET", f"{INVENTORY_URL}/stock/{sku}", timeout=5)
+    if response.status != 200:
+        raise RuntimeError("Inventory lookup failed")
+    return response.json()
diff --git a/examples/auth_alpha/oauth2/src/diagnostics.py b/examples/auth_alpha/oauth2/src/diagnostics.py
new file mode 100644
index 00000000000..345d4385715
--- /dev/null
+++ b/examples/auth_alpha/oauth2/src/diagnostics.py
@@ -0,0 +1,29 @@
+import os
+from urllib.parse import quote
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.auth_alpha import OAuth2Client
+from aws_lambda_powertools.utilities.auth_alpha.oauth2.exceptions import DownstreamRequestError, TokenExchangeError
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+logger = Logger()
+INVENTORY_URL = os.environ["INVENTORY_URL"]
+inventory_api = OAuth2Client(
+    token_url=os.environ["TOKEN_URL"],
+    client_id=os.environ["CLIENT_ID"],
+    client_secret=lambda: os.environ["CLIENT_SECRET"],
+    scopes=["inventory:read"],
+    audience=INVENTORY_URL,
+)
+
+
+def lambda_handler(event: dict, context: LambdaContext):
+    sku = quote(event["sku"], safe="")
+    try:
+        response = inventory_api.request("GET", f"{INVENTORY_URL}/stock/{sku}")
+    except (TokenExchangeError, DownstreamRequestError) as error:
+        logger.warning("Inventory request unavailable", reason=error.reason.value, retryable=error.retryable)
+        return {"statusCode": 502, "body": "Inventory request unavailable"}
+    if response.status != 200:
+        return {"statusCode": 502, "body": "Inventory request unavailable"}
+    return response.json()
diff --git a/examples/auth_alpha/oauth2/src/headers.py b/examples/auth_alpha/oauth2/src/headers.py
new file mode 100644
index 00000000000..4108797aca8
--- /dev/null
+++ b/examples/auth_alpha/oauth2/src/headers.py
@@ -0,0 +1,42 @@
+import os
+from urllib.parse import quote, urlsplit
+
+import urllib3
+
+from aws_lambda_powertools.utilities.auth_alpha import OAuth2Client
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+INVENTORY_URL = os.environ["INVENTORY_URL"]
+inventory_url = urlsplit(INVENTORY_URL)
+if (
+    inventory_url.scheme != "https"
+    or not inventory_url.hostname
+    or inventory_url.username is not None
+    or inventory_url.password is not None
+    or "#" in INVENTORY_URL
+):
+    raise ValueError("INVENTORY_URL must be an HTTPS URL without user information or a fragment")
+
+inventory_api = OAuth2Client(
+    token_url=os.environ["TOKEN_URL"],
+    client_id=os.environ["CLIENT_ID"],
+    client_secret=lambda: os.environ["CLIENT_SECRET"],
+    scopes=["inventory:read"],
+    resource=INVENTORY_URL,
+)
+http = urllib3.PoolManager()
+
+
+def lambda_handler(event: dict, context: LambdaContext):
+    sku = quote(event["sku"], safe="")
+    response = http.request(
+        "GET",
+        f"{INVENTORY_URL}/stock/{sku}",
+        headers=inventory_api.auth_headers(),
+        timeout=urllib3.Timeout(total=5),
+        redirect=False,
+        retries=False,
+    )
+    if response.status != 200:
+        raise RuntimeError("Inventory lookup failed")
+    return response.json()
diff --git a/examples/auth_alpha/oauth2/tests/test_client_credentials.py b/examples/auth_alpha/oauth2/tests/test_client_credentials.py
new file mode 100644
index 00000000000..459e677abb5
--- /dev/null
+++ b/examples/auth_alpha/oauth2/tests/test_client_credentials.py
@@ -0,0 +1,19 @@
+import urllib3
+
+
+def test_inventory_lookup(monkeypatch):
+    monkeypatch.setenv("TOKEN_URL", "https://idp.example.com/token")
+    monkeypatch.setenv("CLIENT_ID", "orders")
+    monkeypatch.setenv("CLIENT_SECRET_NAME", "orders/oauth-secret")
+    monkeypatch.setenv("INVENTORY_URL", "https://inventory.example.com")
+
+    from client_credentials import inventory_api, lambda_handler
+
+    def request(method, url, *, timeout):
+        assert method == "GET"
+        assert url == "https://inventory.example.com/stock/item%2F123"
+        assert timeout == 5
+        return urllib3.HTTPResponse(body=b'{"stock":12}', status=200)
+
+    monkeypatch.setattr(inventory_api, "request", request)
+    assert lambda_handler({"sku": "item/123"}, {}) == {"stock": 12}
diff --git a/examples/auth_alpha/oauth2/tests/test_headers.py b/examples/auth_alpha/oauth2/tests/test_headers.py
new file mode 100644
index 00000000000..64811c4df8e
--- /dev/null
+++ b/examples/auth_alpha/oauth2/tests/test_headers.py
@@ -0,0 +1,70 @@
+import runpy
+from pathlib import Path
+
+import pytest
+import urllib3
+
+from aws_lambda_powertools.utilities.auth_alpha import OAuth2Client
+
+EXAMPLE = Path(__file__).parents[1] / "src" / "headers.py"
+
+
+@pytest.fixture
+def example_environment(monkeypatch):
+    monkeypatch.setenv("TOKEN_URL", "https://idp.example.com/token")
+    monkeypatch.setenv("CLIENT_ID", "orders")
+    monkeypatch.setenv("CLIENT_SECRET", "test-secret")
+
+
+@pytest.mark.parametrize(
+    "destination",
+    [
+        "http://inventory.example.com",
+        "//inventory.example.com",
+        "https:///inventory",
+        "https://user:password@inventory.example.com",
+        "https://inventory.example.com/#fragment",
+    ],
+)
+def test_unsafe_destinations_are_rejected_before_obtaining_headers(example_environment, monkeypatch, destination):
+    monkeypatch.setenv("INVENTORY_URL", destination)
+    acquired = []
+    requests = []
+
+    def auth_headers(self):
+        acquired.append(True)
+        return {"Authorization": "Bearer test-token"}
+
+    def request(self, method, url, **options):
+        requests.append(url)
+        return urllib3.HTTPResponse(body=b'{"stock":12}', status=200)
+
+    monkeypatch.setattr(OAuth2Client, "auth_headers", auth_headers)
+    monkeypatch.setattr(urllib3.PoolManager, "request", request)
+    example_path = str(EXAMPLE)
+
+    with pytest.raises(ValueError, match="INVENTORY_URL"):
+        runpy.run_path(example_path)
+
+    assert acquired == []
+    assert requests == []
+
+
+def test_https_destination_receives_the_token_with_safe_transport_options(example_environment, monkeypatch):
+    monkeypatch.setenv("INVENTORY_URL", "https://inventory.example.com")
+    monkeypatch.setattr(OAuth2Client, "auth_headers", lambda self: {"Authorization": "Bearer test-token"})
+    requests = []
+
+    def request(self, method, url, **options):
+        requests.append((method, url))
+        assert options["headers"] == {"Authorization": "Bearer test-token"}
+        assert options["timeout"].total == 5
+        assert options["redirect"] is False
+        assert options["retries"] is False
+        return urllib3.HTTPResponse(body=b'{"stock":12}', status=200)
+
+    monkeypatch.setattr(urllib3.PoolManager, "request", request)
+    example = runpy.run_path(str(EXAMPLE))
+
+    assert example["lambda_handler"]({"sku": "item/123"}, {}) == {"stock": 12}
+    assert requests == [("GET", "https://inventory.example.com/stock/item%2F123")]
diff --git a/mkdocs.yml b/mkdocs.yml
index b9f02a668b8..f240b60d2ca 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -26,7 +26,9 @@ nav:
               - core/event_handler/appsync_events.md
               - core/event_handler/bedrock_agents.md
           - utilities/parameters.md
-          - Auth (alpha): utilities/auth.md
+          - Auth (alpha):
+              - JWT verification: utilities/auth.md
+              - OAuth2 client: utilities/oauth2.md
           - utilities/batch.md
           - utilities/kafka.md
           - utilities/typing.md
diff --git a/noxfile.py b/noxfile.py
index 4414848d785..df42592d866 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -232,6 +232,16 @@ def test_with_auth_required_packages(session: nox.Session):
     """Verify the Auth utility using only its declared optional dependencies."""
     build_and_run_test(
         session,
-        folders=[f"{PREFIX_TESTS_FUNCTIONAL}/auth_alpha/"],
+        folders=[f"{PREFIX_TESTS_FUNCTIONAL}/auth_alpha/jwt/"],
         extras="jwt",
     )
+
+
+@nox.session()
+def test_with_oauth2_required_packages(session: nox.Session):
+    """Verify OAuth token acquisition without JWT or cryptography dependencies."""
+    build_and_run_test(
+        session,
+        folders=[f"{PREFIX_TESTS_FUNCTIONAL}/auth_alpha/oauth2/"],
+        extras="oauth2",
+    )
diff --git a/poetry.lock b/poetry.lock
index 357219db2ca..9324e72b83d 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -24,6 +24,7 @@ files = [
     {file = "anyio-4.14.2-py3-none-any.whl", hash = "sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494"},
     {file = "anyio-4.14.2.tar.gz", hash = "sha256:cfa139f3ed1a23ee8f88a145ddb5ac7605b8bbfd8592baacd7ce3d8bb4313c7f"},
 ]
+markers = {main = "extra == \"valkey\""}
 
 [package.dependencies]
 exceptiongroup = {version = ">=1.0.2", markers = "python_version < \"3.11\""}
@@ -140,7 +141,7 @@ files = [
     {file = "attrs-25.4.0-py3-none-any.whl", hash = "sha256:adcf7e2a1fb3b36ac48d97835bb6d8ade15b8dcce26aba8bf1d14847b57a3373"},
     {file = "attrs-25.4.0.tar.gz", hash = "sha256:16d5969b87f0859ef33a48b35d55ac1be6e42ae49d5e853b597db70c35c57e11"},
 ]
-markers = {main = "extra == \"all\" or extra == \"datamasking\" or extra == \"valkey\""}
+markers = {main = "extra == \"valkey\" or extra == \"all\" or extra == \"datamasking\""}
 
 [[package]]
 name = "avro"
@@ -1203,7 +1204,7 @@ files = [
     {file = "cffi-2.0.0-cp39-cp39-win_amd64.whl", hash = "sha256:b882b3df248017dba09d6b16defe9b5c407fe32fc7c65a9c69798e6175601be9"},
     {file = "cffi-2.0.0.tar.gz", hash = "sha256:44d1b5909021139fe36001ae048dbdde8214afa20200eda0f64c068cac5d5529"},
 ]
-markers = {main = "(extra == \"all\" or extra == \"datamasking\" or extra == \"auth\") and platform_python_implementation != \"PyPy\" or extra == \"valkey\"", dev = "platform_python_implementation != \"PyPy\""}
+markers = {main = "(extra == \"all\" or extra == \"datamasking\" or extra == \"jwt\") and platform_python_implementation != \"PyPy\" or extra == \"valkey\"", dev = "platform_python_implementation != \"PyPy\""}
 
 [package.dependencies]
 pycparser = {version = "*", markers = "implementation_name != \"PyPy\""}
@@ -1613,7 +1614,7 @@ files = [
     {file = "cryptography-50.0.1-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:693c99b49bd37d0d096e4334c10232c77248c415b98d35236094cdf96d57258b"},
     {file = "cryptography-50.0.1.tar.gz", hash = "sha256:5dd9bda1c12b4162f6ff568eeb5e0ff956c28d14406e875cfe8a63a2d414ff20"},
 ]
-markers = {main = "extra == \"all\" or extra == \"datamasking\" or extra == \"auth\""}
+markers = {main = "extra == \"all\" or extra == \"datamasking\" or extra == \"jwt\""}
 
 [package.dependencies]
 cffi = {version = ">=2.0.0", markers = "platform_python_implementation != \"PyPy\""}
@@ -1882,11 +1883,11 @@ description = "Backport of PEP 654 (exception groups)"
 optional = false
 python-versions = ">=3.7"
 groups = ["main", "dev"]
-markers = "python_version == \"3.10\""
 files = [
     {file = "exceptiongroup-1.3.1-py3-none-any.whl", hash = "sha256:a7a39a3bd276781e98394987d3a5701d0c4edffb633bb7a5144577f82c773598"},
     {file = "exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219"},
 ]
+markers = {main = "extra == \"valkey\" and python_version == \"3.10\"", dev = "python_version == \"3.10\""}
 
 [package.dependencies]
 typing-extensions = {version = ">=4.6.0", markers = "python_version < \"3.13\""}
@@ -2107,6 +2108,7 @@ files = [
     {file = "idna-3.15-py3-none-any.whl", hash = "sha256:048adeaf8c2d788c40fee287673ccaa74c24ffd8dcf09ffa555a2fbb59f10ac8"},
     {file = "idna-3.15.tar.gz", hash = "sha256:ca962446ea538f7092a95e057da437618e886f4d349216d2b1e294abfdb65fdc"},
 ]
+markers = {main = "extra == \"valkey\" or extra == \"datadog\""}
 
 [package.extras]
 all = ["mypy (>=1.11.2)", "pytest (>=8.3.2)", "ruff (>=0.6.2)"]
@@ -3534,7 +3536,7 @@ files = [
     {file = "pycparser-3.0-py3-none-any.whl", hash = "sha256:b727414169a36b7d524c1c3e31839a521725078d7b2ff038656844266160a992"},
     {file = "pycparser-3.0.tar.gz", hash = "sha256:600f49d217304a5902ac3c37e1281c9fe94e4d0489de643a9504c5cdfdfc6b29"},
 ]
-markers = {main = "((extra == \"all\" or extra == \"datamasking\" or extra == \"auth\") and platform_python_implementation != \"PyPy\" or extra == \"valkey\") and implementation_name != \"PyPy\"", dev = "platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\""}
+markers = {main = "((extra == \"all\" or extra == \"datamasking\" or extra == \"jwt\") and platform_python_implementation != \"PyPy\" or extra == \"valkey\") and implementation_name != \"PyPy\"", dev = "platform_python_implementation != \"PyPy\" and implementation_name != \"PyPy\""}
 
 [[package]]
 name = "pydantic"
@@ -3740,7 +3742,7 @@ description = "JSON Web Token implementation in Python"
 optional = true
 python-versions = ">=3.9"
 groups = ["main"]
-markers = "extra == \"auth\" or extra == \"all\""
+markers = "extra == \"jwt\" or extra == \"all\""
 files = [
     {file = "pyjwt-2.14.0-py3-none-any.whl", hash = "sha256:ad0cef71c756a56e74863c2919cf0985f72decbcfcb550ee2f422e7c62b5eedc"},
     {file = "pyjwt-2.14.0.tar.gz", hash = "sha256:77283c83fb56ecf566a886c757a714bc83668e38156de2cce8263302f42e0b86"},
@@ -5123,7 +5125,7 @@ files = [
     {file = "urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3"},
     {file = "urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63"},
 ]
-markers = {main = "extra == \"tracer\" or extra == \"all\" or extra == \"aws-sdk\" or extra == \"datamasking\" or extra == \"datadog\" or extra == \"auth\""}
+markers = {main = "extra == \"tracer\" or extra == \"all\" or extra == \"aws-sdk\" or extra == \"datamasking\" or extra == \"datadog\" or extra == \"jwt\" or extra == \"oauth2\""}
 
 [package.extras]
 brotli = ["brotli (>=1.2.0) ; platform_python_implementation == \"CPython\"", "brotlicffi (>=1.2.0.0) ; platform_python_implementation != \"CPython\""]
@@ -5372,12 +5374,13 @@ type = ["pytest-mypy"]
 
 [extras]
 all = ["aws-encryption-sdk", "aws-xray-sdk", "cryptography", "fastjsonschema", "jsonpath-ng", "pydantic", "pydantic-settings", "pyjwt", "urllib3"]
-jwt = ["cryptography", "pyjwt", "urllib3"]
 aws-sdk = ["boto3"]
 datadog = ["datadog-lambda"]
 datamasking = ["aws-encryption-sdk", "jsonpath-ng"]
+jwt = ["cryptography", "pyjwt", "urllib3"]
 kafka-consumer-avro = ["avro"]
 kafka-consumer-protobuf = ["protobuf"]
+oauth2 = ["urllib3"]
 parser = ["pydantic"]
 redis = ["redis"]
 tracer = ["aws-xray-sdk"]
@@ -5387,4 +5390,4 @@ valkey = ["valkey-glide"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.10,<4.0.0"
-content-hash = "906b66b0cba452e7bed8de1a7557359fd454b60525c1873964c3ca25a299887b"
+content-hash = "2d3adcc97137c392a7631148323a1be812b5beaba36b266b7b753beb382d8358"
diff --git a/pyproject.toml b/pyproject.toml
index e6a0d08ddd1..e35f3b2f7bf 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -68,6 +68,7 @@ tracer = ["aws-xray-sdk"]
 redis = ["redis"]
 valkey = ["valkey-glide"]
 jwt = ["pyjwt", "cryptography", "urllib3"]
+oauth2 = ["urllib3"]
 all = [
   "pydantic",
   "pydantic-settings",
diff --git a/tests/functional/auth_alpha/conftest.py b/tests/functional/auth_alpha/conftest.py
new file mode 100644
index 00000000000..f6873ffbc10
--- /dev/null
+++ b/tests/functional/auth_alpha/conftest.py
@@ -0,0 +1,57 @@
+import io
+import json
+import time
+from collections import deque
+
+import pytest
+import urllib3
+
+
+class FakeHTTP:
+    """In-memory authentication endpoints at the HTTP transport boundary."""
+
+    def __init__(self):
+        self.responses = {}
+        self.requests = []
+
+    def serve(self, url, body, *, status=200, method="GET"):
+        self.responses[(method, url)] = deque([(status, body)])
+
+    def request(self, method, url, **kwargs):
+        self.requests.append((method, url, kwargs))
+        responses = self.responses[(method, url)]
+        status, body = responses[0] if len(responses) == 1 else responses.popleft()
+        if callable(body):
+            body = body()
+        if isinstance(body, Exception):
+            raise body
+        payload = body if isinstance(body, bytes) else json.dumps(body).encode()
+        return urllib3.HTTPResponse(
+            body=io.BytesIO(payload),
+            headers={"content-type": "application/json"},
+            status=status,
+            preload_content=False,
+        )
+
+
+@pytest.fixture
+def http(monkeypatch):
+    transport = FakeHTTP()
+    monkeypatch.setattr(urllib3, "PoolManager", lambda **kwargs: transport)
+    return transport
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    class Clock:
+        now = 1000.0
+
+        def __call__(self):
+            return self.now
+
+        def advance(self, seconds):
+            self.now += seconds
+
+    clock = Clock()
+    monkeypatch.setattr(time, "monotonic", clock)
+    return clock
diff --git a/tests/functional/auth_alpha/jwt/conftest.py b/tests/functional/auth_alpha/jwt/conftest.py
index 69c67d98155..e5259dab923 100644
--- a/tests/functional/auth_alpha/jwt/conftest.py
+++ b/tests/functional/auth_alpha/jwt/conftest.py
@@ -1,12 +1,8 @@
-import io
-import json
 import time
 import weakref
-from collections import deque
 
 import jwt
 import pytest
-import urllib3
 from cryptography.hazmat.primitives.asymmetric import rsa
 
 from aws_lambda_powertools.utilities.auth_alpha.jwt._internal import jwks as jwks_module
@@ -47,54 +43,7 @@ def issue(payload=None, *, key=None, kid="key-1", algorithm="RS256", headers=Non
     return issue
 
 
-class FakeHTTP:
-    """In-memory JWKS endpoints at the HTTP transport boundary."""
-
-    def __init__(self):
-        self.responses = {}
-        self.requests = []
-
-    def serve(self, url, body, *, status=200, method="GET"):
-        self.responses[(method, url)] = deque([(status, body)])
-
-    def request(self, method, url, **kwargs):
-        self.requests.append((method, url, kwargs))
-        responses = self.responses[(method, url)]
-        status, body = responses[0] if len(responses) == 1 else responses.popleft()
-        if callable(body):
-            body = body()
-        if isinstance(body, Exception):
-            raise body
-        payload = body if isinstance(body, bytes) else json.dumps(body).encode()
-        return urllib3.HTTPResponse(
-            body=io.BytesIO(payload),
-            headers={"content-type": "application/json"},
-            status=status,
-            preload_content=False,
-        )
-
-
-@pytest.fixture
-def http(monkeypatch):
-    # Each fake provider belongs to one test. Error tracebacks can keep a
-    # previous verifier alive; retain sharing only within the current test.
+@pytest.fixture(autouse=True)
+def isolated_jwks_caches(monkeypatch):
+    # Each test's fake provider owns its cache, independent of retained tracebacks.
     monkeypatch.setattr(jwks_module, "_caches", weakref.WeakValueDictionary())
-    transport = FakeHTTP()
-    monkeypatch.setattr(urllib3, "PoolManager", lambda **kwargs: transport)
-    return transport
-
-
-@pytest.fixture
-def clock(monkeypatch):
-    class Clock:
-        now = 1000.0
-
-        def __call__(self):
-            return self.now
-
-        def advance(self, seconds):
-            self.now += seconds
-
-    clock = Clock()
-    monkeypatch.setattr(time, "monotonic", clock)
-    return clock
diff --git a/tests/functional/auth_alpha/oauth2/__init__.py b/tests/functional/auth_alpha/oauth2/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/tests/functional/auth_alpha/oauth2/test_client.py b/tests/functional/auth_alpha/oauth2/test_client.py
new file mode 100644
index 00000000000..91121866c6d
--- /dev/null
+++ b/tests/functional/auth_alpha/oauth2/test_client.py
@@ -0,0 +1,716 @@
+import base64
+import threading
+import time
+import traceback
+from collections import deque
+from concurrent.futures import ThreadPoolExecutor
+from urllib.parse import parse_qs
+
+import pytest
+
+from aws_lambda_powertools.utilities.auth_alpha import OAuth2Client
+from aws_lambda_powertools.utilities.auth_alpha.oauth2.exceptions import TokenExchangeError
+
+TOKEN_URL = "https://idp.example.com/oauth/token"
+
+
+def client(**options):
+    config = {
+        "token_url": TOKEN_URL,
+        "client_id": "orders-client",
+        "client_secret": "test-client-secret",
+        "scopes": ["orders:read"],
+    }
+    return OAuth2Client(**{**config, **options})
+
+
+def test_client_credentials_exchange_selects_resource_and_caches_token(http):
+    http.serve(
+        TOKEN_URL,
+        {"access_token": "opaque-access-token", "token_type": "Bearer", "expires_in": 3600},
+        method="POST",
+    )
+    subject = client(audience="https://api.example.com")
+
+    assert subject.auth_headers() == {"Authorization": "Bearer opaque-access-token"}
+    assert subject.auth_headers() == {"Authorization": "Bearer opaque-access-token"}
+    assert len(http.requests) == 1
+    method, url, request = http.requests[0]
+    assert method == "POST"
+    assert url == TOKEN_URL
+    assert parse_qs(request["body"].decode()) == {
+        "grant_type": ["client_credentials"],
+        "scope": ["orders:read"],
+        "audience": ["https://api.example.com"],
+    }
+    assert request["headers"]["Content-Type"] == "application/x-www-form-urlencoded"
+
+
+def test_basic_auth_encodes_each_credential_before_base64(http):
+    http.serve(TOKEN_URL, {"access_token": "token", "token_type": "bearer", "expires_in": 3600}, method="POST")
+    subject = client(client_id="client:id", client_secret="secret:value with space")
+    subject.auth_headers()
+    request = http.requests[0][2]
+    encoded = request["headers"]["Authorization"].removeprefix("Basic ")
+
+    assert base64.b64decode(encoded).decode() == "client%3Aid:secret%3Avalue+with+space"
+    assert "client_secret" not in parse_qs(request["body"].decode())
+
+
+def test_token_is_reacquired_before_expiry_using_the_current_secret(http, clock):
+    secret = ["initial-secret"]
+    observed = []
+
+    def load_secret():
+        observed.append(secret[0])
+        return secret[0]
+
+    http.serve(TOKEN_URL, {"access_token": "first", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    subject = client(client_secret=load_secret)
+    assert subject.auth_headers()["Authorization"] == "Bearer first"
+    clock.advance(69)
+    assert subject.auth_headers()["Authorization"] == "Bearer first"
+    assert observed == ["initial-secret"]
+    secret[0] = "rotated-secret"
+    http.serve(TOKEN_URL, {"access_token": "second", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    clock.advance(1)
+
+    assert subject.auth_headers()["Authorization"] == "Bearer second"
+    assert observed == ["initial-secret", "rotated-secret"]
+
+
+@pytest.mark.parametrize("lifetime", [1, 30, None])
+def test_short_lived_tokens_and_tokens_without_lifetimes_are_not_cached(http, lifetime):
+    payload = {"access_token": "first", "token_type": "Bearer"}
+    if lifetime is not None:
+        payload["expires_in"] = lifetime
+    http.serve(TOKEN_URL, payload, method="POST")
+    subject = client()
+    assert subject.auth_headers()["Authorization"] == "Bearer first"
+    http.serve(TOKEN_URL, {**payload, "access_token": "second"}, method="POST")
+
+    assert subject.auth_headers()["Authorization"] == "Bearer second"
+    assert len(http.requests) == 2
+
+
+@pytest.mark.parametrize(
+    "override",
+    [
+        {"access_token": ""},
+        {"access_token": None},
+        {"access_token": "token\r\ninjected"},
+        {"token_type": "DPoP"},
+        {"token_type": None},
+        {"expires_in": "3600"},
+        {"expires_in": 0},
+        {"expires_in": -1},
+        {"expires_in": True},
+        {"expires_in": None},
+        {"expires_in": float("inf")},
+    ],
+)
+def test_invalid_token_responses_are_rejected_without_retry(http, override):
+    payload = {"access_token": "token", "token_type": "Bearer", "expires_in": 3600, **override}
+    http.serve(TOKEN_URL, payload, method="POST")
+
+    subject = client()
+    with pytest.raises(TokenExchangeError):
+        subject.auth_headers()
+    assert len(http.requests) == 1
+
+
+def test_resources_have_separate_token_caches_and_request_parameters(http):
+    http.responses[("POST", TOKEN_URL)] = deque(
+        [
+            (200, {"access_token": "orders-token", "token_type": "Bearer", "expires_in": 3600}),
+            (200, {"access_token": "inventory-token", "token_type": "Bearer", "expires_in": 3600}),
+        ],
+    )
+    orders = client(audience="https://orders.example.com")
+    inventory = client(resource="https://inventory.example.com")
+
+    assert orders.auth_headers()["Authorization"] == "Bearer orders-token"
+    assert inventory.auth_headers()["Authorization"] == "Bearer inventory-token"
+    assert orders.auth_headers()["Authorization"] == "Bearer orders-token"
+    assert len(http.requests) == 2
+    assert parse_qs(http.requests[0][2]["body"].decode())["audience"] == ["https://orders.example.com"]
+    assert parse_qs(http.requests[1][2]["body"].decode())["resource"] == ["https://inventory.example.com"]
+
+
+@pytest.mark.parametrize("status", [400, 401, 403])
+def test_permanent_exchange_errors_are_not_retried(http, status):
+    http.serve(
+        TOKEN_URL,
+        {"error": "invalid_client", "error_description": "private details"},
+        status=status,
+        method="POST",
+    )
+
+    subject = client()
+    with pytest.raises(TokenExchangeError):
+        subject.auth_headers()
+    assert len(http.requests) == 1
+
+
+def test_transient_exchange_errors_have_at_most_two_retries(http, clock, monkeypatch):
+    http.serve(TOKEN_URL, b"temporarily unavailable", status=503, method="POST")
+    monkeypatch.setattr(time, "sleep", clock.advance)
+    secrets = []
+
+    def load_secret():
+        secrets.append("secret")
+        return secrets[-1]
+
+    subject = client(client_secret=load_secret)
+    with pytest.raises(TokenExchangeError):
+        subject.auth_headers()
+    assert len(http.requests) == 3
+    assert len(secrets) == 3
+
+
+def test_transient_exchange_can_recover_within_the_same_budget(http, clock, monkeypatch):
+    http.responses[("POST", TOKEN_URL)] = deque(
+        [(429, {}), (200, {"access_token": "recovered", "token_type": "Bearer", "expires_in": 100})],
+    )
+    monkeypatch.setattr(time, "sleep", clock.advance)
+
+    assert client().auth_headers() == {"Authorization": "Bearer recovered"}
+    assert len(http.requests) == 2
+
+
+def test_exchange_cannot_accept_a_response_after_its_deadline(http, clock):
+    def slow_endpoint():
+        clock.advance(4)
+        return {"access_token": "too-late", "token_type": "Bearer", "expires_in": 3600}
+
+    http.serve(TOKEN_URL, slow_endpoint, method="POST")
+    subject = client(timeout_seconds=3)
+    with pytest.raises(TokenExchangeError):
+        subject.auth_headers()
+    assert len(http.requests) == 1
+
+
+def test_exchange_cannot_return_a_token_that_expired_during_the_request(http, clock):
+    def slow_endpoint():
+        clock.advance(2)
+        return {"access_token": "already-expired", "token_type": "Bearer", "expires_in": 1}
+
+    http.serve(TOKEN_URL, slow_endpoint, method="POST")
+    subject = client()
+    with pytest.raises(TokenExchangeError):
+        subject.auth_headers()
+
+
+def test_concurrent_requests_share_one_token_exchange(http):
+    entered = threading.Event()
+    release = threading.Event()
+
+    def exchange():
+        entered.set()
+        assert release.wait(2)
+        return {"access_token": "shared-token", "token_type": "Bearer", "expires_in": 100}
+
+    http.serve(TOKEN_URL, exchange, method="POST")
+    subject = client()
+    with ThreadPoolExecutor(max_workers=8) as executor:
+        results = [executor.submit(subject.auth_headers) for _ in range(8)]
+        assert entered.wait(2)
+        release.set()
+        assert all(result.result(timeout=2) == {"Authorization": "Bearer shared-token"} for result in results)
+    assert len(http.requests) == 1
+
+
+@pytest.mark.parametrize("status", [200, 401], ids=["success", "failure"])
+def test_concurrent_callers_share_reacquisition_at_the_refresh_boundary(http, clock, monkeypatch, status):
+    http.serve(TOKEN_URL, {"access_token": "old", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    subject = client()
+    assert subject.auth_headers() == {"Authorization": "Bearer old"}
+    clock.advance(69)
+    assert subject.auth_headers() == {"Authorization": "Bearer old"}
+    assert len(http.requests) == 1
+    clock.advance(1)  # The old token is still valid, with exactly 30 seconds remaining.
+
+    entered = threading.Event()
+    release = threading.Event()
+    joined = threading.Event()
+
+    def exchange():
+        entered.set()
+        assert release.wait(5)
+        if status == 200:
+            return {"access_token": "replacement", "token_type": "Bearer", "expires_in": 100}
+        return {"error": "invalid_client"}
+
+    http.serve(TOKEN_URL, exchange, status=status, method="POST")
+    with ThreadPoolExecutor(max_workers=3) as executor:
+        owner = executor.submit(subject.auth_headers)
+        waiters = []
+        try:
+            assert entered.wait(5)
+            flight = subject._flight
+            assert flight is not None
+            wait = flight.done.wait
+
+            def observe_wait(timeout):
+                joined.set()
+                return wait(timeout)
+
+            monkeypatch.setattr(flight.done, "wait", observe_wait)
+            for _ in range(2):
+                joined.clear()
+                waiters.append(executor.submit(subject.auth_headers))
+                assert joined.wait(5)
+            assert len(http.requests) == 2
+            assert not owner.done()
+            assert all(not waiter.done() for waiter in waiters)
+        finally:
+            release.set()
+
+        for result in (owner, *waiters):
+            if status == 200:
+                assert result.result(timeout=5) == {"Authorization": "Bearer replacement"}
+            else:
+                with pytest.raises(TokenExchangeError):
+                    result.result(timeout=5)
+    assert len(http.requests) == 2
+
+    if status == 200:
+        assert subject.auth_headers() == {"Authorization": "Bearer replacement"}
+        assert len(http.requests) == 2
+    else:
+        # A later caller must also exchange again instead of using the still-valid old token.
+        with pytest.raises(TokenExchangeError):
+            subject.auth_headers()
+        assert len(http.requests) == 3
+
+
+def test_secret_loader_errors_and_representations_are_redacted(http):
+    def load_secret():
+        raise RuntimeError("sensitive-loader-data")
+
+    subject = client(client_secret=load_secret)
+    with pytest.raises(TokenExchangeError) as error:
+        subject.auth_headers()
+    assert "sensitive-loader-data" not in "".join(traceback.format_exception(error.value))
+    assert repr(subject) == "<OAuth2Client>"
+
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        {"audience": "one", "resource": "two"},
+        {"audience": " "},
+        {"resource": ""},
+        {"resource": 42},
+        {"token_url": "http://idp.example.com/token"},
+        {"token_url": "https://user:secret@idp.example.com/token"},
+        {"client_id": ""},
+        {"client_secret": ""},
+        {"timeout_seconds": 0},
+        {"timeout_seconds": float("inf")},
+        {"scopes": ["scope\ninjection"]},
+    ],
+)
+def test_invalid_client_configuration_is_rejected(options):
+    with pytest.raises(ValueError):
+        client(**options)
+
+
+def test_request_attaches_resource_token_without_forwarding_client_credentials(http):
+    http.serve(TOKEN_URL, {"access_token": "resource-token", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    http.serve("https://api.example.com/orders", {"orders": [123]})
+    subject = client(audience="https://api.example.com")
+
+    response = subject.request(
+        "GET",
+        "https://api.example.com/orders",
+        headers={"Accept": "application/json"},
+        timeout=5,
+    )
+
+    assert response.json() == {"orders": [123]}
+    request = http.requests[-1][2]
+    assert request["headers"] == {"Accept": "application/json", "Authorization": "Bearer resource-token"}
+    assert request["redirect"] is False
+    assert request["retries"] is False
+    assert "test-client-secret" not in repr(subject)
+    assert "resource-token" not in repr(subject)
+
+
+@pytest.mark.parametrize(
+    "url,options",
+    [
+        ("http://api.example.com/orders", {}),
+        ("https://api.example.com/orders", {"headers": {"authorization": "other-token"}}),
+        ("https://api.example.com/orders", {"redirect": True}),
+        ("https://api.example.com/orders", {"retries": 3}),
+        ("https://api.example.com/orders", {"timeout": 0}),
+        ("https://api.example.com/orders", {"headers": [("Accept", "application/json")]}),
+    ],
+)
+def test_request_rejects_unsafe_overrides_before_acquiring_credentials(http, url, options):
+    subject = client()
+    with pytest.raises(ValueError):
+        subject.request("GET", url, **options)
+    assert http.requests == []
+
+
+def test_request_does_not_follow_redirects_or_retry_downstream_failures(http):
+    http.serve(TOKEN_URL, {"access_token": "token", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    http.serve("https://api.example.com/orders", {}, status=302)
+    subject = client()
+
+    assert subject.request("GET", "https://api.example.com/orders").status == 302
+    http.serve("https://api.example.com/orders", {}, status=503)
+    assert subject.request("GET", "https://api.example.com/orders").status == 503
+    assert len(http.requests) == 3
+
+
+@pytest.mark.parametrize("method", [None, "", "GET /", "GET\r\nInjected"])
+def test_invalid_http_methods_are_rejected_before_loading_credentials(http, method):
+    calls = []
+
+    def load_secret():
+        calls.append(True)
+        return "test-secret"
+
+    subject = client(client_secret=load_secret)
+    with pytest.raises(ValueError, match="HTTP method"):
+        subject.request(method, "https://api.example.com/orders")
+    assert calls == []
+    assert http.requests == []
+
+
+@pytest.mark.parametrize("secret", [None, "", 42])
+def test_invalid_secret_loader_results_are_rejected_before_sending_credentials(http, secret):
+    subject = client(client_secret=lambda: secret)
+    with pytest.raises(TokenExchangeError) as error:
+        subject.auth_headers()
+    assert error.value.__context__ is None
+    assert http.requests == []
+
+
+def test_retry_stops_when_the_backoff_exceeds_the_remaining_budget(http, clock, monkeypatch):
+    sleeps = []
+    monkeypatch.setattr(time, "sleep", sleeps.append)
+    http.serve(TOKEN_URL, {}, status=503, method="POST")
+
+    subject = client(timeout_seconds=0.05)
+    with pytest.raises(TokenExchangeError):
+        subject.auth_headers()
+    assert sleeps == []
+    assert len(http.requests) == 1
+
+
+def test_waiting_callers_share_a_failed_exchange_and_can_recover(http, monkeypatch):
+    entered = threading.Event()
+    release = threading.Event()
+    joined = threading.Event()
+
+    def exchange():
+        entered.set()
+        assert release.wait(5)
+        return {"error": "invalid_client"}
+
+    http.serve(TOKEN_URL, exchange, status=401, method="POST")
+    subject = client()
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        owner = executor.submit(subject.auth_headers)
+        try:
+            assert entered.wait(5)
+            flight = subject._flight
+            assert flight is not None
+            wait = flight.done.wait
+
+            def observe_wait(timeout):
+                joined.set()
+                return wait(timeout)
+
+            # Keep the real Event; observe it so the provider is released only
+            # after the second caller has joined the active exchange.
+            monkeypatch.setattr(flight.done, "wait", observe_wait)
+            waiter = executor.submit(subject.auth_headers)
+            assert joined.wait(5)
+        finally:
+            release.set()
+        for result in (owner, waiter):
+            with pytest.raises(TokenExchangeError) as error:
+                result.result(timeout=5)
+            assert error.value.__context__ is None
+    assert len(http.requests) == 1
+
+    http.serve(TOKEN_URL, {"access_token": "recovered", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    assert subject.auth_headers() == {"Authorization": "Bearer recovered"}
+    assert len(http.requests) == 2
+
+
+def test_waiting_callers_timeout_without_returning_the_late_token(http):
+    entered = threading.Event()
+    release = threading.Event()
+
+    def exchange():
+        entered.set()
+        assert release.wait(5)
+        return {"access_token": "too-late", "token_type": "Bearer", "expires_in": 100}
+
+    http.serve(TOKEN_URL, exchange, method="POST")
+    subject = client(timeout_seconds=0.1)
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        owner = executor.submit(subject.auth_headers)
+        try:
+            assert entered.wait(5)
+            with pytest.raises(TokenExchangeError):
+                subject.auth_headers()
+            assert not owner.done()
+            assert len(http.requests) == 1
+        finally:
+            release.set()
+        with pytest.raises(TokenExchangeError):
+            owner.result(timeout=5)
+
+    http.serve(TOKEN_URL, {"access_token": "recovered", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    assert subject.auth_headers() == {"Authorization": "Bearer recovered"}
+    assert len(http.requests) == 2
+
+
+def test_configuration_and_returned_headers_do_not_mutate_the_token_cache(http):
+    scopes = ["orders:read"]
+    subject = client(scopes=scopes)
+    scopes.append("orders:write")
+    http.serve(TOKEN_URL, {"access_token": "token", "token_type": "Bearer", "expires_in": 100}, method="POST")
+
+    headers = subject.auth_headers()
+    headers["Authorization"] = "Bearer replacement"
+    assert subject.auth_headers() == {"Authorization": "Bearer token"}
+    assert parse_qs(http.requests[0][2]["body"].decode())["scope"] == ["orders:read"]
+    assert len(http.requests) == 1
+
+
+def test_unselected_resource_and_scopes_are_not_added_to_the_exchange(http):
+    subject = client(scopes=[])
+    http.serve(TOKEN_URL, {"access_token": "token", "token_type": "Bearer", "expires_in": 100}, method="POST")
+
+    subject.auth_headers()
+    assert parse_qs(http.requests[0][2]["body"].decode()) == {"grant_type": ["client_credentials"]}
+
+
+@pytest.mark.parametrize("lifetime", [1, 30, None])
+def test_concurrent_callers_share_an_uncacheable_token(http, monkeypatch, lifetime):
+    entered = threading.Event()
+    release = threading.Event()
+    joined = threading.Event()
+    payload = {"access_token": "shared", "token_type": "Bearer"}
+    if lifetime is not None:
+        payload["expires_in"] = lifetime
+
+    def exchange():
+        entered.set()
+        assert release.wait(5)
+        return payload
+
+    subject = client()
+    http.serve(TOKEN_URL, exchange, method="POST")
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        owner = executor.submit(subject.auth_headers)
+        try:
+            assert entered.wait(5)
+            flight = subject._flight
+            assert flight is not None
+            wait = flight.done.wait
+
+            def observe_wait(timeout):
+                joined.set()
+                return wait(timeout)
+
+            monkeypatch.setattr(flight.done, "wait", observe_wait)
+            waiter = executor.submit(subject.auth_headers)
+            assert joined.wait(5)
+        finally:
+            release.set()
+        assert owner.result(timeout=5) == {"Authorization": "Bearer shared"}
+        assert waiter.result(timeout=5) == {"Authorization": "Bearer shared"}
+    assert len(http.requests) == 1
+    subject.auth_headers()
+    assert len(http.requests) == 2
+
+
+def test_invalid_unicode_from_a_secret_loader_is_sanitized(http):
+    subject = client(client_secret=lambda: "private-secret-\ud800")
+    with pytest.raises(TokenExchangeError) as error:
+        subject.auth_headers()
+    assert error.value.__context__ is None
+    assert "private-secret" not in "".join(traceback.format_exception(error.value))
+    assert http.requests == []
+
+
+@pytest.mark.parametrize("field", ["client_id", "audience", "resource"])
+def test_invalid_configuration_encoding_is_rejected(field):
+    options = {field: "\ud800"}
+    with pytest.raises(ValueError, match="UTF-8"):
+        client(**options)
+
+
+@pytest.mark.parametrize(
+    "resource",
+    [
+        "inventory",
+        "/inventory",
+        "//inventory.example.com",
+        "https://inventory.example.com/#fragment",
+        "https://inventory.example.com/#",
+        "urn:example:inventory#fragment",
+        " https://inventory.example.com",
+        "https://inventory.example.com/\nstock",
+        "https://inventory.example.com/\x00stock",
+        "https://inventory.example.com/stock item",
+        "https://inventory.example.com/%invalid",
+        "https://[invalid",
+        "1inventory:stock",
+        "urn%3Ainventory",
+        "urn:inventory%",
+        "urn:inventory%2",
+        "urn:inventory%2G",
+        "urn:inventory%G2",
+        "urn:inventory%%20",
+        "urn:inventorý",
+    ],
+)
+def test_resource_must_be_an_absolute_uri_without_a_fragment(resource):
+    with pytest.raises(ValueError, match="absolute URI without a fragment"):
+        client(resource=resource)
+
+
+@pytest.mark.parametrize(
+    "selection",
+    [
+        {"resource": "urn:example:inventory"},
+        {"resource": "https://inventory.example.com/stock?region=eu&category=%23parts"},
+        {"resource": "http://inventory.example.com"},
+        {"resource": "URN:example:inventory"},
+        {"resource": "inventory+v1.2-test:stock%2fitems%20eu?category=%23"},
+        {"resource": "https://[::1]/caf%C3%A9?encoded=%00%ff"},
+        {"audience": "inventory"},
+    ],
+)
+def test_valid_resource_identifiers_and_provider_audiences_are_preserved(http, selection):
+    http.serve(TOKEN_URL, {"access_token": "token", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    subject = client(**selection)
+
+    subject.auth_headers()
+
+    name, value = next(iter(selection.items()))
+    fields = parse_qs(http.requests[0][2]["body"].decode())
+    assert fields[name] == [value]
+    other = "audience" if name == "resource" else "resource"
+    assert other not in fields
+
+
+@pytest.mark.parametrize("suffix", ["%", "%4", "%4G", "#"])
+def test_long_resources_with_invalid_suffixes_are_rejected_before_loading_credentials(http, suffix):
+    secrets = []
+
+    def load_secret():
+        secrets.append("test-secret")
+        return secrets[-1]
+
+    resource = "urn:inventory:" + "a%41" * 25_000 + suffix
+    with pytest.raises(ValueError, match="absolute URI without a fragment"):
+        client(resource=resource, client_secret=load_secret)
+
+    assert secrets == []
+    assert http.requests == []
+
+
+def test_secret_lookup_does_not_expire_a_new_short_lived_token(http, clock):
+    def load_secret():
+        clock.advance(2)
+        return "test-secret"
+
+    http.serve(TOKEN_URL, {"access_token": "fresh", "token_type": "Bearer", "expires_in": 1}, method="POST")
+    subject = client(client_secret=load_secret, timeout_seconds=3)
+
+    assert subject.auth_headers() == {"Authorization": "Bearer fresh"}
+    assert subject.auth_headers() == {"Authorization": "Bearer fresh"}
+    assert len(http.requests) == 2
+
+
+def test_secret_lookup_does_not_move_the_cached_tokens_refresh_boundary(http, clock):
+    def load_secret():
+        clock.advance(2)
+        return "test-secret"
+
+    http.serve(TOKEN_URL, {"access_token": "first", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    subject = client(client_secret=load_secret)
+    assert subject.auth_headers() == {"Authorization": "Bearer first"}
+    clock.advance(69)
+    assert subject.auth_headers() == {"Authorization": "Bearer first"}
+    assert len(http.requests) == 1
+
+    http.serve(TOKEN_URL, {"access_token": "second", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    clock.advance(1)
+    assert subject.auth_headers() == {"Authorization": "Bearer second"}
+    assert len(http.requests) == 2
+
+
+def test_secret_lookup_still_consumes_the_acquisition_budget(http, clock):
+    def load_secret():
+        clock.advance(4)
+        return "test-secret"
+
+    subject = client(client_secret=load_secret, timeout_seconds=3)
+    with pytest.raises(TokenExchangeError) as error:
+        subject.auth_headers()
+
+    assert error.value.retryable
+    assert http.requests == []
+
+
+@pytest.mark.parametrize(
+    "name",
+    [
+        "Authorization",
+        "aUtHoRiZaTiOn",
+        "Authorization ",
+        "Authorization\t",
+        " Authorization",
+        "",
+        ":",
+        "X:Trace",
+        "X Trace",
+        "X\tTrace",
+        "X/Trace",
+        "X(Trace)",
+        "X\x00Trace",
+        "X\x7fTrace",
+        "X-Ünicode",
+        "X-Trace\r\nInjected",
+    ],
+)
+def test_invalid_header_names_are_rejected_before_token_acquisition(http, name):
+    secrets = []
+
+    def load_secret():
+        secrets.append("test-secret")
+        return secrets[-1]
+
+    http.serve(TOKEN_URL, {"access_token": "token", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    http.serve("https://api.example.com/orders", {"orders": []})
+    subject = client(client_secret=load_secret)
+
+    with pytest.raises(ValueError, match="Request headers"):
+        subject.request("GET", "https://api.example.com/orders", headers={name: "test-value"})
+
+    assert secrets == []
+    assert http.requests == []
+
+
+def test_http_token_punctuation_is_allowed_in_header_names(http):
+    http.serve(TOKEN_URL, {"access_token": "token", "token_type": "Bearer", "expires_in": 100}, method="POST")
+    http.serve("https://api.example.com/orders", {"orders": []})
+    subject = client()
+    headers = {"X-Trace!#$%&'*+.^_`|~09": "trace-id"}
+
+    response = subject.request("GET", "https://api.example.com/orders", headers=headers)
+
+    assert response.status == 200
+    assert http.requests[-1][2]["headers"] == {**headers, "Authorization": "Bearer token"}
diff --git a/tests/functional/auth_alpha/oauth2/test_errors.py b/tests/functional/auth_alpha/oauth2/test_errors.py
new file mode 100644
index 00000000000..cb60388acc8
--- /dev/null
+++ b/tests/functional/auth_alpha/oauth2/test_errors.py
@@ -0,0 +1,89 @@
+import io
+import json
+import traceback
+from uuid import uuid4
+
+import pytest
+import urllib3
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.auth_alpha import AuthFailureReason, OAuth2Client
+from aws_lambda_powertools.utilities.auth_alpha.exceptions import AuthError
+from aws_lambda_powertools.utilities.auth_alpha.oauth2.exceptions import DownstreamRequestError, TokenExchangeError
+
+TOKEN_URL = "https://idp.example.com/token"
+RESOURCE_URL = "https://api.example.com/orders"
+PRIVATE_DATA = "test-only-sensitive-provider-data"
+
+
+@pytest.mark.parametrize("operation", ["auth_headers", "request"])
+@pytest.mark.parametrize("failure", ["secret", "transport", "json", "expires_in", "downstream"])
+def test_errors_never_expose_credentials_or_active_exception_chains(http, operation, failure, monkeypatch):
+    monkeypatch.setattr("time.sleep", lambda seconds: None)
+
+    def load_secret():
+        if failure == "secret":
+            raise RuntimeError(PRIVATE_DATA)
+        return PRIVATE_DATA
+
+    subject = OAuth2Client(token_url=TOKEN_URL, client_id="orders", client_secret=load_secret)
+    payload = {"access_token": "test-token", "token_type": "Bearer", "expires_in": 600}
+    if failure == "transport":
+        payload = urllib3.exceptions.SSLError(PRIVATE_DATA)
+    elif failure == "json":
+        payload = PRIVATE_DATA.encode()
+    elif failure == "expires_in":
+        payload["expires_in"] = PRIVATE_DATA
+    elif operation == "auth_headers" and failure == "downstream":
+        # This operation has no downstream request, so exercise a bad token instead.
+        payload["access_token"] = PRIVATE_DATA + "\r\ninvalid"
+    http.serve(TOKEN_URL, payload, method="POST")
+    http.serve(RESOURCE_URL, urllib3.exceptions.SSLError(PRIVATE_DATA))
+    stream = io.StringIO()
+    logger = Logger(service=f"oauth-error-test-{uuid4()}", stream=stream)
+    expected = DownstreamRequestError if operation == "request" and failure == "downstream" else TokenExchangeError
+    invoke = subject.request if operation == "request" else subject.auth_headers
+    args = ("GET", RESOURCE_URL) if operation == "request" else ()
+
+    try:
+        raise LookupError(PRIVATE_DATA)
+    except LookupError:
+        with pytest.raises(expected) as captured:
+            invoke(*args)
+        error = captured.value
+        logger.exception("Authentication failed", exc_info=(type(error), error, error.__traceback__))
+    assert isinstance(error, AuthError)
+    assert error.__context__ is None
+    assert error.__cause__ is None
+    assert PRIVATE_DATA not in str(error)
+    assert PRIVATE_DATA not in repr(error)
+    assert PRIVATE_DATA not in "".join(traceback.format_exception(error))
+    assert PRIVATE_DATA not in stream.getvalue()
+    assert json.loads(stream.getvalue())["exception_name"] == expected.__name__
+
+
+@pytest.mark.parametrize(
+    "status,retryable,attempts",
+    [(400, False, 1), (401, False, 1), (429, True, 3), (503, True, 3)],
+)
+def test_exchange_errors_expose_fixed_reason_and_retryability(http, clock, monkeypatch, status, retryable, attempts):
+    monkeypatch.setattr("time.sleep", clock.advance)
+    subject = OAuth2Client(token_url=TOKEN_URL, client_id="orders", client_secret="test-secret")
+    http.serve(TOKEN_URL, {"error_description": PRIVATE_DATA}, method="POST", status=status)
+
+    with pytest.raises(TokenExchangeError) as error:
+        subject.auth_headers()
+    assert error.value.reason is AuthFailureReason.TOKEN_EXCHANGE_FAILED
+    assert error.value.retryable is retryable
+    assert len(http.requests) == attempts
+
+
+def test_outbound_and_jwt_errors_share_the_public_base():
+    from aws_lambda_powertools.utilities.auth_alpha.jwt.exceptions import AuthError as JWTAuthError
+    from aws_lambda_powertools.utilities.auth_alpha.jwt.exceptions import AuthFailureReason as JWTFailureReason
+
+    assert JWTAuthError is AuthError
+    assert JWTFailureReason is AuthFailureReason
+    assert issubclass(TokenExchangeError, AuthError)
+    assert DownstreamRequestError().reason is AuthFailureReason.DOWNSTREAM_REQUEST_FAILED
+    assert not DownstreamRequestError().retryable
diff --git a/tests/functional/auth_alpha/oauth2/test_imports.py b/tests/functional/auth_alpha/oauth2/test_imports.py
new file mode 100644
index 00000000000..aaf854769dd
--- /dev/null
+++ b/tests/functional/auth_alpha/oauth2/test_imports.py
@@ -0,0 +1,42 @@
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+
+def test_oauth_client_does_not_import_jwt_or_cryptography():
+    root = Path(__file__).parents[4]
+    probe = """
+import importlib.abc
+import sys
+
+class BlockJWT(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname.split(".")[0] in {"jwt", "cryptography"}:
+            raise AssertionError("OAuth imported JWT dependencies")
+
+sys.meta_path.insert(0, BlockJWT())
+import aws_lambda_powertools.utilities.auth_alpha as auth
+import aws_lambda_powertools.utilities.auth_alpha.oauth2 as oauth
+assert "urllib3" not in sys.modules
+assert "OAuth2Client" in dir(auth)
+assert "OAuth2Client" in dir(oauth)
+assert auth.OAuth2Client is oauth.OAuth2Client
+client = oauth.OAuth2Client(
+    token_url="https://idp.example.com/token",
+    client_id="orders",
+    client_secret="test-secret",
+)
+assert repr(client) == "<OAuth2Client>"
+try:
+    oauth.unknown_attribute
+except AttributeError:
+    pass
+else:
+    raise AssertionError("Unexpected module attribute")
+assert not {"jwt", "cryptography"} & sys.modules.keys()
+"""
+    env = os.environ.copy()
+    env["PYTHONPATH"] = os.pathsep.join((str(root), env.get("PYTHONPATH", "")))
+    result = subprocess.run([sys.executable, "-c", probe], capture_output=True, text=True, env=env, check=False)
+    assert result.returncode == 0, result.stderr
diff --git a/tests/integration/auth_alpha/jwt/conftest.py b/tests/integration/auth_alpha/conftest.py
similarity index 64%
rename from tests/integration/auth_alpha/jwt/conftest.py
rename to tests/integration/auth_alpha/conftest.py
index 78304d5c565..2913634be1e 100644
--- a/tests/integration/auth_alpha/jwt/conftest.py
+++ b/tests/integration/auth_alpha/conftest.py
@@ -22,6 +22,49 @@ class Reply:
     headers: dict = field(default_factory=dict)
     interval: float = 0
     stall: bool = False
+    header_interval: float = 0
+    chunked: bool = False
+    chunk_size_interval: float = 0
+    trailer_interval: float = 0
+
+
+def _write_bytes(stream, payload, stop, interval=0):
+    if not interval:
+        stream.write(payload)
+        return True
+    for value in payload:
+        if stop.wait(interval):
+            return False
+        stream.write(bytes([value]))
+        stream.flush()
+    return True
+
+
+def _write_body(stream, reply, stop):
+    if reply.stall:
+        stop.wait(5)
+        return
+    parts = [(reply.body, reply.interval)]
+    if reply.chunked:
+        parts = []
+        if reply.body:
+            parts.extend(
+                [
+                    (f"{len(reply.body):x};padding=".encode() + b"x" * 32 + b"\r\n", reply.chunk_size_interval),
+                    (reply.body, reply.interval),
+                    (b"\r\n", 0),
+                ],
+            )
+        parts.extend(
+            [
+                (b"0\r\n", 0),
+                (b"X-Trailer: " + b"x" * 32 + b"\r\n", reply.trailer_interval),
+                (b"\r\n", 0),
+            ],
+        )
+    for payload, interval in parts:
+        if not _write_bytes(stream, payload, stop, interval):
+            return
 
 
 class LocalHTTPS:
@@ -31,9 +74,32 @@ def __init__(self):
         self.stop = threading.Event()
         self.url = ""
 
-    def serve(self, path, payload, *, status=200, headers=None, interval=0, stall=False):
+    def serve(
+        self,
+        path,
+        payload,
+        *,
+        status=200,
+        headers=None,
+        interval=0,
+        stall=False,
+        header_interval=0,
+        chunked=False,
+        chunk_size_interval=0,
+        trailer_interval=0,
+    ):
         body = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
-        self.routes[path] = Reply(body, status, headers or {}, interval, stall)
+        self.routes[path] = Reply(
+            body=body,
+            status=status,
+            headers=headers or {},
+            interval=interval,
+            stall=stall,
+            header_interval=header_interval,
+            chunked=chunked,
+            chunk_size_interval=chunk_size_interval,
+            trailer_interval=trailer_interval,
+        )
 
 
 @pytest.fixture(scope="session")
@@ -96,22 +162,25 @@ def respond(self):
             reply = endpoint.routes.get(self.path, Reply(b"{}", status=404))
             self.send_response(reply.status)
             self.send_header("Content-Type", "application/json")
-            self.send_header("Content-Length", str(len(reply.body)))
+            if reply.chunked:
+                self.send_header("Transfer-Encoding", "chunked")
+            else:
+                self.send_header("Content-Length", str(len(reply.body)))
             self.send_header("Connection", "keep-alive" if request.param else "close")
             for name, value in reply.headers.items():
                 self.send_header(name, value)
-            self.end_headers()
             try:
-                if reply.stall:
-                    endpoint.stop.wait(5)
-                elif reply.interval:
-                    for value in reply.body:
-                        if endpoint.stop.wait(reply.interval):
-                            break
-                        self.wfile.write(bytes([value]))
-                        self.wfile.flush()
-                else:
-                    self.wfile.write(reply.body)
+                if reply.header_interval:
+                    self.flush_headers()
+                    if not _write_bytes(
+                        self.wfile,
+                        b"X-Slow: " + b"x" * 32 + b"\r\n",
+                        endpoint.stop,
+                        reply.header_interval,
+                    ):
+                        return
+                self.end_headers()
+                _write_body(self.wfile, reply, endpoint.stop)
             except (OSError, ssl.SSLError):
                 # Timeout and oversized-body tests deliberately close early.
                 pass
diff --git a/tests/integration/auth_alpha/oauth2/__init__.py b/tests/integration/auth_alpha/oauth2/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/tests/integration/auth_alpha/oauth2/test_https.py b/tests/integration/auth_alpha/oauth2/test_https.py
new file mode 100644
index 00000000000..c5a1f35b874
--- /dev/null
+++ b/tests/integration/auth_alpha/oauth2/test_https.py
@@ -0,0 +1,225 @@
+import base64
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from urllib.parse import parse_qs
+
+import pytest
+
+from aws_lambda_powertools.utilities.auth_alpha import OAuth2Client
+from aws_lambda_powertools.utilities.auth_alpha.oauth2.exceptions import DownstreamRequestError, TokenExchangeError
+
+TOKEN_RESPONSE = {"access_token": "local-test-token", "token_type": "Bearer", "expires_in": 600}
+
+
+def client(endpoint, **options):
+    return OAuth2Client(
+        token_url=endpoint.url + "/token",
+        client_id="orders",
+        client_secret="test-only-secret",
+        scopes=["inventory:read"],
+        **options,
+    )
+
+
+@pytest.mark.parametrize("chunked", [False, True])
+def test_token_exchange_and_authenticated_request_over_trusted_tls(https_server, chunked):
+    https_server.serve("/token", TOKEN_RESPONSE, chunked=chunked)
+    https_server.serve("/inventory", {"items": [123]}, chunked=chunked)
+    subject = client(https_server)
+
+    response = subject.request("GET", https_server.url + "/inventory")
+
+    assert response.status == 200
+    assert response.json() == {"items": [123]}
+    assert subject.auth_headers() == {"Authorization": "Bearer local-test-token"}
+    exchange, resource = https_server.requests
+    assert exchange[:2] == ("POST", "/token")
+    assert base64.b64decode(exchange[2]["Authorization"].removeprefix("Basic ")).decode() == "orders:test-only-secret"
+    assert parse_qs(exchange[3].decode()) == {"grant_type": ["client_credentials"], "scope": ["inventory:read"]}
+    assert resource[2]["Authorization"] == "Bearer local-test-token"
+    assert "test-only-secret" not in str(resource)
+
+
+def test_downstream_redirects_are_returned_without_forwarding_bearer_tokens(https_server):
+    https_server.serve("/token", TOKEN_RESPONSE)
+    https_server.serve("/inventory", {}, status=307, headers={"Location": https_server.url + "/other"})
+    https_server.serve("/other", {})
+
+    response = client(https_server).request("GET", https_server.url + "/inventory")
+
+    assert response.status == 307
+    assert [request[1] for request in https_server.requests] == ["/token", "/inventory"]
+
+
+def test_downstream_failures_are_not_retried(https_server):
+    https_server.serve("/token", TOKEN_RESPONSE)
+    https_server.serve("/inventory", {}, status=503)
+    assert client(https_server).request("POST", https_server.url + "/inventory").status == 503
+    assert [request[1] for request in https_server.requests] == ["/token", "/inventory"]
+
+
+def test_downstream_timeout_has_a_separate_budget_and_a_sanitized_error(https_server):
+    https_server.serve("/token", TOKEN_RESPONSE)
+    https_server.serve("/inventory", {"items": []}, stall=True)
+    subject = client(https_server, timeout_seconds=3)
+    started = time.monotonic()
+
+    with pytest.raises(DownstreamRequestError) as error:
+        subject.request("GET", https_server.url + "/inventory", timeout=0.2)
+
+    assert time.monotonic() - started < 1
+    assert error.value.__context__ is None
+    assert [request[1] for request in https_server.requests] == ["/token", "/inventory"]
+
+
+def test_untrusted_tls_never_sends_client_credentials(https_server, monkeypatch):
+    monkeypatch.delenv("SSL_CERT_FILE")
+    https_server.serve("/token", TOKEN_RESPONSE)
+    subject = client(https_server, timeout_seconds=0.15)
+
+    with pytest.raises(TokenExchangeError) as error:
+        subject.auth_headers()
+    assert error.value.__context__ is None
+    assert https_server.requests == []
+
+
+@pytest.mark.parametrize("failure", ["oversized", "redirect", "stall", "trickle"])
+def test_exchange_failures_are_bounded_without_forwarding_credentials(https_server, failure):
+    if failure == "oversized":
+        https_server.serve("/token", b'{"padding":"' + b"x" * (1024 * 1024) + b'"}')
+    elif failure == "redirect":
+        https_server.serve("/token", {}, status=307, headers={"Location": https_server.url + "/redirected"})
+        https_server.serve("/redirected", TOKEN_RESPONSE)
+    else:
+        https_server.serve(
+            "/token",
+            TOKEN_RESPONSE,
+            stall=failure == "stall",
+            interval=0.04 if failure == "trickle" else 0,
+        )
+    subject = client(https_server, timeout_seconds=0.2)
+    started = time.monotonic()
+
+    with pytest.raises(TokenExchangeError) as error:
+        subject.auth_headers()
+    assert time.monotonic() - started < 1
+    assert error.value.__context__ is None
+    assert [request[1] for request in https_server.requests] == ["/token"]
+
+
+def test_slow_downstream_body_cannot_extend_the_timeout(https_server):
+    https_server.serve("/token", TOKEN_RESPONSE)
+    https_server.serve("/inventory", {"items": list(range(100))}, interval=0.04)
+    subject = client(https_server)
+    started = time.monotonic()
+
+    with pytest.raises(DownstreamRequestError):
+        subject.request("GET", https_server.url + "/inventory", timeout=0.2)
+    assert time.monotonic() - started < 1
+    assert [request[1] for request in https_server.requests] == ["/token", "/inventory"]
+
+
+@pytest.fixture(params=["header", "chunk_size", "trailer"])
+def slow_framing(request):
+    return {f"{request.param}_interval": 0.04, "chunked": request.param != "header"}
+
+
+def test_slow_token_framing_cannot_extend_the_acquisition_timeout(https_server, slow_framing):
+    https_server.serve("/token", TOKEN_RESPONSE, **slow_framing)
+    subject = client(https_server, timeout_seconds=0.2)
+    started = time.monotonic()
+
+    with pytest.raises(TokenExchangeError) as error:
+        subject.auth_headers()
+
+    assert time.monotonic() - started < 1
+    assert error.value.retryable
+    assert error.value.__context__ is None
+    assert error.value.__cause__ is None
+    assert [request[1] for request in https_server.requests] == ["/token"]
+
+    https_server.serve("/token", TOKEN_RESPONSE)
+    assert subject.auth_headers() == {"Authorization": "Bearer local-test-token"}
+    assert [request[1] for request in https_server.requests] == ["/token", "/token"]
+
+
+def test_slow_downstream_framing_cannot_extend_the_request_timeout(https_server, slow_framing):
+    https_server.serve("/token", TOKEN_RESPONSE)
+    https_server.serve("/inventory", {"items": [123]}, **slow_framing)
+    subject = client(https_server)
+    subject.auth_headers()
+    started = time.monotonic()
+
+    with pytest.raises(DownstreamRequestError) as error:
+        subject.request("GET", https_server.url + "/inventory", timeout=0.2)
+
+    assert time.monotonic() - started < 1
+    assert not error.value.retryable
+    assert error.value.__context__ is None
+    assert error.value.__cause__ is None
+    assert [request[1] for request in https_server.requests] == ["/token", "/inventory"]
+
+    https_server.serve("/inventory", {"items": [123]})
+    assert subject.request("GET", https_server.url + "/inventory").json() == {"items": [123]}
+    assert [request[1] for request in https_server.requests] == ["/token", "/inventory", "/inventory"]
+
+
+def test_concurrent_downstream_requests_keep_separate_deadlines(https_server):
+    https_server.serve("/token", TOKEN_RESPONSE)
+    https_server.serve("/inventory", {"items": [123]}, header_interval=0.01)
+    subject = client(https_server)
+    subject.auth_headers()
+    ready = threading.Barrier(2)
+
+    def request(timeout):
+        ready.wait(timeout=2)
+        return subject.request("GET", https_server.url + "/inventory", timeout=timeout)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        short = executor.submit(request, 0.15)
+        long = executor.submit(request, 2)
+        with pytest.raises(DownstreamRequestError):
+            short.result(timeout=3)
+        assert long.result(timeout=3).json() == {"items": [123]}
+
+    assert [request[1] for request in https_server.requests] == ["/token", "/inventory", "/inventory"]
+
+
+def test_body_reads_use_the_budget_remaining_after_headers(https_server):
+    https_server.serve("/token", TOKEN_RESPONSE)
+    # Header receipt consumes about 0.2s; the trailer must use what remains,
+    # rather than starting another downstream timeout after the headers.
+    https_server.serve(
+        "/inventory",
+        {"items": [123]},
+        header_interval=0.005,
+        chunked=True,
+        trailer_interval=0.04,
+    )
+    subject = client(https_server)
+    subject.auth_headers()
+    started = time.monotonic()
+
+    with pytest.raises(DownstreamRequestError):
+        subject.request("GET", https_server.url + "/inventory", timeout=0.4)
+
+    assert time.monotonic() - started < 0.6
+    assert [request[1] for request in https_server.requests] == ["/token", "/inventory"]
+
+
+@pytest.mark.parametrize("chunked", [False, True])
+def test_downstream_gzip_response_remains_readable(https_server, chunked):
+    import gzip
+
+    https_server.serve("/token", TOKEN_RESPONSE)
+    https_server.serve(
+        "/inventory",
+        gzip.compress(b'{"items":[123]}'),
+        headers={"Content-Encoding": "gzip"},
+        chunked=chunked,
+    )
+    subject = client(https_server)
+
+    response = subject.request("GET", https_server.url + "/inventory")
+    assert response.json() == {"items": [123]}