diff --git a/gapic/templates/%namespace/%name_%version/%sub/__init__.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/__init__.py.j2
index 8ee1670431..fcde4fac27 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/__init__.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/__init__.py.j2
@@ -7,6 +7,84 @@ from {{package_path}} import gapic_version as package_version
 
 __version__ = package_version.__version__
 
+
+import google.api_core as api_core
+
+if hasattr(api_core, "check_python_version") and hasattr(api_core, "check_dependency_versions"):   # pragma: NO COVER
+    {# TODO(api_core): remove `type:ignore` below when minimum version of api_core makes the else clause unnecessary. #}
+    api_core.check_python_version("{{package_path}}") # type: ignore
+    api_core.check_dependency_versions("{{package_path}}") # type: ignore
+else:   # pragma: NO COVER
+{# TODO(api_core): Remove this try-catch when we require api-core at a version that
+   supports the changes in https://github.com/googleapis/python-api-core/pull/832
+
+    In the meantime, please ensure the functionality here mirrors the
+    equivalent functionality in api_core, in those two functions above.
+#}
+    # An older version of api_core is installed, which does not define the
+    # functions above. We do equivalent checks manually.
+
+    import warnings
+    import sys
+
+    _py_version_str = sys.version.split()[0]
+    _package_label = "{{package_path}}"
+    if sys.version_info < (3, 9):
+        warnings.warn("You are using a non-supported Python version " +
+                      f"({_py_version_str}).  Google will not post any further " +
+                      f"updates to {_package_label} supporting this Python version. " +
+                      "Please upgrade to the latest Python version, or at " +
+                      f"least to Python 3.9, and then update {_package_label}.",
+                      FutureWarning)
+    if sys.version_info[:2] == (3, 9):
+        warnings.warn(f"You are using a Python version ({_py_version_str}) " +
+                      f"which Google will stop supporting in {_package_label} when " +
+                      "it reaches its end of life (October 2025). Please " +
+                      "upgrade to the latest Python version, or at " +
+                      "least Python 3.10, before then, and " +
+                      f"then update {_package_label}.",
+                      FutureWarning)
+
+    from packaging.version import parse as parse_version
+
+    if sys.version_info < (3, 8):
+        import pkg_resources
+
+        def _get_version(dependency_name):
+          try:
+            version_string = pkg_resources.get_distribution(dependency_name).version
+            return (parse_version(version_string), version_string)
+          except pkg_resources.DistributionNotFound:
+            return (None, "--")
+    else:
+        from importlib import metadata
+
+        def _get_version(dependency_name):
+            try:
+                version_string = metadata.version("requests")
+                parsed_version = parse_version(version_string)
+                return (parsed_version.release, version_string)
+            except metadata.PackageNotFoundError:
+                return (None, "--")
+
+    _dependency_package = "google.protobuf"
+    _next_supported_version = "4.25.8"
+    _next_supported_version_tuple = (4, 25, 8)
+    (_version_used, _version_used_string) = _get_version(_dependency_package)
+    if _version_used and _version_used < _next_supported_version_tuple:
+        warnings.warn(f"Package {_package_label} depends on " +
+                      f"{_dependency_package}, currently installed at version " +
+                      f"{_version_used_string}. Future updates to " +
+                      f"{_package_label} will require {_dependency_package} at " +
+                      f"version {_next_supported_version} or higher. Please ensure " +
+                      "that either (a) your Python environment doesn't pin the " +
+                      f"version of {_dependency_package}, so that updates to " +
+                      f"{_package_label} can require the higher version, or " +
+                      "(b) you manually update your Python environment to use at " +
+                      f"least version {_next_supported_version} of " +
+                      f"{_dependency_package}.",
+                      FutureWarning)
+
 {#  Import subpackages. -#}
 {% for subpackage, _ in api.subpackages|dictsort %}
 from . import {{ subpackage }}
diff --git a/gapic/templates/setup.py.j2 b/gapic/templates/setup.py.j2
index 2f265dd994..79ffd1885a 100644
--- a/gapic/templates/setup.py.j2
+++ b/gapic/templates/setup.py.j2
@@ -37,6 +37,7 @@ dependencies = [
     # Exclude incompatible versions of `google-auth`
     # See https://github.com/googleapis/google-cloud-python/issues/12364
     "google-auth >= 2.14.1, <3.0.0,!=2.24.0,!=2.25.0",
+    "packaging", # TODO: Remove once we require versions of api core that include this
     "proto-plus >= 1.22.3, <2.0.0",
     "proto-plus >= 1.25.0, <2.0.0; python_version >= '3.13'",
     {# Explicitly exclude protobuf versions mentioned in https://cloud.google.com/support/bulletins#GCP-2022-019 #}
diff --git a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index 31ab20095b..6391c95768 100644
--- a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -1472,6 +1472,7 @@ def test_{{ service.name|snake_case }}_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.{{ service.grpc_transport_name }}, transports.{{ service.grpc_asyncio_transport_name }}])
 def test_{{ service.name|snake_case }}_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/noxfile.py b/noxfile.py
index 2d688e139b..870949fcc5 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -60,6 +60,8 @@ def unit(session):
         "pyfakefs",
         "grpcio-status",
         "proto-plus",
+        "setuptools",  # TODO: Remove when not needed in __init__.py.j2
+        "packaging",  # TODO: Remove when not needed in __init__.py.j2
     )
     session.install("-e", ".")
     session.run(
@@ -481,6 +483,8 @@ def run_showcase_unit_tests(session, fail_under=100, rest_async_io_enabled=False
         "pytest-xdist",
         "asyncmock; python_version < '3.8'",
         "pytest-asyncio",
+        "setuptools",  # TODO: Remove when not needed in __init__.py.j2
+        "packaging",  # TODO: Remove when not needed in __init__.py.j2
     )
     # Run the tests.
     # NOTE: async rest is not supported against the minimum supported version of google-api-core.
@@ -595,6 +599,8 @@ def showcase_mypy(
         "types-protobuf",
         "types-requests",
         "types-dataclasses",
+        "setuptools",  # TODO: Remove when not needed in __init__.py.j2
+        "packaging",  # TODO: Remove when not needed in __init__.py.j2
     )
 
     with showcase_library(session, templates=templates, other_opts=other_opts) as lib:
@@ -725,6 +731,8 @@ def mypy(session):
         "types-PyYAML",
         "types-dataclasses",
         "click==8.1.3",
+        "setuptools",  # TODO: Remove when not needed in __init__.py.j2
+        "packaging",  # TODO: Remove when not needed in __init__.py.j2
     )
     session.install(".")
     session.run("mypy", "-p", "gapic")
diff --git a/tests/integration/goldens/asset/google/cloud/asset_v1/__init__.py b/tests/integration/goldens/asset/google/cloud/asset_v1/__init__.py
index 03ee9ec521..bcf3b204be 100755
--- a/tests/integration/goldens/asset/google/cloud/asset_v1/__init__.py
+++ b/tests/integration/goldens/asset/google/cloud/asset_v1/__init__.py
@@ -18,6 +18,77 @@
 __version__ = package_version.__version__
 
 
+import google.api_core as api_core
+
+if hasattr(api_core, "check_python_version") and hasattr(api_core, "check_dependency_versions"):   # pragma: NO COVER
+    api_core.check_python_version("google.cloud.asset_v1") # type: ignore
+    api_core.check_dependency_versions("google.cloud.asset_v1") # type: ignore
+else:   # pragma: NO COVER
+    # An older version of api_core is installed, which does not define the
+    # functions above. We do equivalent checks manually.
+
+    import warnings
+    import sys
+
+    _py_version_str = sys.version.split()[0]
+    _package_label = "google.cloud.asset_v1"
+    if sys.version_info < (3, 9):
+        warnings.warn("You are using a non-supported Python version " +
+                      f"({_py_version_str}).  Google will not post any further " +
+                      f"updates to {_package_label} supporting this Python version. " +
+                      "Please upgrade to the latest Python version, or at " +
+                      f"least to Python 3.9, and then update {_package_label}.",
+                      FutureWarning)
+    if sys.version_info[:2] == (3, 9):
+        warnings.warn(f"You are using a Python version ({_py_version_str}) " +
+                      f"which Google will stop supporting in {_package_label} when " +
+                      "it reaches its end of life (October 2025). Please " +
+                      "upgrade to the latest Python version, or at " +
+                      "least Python 3.10, before then, and " +
+                      f"then update {_package_label}.",
+                      FutureWarning)
+
+    from packaging.version import parse as parse_version
+
+    if sys.version_info < (3, 8):
+        import pkg_resources
+
+        def _get_version(dependency_name):
+          try:
+            version_string = pkg_resources.get_distribution(dependency_name).version
+            return (parse_version(version_string), version_string)
+          except pkg_resources.DistributionNotFound:
+            return (None, "--")
+    else:
+        from importlib import metadata
+
+        def _get_version(dependency_name):
+            try:
+                version_string = metadata.version("requests")
+                parsed_version = parse_version(version_string)
+                return (parsed_version.release, version_string)
+            except metadata.PackageNotFoundError:
+                return (None, "--")
+
+    _dependency_package = "google.protobuf"
+    _next_supported_version = "4.25.8"
+    _next_supported_version_tuple = (4, 25, 8)
+    (_version_used, _version_used_string) = _get_version(_dependency_package)
+    if _version_used and _version_used < _next_supported_version_tuple:
+        warnings.warn(f"Package {_package_label} depends on " +
+                      f"{_dependency_package}, currently installed at version " +
+                      f"{_version_used_string}. Future updates to " +
+                      f"{_package_label} will require {_dependency_package} at " +
+                      f"version {_next_supported_version} or higher. Please ensure " +
+                      "that either (a) your Python environment doesn't pin the " +
+                      f"version of {_dependency_package}, so that updates to " +
+                      f"{_package_label} can require the higher version, or " +
+                      "(b) you manually update your Python environment to use at " +
+                      f"least version {_next_supported_version} of " +
+                      f"{_dependency_package}.",
+                      FutureWarning)
+
+
 from .services.asset_service import AssetServiceClient
 from .services.asset_service import AssetServiceAsyncClient
 
diff --git a/tests/integration/goldens/asset/setup.py b/tests/integration/goldens/asset/setup.py
index 5d506dc2d9..2ae1de860b 100755
--- a/tests/integration/goldens/asset/setup.py
+++ b/tests/integration/goldens/asset/setup.py
@@ -43,6 +43,7 @@
     # Exclude incompatible versions of `google-auth`
     # See https://github.com/googleapis/google-cloud-python/issues/12364
     "google-auth >= 2.14.1, <3.0.0,!=2.24.0,!=2.25.0",
+    "packaging", # TODO: Remove once we require versions of api core that include this
     "proto-plus >= 1.22.3, <2.0.0",
     "proto-plus >= 1.25.0, <2.0.0; python_version >= '3.13'",
     "protobuf>=3.20.2,<7.0.0,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5",
diff --git a/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py b/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
index 10cb01ca87..edd142a808 100755
--- a/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
+++ b/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py
@@ -17712,6 +17712,7 @@ def test_asset_service_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.AssetServiceGrpcTransport, transports.AssetServiceGrpcAsyncIOTransport])
 def test_asset_service_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/credentials/google/iam/credentials_v1/__init__.py b/tests/integration/goldens/credentials/google/iam/credentials_v1/__init__.py
