diff --git a/datadog_checks_base/changelog.d/22676.added b/datadog_checks_base/changelog.d/22676.added
index ff63715e3c69d..6a7dc34f16936 100644
--- a/datadog_checks_base/changelog.d/22676.added
+++ b/datadog_checks_base/changelog.d/22676.added
@@ -1 +1 @@
-Add library-agnostic HTTP mocks/proto/exceptions and migrate intg tests.
+Add library-agnostic HTTP mocks/proto/exceptions and migrate intg tests. Add ``HTTPX2Wrapper``, an httpx2-backed HTTP client opt-in via the ``use_httpx2`` instance config flag. The default remains the existing requests-based client.
diff --git a/datadog_checks_base/datadog_checks/base/checks/base.py b/datadog_checks_base/datadog_checks/base/checks/base.py
index 77da814ad0b45..49844c6d5dc3d 100644
--- a/datadog_checks_base/datadog_checks/base/checks/base.py
+++ b/datadog_checks_base/datadog_checks/base/checks/base.py
@@ -400,6 +400,16 @@ def _get_metric_limit(self, instance=None):
 
         return limit
 
+    def _build_http_client(self, instance: dict[str, Any]) -> HTTPClientProtocol:
+        if is_affirmative(instance.get('use_httpx2', False)):
+            from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper
+
+            return HTTPX2Wrapper(instance, self.init_config, self.HTTP_CONFIG_REMAPPER, self.log)
+        # See Performance Optimizations in this package's README.md.
+        from datadog_checks.base.utils.http import RequestsWrapper
+
+        return RequestsWrapper(instance, self.init_config, self.HTTP_CONFIG_REMAPPER, self.log)
+
     @property
     def http(self) -> HTTPClientProtocol:
         """
@@ -408,10 +418,7 @@ def http(self) -> HTTPClientProtocol:
         Only new checks or checks on Agent 6.13+ can and should use this for HTTP requests.
         """
         if not hasattr(self, '_http'):
-            # See Performance Optimizations in this package's README.md.
-            from datadog_checks.base.utils.http import RequestsWrapper
-
-            self._http = RequestsWrapper(self.instance or {}, self.init_config, self.HTTP_CONFIG_REMAPPER, self.log)
+            self._http = self._build_http_client(self.instance or {})
 
         return self._http
 
diff --git a/datadog_checks_base/datadog_checks/base/checks/openmetrics/v2/scraper/base_scraper.py b/datadog_checks_base/datadog_checks/base/checks/openmetrics/v2/scraper/base_scraper.py
index 24719c329efef..3ba8f717d09d6 100644
--- a/datadog_checks_base/datadog_checks/base/checks/openmetrics/v2/scraper/base_scraper.py
+++ b/datadog_checks_base/datadog_checks/base/checks/openmetrics/v2/scraper/base_scraper.py
@@ -13,7 +13,7 @@
 from prometheus_client import Metric
 from prometheus_client.openmetrics.parser import text_fd_to_metric_families as parse_openmetrics
 from prometheus_client.parser import text_fd_to_metric_families as parse_prometheus
-from requests.exceptions import ConnectionError
+from requests.exceptions import ConnectionError as RequestsConnectionError
 
 from datadog_checks.base.agent import datadog_agent
 from datadog_checks.base.checks.openmetrics.v2.first_scrape_handler import first_scrape_handler
@@ -23,6 +23,7 @@
 from datadog_checks.base.constants import ServiceCheck
 from datadog_checks.base.errors import ConfigurationError
 from datadog_checks.base.utils.functions import no_op, return_true
+from datadog_checks.base.utils.http_exceptions import HTTPConnectionError
 
 
 class OpenMetricsScraper:
@@ -405,11 +406,11 @@ def stream_connection_lines(self):
                 self._content_type = connection.headers.get('Content-Type', '')
                 for line in connection.iter_lines(decode_unicode=True):
                     yield line
-        except ConnectionError as e:
+        except (RequestsConnectionError, HTTPConnectionError) as e:
             if self.ignore_connection_errors:
-                self.log.warning("OpenMetrics endpoint %s is not accessible", self.endpoint)
+                self.log.warning("OpenMetrics endpoint %s is not accessible: %s", self.endpoint, e)
             else:
-                raise e
+                raise
 
     def filter_connection_lines(self, line_streamer):
         """
diff --git a/datadog_checks_base/datadog_checks/base/utils/http_protocol.py b/datadog_checks_base/datadog_checks/base/utils/http_protocol.py
index 4d103de5930c3..2d31c24d1b5fb 100644
--- a/datadog_checks_base/datadog_checks/base/utils/http_protocol.py
+++ b/datadog_checks_base/datadog_checks/base/utils/http_protocol.py
@@ -4,9 +4,11 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from typing import Any, Iterator, Protocol
+from datetime import timedelta
+from typing import Any, Iterator, Protocol, runtime_checkable
 
 
+@runtime_checkable
 class HTTPResponseProtocol(Protocol):
     status_code: int
     content: bytes
@@ -17,6 +19,16 @@ class HTTPResponseProtocol(Protocol):
     def ok(self) -> bool: ...
     @property
     def reason(self) -> str: ...
+    @property
+    def encoding(self) -> str | None: ...
+    @encoding.setter
+    def encoding(self, value: str | None) -> None: ...
+    @property
+    def url(self) -> str: ...
+    @property
+    def cookies(self) -> Mapping[str, str]: ...
+    @property
+    def elapsed(self) -> timedelta: ...
 
     def json(self, **kwargs: Any) -> Any: ...
     def raise_for_status(self) -> None: ...
diff --git a/datadog_checks_base/datadog_checks/base/utils/httpx2.py b/datadog_checks_base/datadog_checks/base/utils/httpx2.py
new file mode 100644
index 0000000000000..85fbdf390a6a0
--- /dev/null
+++ b/datadog_checks_base/datadog_checks/base/utils/httpx2.py
@@ -0,0 +1,486 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+import logging
+from collections.abc import Iterator, Mapping
+from datetime import timedelta
+from typing import Any, Self
+
+import httpx2
+from binary import KIBIBYTE
+
+from datadog_checks.base.config import is_affirmative
+
+from .headers import get_default_headers, update_headers
+from .http_exceptions import (
+    HTTPConnectionError,
+    HTTPError,
+    HTTPInvalidURLError,
+    HTTPRequestError,
+    HTTPStatusError,
+    HTTPTimeoutError,
+)
+
+LOGGER = logging.getLogger(__name__)
+
+DEFAULT_TIMEOUT = 10
+# Matches the effective chunk size used by RequestsWrapper.iter_content (http.py:415 multiplies by KIBIBYTE).
+DEFAULT_CHUNK_SIZE = 16 * KIBIBYTE
+
+STANDARD_FIELDS = {
+    'allow_redirects': True,
+    'connect_timeout': None,
+    'extra_headers': None,
+    'headers': None,
+    'log_requests': False,
+    'password': None,
+    'read_timeout': None,
+    'timeout': DEFAULT_TIMEOUT,
+    'tls_ca_cert': None,
+    'tls_cert': None,
+    'tls_private_key': None,
+    'tls_verify': True,
+    'username': None,
+}
+
+REQUEST_KWARGS = frozenset(
+    {
+        'params',
+        'json',
+        'data',
+        'content',
+        'files',
+        'cookies',
+        'headers',
+        'extra_headers',
+        'timeout',
+        'follow_redirects',
+        'auth',
+    }
+)
+REQUEST_KWARGS_SPECIAL = frozenset({'headers', 'extra_headers', 'timeout', 'follow_redirects', 'auth'})
+REQUEST_KWARGS_PASSTHROUGH = REQUEST_KWARGS - REQUEST_KWARGS_SPECIAL
+assert REQUEST_KWARGS_PASSTHROUGH | REQUEST_KWARGS_SPECIAL == REQUEST_KWARGS
+assert REQUEST_KWARGS_PASSTHROUGH & REQUEST_KWARGS_SPECIAL == frozenset()
+
+UNKNOWN_KWARG_HINTS: dict[str, str] = {
+    'allow_redirects': "use 'follow_redirects' instead (httpx2 name for this kwarg)",
+    'verify': "configure 'tls_verify' in instance config, or drop the per-call kwarg",
+    'persist': 'drop the kwarg, httpx2 pools connections by default',
+    'cert': "configure 'tls_cert' and 'tls_private_key' in instance config",
+    'proxies': 'proxy support is not yet wired through to httpx2, see TODO in httpx2.py',
+}
+
+
+def _make_timeout(connect: float, read: float) -> httpx2.Timeout:
+    # Pass the read timeout to every phase so write and pool aren't left unbounded.
+    return httpx2.Timeout(connect=connect, read=read, write=read, pool=read)
+
+
+def _build_basic_auth(config: dict[str, Any]) -> httpx2.BasicAuth | None:
+    if config['username'] is not None and config['password'] is not None:
+        return httpx2.BasicAuth(config['username'], config['password'])
+    return None
+
+
+def _build_verify(config: dict[str, Any]) -> bool | str:
+    if isinstance(config['tls_ca_cert'], str):
+        return config['tls_ca_cert']
+    if not is_affirmative(config['tls_verify']):
+        return False
+    return True
+
+
+def _build_cert(config: dict[str, Any]) -> str | tuple[str, str] | None:
+    cert = config['tls_cert']
+    if not isinstance(cert, str):
+        return None
+    private_key = config['tls_private_key']
+    if isinstance(private_key, str):
+        return (cert, private_key)
+    return cert
+
+
+def _build_timeout(config: dict[str, Any]) -> tuple[float, float]:
+    base = float(config['timeout'])
+    connect = float(config['connect_timeout']) if config['connect_timeout'] is not None else base
+    read = float(config['read_timeout']) if config['read_timeout'] is not None else base
+    return connect, read
+
+
+def _map_httpx2_exception(exc: httpx2.HTTPError | httpx2.InvalidURL) -> HTTPError:
+    """Translate an httpx2 exception into the library-agnostic equivalent."""
+    if isinstance(exc, httpx2.InvalidURL):
+        return HTTPInvalidURLError(str(exc) or exc.__class__.__name__, request=getattr(exc, 'request', None))
+    if isinstance(exc, httpx2.TimeoutException):
+        return HTTPTimeoutError(str(exc) or exc.__class__.__name__, request=getattr(exc, 'request', None))
+    # httpx2 has no dedicated SSL/cert exception: TLS failures surface as ConnectError (a NetworkError),
+    # so they fold into the generic HTTPConnectionError. The HTTPSSLError subtype is intentionally unproduced here.
+    if isinstance(exc, httpx2.NetworkError):
+        return HTTPConnectionError(str(exc) or exc.__class__.__name__, request=getattr(exc, 'request', None))
+    if isinstance(exc, httpx2.HTTPStatusError):
+        return HTTPStatusError(
+            str(exc) or exc.__class__.__name__,
+            request=getattr(exc, 'request', None),
+            response=getattr(exc, 'response', None),
+        )
+    if isinstance(exc, httpx2.RequestError):
+        return HTTPRequestError(str(exc) or exc.__class__.__name__, request=getattr(exc, 'request', None))
+    return HTTPError(str(exc) or exc.__class__.__name__)
+
+
+class HTTPX2ResponseAdapter:
+    """Wraps an httpx2.Response to satisfy HTTPResponseProtocol."""
+
+    __slots__ = ('_response',)
+
+    def __init__(self, response: httpx2.Response) -> None:
+        self._response = response
+
+    @property
+    def status_code(self) -> int:
+        return self._response.status_code
+
+    @property
+    def content(self) -> bytes:
+        try:
+            return self._response.read()
+        except (httpx2.HTTPError, httpx2.InvalidURL) as exc:
+            raise _map_httpx2_exception(exc) from exc
+
+    @property
+    def text(self) -> str:
+        try:
+            self._response.read()
+            return self._response.text
+        except (httpx2.HTTPError, httpx2.InvalidURL) as exc:
+            raise _map_httpx2_exception(exc) from exc
+
+    @property
+    def headers(self) -> Mapping[str, str]:
+        return self._response.headers
+
+    @property
+    def ok(self) -> bool:
+        return self._response.status_code < 400
+
+    @property
+    def reason(self) -> str:
+        return self._response.reason_phrase
+
+    @property
+    def encoding(self) -> str | None:
+        return self._response.encoding
+
+    @encoding.setter
+    def encoding(self, value: str | None) -> None:
+        # Encoding is consulted at decode time, not when set. Bytes already read by
+        # iter_text or iter_lines stay decoded under the previous encoding.
+        self._response.encoding = value
+
+    @property
+    def url(self) -> str:
+        return str(self._response.url)
+
+    @property
+    def cookies(self) -> Mapping[str, str]:
+        # Narrowed to Mapping to keep httpx2 out of the wrapper surface. Callers needing
+        # the richer httpx2.Cookies API can reach self._response.cookies directly.
+        return self._response.cookies
+
+    @property
+    def elapsed(self) -> timedelta:
+        try:
+            return self._response.elapsed
+        except RuntimeError:
+            LOGGER.debug('elapsed unavailable for response from %s', self._response.url)
+            return timedelta(0)
+
+    def json(self, **kwargs: Any) -> Any:
+        try:
+            self._response.read()
+            return self._response.json(**kwargs)
+        except (httpx2.HTTPError, httpx2.InvalidURL) as exc:
+            raise _map_httpx2_exception(exc) from exc
+
+    def raise_for_status(self) -> None:
+        # Mirror requests semantics (4xx/5xx only); httpx2 also raises on 3xx.
+        if self._response.status_code < 400:
+            return
+        try:
+            self._response.raise_for_status()
+        except httpx2.HTTPStatusError as exc:
+            raise _map_httpx2_exception(exc) from exc
+
+    def close(self) -> None:
+        self._response.close()
+
+    def iter_content(self, chunk_size: int | None = None, decode_unicode: bool = False) -> Iterator[bytes | str]:
+        effective_size = chunk_size if chunk_size is not None else DEFAULT_CHUNK_SIZE
+        try:
+            if decode_unicode:
+                # iter_text uses an incremental decoder. Manual chunk decoding breaks on
+                # multibyte chars split across boundaries.
+                yield from self._response.iter_text(chunk_size=effective_size)
+                return
+            yield from self._response.iter_bytes(chunk_size=effective_size)
+        except (httpx2.HTTPError, httpx2.InvalidURL) as exc:
+            raise _map_httpx2_exception(exc) from exc
+
+    def iter_lines(
+        self,
+        chunk_size: int | None = None,  # noqa: ARG002 - httpx2 buffers lines internally. Kept for HTTPResponseProtocol parity
+        decode_unicode: bool = False,
+        delimiter: bytes | str | None = None,
+    ) -> Iterator[bytes | str]:
+        if delimiter is not None:
+            raise NotImplementedError("HTTPX2ResponseAdapter.iter_lines does not support custom delimiters")
+        try:
+            if decode_unicode:
+                yield from self._response.iter_lines()
+                return
+            buf = bytearray()
+            for chunk in self._response.iter_bytes():
+                buf.extend(chunk)
+                while (idx := buf.find(b'\n')) != -1:
+                    yield bytes(buf[:idx]).rstrip(b'\r')
+                    del buf[: idx + 1]
+            if buf:
+                yield bytes(buf).rstrip(b'\r')
+        except (httpx2.HTTPError, httpx2.InvalidURL) as exc:
+            raise _map_httpx2_exception(exc) from exc
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool | None:
+        self.close()
+        return None
+
+
+class HTTPX2Wrapper:
+    """Implements HTTPClientProtocol using a single shared httpx2.Client per wrapper."""
+
+    __slots__ = (
+        '_client',
+        '_log_requests',
+        'logger',
+        'options',
+    )
+
+    def __init__(
+        self,
+        instance: dict[str, Any],
+        init_config: dict[str, Any] | None = None,
+        remapper: dict[str, dict[str, Any]] | None = None,
+        logger: logging.Logger | None = None,
+        transport: httpx2.BaseTransport | None = None,
+    ) -> None:
+        self.logger = logger or LOGGER
+        init_config = init_config or {}
+
+        config = self._resolve_config(instance, init_config, remapper)
+
+        # options['headers'] is a plain dict for parity with RequestsWrapper (http.py:319).
+        # Case-folding is applied at request time via httpx2.Headers in _build_request_kwargs;
+        # get_header/set_header below do case-insensitive iteration over this dict.
+        headers = get_default_headers()
+        if config['headers']:
+            headers.clear()
+            update_headers(headers, config['headers'])
+        if config['extra_headers']:
+            update_headers(headers, config['extra_headers'])
+
+        auth = _build_basic_auth(config)
+        verify = _build_verify(config)
+        cert = _build_cert(config)
+        timeout = _build_timeout(config)
+        allow_redirects = is_affirmative(config['allow_redirects'])
+
+        # proxies=None mirrors RequestsWrapper.options for consumers (e.g. http_check).
+        # TODO: wire proxies through to httpx2 as part of the ongoing httpx migration.
+        # options['headers'] is the per-request source of truth and is re-read in _build_request_kwargs,
+        # so direct mutation (__setitem__, update(), whole-dict replacement) reaches the wire.
+        # set_header mutates options['headers'] in place; the per-request merge above re-reads it.
+        self.options: dict[str, Any] = {
+            'auth': auth,
+            'cert': cert,
+            'headers': headers,
+            'proxies': None,
+            'timeout': timeout,
+            'verify': verify,
+            'allow_redirects': allow_redirects,
+        }
+
+        self._log_requests = is_affirmative(config['log_requests'])
+        self._client = self._build_client(transport)
+
+    @staticmethod
+    def _resolve_config(
+        instance: dict[str, Any],
+        init_config: dict[str, Any],
+        remapper: dict[str, dict[str, Any]] | None,
+    ) -> dict[str, Any]:
+        default_fields = dict(STANDARD_FIELDS)
+        default_fields['log_requests'] = init_config.get('log_requests', default_fields['log_requests'])
+        default_fields['timeout'] = init_config.get('timeout', default_fields['timeout'])
+
+        config = {field: instance.get(field, value) for field, value in default_fields.items()}
+
+        remapper = dict(remapper) if remapper else {}
+
+        for remapped_field, data in remapper.items():
+            field = data.get('name')
+            if field not in STANDARD_FIELDS:
+                continue
+            if field in instance:
+                continue
+
+            default = default_fields[field]
+            if data.get('invert'):
+                default = not default
+
+            value = instance.get(remapped_field, data.get('default', default))
+            if data.get('invert'):
+                value = not is_affirmative(value)
+
+            config[field] = value
+        return config
+
+    def _build_client(self, transport: httpx2.BaseTransport | None) -> httpx2.Client:
+        kwargs: dict[str, Any] = {
+            'headers': self.options['headers'],
+            'timeout': _make_timeout(self.options['timeout'][0], self.options['timeout'][1]),
+            'follow_redirects': self.options['allow_redirects'],
+            'verify': self.options['verify'],
+        }
+        if self.options['cert'] is not None:
+            kwargs['cert'] = self.options['cert']
+        if self.options['auth'] is not None:
+            kwargs['auth'] = self.options['auth']
+        if transport is not None:
+            kwargs['transport'] = transport
+        return httpx2.Client(**kwargs)
+
+    def get_header(self, name: str, default: str | None = None) -> str | None:
+        for key, value in self.options['headers'].items():
+            if key.lower() == name.lower():
+                return value
+        return default
+
+    def set_header(self, name: str, value: str) -> None:
+        for key in list(self.options['headers']):
+            if key.lower() == name.lower():
+                self.options['headers'][key] = value
+                return
+        self.options['headers'][name] = value
+
+    def get(self, url: str, **options: Any) -> HTTPX2ResponseAdapter:
+        return self._request('GET', url, options)
+
+    def post(self, url: str, **options: Any) -> HTTPX2ResponseAdapter:
+        return self._request('POST', url, options)
+
+    def put(self, url: str, **options: Any) -> HTTPX2ResponseAdapter:
+        return self._request('PUT', url, options)
+
+    def delete(self, url: str, **options: Any) -> HTTPX2ResponseAdapter:
+        return self._request('DELETE', url, options)
+
+    def head(self, url: str, **options: Any) -> HTTPX2ResponseAdapter:
+        return self._request('HEAD', url, options)
+
+    def patch(self, url: str, **options: Any) -> HTTPX2ResponseAdapter:
+        return self._request('PATCH', url, options)
+
+    def options_method(self, url: str, **options: Any) -> HTTPX2ResponseAdapter:
+        return self._request('OPTIONS', url, options)
+
+    def _request(self, method: str, url: str, options: dict[str, Any]) -> HTTPX2ResponseAdapter:
+        # stream=True keeps the connection open until the body is consumed (.content/.text/.iter_*)
+        # or close() is called. Callers should use it as a context manager when not reading the body.
+        if self._log_requests:
+            self.logger.debug('Sending %s request to %s', method, url)
+
+        follow_redirects = (
+            bool(options['follow_redirects']) if 'follow_redirects' in options else httpx2.USE_CLIENT_DEFAULT
+        )
+        auth = options['auth'] if 'auth' in options else httpx2.USE_CLIENT_DEFAULT
+        request_kwargs = self._build_request_kwargs(options, method=method, url=url)
+        try:
+            request = self._client.build_request(method, url, **request_kwargs)
+            response = self._client.send(request, stream=True, follow_redirects=follow_redirects, auth=auth)
+        except (httpx2.HTTPError, httpx2.InvalidURL) as exc:
+            raise _map_httpx2_exception(exc) from exc
+        try:
+            return HTTPX2ResponseAdapter(response)
+        except BaseException:
+            # Close the open stream if anything between send(stream=True) and the wrap raises.
+            response.close()
+            raise
+
+    def _build_request_kwargs(self, options: dict[str, Any], *, method: str = '', url: str = '') -> dict[str, Any]:
+        """Translate call-site options to httpx2.Client.request kwargs."""
+        # Drop stream silently because OM v2 scraper passes it unconditionally (base_scraper.py:459).
+        # The wrapper streams internally, so the kwarg is meaningless.
+        if 'stream' in options:
+            self.logger.debug('HTTPX2Wrapper dropping unsupported per-request kwarg: stream')
+        options = {k: v for k, v in options.items() if k != 'stream'}
+
+        unknown = set(options) - REQUEST_KWARGS
+        if unknown:
+            context = f' for {method} {url}' if method or url else ''
+            sorted_unknown = sorted(unknown)
+            hints = [f'  {k}: {UNKNOWN_KWARG_HINTS[k]}' for k in sorted_unknown if k in UNKNOWN_KWARG_HINTS]
+            hint_block = ('\n' + '\n'.join(hints)) if hints else ''
+            raise TypeError(f"HTTPX2Wrapper does not support per-request kwargs{context}: {sorted_unknown}{hint_block}")
+        kwargs: dict[str, Any] = {}
+        for key in REQUEST_KWARGS_PASSTHROUGH:
+            if key in options:
+                kwargs[key] = options[key]
+
+        # Re-read self.options['headers'] per request so post-init mutations (__setitem__, update(),
+        # whole-dict replacement) reach the wire, mirroring RequestsWrapper's ChainMap pattern at http.py:497.
+        # Layered order: self.options['headers'] < per-call headers < per-call extra_headers.
+        # httpx2.Headers folds keys case-insensitively, matching RequestsWrapper's CaseInsensitiveDict.
+        merged = httpx2.Headers(self.options.get('headers') or {})
+        headers = options.get('headers')
+        if headers:
+            merged.update(headers)
+        extra_headers = options.get('extra_headers')
+        if extra_headers:
+            merged.update(extra_headers)
+        kwargs['headers'] = merged
+
+        if 'timeout' in options:
+            timeout_value = options['timeout']
+            if timeout_value is None:
+                kwargs['timeout'] = None
+            elif isinstance(timeout_value, (tuple, list)) and len(timeout_value) == 2:
+                kwargs['timeout'] = _make_timeout(float(timeout_value[0]), float(timeout_value[1]))
+            else:
+                value = float(timeout_value)
+                kwargs['timeout'] = _make_timeout(value, value)
+
+        return kwargs
+
+    def close(self) -> None:
+        client = getattr(self, '_client', None)
+        if client is not None:
+            client.close()
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool | None:
+        self.close()
+        return None
+
+    def __del__(self) -> None:  # no cov
+        try:
+            self.close()
+        except Exception as exc:
+            # Don't propagate from __del__ (interpreter handles it). Log at debug so the swallow is observable.
+            self.logger.debug('HTTPX2Wrapper.close raised during __del__: %r', exc)
diff --git a/datadog_checks_base/pyproject.toml b/datadog_checks_base/pyproject.toml
index 9fa0f064496f4..d836c5a6fa046 100644
--- a/datadog_checks_base/pyproject.toml
+++ b/datadog_checks_base/pyproject.toml
@@ -38,6 +38,7 @@ deps = [
     "cachetools==7.0.5",
     "cryptography==48.0.1",
     "ddtrace==3.19.5",
+    "httpx2==2.2.0",
     "jellyfish==1.2.1",
     "lazy-loader==0.5",
     "prometheus-client==0.24.1",
diff --git a/datadog_checks_base/tests/base/checks/openmetrics/test_v2/scraper/test_stream_connection_lines.py b/datadog_checks_base/tests/base/checks/openmetrics/test_v2/scraper/test_stream_connection_lines.py
new file mode 100644
index 0000000000000..4e35c8171ac23
--- /dev/null
+++ b/datadog_checks_base/tests/base/checks/openmetrics/test_v2/scraper/test_stream_connection_lines.py
@@ -0,0 +1,50 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import logging
+from unittest import mock
+
+import pytest
+from requests.exceptions import ConnectionError as RequestsConnectionError
+
+from datadog_checks.base.checks.openmetrics.v2.scraper.base_scraper import OpenMetricsScraper
+from datadog_checks.base.utils.http_exceptions import HTTPConnectionError
+
+
+def _scraper_with_connection_error(exception, *, ignore_connection_errors):
+    scraper = OpenMetricsScraper.__new__(OpenMetricsScraper)
+    scraper.endpoint = 'http://example.test/metrics'
+    scraper.ignore_connection_errors = ignore_connection_errors
+    scraper.log = logging.getLogger('test_stream_connection_lines')
+    scraper.get_connection = mock.Mock(side_effect=exception)
+    return scraper
+
+
+@pytest.mark.parametrize(
+    'exception',
+    [
+        pytest.param(HTTPConnectionError('refused'), id='http-connection-error'),
+        pytest.param(RequestsConnectionError('boom'), id='requests-connection-error-backcompat'),
+    ],
+)
+def test_connection_error_warning_path(caplog, exception):
+    scraper = _scraper_with_connection_error(
+        exception,
+        ignore_connection_errors=True,
+    )
+    with caplog.at_level(logging.WARNING, logger='test_stream_connection_lines'):
+        assert list(scraper.stream_connection_lines()) == []
+    assert any(
+        'OpenMetrics endpoint http://example.test/metrics is not accessible' in record.message
+        for record in caplog.records
+    )
+    assert any(str(exception) in record.message for record in caplog.records)
+
+
+def test_connection_error_reraises_when_not_ignored():
+    scraper = _scraper_with_connection_error(
+        HTTPConnectionError('refused'),
+        ignore_connection_errors=False,
+    )
+    with pytest.raises(HTTPConnectionError):
+        list(scraper.stream_connection_lines())
diff --git a/datadog_checks_base/tests/base/utils/httpx2/__init__.py b/datadog_checks_base/tests/base/utils/httpx2/__init__.py
new file mode 100644
index 0000000000000..75c6647cb9233
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/__init__.py
@@ -0,0 +1,3 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
diff --git a/datadog_checks_base/tests/base/utils/httpx2/common.py b/datadog_checks_base/tests/base/utils/httpx2/common.py
new file mode 100644
index 0000000000000..d63c9b58cb84a
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/common.py
@@ -0,0 +1,12 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import base64
+
+
+def parse_basic_auth(header_value: str) -> tuple[str, str]:
+    scheme, _, b64 = header_value.partition(' ')
+    assert scheme.lower() == 'basic'
+    user_pass = base64.b64decode(b64).decode('utf-8')
+    user, _, password = user_pass.partition(':')
+    return user, password
diff --git a/datadog_checks_base/tests/base/utils/httpx2/conftest.py b/datadog_checks_base/tests/base/utils/httpx2/conftest.py
new file mode 100644
index 0000000000000..cf99a2817779b
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/conftest.py
@@ -0,0 +1,61 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from collections.abc import Callable
+
+import httpx2
+import pytest
+
+
+@pytest.fixture
+def status_transport_factory() -> Callable[[int, bytes | str], httpx2.MockTransport]:
+    def _factory(status_code: int, body: bytes | str = b''):
+        def handler(_request: httpx2.Request) -> httpx2.Response:
+            if isinstance(body, str):
+                return httpx2.Response(status_code, text=body)
+            return httpx2.Response(status_code, content=body)
+
+        return httpx2.MockTransport(handler)
+
+    return _factory
+
+
+@pytest.fixture
+def raising_transport_factory() -> Callable[[Exception], httpx2.MockTransport]:
+    def _factory(exc: Exception):
+        def handler(_request: httpx2.Request) -> httpx2.Response:
+            raise exc
+
+        return httpx2.MockTransport(handler)
+
+    return _factory
+
+
+@pytest.fixture
+def mid_stream_raising_transport_factory() -> Callable[[Exception], httpx2.MockTransport]:
+    def _factory(exc: Exception):
+        def body():
+            yield b'first\n'
+            raise exc
+
+        def handler(_request: httpx2.Request) -> httpx2.Response:
+            return httpx2.Response(200, content=body())
+
+        return httpx2.MockTransport(handler)
+
+    return _factory
+
+
+@pytest.fixture
+def captured_requests() -> list[httpx2.Request]:
+    return []
+
+
+@pytest.fixture
+def capturing_transport(captured_requests: list[httpx2.Request]) -> httpx2.MockTransport:
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        _ = request.content
+        captured_requests.append(request)
+        return httpx2.Response(200, json={'ok': True})
+
+    return httpx2.MockTransport(handler)
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_auth_basic.py b/datadog_checks_base/tests/base/utils/httpx2/test_auth_basic.py
new file mode 100644
index 0000000000000..034ce388ffb52
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_auth_basic.py
@@ -0,0 +1,34 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import pytest
+
+from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper
+
+from .common import parse_basic_auth
+
+
+def test_basic_auth_sent_on_request(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper(
+        {'username': 'alice', 'password': 'secret'},
+        {},
+        transport=capturing_transport,
+    )
+    http.get('http://example.test/')
+    user, password = parse_basic_auth(captured_requests[0].headers['authorization'])
+    assert user == 'alice'
+    assert password == 'secret'
+
+
+@pytest.mark.parametrize(
+    'instance',
+    [
+        pytest.param({}, id='no-credentials'),
+        pytest.param({'username': 'alice'}, id='username-only'),
+        pytest.param({'password': 'secret'}, id='password-only'),
+    ],
+)
+def test_no_authorization_header_set(instance, capturing_transport, captured_requests):
+    http = HTTPX2Wrapper(instance, {}, transport=capturing_transport)
+    http.get('http://example.test/')
+    assert 'authorization' not in captured_requests[0].headers
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_config.py b/datadog_checks_base/tests/base/utils/httpx2/test_config.py
new file mode 100644
index 0000000000000..7bb5baa659587
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_config.py
@@ -0,0 +1,267 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import pytest
+
+from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper
+
+
+def test_default_headers_include_user_agent(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    assert any(key.lower() == 'user-agent' for key in http.options['headers'])
+
+
+def test_extra_headers_merge(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({'extra_headers': {'X-Extra': 'value'}}, {}, transport=capturing_transport)
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['x-extra'] == 'value'
+
+
+def test_headers_override_defaults(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({'headers': {'User-Agent': 'custom-agent/1.0'}}, {}, transport=capturing_transport)
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['user-agent'] == 'custom-agent/1.0'
+
+
+def test_per_request_headers_merge_into_request(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.get('http://example.test/', headers={'X-Per-Request': 'yes'})
+    assert captured_requests[0].headers['x-per-request'] == 'yes'
+
+
+def test_per_request_empty_headers_dict_preserves_client_defaults(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({'extra_headers': {'X-Default': 'kept'}}, {}, transport=capturing_transport)
+    http.get('http://example.test/', headers={})
+    assert captured_requests[0].headers['x-default'] == 'kept'
+
+
+@pytest.mark.parametrize(
+    'headers,extra_headers,canonical_key,expected_value',
+    [
+        pytest.param({'X-Foo': 'a'}, {'x-foo': 'b'}, 'x-foo', 'b', id='extra-wins-on-case-different-overlap'),
+        pytest.param(
+            {'Content-Type': 'application/json'},
+            {'content-type': 'text/plain'},
+            'content-type',
+            'text/plain',
+            id='extra-wins-same-case',
+        ),
+        pytest.param({'X-Foo': 'a'}, None, 'x-foo', 'a', id='headers-only'),
+        pytest.param(None, {'X-Foo': 'b'}, 'x-foo', 'b', id='extra-only'),
+    ],
+)
+def test_per_request_headers_extra_headers_case_fold(
+    capturing_transport, captured_requests, headers, extra_headers, canonical_key, expected_value
+):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    kwargs = {}
+    if headers is not None:
+        kwargs['headers'] = headers
+    if extra_headers is not None:
+        kwargs['extra_headers'] = extra_headers
+    http.get('http://example.test/', **kwargs)
+
+    sent = captured_requests[0].headers
+    # Single case-folded entry on the outgoing request.
+    assert sent.get_list(canonical_key) == [expected_value]
+
+
+def test_init_header_get_set_are_case_insensitive_with_case_different_overlap(capturing_transport):
+    http = HTTPX2Wrapper(
+        {'headers': {'X-Foo': 'a'}, 'extra_headers': {'x-foo': 'b'}},
+        {},
+        transport=capturing_transport,
+    )
+    assert http.get_header('X-FOO') == http.get_header('x-foo')
+    http.set_header('X-Foo', 'c')
+    assert http.get_header('X-Foo') == 'c'
+    assert http.get_header('x-foo') == 'c'
+
+
+def test_timeout_from_instance(capturing_transport):
+    http = HTTPX2Wrapper({'timeout': 25}, {}, transport=capturing_transport)
+    connect, read = http.options['timeout']
+    assert connect == 25.0
+    assert read == 25.0
+
+
+def test_connect_and_read_timeout_split(capturing_transport):
+    http = HTTPX2Wrapper({'connect_timeout': 5, 'read_timeout': 30}, {}, transport=capturing_transport)
+    connect, read = http.options['timeout']
+    assert connect == 5.0
+    assert read == 30.0
+
+
+@pytest.mark.parametrize(
+    'timeout_value,expected',
+    [
+        pytest.param(
+            None,
+            {'connect': None, 'read': None, 'write': None, 'pool': None},
+            id='none-means-no-timeout',
+        ),
+        pytest.param(
+            7,
+            {'connect': 7.0, 'read': 7.0, 'write': 7.0, 'pool': 7.0},
+            id='scalar-pins-all-phases',
+        ),
+        pytest.param(
+            (3, 5),
+            {'connect': 3.0, 'read': 5.0, 'write': 5.0, 'pool': 5.0},
+            id='tuple-connect-read-pins-write-pool',
+        ),
+    ],
+)
+def test_per_request_timeout_value_construction(capturing_transport, captured_requests, timeout_value, expected):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.get('http://example.test/', timeout=timeout_value)
+    assert captured_requests[0].extensions['timeout'] == expected
+
+
+def test_verify_defaults_to_true(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    assert http.options['verify'] is True
+
+
+def test_verify_false_when_tls_verify_off(capturing_transport):
+    http = HTTPX2Wrapper({'tls_verify': False}, {}, transport=capturing_transport)
+    assert http.options['verify'] is False
+
+
+def test_tls_ca_cert_uses_path(capturing_transport):
+    http = HTTPX2Wrapper({'tls_ca_cert': '/etc/ssl/ca.pem'}, {}, transport=capturing_transport)
+    assert http.options['verify'] == '/etc/ssl/ca.pem'
+
+
+def test_tls_client_cert_string(capturing_transport):
+    http = HTTPX2Wrapper({'tls_cert': '/etc/ssl/client.pem'}, {}, transport=capturing_transport)
+    assert http.options['cert'] == '/etc/ssl/client.pem'
+
+
+def test_tls_client_cert_with_key(capturing_transport):
+    http = HTTPX2Wrapper(
+        {'tls_cert': '/etc/ssl/client.pem', 'tls_private_key': '/etc/ssl/client.key'},
+        {},
+        transport=capturing_transport,
+    )
+    assert http.options['cert'] == ('/etc/ssl/client.pem', '/etc/ssl/client.key')
+
+
+def test_tls_no_cert_when_not_configured(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    assert http.options['cert'] is None
+
+
+def test_options_proxies_is_none(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    assert http.options['proxies'] is None
+
+
+@pytest.mark.parametrize(
+    'lookup_name,default,expected',
+    [
+        pytest.param('x-foo', None, 'bar', id='lowercase-lookup'),
+        pytest.param('X-FOO', None, 'bar', id='uppercase-lookup'),
+        pytest.param('missing', None, None, id='missing-no-default'),
+        pytest.param('missing', 'fallback', 'fallback', id='missing-with-default'),
+    ],
+)
+def test_get_header(capturing_transport, lookup_name, default, expected):
+    http = HTTPX2Wrapper({'extra_headers': {'X-Foo': 'bar'}}, {}, transport=capturing_transport)
+    assert http.get_header(lookup_name, default=default) == expected
+
+
+def test_set_header_overrides_existing(capturing_transport):
+    http = HTTPX2Wrapper({'extra_headers': {'X-Foo': 'bar'}}, {}, transport=capturing_transport)
+    http.set_header('X-FOO', 'new')
+    assert http.get_header('x-foo') == 'new'
+
+
+def test_set_header_propagates_to_outgoing_request(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({'extra_headers': {'X-Foo': 'bar'}}, {}, transport=capturing_transport)
+    http.set_header('X-FOO', 'updated')
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['x-foo'] == 'updated'
+
+
+def test_set_header_adds_new_header(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.set_header('X-New', 'value')
+    assert http.get_header('x-new') == 'value'
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['x-new'] == 'value'
+
+
+def test_remapper_renames_field(capturing_transport):
+    remapper = {'ssl_validation': {'name': 'tls_verify'}}
+    http = HTTPX2Wrapper({'ssl_validation': False}, {}, remapper=remapper, transport=capturing_transport)
+    assert http.options['verify'] is False
+
+
+def test_remapper_invert_flips_value(capturing_transport):
+    # disable_ssl_validation=True means "do NOT verify" -> options['verify'] is False.
+    remapper = {'disable_ssl_validation': {'name': 'tls_verify', 'invert': True}}
+    http = HTTPX2Wrapper({'disable_ssl_validation': True}, {}, remapper=remapper, transport=capturing_transport)
+    assert http.options['verify'] is False
+
+
+def test_remapper_invert_with_explicit_default(capturing_transport):
+    # Remapped field absent from instance, explicit default takes effect, then invert flips it.
+    remapper = {'disable_ssl_validation': {'name': 'tls_verify', 'invert': True, 'default': False}}
+    http = HTTPX2Wrapper({}, {}, remapper=remapper, transport=capturing_transport)
+    # default=False, invert flips to True -> verify is True.
+    assert http.options['verify'] is True
+
+
+def test_remapper_instance_wins_over_remapped_field(capturing_transport):
+    # If the standard field is present in instance, the remapped alternative is ignored.
+    remapper = {'ssl_validation': {'name': 'tls_verify'}}
+    http = HTTPX2Wrapper(
+        {'ssl_validation': False, 'tls_verify': True}, {}, remapper=remapper, transport=capturing_transport
+    )
+    assert http.options['verify'] is True
+
+
+def test_remapper_ignores_unknown_target_field(capturing_transport):
+    # Remapper targeting a non-STANDARD_FIELDS key is silently ignored.
+    remapper = {'some_alias': {'name': 'definitely_not_a_known_field'}}
+    http = HTTPX2Wrapper({'some_alias': 'ignored'}, {}, remapper=remapper, transport=capturing_transport)
+    assert 'definitely_not_a_known_field' not in http.options
+
+
+@pytest.mark.parametrize(
+    'kwarg,value,hint_substring',
+    [
+        pytest.param('proxies', {'http': 'http://proxy:8080'}, 'proxy support', id='proxies'),
+        pytest.param('allow_redirects', False, 'follow_redirects', id='allow-redirects-uses-httpx-name'),
+    ],
+)
+def test_request_rejects_unknown_kwarg(capturing_transport, kwarg, value, hint_substring):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError, match=kwarg) as excinfo:
+        http.get('http://example.test/', **{kwarg: value})
+    assert hint_substring in str(excinfo.value)
+
+
+def test_init_config_timeout_used_when_instance_has_none(capturing_transport):
+    http = HTTPX2Wrapper({}, {'timeout': 42}, transport=capturing_transport)
+    connect, read = http.options['timeout']
+    assert connect == 42.0
+    assert read == 42.0
+
+
+def test_instance_timeout_overrides_init_config_timeout(capturing_transport):
+    http = HTTPX2Wrapper({'timeout': 7}, {'timeout': 42}, transport=capturing_transport)
+    connect, read = http.options['timeout']
+    assert connect == 7.0
+    assert read == 7.0
+
+
+def test_init_config_log_requests_used_when_instance_has_none(capturing_transport):
+    http = HTTPX2Wrapper({}, {'log_requests': True}, transport=capturing_transport)
+    assert http._log_requests is True
+
+
+def test_instance_log_requests_overrides_init_config(capturing_transport):
+    http = HTTPX2Wrapper({'log_requests': False}, {'log_requests': True}, transport=capturing_transport)
+    assert http._log_requests is False
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_exceptions.py b/datadog_checks_base/tests/base/utils/httpx2/test_exceptions.py
new file mode 100644
index 0000000000000..893c8a25d8cad
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_exceptions.py
@@ -0,0 +1,95 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import httpx2
+import pytest
+
+from datadog_checks.base.utils.http_exceptions import (
+    HTTPConnectionError,
+    HTTPInvalidURLError,
+    HTTPRequestError,
+    HTTPTimeoutError,
+)
+from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper, _map_httpx2_exception
+
+
+@pytest.mark.parametrize(
+    'raised,expected',
+    [
+        pytest.param(httpx2.ConnectTimeout('boom'), HTTPTimeoutError, id='connect-timeout'),
+        pytest.param(httpx2.ReadTimeout('slow'), HTTPTimeoutError, id='read-timeout'),
+        pytest.param(httpx2.PoolTimeout('pool'), HTTPTimeoutError, id='pool-timeout'),
+        pytest.param(httpx2.ConnectError('refused'), HTTPConnectionError, id='connect-error'),
+        pytest.param(httpx2.ReadError('mid-stream'), HTTPConnectionError, id='read-error'),
+        pytest.param(httpx2.WriteError('broken-pipe'), HTTPConnectionError, id='write-error'),
+        pytest.param(httpx2.CloseError('half-closed'), HTTPConnectionError, id='close-error'),
+        pytest.param(httpx2.LocalProtocolError('bad'), HTTPRequestError, id='local-protocol-error'),
+        pytest.param(httpx2.RequestError('generic'), HTTPRequestError, id='request-error'),
+    ],
+)
+def test_request_exception_mapping(raising_transport_factory, raised, expected):
+    transport = raising_transport_factory(raised)
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    with pytest.raises(expected):
+        http.get('http://example.test/')
+
+
+def test_map_httpx2_exception_routes_invalid_url():
+    mapped = _map_httpx2_exception(httpx2.InvalidURL('bad url'))
+    assert isinstance(mapped, HTTPInvalidURLError)
+
+
+def test_request_raises_invalid_url_error(raising_transport_factory):
+    transport = raising_transport_factory(httpx2.InvalidURL('bad url'))
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    with pytest.raises(HTTPInvalidURLError):
+        http.get('http://example.test/')
+
+
+@pytest.mark.parametrize(
+    'raised,expected',
+    [
+        pytest.param(httpx2.ReadError('mid-stream'), HTTPConnectionError, id='read-error'),
+        pytest.param(httpx2.ReadTimeout('slow'), HTTPTimeoutError, id='read-timeout'),
+    ],
+)
+@pytest.mark.parametrize(
+    'iter_method,iter_kwargs',
+    [
+        pytest.param('iter_lines', {}, id='iter_lines'),
+        pytest.param('iter_lines', {'decode_unicode': True}, id='iter_lines-decoded'),
+        pytest.param('iter_content', {}, id='iter_content-bytes'),
+        pytest.param('iter_content', {'decode_unicode': True}, id='iter_content-decoded'),
+    ],
+)
+def test_iter_methods_map_mid_stream_exceptions(
+    mid_stream_raising_transport_factory, raised, expected, iter_method, iter_kwargs
+):
+    transport = mid_stream_raising_transport_factory(raised)
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    with pytest.raises(expected):
+        list(getattr(response, iter_method)(**iter_kwargs))
+
+
+@pytest.mark.parametrize(
+    'raised,expected',
+    [
+        pytest.param(httpx2.ReadError('mid-stream'), HTTPConnectionError, id='read-error'),
+        pytest.param(httpx2.ReadTimeout('slow'), HTTPTimeoutError, id='read-timeout'),
+    ],
+)
+@pytest.mark.parametrize(
+    'reader',
+    [
+        pytest.param(lambda r: r.content, id='content'),
+        pytest.param(lambda r: r.text, id='text'),
+        pytest.param(lambda r: r.json(), id='json'),
+    ],
+)
+def test_buffered_reads_map_exceptions(mid_stream_raising_transport_factory, raised, expected, reader):
+    transport = mid_stream_raising_transport_factory(raised)
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    with pytest.raises(expected):
+        reader(response)
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_lifecycle.py b/datadog_checks_base/tests/base/utils/httpx2/test_lifecycle.py
new file mode 100644
index 0000000000000..617e1225565ad
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_lifecycle.py
@@ -0,0 +1,91 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import sys
+
+import pytest
+
+from datadog_checks.base.utils import httpx2 as httpx2_module
+from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper
+
+
+def test_close_is_idempotent(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.close()
+    http.close()
+
+
+def test_context_manager(capturing_transport):
+    with HTTPX2Wrapper({}, {}, transport=capturing_transport) as http:
+        response = http.get('http://example.test/')
+    assert response.status_code == 200
+
+
+def test_request_closes_response_when_adapter_wrap_fails(capturing_transport, monkeypatch):
+    closed = []
+
+    def failing_init(self, response):
+        # Install the spy at adapter-init time so the close() calls that httpx2 makes
+        # while eagerly reading the in-memory MockTransport body during Response.__init__
+        # are not counted; only the production guard's close() is observed.
+        original_close = response.close
+        response.close = lambda: (closed.append(True), original_close())[-1]
+        raise RuntimeError('boom')
+
+    monkeypatch.setattr(httpx2_module.HTTPX2ResponseAdapter, '__init__', failing_init)
+
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(RuntimeError, match='boom'):
+        http.get('http://example.test/')
+
+    assert closed == [True]
+
+
+def test_module_import_fails_without_httpx2(monkeypatch):
+    monkeypatch.setitem(sys.modules, 'httpx2', None)
+    monkeypatch.delitem(sys.modules, 'datadog_checks.base.utils.httpx2', raising=False)
+    with pytest.raises(ImportError) as exc_info:
+        import datadog_checks.base.utils.httpx2  # noqa: F401
+    assert exc_info.value.name == 'httpx2'
+
+
+@pytest.mark.parametrize(
+    'instance,expected_cls_name',
+    [
+        pytest.param({'use_httpx2': True}, 'HTTPX2Wrapper', id='opt-in'),
+        pytest.param({'use_httpx2': False}, 'RequestsWrapper', id='explicit-default'),
+        pytest.param({}, 'RequestsWrapper', id='unset-default'),
+    ],
+)
+def test_agentcheck_http_dispatch(instance, expected_cls_name):
+    from datadog_checks.base import AgentCheck
+
+    check = AgentCheck('test', {}, [instance])
+    try:
+        assert type(check.http).__name__ == expected_cls_name
+    finally:
+        # Read the cached attribute directly (AgentCheck.http is a property that may rebuild).
+        http = getattr(check, '_http', None)
+        if http is not None:
+            close = getattr(http, 'close', None)
+            if close is not None:
+                close()
+
+
+def test_teardown_guard_preserves_build_error():
+    from datadog_checks.base import AgentCheck
+
+    class BoomCheck(AgentCheck):
+        def _build_http_client(self, instance):
+            raise RuntimeError('build failed')
+
+    check = BoomCheck('test', {}, [{}])
+    with pytest.raises(RuntimeError, match='build failed'):
+        try:
+            _ = check.http
+        finally:
+            http = getattr(check, '_http', None)
+            if http is not None:
+                close = getattr(http, 'close', None)
+                if close is not None:
+                    close()
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_methods.py b/datadog_checks_base/tests/base/utils/httpx2/test_methods.py
new file mode 100644
index 0000000000000..4d88506b516e1
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_methods.py
@@ -0,0 +1,72 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import json
+import logging
+
+import httpx2
+import pytest
+
+from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper
+
+HTTP_VERBS = {
+    'get': 'GET',
+    'post': 'POST',
+    'put': 'PUT',
+    'delete': 'DELETE',
+    'head': 'HEAD',
+    'patch': 'PATCH',
+    'options_method': 'OPTIONS',
+}
+
+
+@pytest.mark.parametrize('method,verb', HTTP_VERBS.items())
+def test_method_dispatches_with_correct_verb(method, verb, captured_requests, capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    fn = getattr(http, method)
+    response = fn('http://example.test/path', headers={'X-Test': '1'})
+
+    assert response.status_code == 200
+    assert captured_requests[0].method == verb
+
+
+def test_post_json_body_is_serialized(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.post('http://example.test/path', json={'a': 1, 'b': 'two'})
+    req = captured_requests[0]
+    assert req.headers['content-type'] == 'application/json'
+    assert json.loads(req.content) == {'a': 1, 'b': 'two'}
+
+
+def test_stream_kwarg_logged_and_dropped(capturing_transport, caplog):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with caplog.at_level(logging.DEBUG, logger='datadog_checks.base.utils.httpx2'):
+        response = http.get('http://example.test/', stream=True)
+    assert response.status_code == 200
+    assert any('dropping unsupported per-request kwarg: stream' in r.message for r in caplog.records)
+
+
+def _spy_on_send(http):
+    captured = {}
+    original_send = http._client.send
+
+    def spy(request, **kwargs):
+        captured.update(kwargs)
+        return original_send(request, **kwargs)
+
+    http._client.send = spy
+    return captured
+
+
+def test_request_passes_follow_redirects_per_request(capturing_transport):
+    http = HTTPX2Wrapper({'allow_redirects': False}, {}, transport=capturing_transport)
+    captured = _spy_on_send(http)
+    http.get('http://example.test/', follow_redirects=True)
+    assert captured['follow_redirects'] is True
+
+
+def test_request_uses_client_default_follow_redirects_when_omitted(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    captured = _spy_on_send(http)
+    http.get('http://example.test/')
+    assert captured['follow_redirects'] is httpx2.USE_CLIENT_DEFAULT
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_options_headers_mutation.py b/datadog_checks_base/tests/base/utils/httpx2/test_options_headers_mutation.py
new file mode 100644
index 0000000000000..ada7c1d22771f
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_options_headers_mutation.py
@@ -0,0 +1,69 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper
+
+
+def test_options_headers_setitem_reaches_wire(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.options['headers']['X-Token'] = 'after-init'
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['x-token'] == 'after-init'
+
+
+def test_options_headers_update_reaches_wire(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.options['headers'].update({'X-Token': 'after-init', 'X-Extra': 'two'})
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['x-token'] == 'after-init'
+    assert captured_requests[0].headers['x-extra'] == 'two'
+
+
+def test_options_headers_whole_dict_replace_reaches_wire(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.options['headers'] = {'X-Token': 'after-init'}
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['x-token'] == 'after-init'
+
+
+def test_options_headers_mid_stream_mutation_observed(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.options['headers']['X-Token'] = 'first'
+    http.get('http://example.test/')
+    http.options['headers']['X-Token'] = 'second'
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['x-token'] == 'first'
+    assert captured_requests[1].headers['x-token'] == 'second'
+
+
+def test_set_header_still_works_after_change(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.set_header('X-Token', 'set-via-helper')
+    http.get('http://example.test/')
+    assert captured_requests[0].headers['x-token'] == 'set-via-helper'
+
+
+def test_per_call_headers_override_options_headers(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.options['headers']['X-Token'] = 'wrapper-default'
+    http.get('http://example.test/', headers={'X-Token': 'per-call'})
+    assert captured_requests[0].headers['x-token'] == 'per-call'
+
+
+def test_extra_headers_override_per_call_headers_with_options_default(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.options['headers']['X-Token'] = 'wrapper-default'
+    http.get(
+        'http://example.test/',
+        headers={'X-Token': 'per-call'},
+        extra_headers={'X-Token': 'per-call-extra'},
+    )
+    assert captured_requests[0].headers['x-token'] == 'per-call-extra'
+
+
+def test_options_headers_case_insensitive_collapse_with_per_call(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.options['headers']['X-FOO'] = 'a'
+    http.get('http://example.test/', headers={'x-foo': 'b'})
+    sent = captured_requests[0].headers
+    assert sent.get_list('x-foo') == ['b']
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_request_auth.py b/datadog_checks_base/tests/base/utils/httpx2/test_request_auth.py
new file mode 100644
index 0000000000000..7092fe39276bc
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_request_auth.py
@@ -0,0 +1,66 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import httpx2
+import pytest
+
+from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper
+
+from .common import parse_basic_auth
+
+
+def test_per_request_auth_tuple_sets_basic_auth_header(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.get('http://example.test/', auth=('alice', 'secret'))
+    user, password = parse_basic_auth(captured_requests[0].headers['authorization'])
+    assert user == 'alice'
+    assert password == 'secret'
+
+
+def test_per_request_auth_basicauth_instance_sets_basic_auth_header(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.get('http://example.test/', auth=httpx2.BasicAuth('bob', 'hunter2'))
+    user, password = parse_basic_auth(captured_requests[0].headers['authorization'])
+    assert user == 'bob'
+    assert password == 'hunter2'
+
+
+def test_per_request_auth_shadows_client_level_auth(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper(
+        {'username': 'alice', 'password': 'secret'},
+        {},
+        transport=capturing_transport,
+    )
+    http.get('http://example.test/', auth=('carol', 'override'))
+    user, password = parse_basic_auth(captured_requests[0].headers['authorization'])
+    assert user == 'carol'
+    assert password == 'override'
+
+
+def test_per_request_auth_alongside_other_kwargs_reaches_wire(capturing_transport, captured_requests):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    http.get(
+        'http://example.test/resource',
+        auth=('alice', 'secret'),
+        params={'q': 'value'},
+        headers={'X-Trace': 'abc'},
+    )
+    request = captured_requests[0]
+    user, password = parse_basic_auth(request.headers['authorization'])
+    assert user == 'alice'
+    assert password == 'secret'
+    assert request.headers['x-trace'] == 'abc'
+    assert request.url.params['q'] == 'value'
+
+
+@pytest.mark.parametrize(
+    'auth_value',
+    [
+        pytest.param(('alice', 'secret'), id='tuple'),
+        pytest.param(httpx2.BasicAuth('alice', 'secret'), id='basicauth'),
+    ],
+)
+def test_per_request_auth_does_not_raise_typeerror(auth_value, capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    response = http.get('http://example.test/', auth=auth_value)
+    assert response.status_code == 200
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_response.py b/datadog_checks_base/tests/base/utils/httpx2/test_response.py
new file mode 100644
index 0000000000000..b99972484d6e9
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_response.py
@@ -0,0 +1,264 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import logging
+from collections.abc import Mapping
+from datetime import timedelta
+
+import httpx2
+import pytest
+
+from datadog_checks.base.utils.http_exceptions import HTTPStatusError
+from datadog_checks.base.utils.http_protocol import HTTPResponseProtocol
+from datadog_checks.base.utils.httpx2 import DEFAULT_CHUNK_SIZE, HTTPX2ResponseAdapter, HTTPX2Wrapper
+
+
+@pytest.mark.parametrize('status_code', [404, 500])
+def test_response_raise_for_status_raises_on_error_codes(status_transport_factory, status_code):
+    transport = status_transport_factory(status_code, b'')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    with pytest.raises(HTTPStatusError):
+        response.raise_for_status()
+
+
+def test_response_iter_content_bytes(status_transport_factory):
+    transport = status_transport_factory(200, b'abcdef')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    chunks = list(response.iter_content(chunk_size=2))
+    assert b''.join(chunks) == b'abcdef'
+
+
+def test_response_iter_content_decode_unicode(status_transport_factory):
+    transport = status_transport_factory(200, b'abcdef')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    chunks = list(response.iter_content(chunk_size=3, decode_unicode=True))
+    assert ''.join(chunks) == 'abcdef'
+
+
+@pytest.mark.parametrize(
+    'decode_unicode,expected',
+    [
+        pytest.param(False, [b'a', b'b', b'c'], id='bytes'),
+        pytest.param(True, ['a', 'b', 'c'], id='decoded-unicode'),
+    ],
+)
+def test_response_iter_lines(status_transport_factory, decode_unicode, expected):
+    transport = status_transport_factory(200, b'a\nb\nc')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert list(response.iter_lines(decode_unicode=decode_unicode)) == expected
+
+
+def test_response_iter_content_empty_body_yields_nothing(status_transport_factory):
+    transport = status_transport_factory(200, b'')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert list(response.iter_content()) == []
+
+
+def test_response_iter_content_default_chunk_size_uses_default(status_transport_factory):
+    body = b'X' * (DEFAULT_CHUNK_SIZE * 3 + 5)
+    transport = status_transport_factory(200, body)
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    chunks = list(response.iter_content())
+    assert b''.join(chunks) == body
+    assert all(len(chunk) <= DEFAULT_CHUNK_SIZE for chunk in chunks)
+    assert any(len(chunk) == DEFAULT_CHUNK_SIZE for chunk in chunks)
+
+
+@pytest.mark.parametrize(
+    'charset,raw,expected',
+    [
+        pytest.param('utf-8', 'café'.encode('utf-8'), 'café', id='utf-8'),
+        pytest.param('iso-8859-1', 'café'.encode('iso-8859-1'), 'café', id='iso-8859-1'),
+    ],
+)
+def test_response_iter_content_decode_uses_response_encoding(charset, raw, expected):
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, content=raw, headers={'Content-Type': f'text/plain; charset={charset}'})
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    chunks = list(response.iter_content(chunk_size=64, decode_unicode=True))
+    assert ''.join(chunks) == expected
+
+
+@pytest.mark.parametrize(
+    'charset,line',
+    [
+        pytest.param('utf-8', 'café', id='utf-8'),
+        pytest.param('iso-8859-1', 'café', id='iso-8859-1'),
+    ],
+)
+def test_response_iter_lines_bytes_path_is_encoding_agnostic(charset, line):
+    raw = (line + '\n' + line).encode(charset)
+
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, content=raw, headers={'Content-Type': f'text/plain; charset={charset}'})
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    encoded_lines = list(response.iter_lines(decode_unicode=False))
+    assert encoded_lines == [line.encode(charset), line.encode(charset)]
+
+
+def test_response_iter_lines_bytes_through_invalid_sequences():
+    raw = b'invalid\xff\xfe\nsecond\n'
+
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, content=raw, headers={'Content-Type': 'text/plain; charset=utf-8'})
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    assert list(response.iter_lines(decode_unicode=False)) == [b'invalid\xff\xfe', b'second']
+
+
+def test_response_iter_lines_bytes_through_crlf(status_transport_factory):
+    transport = status_transport_factory(200, b'a\r\nb\r\nc')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert list(response.iter_lines(decode_unicode=False)) == [b'a', b'b', b'c']
+
+
+def test_response_iter_lines_decoded_through_crlf(status_transport_factory):
+    transport = status_transport_factory(200, b'a\r\nb\r\nc')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert list(response.iter_lines(decode_unicode=True)) == ['a', 'b', 'c']
+
+
+def test_response_iter_lines_rejects_delimiter(status_transport_factory):
+    transport = status_transport_factory(200, b'a\nb\n')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    with pytest.raises(NotImplementedError):
+        list(response.iter_lines(delimiter=b'|'))
+
+
+def test_response_iter_lines_bytes_across_many_chunks():
+    chunk_size = 1024
+    num_chunks = 6
+    body = b''.join((b'X' * (chunk_size - 1) + b'\n') for _ in range(num_chunks)) + b'tail'
+
+    def streamed():
+        for i in range(0, len(body), chunk_size):
+            yield body[i : i + chunk_size]
+
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, content=streamed())
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    lines = list(response.iter_lines(decode_unicode=False))
+    assert lines == [b'X' * (chunk_size - 1)] * num_chunks + [b'tail']
+
+
+def test_response_elapsed_returns_zero_on_runtime_error(caplog):
+    # httpx2's .elapsed raises RuntimeError for in-memory responses. The real httpx2.Request is
+    # attached so the adapter's debug log can format request info without a secondary error.
+    request = httpx2.Request('GET', 'http://example.test/')
+    response = httpx2.Response(200, content=b'x', request=request)
+    adapter = HTTPX2ResponseAdapter(response)
+    with caplog.at_level(logging.DEBUG, logger='datadog_checks.base.utils.httpx2'):
+        assert adapter.elapsed == timedelta(0)
+    assert any('elapsed unavailable' in record.message for record in caplog.records)
+
+
+@pytest.mark.parametrize('status_code,expected_ok', [(200, True), (204, True), (301, True), (400, False), (500, False)])
+def test_response_ok_property(status_transport_factory, status_code, expected_ok):
+    transport = status_transport_factory(status_code, b'')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert response.ok is expected_ok
+
+
+def test_response_reason_from_httpx_response(status_transport_factory):
+    transport = status_transport_factory(200, b'')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert response.reason == 'OK'
+
+
+def test_response_text_decodes_body(status_transport_factory):
+    transport = status_transport_factory(200, b'hello')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert response.text == 'hello'
+
+
+def test_response_json_returns_decoded_object():
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, json={'a': 1})
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    assert response.json() == {'a': 1}
+
+
+def test_response_content_returns_raw_bytes(status_transport_factory):
+    transport = status_transport_factory(200, b'\x00\x01\x02')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert response.content == b'\x00\x01\x02'
+
+
+def test_response_headers_exposed():
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, headers={'X-Custom': 'value'}, content=b'')
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    assert response.headers['X-Custom'] == 'value'
+
+
+def test_response_url_reflects_request_url(status_transport_factory):
+    transport = status_transport_factory(200, b'')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/path')
+    assert str(response.url) == 'http://example.test/path'
+
+
+def test_response_encoding_property_reflects_declared_charset():
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, content=b'data', headers={'Content-Type': 'text/plain; charset=iso-8859-1'})
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    assert response.encoding == 'iso-8859-1'
+
+
+def test_response_encoding_property_settable_through_adapter():
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, content=b'data', headers={'Content-Type': 'text/plain; charset=utf-8'})
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    response.encoding = 'iso-8859-1'
+    assert response.encoding == 'iso-8859-1'
+
+
+def test_response_cookies_exposed():
+    def handler(_request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, headers={'Set-Cookie': 'session=abc123'}, content=b'')
+
+    http = HTTPX2Wrapper({}, {}, transport=httpx2.MockTransport(handler))
+    response = http.get('http://example.test/')
+    assert response.cookies['session'] == 'abc123'
+
+
+def test_response_cookies_is_mapping(status_transport_factory):
+    transport = status_transport_factory(200, b'')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert isinstance(response.cookies, Mapping)
+
+
+def test_response_adapter_satisfies_protocol(status_transport_factory):
+    transport = status_transport_factory(200, b'')
+    http = HTTPX2Wrapper({}, {}, transport=transport)
+    response = http.get('http://example.test/')
+    assert isinstance(response, HTTPResponseProtocol)
diff --git a/datadog_checks_base/tests/base/utils/httpx2/test_unknown_kwarg_hints.py b/datadog_checks_base/tests/base/utils/httpx2/test_unknown_kwarg_hints.py
new file mode 100644
index 0000000000000..22a13882a6d55
--- /dev/null
+++ b/datadog_checks_base/tests/base/utils/httpx2/test_unknown_kwarg_hints.py
@@ -0,0 +1,85 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import pytest
+
+from datadog_checks.base.utils.httpx2 import HTTPX2Wrapper
+
+
+def test_unknown_kwarg_typeerror_includes_kwarg_name(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError, match='persist'):
+        http.get('http://example.test/', persist=True)
+
+
+def test_unknown_kwarg_typeerror_includes_remediation_for_verify(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError) as excinfo:
+        http.get('http://example.test/', verify=False)
+    message = str(excinfo.value)
+    assert 'verify' in message
+    assert 'tls_verify' in message
+
+
+def test_unknown_kwarg_typeerror_includes_remediation_for_persist(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError) as excinfo:
+        http.get('http://example.test/', persist=True)
+    message = str(excinfo.value)
+    assert 'persist' in message
+    assert 'pools connections' in message
+
+
+def test_unknown_kwarg_typeerror_includes_remediation_for_cert(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError) as excinfo:
+        http.get('http://example.test/', cert='/etc/ssl/client.pem')
+    message = str(excinfo.value)
+    assert 'cert' in message
+    assert 'tls_cert' in message
+    assert 'tls_private_key' in message
+
+
+def test_unknown_kwarg_typeerror_includes_remediation_for_proxies(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError) as excinfo:
+        http.get('http://example.test/', proxies={'http': 'http://proxy:8080'})
+    message = str(excinfo.value)
+    assert 'proxies' in message
+    assert 'proxy support' in message
+
+
+def test_unknown_kwarg_typeerror_includes_remediation_for_allow_redirects(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError) as excinfo:
+        http.get('http://example.test/', allow_redirects=False)
+    message = str(excinfo.value)
+    assert 'allow_redirects' in message
+    assert 'follow_redirects' in message
+
+
+def test_unknown_kwarg_without_hint_falls_back_gracefully(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError) as excinfo:
+        http.get('http://example.test/', foobar=1)
+    message = str(excinfo.value)
+    assert 'foobar' in message
+    assert 'tls_verify' not in message
+    assert 'pools connections' not in message
+
+
+def test_unknown_kwarg_multi_lists_each_hint(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    with pytest.raises(TypeError) as excinfo:
+        http.get('http://example.test/', verify=False, persist=True)
+    message = str(excinfo.value)
+    assert 'verify' in message
+    assert 'persist' in message
+    assert 'tls_verify' in message
+    assert 'pools connections' in message
+
+
+def test_stream_kwarg_still_silently_dropped(capturing_transport):
+    http = HTTPX2Wrapper({}, {}, transport=capturing_transport)
+    response = http.get('http://example.test/', stream=True)
+    assert response.status_code == 200