index 2be4471eb8..f4f6d88262 100755
--- a/tests/integration/goldens/credentials/google/iam/credentials_v1/__init__.py
+++ b/tests/integration/goldens/credentials/google/iam/credentials_v1/__init__.py
@@ -18,6 +18,77 @@
 __version__ = package_version.__version__
 
 
+import google.api_core as api_core
+
+if hasattr(api_core, "check_python_version") and hasattr(api_core, "check_dependency_versions"):   # pragma: NO COVER
+    api_core.check_python_version("google.iam.credentials_v1") # type: ignore
+    api_core.check_dependency_versions("google.iam.credentials_v1") # type: ignore
+else:   # pragma: NO COVER
+    # An older version of api_core is installed, which does not define the
+    # functions above. We do equivalent checks manually.
+
+    import warnings
+    import sys
+
+    _py_version_str = sys.version.split()[0]
+    _package_label = "google.iam.credentials_v1"
+    if sys.version_info < (3, 9):
+        warnings.warn("You are using a non-supported Python version " +
+                      f"({_py_version_str}).  Google will not post any further " +
+                      f"updates to {_package_label} supporting this Python version. " +
+                      "Please upgrade to the latest Python version, or at " +
+                      f"least to Python 3.9, and then update {_package_label}.",
+                      FutureWarning)
+    if sys.version_info[:2] == (3, 9):
+        warnings.warn(f"You are using a Python version ({_py_version_str}) " +
+                      f"which Google will stop supporting in {_package_label} when " +
+                      "it reaches its end of life (October 2025). Please " +
+                      "upgrade to the latest Python version, or at " +
+                      "least Python 3.10, before then, and " +
+                      f"then update {_package_label}.",
+                      FutureWarning)
+
+    from packaging.version import parse as parse_version
+
+    if sys.version_info < (3, 8):
+        import pkg_resources
+
+        def _get_version(dependency_name):
+          try:
+            version_string = pkg_resources.get_distribution(dependency_name).version
+            return (parse_version(version_string), version_string)
+          except pkg_resources.DistributionNotFound:
+            return (None, "--")
+    else:
+        from importlib import metadata
+
+        def _get_version(dependency_name):
+            try:
+                version_string = metadata.version("requests")
+                parsed_version = parse_version(version_string)
+                return (parsed_version.release, version_string)
+            except metadata.PackageNotFoundError:
+                return (None, "--")
+
+    _dependency_package = "google.protobuf"
+    _next_supported_version = "4.25.8"
+    _next_supported_version_tuple = (4, 25, 8)
+    (_version_used, _version_used_string) = _get_version(_dependency_package)
+    if _version_used and _version_used < _next_supported_version_tuple:
+        warnings.warn(f"Package {_package_label} depends on " +
+                      f"{_dependency_package}, currently installed at version " +
+                      f"{_version_used_string}. Future updates to " +
+                      f"{_package_label} will require {_dependency_package} at " +
+                      f"version {_next_supported_version} or higher. Please ensure " +
+                      "that either (a) your Python environment doesn't pin the " +
+                      f"version of {_dependency_package}, so that updates to " +
+                      f"{_package_label} can require the higher version, or " +
+                      "(b) you manually update your Python environment to use at " +
+                      f"least version {_next_supported_version} of " +
+                      f"{_dependency_package}.",
+                      FutureWarning)
+
+
 from .services.iam_credentials import IAMCredentialsClient
 from .services.iam_credentials import IAMCredentialsAsyncClient
 
diff --git a/tests/integration/goldens/credentials/setup.py b/tests/integration/goldens/credentials/setup.py
index 6ffe75b3ba..e1728860d3 100755
--- a/tests/integration/goldens/credentials/setup.py
+++ b/tests/integration/goldens/credentials/setup.py
@@ -43,6 +43,7 @@
     # Exclude incompatible versions of `google-auth`
     # See https://github.com/googleapis/google-cloud-python/issues/12364
     "google-auth >= 2.14.1, <3.0.0,!=2.24.0,!=2.25.0",
+    "packaging", # TODO: Remove once we require versions of api core that include this
     "proto-plus >= 1.22.3, <2.0.0",
     "proto-plus >= 1.25.0, <2.0.0; python_version >= '3.13'",
     "protobuf>=3.20.2,<7.0.0,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5",
diff --git a/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py b/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
index 62b329f3e5..90ddb8c449 100755
--- a/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
+++ b/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py
@@ -3972,6 +3972,7 @@ def test_iam_credentials_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.IAMCredentialsGrpcTransport, transports.IAMCredentialsGrpcAsyncIOTransport])
 def test_iam_credentials_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/__init__.py b/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/__init__.py
index a343867284..ff2ffc509d 100755
--- a/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/__init__.py
+++ b/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/__init__.py
@@ -18,6 +18,77 @@
 __version__ = package_version.__version__
 
 
+import google.api_core as api_core
+
+if hasattr(api_core, "check_python_version") and hasattr(api_core, "check_dependency_versions"):   # pragma: NO COVER
+    api_core.check_python_version("google.cloud.eventarc_v1") # type: ignore
+    api_core.check_dependency_versions("google.cloud.eventarc_v1") # type: ignore
+else:   # pragma: NO COVER
+    # An older version of api_core is installed, which does not define the
+    # functions above. We do equivalent checks manually.
+
+    import warnings
+    import sys
+
+    _py_version_str = sys.version.split()[0]
+    _package_label = "google.cloud.eventarc_v1"
+    if sys.version_info < (3, 9):
+        warnings.warn("You are using a non-supported Python version " +
+                      f"({_py_version_str}).  Google will not post any further " +
+                      f"updates to {_package_label} supporting this Python version. " +
+                      "Please upgrade to the latest Python version, or at " +
+                      f"least to Python 3.9, and then update {_package_label}.",
+                      FutureWarning)
+    if sys.version_info[:2] == (3, 9):
+        warnings.warn(f"You are using a Python version ({_py_version_str}) " +
+                      f"which Google will stop supporting in {_package_label} when " +
+                      "it reaches its end of life (October 2025). Please " +
+                      "upgrade to the latest Python version, or at " +
+                      "least Python 3.10, before then, and " +
+                      f"then update {_package_label}.",
+                      FutureWarning)
+
+    from packaging.version import parse as parse_version
+
+    if sys.version_info < (3, 8):
+        import pkg_resources
+
+        def _get_version(dependency_name):
+          try:
+            version_string = pkg_resources.get_distribution(dependency_name).version
+            return (parse_version(version_string), version_string)
+          except pkg_resources.DistributionNotFound:
+            return (None, "--")
+    else:
+        from importlib import metadata
+
+        def _get_version(dependency_name):
+            try:
+                version_string = metadata.version("requests")
+                parsed_version = parse_version(version_string)
+                return (parsed_version.release, version_string)
+            except metadata.PackageNotFoundError:
+                return (None, "--")
+
+    _dependency_package = "google.protobuf"
+    _next_supported_version = "4.25.8"
+    _next_supported_version_tuple = (4, 25, 8)
+    (_version_used, _version_used_string) = _get_version(_dependency_package)
+    if _version_used and _version_used < _next_supported_version_tuple:
+        warnings.warn(f"Package {_package_label} depends on " +
+                      f"{_dependency_package}, currently installed at version " +
+                      f"{_version_used_string}. Future updates to " +
+                      f"{_package_label} will require {_dependency_package} at " +
+                      f"version {_next_supported_version} or higher. Please ensure " +
+                      "that either (a) your Python environment doesn't pin the " +
+                      f"version of {_dependency_package}, so that updates to " +
+                      f"{_package_label} can require the higher version, or " +
+                      "(b) you manually update your Python environment to use at " +
+                      f"least version {_next_supported_version} of " +
+                      f"{_dependency_package}.",
+                      FutureWarning)
+
+
 from .services.eventarc import EventarcClient
 from .services.eventarc import EventarcAsyncClient
 
diff --git a/tests/integration/goldens/eventarc/setup.py b/tests/integration/goldens/eventarc/setup.py
index 4cba69da02..50e26c8ca3 100755
--- a/tests/integration/goldens/eventarc/setup.py
+++ b/tests/integration/goldens/eventarc/setup.py
@@ -43,6 +43,7 @@
     # Exclude incompatible versions of `google-auth`
     # See https://github.com/googleapis/google-cloud-python/issues/12364
     "google-auth >= 2.14.1, <3.0.0,!=2.24.0,!=2.25.0",
+    "packaging", # TODO: Remove once we require versions of api core that include this
     "proto-plus >= 1.22.3, <2.0.0",
     "proto-plus >= 1.25.0, <2.0.0; python_version >= '3.13'",
     "protobuf>=3.20.2,<7.0.0,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5",
diff --git a/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py b/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py
index 66ac1829f1..c815caafc5 100755
--- a/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py
+++ b/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py
@@ -15378,6 +15378,7 @@ def test_eventarc_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.EventarcGrpcTransport, transports.EventarcGrpcAsyncIOTransport])
 def test_eventarc_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/__init__.py b/tests/integration/goldens/logging/google/cloud/logging_v2/__init__.py
index e8b59e0bb5..6bc52ef1fd 100755
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/__init__.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/__init__.py
@@ -18,6 +18,77 @@
 __version__ = package_version.__version__
 
 
+import google.api_core as api_core
+
+if hasattr(api_core, "check_python_version") and hasattr(api_core, "check_dependency_versions"):   # pragma: NO COVER
+    api_core.check_python_version("google.cloud.logging_v2") # type: ignore
+    api_core.check_dependency_versions("google.cloud.logging_v2") # type: ignore
+else:   # pragma: NO COVER
+    # An older version of api_core is installed, which does not define the
+    # functions above. We do equivalent checks manually.
+
+    import warnings
+    import sys
+
+    _py_version_str = sys.version.split()[0]
+    _package_label = "google.cloud.logging_v2"
+    if sys.version_info < (3, 9):
+        warnings.warn("You are using a non-supported Python version " +
+                      f"({_py_version_str}).  Google will not post any further " +
+                      f"updates to {_package_label} supporting this Python version. " +
+                      "Please upgrade to the latest Python version, or at " +
+                      f"least to Python 3.9, and then update {_package_label}.",
+                      FutureWarning)
+    if sys.version_info[:2] == (3, 9):
+        warnings.warn(f"You are using a Python version ({_py_version_str}) " +
+                      f"which Google will stop supporting in {_package_label} when " +
+                      "it reaches its end of life (October 2025). Please " +
+                      "upgrade to the latest Python version, or at " +
+                      "least Python 3.10, before then, and " +
+                      f"then update {_package_label}.",
+                      FutureWarning)
+
+    from packaging.version import parse as parse_version
+
+    if sys.version_info < (3, 8):
+        import pkg_resources
+
+        def _get_version(dependency_name):
+          try:
+            version_string = pkg_resources.get_distribution(dependency_name).version
+            return (parse_version(version_string), version_string)
+          except pkg_resources.DistributionNotFound:
+            return (None, "--")
+    else:
+        from importlib import metadata
+
+        def _get_version(dependency_name):
+            try:
+                version_string = metadata.version("requests")
+                parsed_version = parse_version(version_string)
+                return (parsed_version.release, version_string)
+            except metadata.PackageNotFoundError:
+                return (None, "--")
+
+    _dependency_package = "google.protobuf"
+    _next_supported_version = "4.25.8"
+    _next_supported_version_tuple = (4, 25, 8)
+    (_version_used, _version_used_string) = _get_version(_dependency_package)
+    if _version_used and _version_used < _next_supported_version_tuple:
+        warnings.warn(f"Package {_package_label} depends on " +
+                      f"{_dependency_package}, currently installed at version " +
+                      f"{_version_used_string}. Future updates to " +
+                      f"{_package_label} will require {_dependency_package} at " +
+                      f"version {_next_supported_version} or higher. Please ensure " +
+                      "that either (a) your Python environment doesn't pin the " +
+                      f"version of {_dependency_package}, so that updates to " +
+                      f"{_package_label} can require the higher version, or " +
+                      "(b) you manually update your Python environment to use at " +
+                      f"least version {_next_supported_version} of " +
+                      f"{_dependency_package}.",
+                      FutureWarning)
+
+
 from .services.config_service_v2 import ConfigServiceV2Client
 from .services.config_service_v2 import ConfigServiceV2AsyncClient
 from .services.logging_service_v2 import LoggingServiceV2Client
diff --git a/tests/integration/goldens/logging/setup.py b/tests/integration/goldens/logging/setup.py
index 23a3a12a2b..f991eec737 100755
--- a/tests/integration/goldens/logging/setup.py
+++ b/tests/integration/goldens/logging/setup.py
@@ -43,6 +43,7 @@
     # Exclude incompatible versions of `google-auth`
     # See https://github.com/googleapis/google-cloud-python/issues/12364
     "google-auth >= 2.14.1, <3.0.0,!=2.24.0,!=2.25.0",
+    "packaging", # TODO: Remove once we require versions of api core that include this
     "proto-plus >= 1.22.3, <2.0.0",
     "proto-plus >= 1.25.0, <2.0.0; python_version >= '3.13'",
     "protobuf>=3.20.2,<7.0.0,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5",
diff --git a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
index 5de14eaafe..c284e58f8b 100755
--- a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
+++ b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py
@@ -12897,6 +12897,7 @@ def test_config_service_v2_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.ConfigServiceV2GrpcTransport, transports.ConfigServiceV2GrpcAsyncIOTransport])
 def test_config_service_v2_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
index 195d456017..3a3740300b 100755
--- a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
+++ b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py
@@ -3445,6 +3445,7 @@ def test_logging_service_v2_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.LoggingServiceV2GrpcTransport, transports.LoggingServiceV2GrpcAsyncIOTransport])
 def test_logging_service_v2_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
index a7b6c70e07..4633e315a1 100755
--- a/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
+++ b/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
@@ -3249,6 +3249,7 @@ def test_metrics_service_v2_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.MetricsServiceV2GrpcTransport, transports.MetricsServiceV2GrpcAsyncIOTransport])
 def test_metrics_service_v2_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/logging_internal/google/cloud/logging_v2/__init__.py b/tests/integration/goldens/logging_internal/google/cloud/logging_v2/__init__.py
index 3e0f9c2155..3bc21f629e 100755
--- a/tests/integration/goldens/logging_internal/google/cloud/logging_v2/__init__.py
+++ b/tests/integration/goldens/logging_internal/google/cloud/logging_v2/__init__.py
@@ -18,6 +18,77 @@
 __version__ = package_version.__version__
 
 
+import google.api_core as api_core
+
+if hasattr(api_core, "check_python_version") and hasattr(api_core, "check_dependency_versions"):   # pragma: NO COVER
+    api_core.check_python_version("google.cloud.logging_v2") # type: ignore
+    api_core.check_dependency_versions("google.cloud.logging_v2") # type: ignore
+else:   # pragma: NO COVER
+    # An older version of api_core is installed, which does not define the
+    # functions above. We do equivalent checks manually.
+
+    import warnings
+    import sys
+
+    _py_version_str = sys.version.split()[0]
+    _package_label = "google.cloud.logging_v2"
+    if sys.version_info < (3, 9):
+        warnings.warn("You are using a non-supported Python version " +
+                      f"({_py_version_str}).  Google will not post any further " +
+                      f"updates to {_package_label} supporting this Python version. " +
+                      "Please upgrade to the latest Python version, or at " +
+                      f"least to Python 3.9, and then update {_package_label}.",
+                      FutureWarning)
+    if sys.version_info[:2] == (3, 9):
+        warnings.warn(f"You are using a Python version ({_py_version_str}) " +
+                      f"which Google will stop supporting in {_package_label} when " +
+                      "it reaches its end of life (October 2025). Please " +
+                      "upgrade to the latest Python version, or at " +
+                      "least Python 3.10, before then, and " +
+                      f"then update {_package_label}.",
+                      FutureWarning)
+
+    from packaging.version import parse as parse_version
+
+    if sys.version_info < (3, 8):
+        import pkg_resources
+
+        def _get_version(dependency_name):
+          try:
+            version_string = pkg_resources.get_distribution(dependency_name).version
+            return (parse_version(version_string), version_string)
+          except pkg_resources.DistributionNotFound:
+            return (None, "--")
+    else:
+        from importlib import metadata
+
+        def _get_version(dependency_name):
+            try:
+                version_string = metadata.version("requests")
+                parsed_version = parse_version(version_string)
+                return (parsed_version.release, version_string)
+            except metadata.PackageNotFoundError:
+                return (None, "--")
+
+    _dependency_package = "google.protobuf"
+    _next_supported_version = "4.25.8"
+    _next_supported_version_tuple = (4, 25, 8)
+    (_version_used, _version_used_string) = _get_version(_dependency_package)
+    if _version_used and _version_used < _next_supported_version_tuple:
+        warnings.warn(f"Package {_package_label} depends on " +
+                      f"{_dependency_package}, currently installed at version " +
+                      f"{_version_used_string}. Future updates to " +
+                      f"{_package_label} will require {_dependency_package} at " +
+                      f"version {_next_supported_version} or higher. Please ensure " +
+                      "that either (a) your Python environment doesn't pin the " +
+                      f"version of {_dependency_package}, so that updates to " +
+                      f"{_package_label} can require the higher version, or " +
+                      "(b) you manually update your Python environment to use at " +
+                      f"least version {_next_supported_version} of " +
+                      f"{_dependency_package}.",
+                      FutureWarning)
+
+
 from .services.config_service_v2 import BaseConfigServiceV2Client
 from .services.config_service_v2 import BaseConfigServiceV2AsyncClient
 from .services.logging_service_v2 import LoggingServiceV2Client
diff --git a/tests/integration/goldens/logging_internal/setup.py b/tests/integration/goldens/logging_internal/setup.py
index 23a3a12a2b..f991eec737 100755
--- a/tests/integration/goldens/logging_internal/setup.py
+++ b/tests/integration/goldens/logging_internal/setup.py
@@ -43,6 +43,7 @@
     # Exclude incompatible versions of `google-auth`
     # See https://github.com/googleapis/google-cloud-python/issues/12364
     "google-auth >= 2.14.1, <3.0.0,!=2.24.0,!=2.25.0",
+    "packaging", # TODO: Remove once we require versions of api core that include this
     "proto-plus >= 1.22.3, <2.0.0",
     "proto-plus >= 1.25.0, <2.0.0; python_version >= '3.13'",
     "protobuf>=3.20.2,<7.0.0,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5",
diff --git a/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py b/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py
index e206a2486b..97d56ab135 100755
--- a/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py
+++ b/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py
@@ -12897,6 +12897,7 @@ def test_config_service_v2_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.ConfigServiceV2GrpcTransport, transports.ConfigServiceV2GrpcAsyncIOTransport])
 def test_config_service_v2_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py b/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py
index 195d456017..3a3740300b 100755
--- a/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py
+++ b/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py
@@ -3445,6 +3445,7 @@ def test_logging_service_v2_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.LoggingServiceV2GrpcTransport, transports.LoggingServiceV2GrpcAsyncIOTransport])
 def test_logging_service_v2_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py b/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
index 35cb37f70d..bcb2c8346f 100755
--- a/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
+++ b/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py
@@ -3249,6 +3249,7 @@ def test_metrics_service_v2_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.MetricsServiceV2GrpcTransport, transports.MetricsServiceV2GrpcAsyncIOTransport])
 def test_metrics_service_v2_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/redis/google/cloud/redis_v1/__init__.py b/tests/integration/goldens/redis/google/cloud/redis_v1/__init__.py
index af3d250ccd..994427c2c9 100755
--- a/tests/integration/goldens/redis/google/cloud/redis_v1/__init__.py
+++ b/tests/integration/goldens/redis/google/cloud/redis_v1/__init__.py
@@ -18,6 +18,77 @@
 __version__ = package_version.__version__
 
 
+import google.api_core as api_core
+
+if hasattr(api_core, "check_python_version") and hasattr(api_core, "check_dependency_versions"):   # pragma: NO COVER
+    api_core.check_python_version("google.cloud.redis_v1") # type: ignore
+    api_core.check_dependency_versions("google.cloud.redis_v1") # type: ignore
+else:   # pragma: NO COVER
+    # An older version of api_core is installed, which does not define the
+    # functions above. We do equivalent checks manually.
+
+    import warnings
+    import sys
+
+    _py_version_str = sys.version.split()[0]
+    _package_label = "google.cloud.redis_v1"
+    if sys.version_info < (3, 9):
+        warnings.warn("You are using a non-supported Python version " +
+                      f"({_py_version_str}).  Google will not post any further " +
+                      f"updates to {_package_label} supporting this Python version. " +
+                      "Please upgrade to the latest Python version, or at " +
+                      f"least to Python 3.9, and then update {_package_label}.",
+                      FutureWarning)
+    if sys.version_info[:2] == (3, 9):
+        warnings.warn(f"You are using a Python version ({_py_version_str}) " +
+                      f"which Google will stop supporting in {_package_label} when " +
+                      "it reaches its end of life (October 2025). Please " +
+                      "upgrade to the latest Python version, or at " +
+                      "least Python 3.10, before then, and " +
+                      f"then update {_package_label}.",
+                      FutureWarning)
+
+    from packaging.version import parse as parse_version
+
+    if sys.version_info < (3, 8):
+        import pkg_resources
+
+        def _get_version(dependency_name):
+          try:
+            version_string = pkg_resources.get_distribution(dependency_name).version
+            return (parse_version(version_string), version_string)
+          except pkg_resources.DistributionNotFound:
+            return (None, "--")
+    else:
+        from importlib import metadata
+
+        def _get_version(dependency_name):
+            try:
+                version_string = metadata.version("requests")
+                parsed_version = parse_version(version_string)
+                return (parsed_version.release, version_string)
+            except metadata.PackageNotFoundError:
+                return (None, "--")
+
+    _dependency_package = "google.protobuf"
+    _next_supported_version = "4.25.8"
+    _next_supported_version_tuple = (4, 25, 8)
+    (_version_used, _version_used_string) = _get_version(_dependency_package)
+    if _version_used and _version_used < _next_supported_version_tuple:
+        warnings.warn(f"Package {_package_label} depends on " +
+                      f"{_dependency_package}, currently installed at version " +
+                      f"{_version_used_string}. Future updates to " +
+                      f"{_package_label} will require {_dependency_package} at " +
+                      f"version {_next_supported_version} or higher. Please ensure " +
+                      "that either (a) your Python environment doesn't pin the " +
+                      f"version of {_dependency_package}, so that updates to " +
+                      f"{_package_label} can require the higher version, or " +
+                      "(b) you manually update your Python environment to use at " +
+                      f"least version {_next_supported_version} of " +
+                      f"{_dependency_package}.",
+                      FutureWarning)
+
+
 from .services.cloud_redis import CloudRedisClient
 from .services.cloud_redis import CloudRedisAsyncClient
 
diff --git a/tests/integration/goldens/redis/setup.py b/tests/integration/goldens/redis/setup.py
index 54fece2f91..0d28f00404 100755
--- a/tests/integration/goldens/redis/setup.py
+++ b/tests/integration/goldens/redis/setup.py
@@ -43,6 +43,7 @@
     # Exclude incompatible versions of `google-auth`
     # See https://github.com/googleapis/google-cloud-python/issues/12364
     "google-auth >= 2.14.1, <3.0.0,!=2.24.0,!=2.25.0",
+    "packaging", # TODO: Remove once we require versions of api core that include this
     "proto-plus >= 1.22.3, <2.0.0",
     "proto-plus >= 1.25.0, <2.0.0; python_version >= '3.13'",
     "protobuf>=3.20.2,<7.0.0,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5",
diff --git a/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py b/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
index 4ff5644cdc..02549e2f2f 100755
--- a/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -11657,6 +11657,7 @@ def test_cloud_redis_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.CloudRedisGrpcTransport, transports.CloudRedisGrpcAsyncIOTransport])
 def test_cloud_redis_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/integration/goldens/redis_selective/google/cloud/redis_v1/__init__.py b/tests/integration/goldens/redis_selective/google/cloud/redis_v1/__init__.py
index 760fe80f6c..b3e98afba4 100755
--- a/tests/integration/goldens/redis_selective/google/cloud/redis_v1/__init__.py
+++ b/tests/integration/goldens/redis_selective/google/cloud/redis_v1/__init__.py
@@ -18,6 +18,77 @@
 __version__ = package_version.__version__
 
 
+import google.api_core as api_core
+
+if hasattr(api_core, "check_python_version") and hasattr(api_core, "check_dependency_versions"):   # pragma: NO COVER
+    api_core.check_python_version("google.cloud.redis_v1") # type: ignore
+    api_core.check_dependency_versions("google.cloud.redis_v1") # type: ignore
+else:   # pragma: NO COVER
+    # An older version of api_core is installed, which does not define the
+    # functions above. We do equivalent checks manually.
+
+    import warnings
+    import sys
+
+    _py_version_str = sys.version.split()[0]
+    _package_label = "google.cloud.redis_v1"
+    if sys.version_info < (3, 9):
+        warnings.warn("You are using a non-supported Python version " +
+                      f"({_py_version_str}).  Google will not post any further " +
+                      f"updates to {_package_label} supporting this Python version. " +
+                      "Please upgrade to the latest Python version, or at " +
+                      f"least to Python 3.9, and then update {_package_label}.",
+                      FutureWarning)
+    if sys.version_info[:2] == (3, 9):
+        warnings.warn(f"You are using a Python version ({_py_version_str}) " +
+                      f"which Google will stop supporting in {_package_label} when " +
+                      "it reaches its end of life (October 2025). Please " +
+                      "upgrade to the latest Python version, or at " +
+                      "least Python 3.10, before then, and " +
+                      f"then update {_package_label}.",
+                      FutureWarning)
+
+    from packaging.version import parse as parse_version
+
+    if sys.version_info < (3, 8):
+        import pkg_resources
+
+        def _get_version(dependency_name):
+          try:
+            version_string = pkg_resources.get_distribution(dependency_name).version
+            return (parse_version(version_string), version_string)
+          except pkg_resources.DistributionNotFound:
+            return (None, "--")
+    else:
+        from importlib import metadata
+
+        def _get_version(dependency_name):
+            try:
+                version_string = metadata.version("requests")
+                parsed_version = parse_version(version_string)
+                return (parsed_version.release, version_string)
+            except metadata.PackageNotFoundError:
+                return (None, "--")
+
+    _dependency_package = "google.protobuf"
+    _next_supported_version = "4.25.8"
+    _next_supported_version_tuple = (4, 25, 8)
+    (_version_used, _version_used_string) = _get_version(_dependency_package)
+    if _version_used and _version_used < _next_supported_version_tuple:
+        warnings.warn(f"Package {_package_label} depends on " +
+                      f"{_dependency_package}, currently installed at version " +
+                      f"{_version_used_string}. Future updates to " +
+                      f"{_package_label} will require {_dependency_package} at " +
+                      f"version {_next_supported_version} or higher. Please ensure " +
+                      "that either (a) your Python environment doesn't pin the " +
+                      f"version of {_dependency_package}, so that updates to " +
+                      f"{_package_label} can require the higher version, or " +
+                      "(b) you manually update your Python environment to use at " +
+                      f"least version {_next_supported_version} of " +
+                      f"{_dependency_package}.",
+                      FutureWarning)
+
+
 from .services.cloud_redis import CloudRedisClient
 from .services.cloud_redis import CloudRedisAsyncClient
 
diff --git a/tests/integration/goldens/redis_selective/setup.py b/tests/integration/goldens/redis_selective/setup.py
index 54fece2f91..0d28f00404 100755
--- a/tests/integration/goldens/redis_selective/setup.py
+++ b/tests/integration/goldens/redis_selective/setup.py
@@ -43,6 +43,7 @@
     # Exclude incompatible versions of `google-auth`
     # See https://github.com/googleapis/google-cloud-python/issues/12364
     "google-auth >= 2.14.1, <3.0.0,!=2.24.0,!=2.25.0",
+    "packaging", # TODO: Remove once we require versions of api core that include this
     "proto-plus >= 1.22.3, <2.0.0",
     "proto-plus >= 1.25.0, <2.0.0; python_version >= '3.13'",
     "protobuf>=3.20.2,<7.0.0,!=4.21.0,!=4.21.1,!=4.21.2,!=4.21.3,!=4.21.4,!=4.21.5",
diff --git a/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py b/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
index fd46390682..2aed5640c0 100755
--- a/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
+++ b/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py
@@ -6823,6 +6823,7 @@ def test_cloud_redis_grpc_asyncio_transport_channel():
 
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 @pytest.mark.parametrize("transport_class", [transports.CloudRedisGrpcTransport, transports.CloudRedisGrpcAsyncIOTransport])
 def test_cloud_redis_transport_channel_mtls_with_client_cert_source(
     transport_class
diff --git a/tests/system/conftest.py b/tests/system/conftest.py
index a7967d4f5d..1858fdb424 100644
--- a/tests/system/conftest.py
+++ b/tests/system/conftest.py
@@ -17,6 +17,7 @@
 from unittest import mock
 import os
 import pytest
+import pytest_asyncio
 
 from typing import Sequence, Tuple
 
@@ -79,7 +80,7 @@ def async_anonymous_credentials():
     def event_loop():
         return asyncio.get_event_loop()
 
-    @pytest.fixture(params=["grpc_asyncio", "rest_asyncio"])
+    @pytest_asyncio.fixture(params=["grpc_asyncio", "rest_asyncio"])
     def async_echo(use_mtls, request, event_loop):
         transport = request.param
         if transport == "rest_asyncio" and not HAS_ASYNC_REST_ECHO_TRANSPORT:
@@ -94,7 +95,7 @@ def async_echo(use_mtls, request, event_loop):
             credentials=async_anonymous_credentials(),
         )
 
-    @pytest.fixture(params=["grpc_asyncio", "rest_asyncio"])
+    @pytest_asyncio.fixture(params=["grpc_asyncio", "rest_asyncio"])
     def async_identity(use_mtls, request, event_loop):
         transport = request.param
         if transport == "rest_asyncio" and not HAS_ASYNC_REST_IDENTITY_TRANSPORT:
@@ -302,41 +303,46 @@ class EchoMetadataClientGrpcInterceptor(
     grpc.StreamUnaryClientInterceptor,
     grpc.StreamStreamClientInterceptor,
 ):
-    def __init__(self, key, value):
-        self._key = key
-        self._value = value
+    def __init__(self):
         self.request_metadata = []
         self.response_metadata = []
 
-    def _add_request_metadata(self, client_call_details):
+    def _read_request_metadata(self, client_call_details):
         if client_call_details.metadata is not None:
-            client_call_details.metadata.append((self._key, self._value))
             self.request_metadata = client_call_details.metadata
 
+    def _read_response_metadata_stream(self):
+        # Access the metadata via the original stream object
+        if hasattr(self, "_original_stream"):
+            return self._original_stream.trailing_metadata()
+        return []
+
     def intercept_unary_unary(self, continuation, client_call_details, request):
-        self._add_request_metadata(client_call_details)
+        self._read_request_metadata(client_call_details)
         response = continuation(client_call_details, request)
         metadata = [(k, str(v)) for k, v in response.trailing_metadata()]
         self.response_metadata = metadata
         return response
 
     def intercept_unary_stream(self, continuation, client_call_details, request):
-        self._add_request_metadata(client_call_details)
+        self._read_request_metadata(client_call_details)
         response_it = continuation(client_call_details, request)
+        self._original_stream = response_it
         return response_it
 
     def intercept_stream_unary(
         self, continuation, client_call_details, request_iterator
     ):
-        self._add_request_metadata(client_call_details)
+        self._read_request_metadata(client_call_details)
         response = continuation(client_call_details, request_iterator)
         return response
 
     def intercept_stream_stream(
         self, continuation, client_call_details, request_iterator
     ):
-        self._add_request_metadata(client_call_details)
+        self._read_request_metadata(client_call_details)
         response_it = continuation(client_call_details, request_iterator)
+        self._original_stream = response_it
         return response_it
 
 
@@ -346,53 +352,46 @@ class EchoMetadataClientGrpcAsyncInterceptor(
     grpc.aio.StreamUnaryClientInterceptor,
     grpc.aio.StreamStreamClientInterceptor,
 ):
-    def __init__(self, key, value):
-        self._key = key
-        self._value = value
+    def __init__(self):
         self.request_metadata = []
         self.response_metadata = []
 
-    async def _add_request_metadata(self, client_call_details):
+    async def _read_request_metadata(self, client_call_details):
         if client_call_details.metadata is not None:
-            client_call_details.metadata.append((self._key, self._value))
-            self.request_metadata = client_call_details.metadata
+            self.request_metadata = list(client_call_details.metadata)
 
     async def intercept_unary_unary(self, continuation, client_call_details, request):
-        await self._add_request_metadata(client_call_details)
+        await self._read_request_metadata(client_call_details)
         response = await continuation(client_call_details, request)
         metadata = [(k, str(v)) for k, v in await response.trailing_metadata()]
         self.response_metadata = metadata
         return response
 
     async def intercept_unary_stream(self, continuation, client_call_details, request):
-        self._add_request_metadata(client_call_details)
+        self._read_request_metadata(client_call_details)
         response_it = continuation(client_call_details, request)
         return response_it
 
     async def intercept_stream_unary(
         self, continuation, client_call_details, request_iterator
     ):
-        self._add_request_metadata(client_call_details)
+        self._read_request_metadata(client_call_details)
         response = continuation(client_call_details, request_iterator)
         return response
 
     async def intercept_stream_stream(
         self, continuation, client_call_details, request_iterator
     ):
-        self._add_request_metadata(client_call_details)
+        self._read_request_metadata(client_call_details)
         response_it = continuation(client_call_details, request_iterator)
         return response_it
 
 
 @pytest.fixture
 def intercepted_echo_grpc(use_mtls):
-    # The interceptor adds 'showcase-trailer' client metadata. Showcase server
-    # echoes any metadata with key 'showcase-trailer', so the same metadata
-    # should appear as trailing metadata in the response.
-    interceptor = EchoMetadataClientGrpcInterceptor(
-        "showcase-trailer",
-        "intercepted",
-    )
+    # The interceptor reads request
+    # and response metadata.
+    interceptor = EchoMetadataClientGrpcInterceptor()
     host = "localhost:7469"
     channel = (
         grpc.secure_channel(host, ssl_credentials)
@@ -407,15 +406,11 @@ def intercepted_echo_grpc(use_mtls):
     return EchoClient(transport=transport), interceptor
 
 
-@pytest.fixture
-def intercepted_echo_grpc_async():
-    # The interceptor adds 'showcase-trailer' client metadata. Showcase server
-    # echoes any metadata with key 'showcase-trailer', so the same metadata
-    # should appear as trailing metadata in the response.
-    interceptor = EchoMetadataClientGrpcAsyncInterceptor(
-        "showcase-trailer",
-        "intercepted",
-    )
+@pytest_asyncio.fixture
+async def intercepted_echo_grpc_async():
+    # The interceptor reads request
+    # and response metadata.
+    interceptor = EchoMetadataClientGrpcAsyncInterceptor()
     host = "localhost:7469"
     channel = grpc.aio.insecure_channel(host, interceptors=[interceptor])
     # intercept_channel = grpc.aio.intercept_channel(channel, interceptor)
diff --git a/tests/system/test_grpc_interceptor_streams.py b/tests/system/test_grpc_interceptor_streams.py
index 50c5bee31d..8899707a4f 100644
--- a/tests/system/test_grpc_interceptor_streams.py
+++ b/tests/system/test_grpc_interceptor_streams.py
@@ -15,10 +15,10 @@
 from google import showcase
 
 
-# intercetped_metadata will be added by the interceptor automatically, and
+# `_METADATA` will be sent as part of the request, and the
 # showcase server will echo it (since it has key 'showcase-trailer') as trailing
 # metadata.
-intercepted_metadata = (("showcase-trailer", "intercepted"),)
+_METADATA = (("showcase-trailer", "intercepted"),)
 
 
 def test_unary_stream(intercepted_echo_grpc):
@@ -27,7 +27,8 @@ def test_unary_stream(intercepted_echo_grpc):
     responses = client.expand(
         {
             "content": content,
-        }
+        },
+        metadata=_METADATA,
     )
 
     for ground_truth, response in zip(content.split(" "), responses):
@@ -37,8 +38,9 @@ def test_unary_stream(intercepted_echo_grpc):
     response_metadata = [
         (metadata.key, metadata.value) for metadata in responses.trailing_metadata()
     ]
-    assert intercepted_metadata[0] in response_metadata
-    interceptor.response_metadata = response_metadata
+    assert _METADATA[0] in interceptor.request_metadata
+    assert _METADATA[0] in response_metadata
+    assert _METADATA[0] in interceptor._read_response_metadata_stream()
 
 
 def test_stream_stream(intercepted_echo_grpc):
@@ -46,7 +48,7 @@ def test_stream_stream(intercepted_echo_grpc):
     requests = []
     requests.append(showcase.EchoRequest(content="hello"))
     requests.append(showcase.EchoRequest(content="world!"))
-    responses = client.chat(iter(requests))
+    responses = client.chat(iter(requests), metadata=_METADATA)
 
     contents = [response.content for response in responses]
     assert contents == ["hello", "world!"]
@@ -54,5 +56,6 @@ def test_stream_stream(intercepted_echo_grpc):
     response_metadata = [
         (metadata.key, metadata.value) for metadata in responses.trailing_metadata()
     ]
-    assert intercepted_metadata[0] in response_metadata
-    interceptor.response_metadata = response_metadata
+    assert _METADATA[0] in interceptor.request_metadata
+    assert _METADATA[0] in response_metadata
+    assert _METADATA[0] in interceptor._read_response_metadata_stream()
diff --git a/tests/system/test_response_metadata.py b/tests/system/test_response_metadata.py
index f99abf3e11..2cc8681455 100644
--- a/tests/system/test_response_metadata.py
+++ b/tests/system/test_response_metadata.py
@@ -84,6 +84,7 @@ async def test_metadata_response_rest_streams_async(intercepted_echo_rest_async)
             ("rest_asyncio", ("X-Showcase-Request-Something3", "something_value3")),
         ],
     )
+    @pytest.mark.filterwarnings("ignore::FutureWarning")
     @pytest.mark.asyncio
     async def test_metadata_response_unary_async(
         intercepted_echo_grpc_async,
diff --git a/tests/system/test_streams.py b/tests/system/test_streams.py
index 38a2e47745..e7fbc1c94b 100644
--- a/tests/system/test_streams.py
+++ b/tests/system/test_streams.py
@@ -19,6 +19,9 @@
 from google import showcase
 
 
+# `_METADATA` will be sent as part of the request, and the
+# showcase server will echo it (since it has key 'showcase-trailer') as trailing
+# metadata.
 _METADATA = (("showcase-trailer", "hello world"),)