diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index e8a914f0e7..8be4f041d2 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -14,7 +14,7 @@ concurrency:
   cancel-in-progress: true
 
 env:
-  SHOWCASE_VERSION: 0.32.0
+  SHOWCASE_VERSION: 0.35.0
   PROTOC_VERSION: 3.20.2
 
 jobs:
diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/_shared_macros.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/_shared_macros.j2
new file mode 100644
index 0000000000..e7f623cfd2
--- /dev/null
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/_shared_macros.j2
@@ -0,0 +1,76 @@
+{#
+ # Copyright (C) 2024  Google LLC
+ #
+ # Licensed under the Apache License, Version 2.0 (the "License");
+ # you may not use this file except in compliance with the License.
+ # You may obtain a copy of the License at
+ #
+ #     http://www.apache.org/licenses/LICENSE-2.0
+ #
+ # Unless required by applicable law or agreed to in writing, software
+ # distributed under the License is distributed on an "AS IS" BASIS,
+ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ # See the License for the specific language governing permissions and
+ # limitations under the License.
+ #
+ # This file is a copy of `_shared_macros.j2` in standard templates located at
+ # `gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2`
+ # It is intended to be a symlink. 
+ # See https://github.com/googleapis/gapic-generator-python/issues/2028
+ # which contains follow up work to convert it to a symlink.
+ # Do not diverge from the copy of `_shared_macros.j2` in standard templates.
+#}
+
+{% macro auto_populate_uuid4_fields(api, method) %}
+{#
+    Automatically populate UUID4 fields according to
+    https://google.aip.dev/client-libraries/4235 when the
+    field satisfies either of:
+    - The field supports explicit presence and has not been set by the user.
+    - The field doesn't support explicit presence, and its value is the empty
+      string (i.e. the default value).
+    When using this macro, ensure the calling template generates a line `import uuid`
+#}
+{% with method_settings = api.all_method_settings.get(method.meta.address.proto) %}
+{% if method_settings is not none %}
+{% for auto_populated_field in method_settings.auto_populated_fields %}
+    {% if method.input.fields[auto_populated_field].proto3_optional %}
+        if '{{ auto_populated_field }}' not in request:
+    {% else %}
+        if not request.{{ auto_populated_field }}:
+    {% endif %}
+            request.{{ auto_populated_field }} = str(uuid.uuid4())
+{% endfor %}
+{% endif %}{# if method_settings is not none #}
+{% endwith %}{# method_settings #}
+{% endmacro %}
+
+{% macro add_google_api_core_version_header_import(service_version) %}
+{#
+The `version_header` module was added to `google-api-core`
+in version 2.19.0.
+https://github.com/googleapis/python-api-core/releases/tag/v2.19.0
+The `try/except` below can be removed once the minimum version of
+`google-api-core` is 2.19.0 or newer.
+#}
+{% if service_version %}
+try:
+    from google.api_core import version_header
+    HAS_GOOGLE_API_CORE_VERSION_HEADER = True  # pragma: NO COVER
+except ImportError:  # pragma: NO COVER
+    HAS_GOOGLE_API_CORE_VERSION_HEADER = False
+{% endif %}{# service_version #}
+{% endmacro %}
+{% macro add_api_version_header_to_metadata(service_version) %}
+{#
+    Add API Version to metadata as per https://github.com/aip-dev/google.aip.dev/pull/1331.
+    When using this macro, ensure the calling template also calls macro
+    `add_google_api_core_version_header_import` to add the necessary import statements.
+#}
+    {% if service_version %}
+        if HAS_GOOGLE_API_CORE_VERSION_HEADER:  # pragma: NO COVER
+            metadata = tuple(metadata) + (
+                version_header.to_api_version_header("{{ service_version }}"),
+            )
+    {% endif %}{# service_version #}
+{% endmacro %}
diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/client.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/client.py.j2
index aff47ad913..e4b0af6dec 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/client.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/client.py.j2
@@ -1,6 +1,7 @@
 {% extends '_base.py.j2' %}
 
 {% block content %}
+{% import "%namespace/%name/%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
 
 from collections import OrderedDict
 import os
@@ -23,6 +24,7 @@ from google.auth.transport.grpc import SslCredentials             # type: ignore
 from google.auth.exceptions import MutualTLSChannelError          # type: ignore
 from google.oauth2 import service_account                         # type: ignore
 
+{{ shared_macros.add_google_api_core_version_header_import(service.version) }}
 {% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
 from {{package_path}} import gapic_version as package_version
 
@@ -94,7 +96,8 @@ class {{ service.client_name }}Meta(type):
 
 
 class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
-    """{{ service.meta.doc|rst(width=72, indent=4) }}"""
+    """{{ service.meta.doc|rst(width=72, indent=4) }}{% if service.version|length %}
+    This class implements API version {{ service.version }}.{% endif %}"""
 
     @staticmethod
     def _get_default_mtls_endpoint(api_endpoint):
@@ -475,27 +478,8 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
             )),
         )
         {% endif %}
-
-{#
-    Automatically populate UUID4 fields according to
-    https://google.aip.dev/client-libraries/4235 when the
-    field satisfies either of:
-    - The field supports explicit presence and has not been set by the user.
-    - The field doesn't support explicit presence, and its value is the empty
-      string (i.e. the default value).
-#}
-{% with method_settings = api.all_method_settings.get(method.meta.address.proto) %}
-{% if method_settings is not none %}
-{% for auto_populated_field in method_settings.auto_populated_fields %}
-    {% if method.input.fields[auto_populated_field].proto3_optional %}
-        if '{{ auto_populated_field }}' not in request:
-    {% else %}
-        if not request.{{ auto_populated_field }}:
-    {% endif %}
-            request.{{ auto_populated_field }} = str(uuid.uuid4())
-{% endfor %}
-{% endif %}{# if method_settings is not none #}
-{% endwith %}{# method_settings #}
+{{ shared_macros.add_api_version_header_to_metadata(service.version) }}
+{{ shared_macros.auto_populate_uuid4_fields(api, method) }}
 
         # Send the request.
         {%+ if not method.void %}response = {% endif %}rpc(
diff --git a/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index af148882f4..2a6af5ac1e 100644
--- a/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/gapic/ads-templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -1,6 +1,7 @@
 {% extends "_base.py.j2" %}
 
 {% block content %}
+{% import "%namespace/%name/%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
 
 import os
 {% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|list %}
@@ -39,6 +40,7 @@ from google.oauth2 import service_account
 from {{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join(".") }}.services.{{ service.name|snake_case }} import {{ service.client_name }}
 from {{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join(".") }}.services.{{ service.name|snake_case }} import transports
 
+from google.api_core import api_core_version
 from google.api_core import client_options
 from google.api_core import exceptions as core_exceptions
 from google.api_core import grpc_helpers
@@ -69,6 +71,8 @@ from google.iam.v1 import options_pb2  # type: ignore
 from google.iam.v1 import policy_pb2  # type: ignore
 {% endif %}
 {% endfilter %}
+{{ shared_macros.add_google_api_core_version_header_import(service.version) }}
+
 
 {% with uuid4_re = "[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}" %}
 
@@ -636,6 +640,35 @@ def test_{{ method_name }}(request_type, transport: str = 'grpc'):
     {% endwith %}{# auto_populated_field_sample_value #}
 
 
+{% if service.version %}
+@pytest.mark.parametrize("transport_name", [
+    {% if 'grpc' in opts.transport %}
+    ("grpc"),
+    {% endif %}
+    {% if 'rest' in opts.transport %}
+    ("rest"),
+    {% endif %}
+])
+def test_{{ method_name }}_api_version_header(transport_name):
+    # TODO: Make this test unconditional once the minimum supported version of
+    # google-api-core becomes 2.19.0 or higher.
+    api_core_major, api_core_minor = [int(part) for part in api_core_version.__version__.split(".")[0:2]]
+    if api_core_major > 2 or (api_core_major == 2 and api_core_minor >= 19):
+        client = {{ service.client_name }}(credentials=ga_credentials.AnonymousCredentials(), transport=transport_name)
+        # Mock the actual call within the gRPC stub, and fake the request.
+        with mock.patch.object(
+                type(client.transport.{{ method.transport_safe_name|snake_case }}),
+                '__call__'
+            ) as call:
+                client.{{ method_name }}()
+
+        # Establish that the api version header was sent.
+        _, _, kw = call.mock_calls[0]
+        assert kw['metadata'][0] == (version_header.API_VERSION_METADATA_KEY, "{{ service.version }}")
+    else:
+        pytest.skip("google-api-core>=2.19.0 is required for `google.api_core.version_header`")
+{% endif %}{# service.version #}
+
 {% if not method.client_streaming %}
 def test_{{ method_name }}_empty_call():
     # This test is a coverage failsafe to make sure that totally empty calls,
@@ -904,9 +937,9 @@ def test_{{ method_name }}_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
+        expected_metadata = ()
         {% if method.field_headers %}
-        metadata = tuple(metadata) + (
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 {% for field_header in method.field_headers %}
                 {% if not method.client_streaming %}
@@ -918,7 +951,13 @@ def test_{{ method_name }}_pager(transport_name: str = "grpc"):
         {% endif %}
         pager = client.{{ method_name }}(request={})
 
-        assert pager._metadata == metadata
+        {% if service.version %}
+        if HAS_GOOGLE_API_CORE_VERSION_HEADER:
+            expected_metadata = tuple(expected_metadata) + (
+                version_header.to_api_version_header("{{ service.version }}"),
+            )
+        {% endif %}
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
diff --git a/gapic/schema/wrappers.py b/gapic/schema/wrappers.py
index c7396ca5e9..fd75e8fd5e 100644
--- a/gapic/schema/wrappers.py
+++ b/gapic/schema/wrappers.py
@@ -1734,6 +1734,17 @@ def host(self) -> str:
             return self.options.Extensions[client_pb2.default_host]
         return ''
 
+    @property
+    def version(self) -> str:
+        """Return the API version for this service, if specified.
+
+        Returns:
+            str: The API version for this service.
+        """
+        if self.options.Extensions[client_pb2.api_version]:
+            return self.options.Extensions[client_pb2.api_version]
+        return ''
+
     @property
     def shortname(self) -> str:
         """Return the API short name. DRIFT uses this to identify
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/_client_macros.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/_client_macros.j2
index 25b3ef6fca..128421655e 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/_client_macros.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/_client_macros.j2
@@ -14,6 +14,8 @@
  # limitations under the License.
 #}
 
+{% import "%namespace/%name_%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
+
 {% macro client_method(method, name, snippet_index, api, service, full_extended_lro=False) %}
     def {{ name }}(self,
             {% if not method.client_streaming %}
@@ -181,7 +183,8 @@
         )
         {% endif %} {# method.explicit_routing #}
 
-{{ auto_populate_uuid4_fields(api, method) }}
+{{ shared_macros.add_api_version_header_to_metadata(service.version) }}
+{{ shared_macros.auto_populate_uuid4_fields(api, method) }}
 
         # Validate the universe domain.
         self._validate_universe_domain()
@@ -265,27 +268,3 @@
 
 {% macro define_extended_operation_subclass(extended_operation) %}
 {% endmacro %}
-
-{% macro auto_populate_uuid4_fields(api, method) %}
-{#
-    Automatically populate UUID4 fields according to
-    https://google.aip.dev/client-libraries/4235 when the
-    field satisfies either of:
-    - The field supports explicit presence and has not been set by the user.
-    - The field doesn't support explicit presence, and its value is the empty
-      string (i.e. the default value).
-    When using this macro, ensure the calling template generates a line `import uuid`
-#}
-{% with method_settings = api.all_method_settings.get(method.meta.address.proto) %}
-{% if method_settings is not none %}
-{% for auto_populated_field in method_settings.auto_populated_fields %}
-    {% if method.input.fields[auto_populated_field].proto3_optional %}
-        if '{{ auto_populated_field }}' not in request:
-    {% else %}
-        if not request.{{ auto_populated_field }}:
-    {% endif %}
-            request.{{ auto_populated_field }} = str(uuid.uuid4())
-{% endfor %}
-{% endif %}{# if method_settings is not none #}
-{% endwith %}{# method_settings #}
-{% endmacro %}
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
new file mode 100644
index 0000000000..14764da17e
--- /dev/null
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2
@@ -0,0 +1,70 @@
+{#
+ # Copyright (C) 2024  Google LLC
+ #
+ # Licensed under the Apache License, Version 2.0 (the "License");
+ # you may not use this file except in compliance with the License.
+ # You may obtain a copy of the License at
+ #
+ #     http://www.apache.org/licenses/LICENSE-2.0
+ #
+ # Unless required by applicable law or agreed to in writing, software
+ # distributed under the License is distributed on an "AS IS" BASIS,
+ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ # See the License for the specific language governing permissions and
+ # limitations under the License.
+#}
+
+{% macro auto_populate_uuid4_fields(api, method) %}
+{#
+    Automatically populate UUID4 fields according to
+    https://google.aip.dev/client-libraries/4235 when the
+    field satisfies either of:
+    - The field supports explicit presence and has not been set by the user.
+    - The field doesn't support explicit presence, and its value is the empty
+      string (i.e. the default value).
+    When using this macro, ensure the calling template generates a line `import uuid`
+#}
+{% with method_settings = api.all_method_settings.get(method.meta.address.proto) %}
+{% if method_settings is not none %}
+{% for auto_populated_field in method_settings.auto_populated_fields %}
+    {% if method.input.fields[auto_populated_field].proto3_optional %}
+        if '{{ auto_populated_field }}' not in request:
+    {% else %}
+        if not request.{{ auto_populated_field }}:
+    {% endif %}
+            request.{{ auto_populated_field }} = str(uuid.uuid4())
+{% endfor %}
+{% endif %}{# if method_settings is not none #}
+{% endwith %}{# method_settings #}
+{% endmacro %}
+
+{% macro add_google_api_core_version_header_import(service_version) %}
+{#
+The `version_header` module was added to `google-api-core`
+in version 2.19.0.
+https://github.com/googleapis/python-api-core/releases/tag/v2.19.0
+The `try/except` below can be removed once the minimum version of
+`google-api-core` is 2.19.0 or newer.
+#}
+{% if service_version %}
+try:
+    from google.api_core import version_header
+    HAS_GOOGLE_API_CORE_VERSION_HEADER = True  # pragma: NO COVER
+except ImportError:  # pragma: NO COVER
+    HAS_GOOGLE_API_CORE_VERSION_HEADER = False
+{% endif %}{# service_version #}
+{% endmacro %}
+
+{% macro add_api_version_header_to_metadata(service_version) %}
+{#
+    Add API Version to metadata as per https://github.com/aip-dev/google.aip.dev/pull/1331.
+    When using this macro, ensure the calling template also calls macro
+    `add_google_api_core_version_header_import` to add the necessary import statements.
+#}
+    {% if service_version %}
+        if HAS_GOOGLE_API_CORE_VERSION_HEADER:  # pragma: NO COVER
+            metadata = tuple(metadata) + (
+                version_header.to_api_version_header("{{ service_version }}"),
+            )
+    {% endif %}{# service_version #}
+{% endmacro %}
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
index 66ac6a70b5..b63cb1e99f 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
@@ -2,6 +2,7 @@
 
 {% block content %}
 {% import "%namespace/%name_%version/%sub/services/%service/_client_macros.j2" as macros %}
+{% import "%namespace/%name_%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
 
 from collections import OrderedDict
 import functools
@@ -24,6 +25,7 @@ from google.api_core import retry_async as retries
 from google.auth import credentials as ga_credentials   # type: ignore
 from google.oauth2 import service_account              # type: ignore
 
+{{ shared_macros.add_google_api_core_version_header_import(service.version) }}
 try:
     OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
@@ -54,7 +56,8 @@ from .client import {{ service.client_name }}
 
 {# TODO(yon-mg): handle rest transport async client interaction #}
 class {{ service.async_client_name }}:
-    """{{ service.meta.doc|rst(width=72, indent=4) }}"""
+    """{{ service.meta.doc|rst(width=72, indent=4) }}{% if service.version|length %}
+    This class implements API version {{ service.version }}.{% endif %}"""
 
     _client: {{ service.client_name }}
 
@@ -376,7 +379,8 @@ class {{ service.async_client_name }}:
         )
         {% endif %}
 
-{{ macros.auto_populate_uuid4_fields(api, method) }}
+{{ shared_macros.add_api_version_header_to_metadata(service.version) }}
+{{ shared_macros.auto_populate_uuid4_fields(api, method) }}
 
         # Validate the universe domain.
         self._client._validate_universe_domain()
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
index e047b0d55a..eaa572f6e3 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
@@ -2,6 +2,7 @@
 
 {% block content %}
 {% import "%namespace/%name_%version/%sub/services/%service/_client_macros.j2" as macros %}
+{% import "%namespace/%name_%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
 
 from collections import OrderedDict
 {% if service.any_extended_operations_methods %}
@@ -30,7 +31,7 @@ from google.auth.transport import mtls                            # type: ignore
 from google.auth.transport.grpc import SslCredentials             # type: ignore
 from google.auth.exceptions import MutualTLSChannelError          # type: ignore
 from google.oauth2 import service_account                         # type: ignore
-
+{{ shared_macros.add_google_api_core_version_header_import(service.version) }}
 try:
     OptionalRetry = Union[retries.Retry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
@@ -102,7 +103,8 @@ class {{ service.client_name }}Meta(type):
 
 
 class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
-    """{{ service.meta.doc|rst(width=72, indent=4) }}"""
+    """{{ service.meta.doc|rst(width=72, indent=4) }}{% if service.version|length %}
+    This class implements API version {{ service.version }}.{% endif %}"""
 
     @staticmethod
     def _get_default_mtls_endpoint(api_endpoint):
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 ebc96487a1..edf2588ecd 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
@@ -2,6 +2,7 @@
 
 {% block content %}
 {% import "tests/unit/gapic/%name_%version/%sub/test_macros.j2" as test_macros %}
+{% import "%namespace/%name_%version/%sub/services/%service/_shared_macros.j2" as shared_macros %}
 
 import os
 {% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|list %}
@@ -77,7 +78,7 @@ from google.iam.v1 import options_pb2  # type: ignore
 from google.iam.v1 import policy_pb2  # type: ignore
 {% endif %}
 {% endfilter %}
-
+{{ shared_macros.add_google_api_core_version_header_import(service.version) }}
 
 def client_cert_source_callback():
     return b"cert bytes", b"key bytes"
@@ -178,6 +179,49 @@ def test__get_api_endpoint():
         {{ service.client_name }}._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
+{% if service.version %}
+{% for method in service.methods.values() %}{% with method_name = method.name|snake_case %}
+{% for mode in ["", "async"] %}
+{% if mode == "async" %}
+async def test_{{ method_name }}_api_version_header_async(transport_name="grpc"):
+    client = {{ service.async_client_name }}(credentials=ga_credentials.AnonymousCredentials(), transport=transport_name)
+{% else %}
+@pytest.mark.parametrize("transport_name", [
+    {% if 'grpc' in opts.transport %}
+    ("grpc"),
+    {% endif %}
+    {% if 'rest' in opts.transport %}
+    ("rest"),
+    {% endif %}
+])
+def test_{{ method_name }}_api_version_header(transport_name):
+    client = {{ service.client_name }}(credentials=ga_credentials.AnonymousCredentials(), transport=transport_name)
+{% endif %}
+    # TODO: Make this test unconditional once the minimum supported version of
+    # google-api-core becomes 2.19.0 or higher.
+    api_core_major, api_core_minor = [int(part) for part in api_core_version.__version__.split(".")[0:2]]
+    if api_core_major > 2 or (api_core_major == 2 and api_core_minor >= 19):
+        # Mock the actual call within the gRPC stub, and fake the request.
+        with mock.patch.object(
+                type(client.transport.{{ method.transport_safe_name|snake_case }}),
+                '__call__'
+            ) as call:
+            {% if mode == "async" %}
+                await client.{{ method_name }}()
+            {% else %}
+                client.{{ method_name }}()
+            {% endif %}
+
+        # Establish that the api version header was sent.
+        _, _, kw = call.mock_calls[0]
+        assert kw['metadata'][0] == (version_header.API_VERSION_METADATA_KEY, "{{ service.version }}")
+    else:
+        pytest.skip("google-api-core>=2.19.0 is required for `google.api_core.version_header`")
+{% endfor %}{# mode #}
+{% endwith %}
+{% endfor %}
+{% endif %}{# service.version #}
+
 def test__get_universe_domain():
     client_universe_domain = "foo.com"
     universe_domain_env = "bar.com"
diff --git a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2 b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
index 9cb64a3d29..3d6fb7fd7d 100644
--- a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
+++ b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
@@ -853,9 +853,9 @@ def test_{{ method_name }}_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
+        expected_metadata = ()
         {% if not method.explicit_routing and method.field_headers %}
-        metadata = tuple(metadata) + (
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 {% for field_header in method.field_headers %}
                 {% if not method.client_streaming %}
@@ -865,9 +865,15 @@ def test_{{ method_name }}_pager(transport_name: str = "grpc"):
             )),
         )
         {% endif %}
+        {% if service.version %}
+        if HAS_GOOGLE_API_CORE_VERSION_HEADER:
+            expected_metadata = tuple(expected_metadata) + (
+                version_header.to_api_version_header("{{ service.version }}"),
+            )
+        {% endif %}
         pager = client.{{ method_name }}(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
diff --git a/noxfile.py b/noxfile.py
index a7e49ac432..9eba5d7c79 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -29,7 +29,7 @@
 nox.options.error_on_missing_interpreters = True
 
 
-showcase_version = os.environ.get("SHOWCASE_VERSION", "0.32.0")
+showcase_version = os.environ.get("SHOWCASE_VERSION", "0.35.0")
 ADS_TEMPLATES = path.join(path.dirname(__file__), "gapic", "ads-templates")
 
 
diff --git a/test_utils/test_utils.py b/test_utils/test_utils.py
index ac186d293e..f23a7cec41 100644
--- a/test_utils/test_utils.py
+++ b/test_utils/test_utils.py
@@ -32,6 +32,7 @@ def make_service(
     visible_resources: typing.Optional[
         typing.Mapping[str, wrappers.CommonResource]
     ] = None,
+    version: str = "",
 ) -> wrappers.Service:
     visible_resources = visible_resources or {}
     # Define a service descriptor, and set a host and oauth scopes if
@@ -40,6 +41,8 @@ def make_service(
     if host:
         service_pb.options.Extensions[client_pb2.default_host] = host
     service_pb.options.Extensions[client_pb2.oauth_scopes] = ','.join(scopes)
+    if version:
+        service_pb.options.Extensions[client_pb2.api_version] = version
 
     # Return a service object to test.
     return wrappers.Service(
diff --git a/tests/fragments/google/api/client.proto b/tests/fragments/google/api/client.proto
index 2102623d30..f7781254b6 100644
--- a/tests/fragments/google/api/client.proto
+++ b/tests/fragments/google/api/client.proto
@@ -96,4 +96,20 @@ extend google.protobuf.ServiceOptions {
   //     ...
   //   }
   string oauth_scopes = 1050;
+
+  // The API version of this service, which should be sent by version-aware
+  // clients to the service. This allows services to abide by the schema and
+  // behavior of the service at the time this API version was deployed.
+  // The format of the API version must be treated as opaque by clients.
+  // Services may use a format with an apparent structure, but clients must
+  // not rely on this to determine components within an API version, or attempt
+  // to construct other valid API versions. Note that this is for upcoming
+  // functionality and may not be implemented for all services.
+  //
+  // Example:
+  //
+  //   service Foo {
+  //     option (google.api.api_version) = "v1_20230821_preview";
+  //   }
+  string api_version = 525000001;
 }
diff --git a/tests/fragments/test_api_version.proto b/tests/fragments/test_api_version.proto
new file mode 100644
index 0000000000..d75134d845
--- /dev/null
+++ b/tests/fragments/test_api_version.proto
@@ -0,0 +1,37 @@
+// Copyright (C) 2024  Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package google.fragment;
+
+import "google/protobuf/struct.proto";
+import "google/api/client.proto";
+
+service MyServiceWithVersion {
+  option (google.api.api_version) = "v1_20230601";
+  option (google.api.default_host) = "my.example.com";
+
+  rpc MyMethod(MethodRequest) returns (MethodResponse) {
+    option (google.api.method_signature) = "parameter";
+  }
+}
+
+message MethodRequest {
+  google.protobuf.Value parameter = 1;
+}
+
+message MethodResponse {
+  google.protobuf.Value result = 1;
+}
diff --git a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/async_client.py b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/async_client.py
index dc47b46ca8..2985521eab 100755
--- a/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/async_client.py
+++ b/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/async_client.py
@@ -27,6 +27,7 @@
 from google.auth import credentials as ga_credentials   # type: ignore
 from google.oauth2 import service_account              # type: ignore
 
+
 try:
     OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
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 22f07d0368..a304fe8da4 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
@@ -160,6 +160,7 @@ def test__get_api_endpoint():
         AssetServiceClient._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
+
 def test__get_universe_domain():
     client_universe_domain = "foo.com"
     universe_domain_env = "bar.com"
@@ -1439,15 +1440,15 @@ def test_list_assets_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_assets(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -4095,15 +4096,15 @@ def test_search_all_resources_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('scope', ''),
             )),
         )
         pager = client.search_all_resources(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -4657,15 +4658,15 @@ def test_search_all_iam_policies_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('scope', ''),
             )),
         )
         pager = client.search_all_iam_policies(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -7067,15 +7068,15 @@ def test_list_saved_queries_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_saved_queries(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -8629,15 +8630,15 @@ def test_analyze_org_policies_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('scope', ''),
             )),
         )
         pager = client.analyze_org_policies(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -9201,15 +9202,15 @@ def test_analyze_org_policy_governed_containers_pager(transport_name: str = "grp
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('scope', ''),
             )),
         )
         pager = client.analyze_org_policy_governed_containers(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -9773,15 +9774,15 @@ def test_analyze_org_policy_governed_assets_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('scope', ''),
             )),
         )
         pager = client.analyze_org_policy_governed_assets(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
diff --git a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/async_client.py b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/async_client.py
index 151f89ed22..d61293a8be 100755
--- a/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/async_client.py
+++ b/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/async_client.py
@@ -27,6 +27,7 @@
 from google.auth import credentials as ga_credentials   # type: ignore
 from google.oauth2 import service_account              # type: ignore
 
+
 try:
     OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
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 9e86be3470..cc8db8a42c 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
@@ -150,6 +150,7 @@ def test__get_api_endpoint():
         IAMCredentialsClient._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
+
 def test__get_universe_domain():
     client_universe_domain = "foo.com"
     universe_domain_env = "bar.com"
diff --git a/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/async_client.py b/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/async_client.py
index 9fc21688ad..b2b3a07b79 100755
--- a/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/async_client.py
+++ b/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/async_client.py
@@ -27,6 +27,7 @@
 from google.auth import credentials as ga_credentials   # type: ignore
 from google.oauth2 import service_account              # type: ignore
 
+
 try:
     OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
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 06b205c1d1..ae832fb35e 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
@@ -170,6 +170,7 @@ def test__get_api_endpoint():
         EventarcClient._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
+
 def test__get_universe_domain():
     client_universe_domain = "foo.com"
     universe_domain_env = "bar.com"
@@ -1556,15 +1557,15 @@ def test_list_triggers_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_triggers(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -3623,15 +3624,15 @@ def test_list_channels_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_channels(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -5649,15 +5650,15 @@ def test_list_providers_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_providers(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -6571,15 +6572,15 @@ def test_list_channel_connections_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_channel_connections(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/async_client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/async_client.py
index e93c0db603..8c73903d7c 100755
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/async_client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/async_client.py
@@ -27,6 +27,7 @@
 from google.auth import credentials as ga_credentials   # type: ignore
 from google.oauth2 import service_account              # type: ignore
 
+
 try:
     OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/async_client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/async_client.py
index 1c844f9ee0..97bd907c93 100755
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/async_client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/async_client.py
@@ -27,6 +27,7 @@
 from google.auth import credentials as ga_credentials   # type: ignore
 from google.oauth2 import service_account              # type: ignore
 
+
 try:
     OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
diff --git a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/async_client.py b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/async_client.py
index 681c965ed9..4b937e1535 100755
--- a/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/async_client.py
+++ b/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/async_client.py
@@ -27,6 +27,7 @@
 from google.auth import credentials as ga_credentials   # type: ignore
 from google.oauth2 import service_account              # type: ignore
 
+
 try:
     OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
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 138133b22d..2219af1173 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
@@ -150,6 +150,7 @@ def test__get_api_endpoint():
         ConfigServiceV2Client._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
+
 def test__get_universe_domain():
     client_universe_domain = "foo.com"
     universe_domain_env = "bar.com"
@@ -1140,15 +1141,15 @@ def test_list_buckets_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_buckets(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -3678,15 +3679,15 @@ def test_list_views_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_views(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -5338,15 +5339,15 @@ def test_list_sinks_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_sinks(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -8171,15 +8172,15 @@ def test_list_links_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_links(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -9083,15 +9084,15 @@ def test_list_exclusions_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_exclusions(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
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 73f543d55d..992c8836f7 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
@@ -151,6 +151,7 @@ def test__get_api_endpoint():
         LoggingServiceV2Client._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
+
 def test__get_universe_domain():
     client_universe_domain = "foo.com"
     universe_domain_env = "bar.com"
@@ -1763,10 +1764,10 @@ def test_list_log_entries_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
+        expected_metadata = ()
         pager = client.list_log_entries(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -2159,10 +2160,10 @@ def test_list_monitored_resource_descriptors_pager(transport_name: str = "grpc")
             RuntimeError,
         )
 
-        metadata = ()
+        expected_metadata = ()
         pager = client.list_monitored_resource_descriptors(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
@@ -2707,15 +2708,15 @@ def test_list_logs_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_logs(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
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 be4325c678..8acc9a12e8 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
@@ -149,6 +149,7 @@ def test__get_api_endpoint():
         MetricsServiceV2Client._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
+
 def test__get_universe_domain():
     client_universe_domain = "foo.com"
     universe_domain_env = "bar.com"
@@ -1140,15 +1141,15 @@ def test_list_log_metrics_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_log_metrics(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
diff --git a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/async_client.py b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/async_client.py
index cd02bb551f..38b7709448 100755
--- a/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/async_client.py
+++ b/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/async_client.py
@@ -27,6 +27,7 @@
 from google.auth import credentials as ga_credentials   # type: ignore
 from google.oauth2 import service_account              # type: ignore
 
+
 try:
     OptionalRetry = Union[retries.AsyncRetry, gapic_v1.method._MethodDefault, None]
 except AttributeError:  # pragma: NO COVER
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 0441f71b12..f22a2c227e 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
@@ -161,6 +161,7 @@ def test__get_api_endpoint():
         CloudRedisClient._get_api_endpoint(None, mock_client_cert_source, mock_universe, "auto")
     assert str(excinfo.value) == "mTLS is not supported in any universe other than googleapis.com."
 
+
 def test__get_universe_domain():
     client_universe_domain = "foo.com"
     universe_domain_env = "bar.com"
@@ -1169,15 +1170,15 @@ def test_list_instances_pager(transport_name: str = "grpc"):
             RuntimeError,
         )
 
-        metadata = ()
-        metadata = tuple(metadata) + (
+        expected_metadata = ()
+        expected_metadata = tuple(expected_metadata) + (
             gapic_v1.routing_header.to_grpc_metadata((
                 ('parent', ''),
             )),
         )
         pager = client.list_instances(request={})
 
-        assert pager._metadata == metadata
+        assert pager._metadata == expected_metadata
 
         results = list(pager)
         assert len(results) == 6
diff --git a/tests/system/test_api_version_header.py b/tests/system/test_api_version_header.py
new file mode 100644
index 0000000000..5fcb4be517
--- /dev/null
+++ b/tests/system/test_api_version_header.py
@@ -0,0 +1,42 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import pytest
+
+try:
+    from google.api_core import version_header
+except ImportError:
+    version_header = None
+
+
+def test_api_version_in_grpc_trailing_metadata(echo):
+    if not version_header:
+        pytest.skip(
+            "google-api-core>=2.19.0 is required for `google.api_core.version_header`"
+        )
+
+    # This feature requires version 0.35.0 of `gapic-showcase` or newer which has the
+    # ability to echo request headers
+    content = 'The hail in Wales falls mainly on the snails.'
+    responses = echo.expand({
+        'content': content,
+    })
+    if isinstance(echo.transport, type(echo).get_transport_class("grpc")):
+        response_metadata = [
+            (metadata.key, metadata.value)
+            for metadata in responses.trailing_metadata()
+        ]
+        assert ("x-goog-api-version", "v1_20240408") in response_metadata
+    else:
+        assert "X-Showcase-Request-X-Goog-Api-Version" in responses._response.headers
+        assert responses._response.headers["X-Showcase-Request-X-Goog-Api-Version"] == "v1_20240408"
diff --git a/tests/system/test_grpc_interceptor_streams.py b/tests/system/test_grpc_interceptor_streams.py
index 4b40b7611f..cbfa72d2ce 100644
--- a/tests/system/test_grpc_interceptor_streams.py
+++ b/tests/system/test_grpc_interceptor_streams.py
@@ -31,7 +31,11 @@ def test_unary_stream(intercepted_echo):
         assert response.content == ground_truth
     assert ground_truth == 'snails.'
 
-    assert responses.trailing_metadata() == intercepted_metadata
+    response_metadata = [
+        (metadata.key, metadata.value)
+        for metadata in responses.trailing_metadata()
+    ]
+    assert intercepted_metadata[0] in response_metadata
 
 
 def test_stream_stream(intercepted_echo):
@@ -43,4 +47,8 @@ def test_stream_stream(intercepted_echo):
     contents = [response.content for response in responses]
     assert contents == ['hello', 'world!']
 
-    assert responses.trailing_metadata() == intercepted_metadata
+    response_metadata = [
+        (metadata.key, metadata.value)
+        for metadata in responses.trailing_metadata()
+    ]
+    assert intercepted_metadata[0] in response_metadata
diff --git a/tests/system/test_streams.py b/tests/system/test_streams.py
index 8294e67958..aa8c84c84c 100644
--- a/tests/system/test_streams.py
+++ b/tests/system/test_streams.py
@@ -19,14 +19,14 @@
 from google import showcase
 
 
-metadata = (("showcase-trailer", "hello world"),)
+_METADATA = (("showcase-trailer", "hello world"),)
 
 
 def test_unary_stream(echo):
     content = 'The hail in Wales falls mainly on the snails.'
     responses = echo.expand({
         'content': content,
-    }, metadata=metadata)
+    }, metadata=_METADATA)
 
     # Consume the response and ensure it matches what we expect.
     # with pytest.raises(exceptions.NotFound) as exc:
@@ -34,7 +34,15 @@ def test_unary_stream(echo):
         assert response.content == ground_truth
     assert ground_truth == 'snails.'
     if isinstance(echo.transport, type(echo).get_transport_class("grpc")):
-        assert responses.trailing_metadata() == metadata
+        response_metadata = [
+            (metadata.key, metadata.value)
+            for metadata in responses.trailing_metadata()
+        ]
+        assert _METADATA[0] in response_metadata
+    else:
+        showcase_header = f"X-Showcase-Request-{_METADATA[0][0]}"
+        assert showcase_header in responses._response.headers
+        assert responses._response.headers[showcase_header] == _METADATA[0][1]
 
 
 def test_stream_unary(echo):
@@ -67,14 +75,18 @@ def test_stream_stream(echo):
     requests = []
     requests.append(showcase.EchoRequest(content="hello"))
     requests.append(showcase.EchoRequest(content="world!"))
-    responses = echo.chat(iter(requests), metadata=metadata)
+    responses = echo.chat(iter(requests), metadata=_METADATA)
 
     contents = []
     for response in responses:
         contents.append(response.content)
     assert contents == ['hello', 'world!']
 
-    assert responses.trailing_metadata() == metadata
+    response_metadata = [
+        (metadata.key, metadata.value)
+        for metadata in responses.trailing_metadata()
+    ]
+    assert _METADATA[0] in response_metadata
 
 
 def test_stream_stream_passing_dict(echo):
@@ -83,14 +95,18 @@ def test_stream_stream_passing_dict(echo):
         return
 
     requests = [{'content': 'hello'}, {'content': 'world!'}]
-    responses = echo.chat(iter(requests), metadata=metadata)
+    responses = echo.chat(iter(requests), metadata=_METADATA)
 
     contents = []
     for response in responses:
         contents.append(response.content)
     assert contents == ['hello', 'world!']
 
-    assert responses.trailing_metadata() == metadata
+    response_metadata = [
+        (metadata.key, metadata.value)
+        for metadata in responses.trailing_metadata()
+    ]
+    assert _METADATA[0] in response_metadata
 
 
 if os.environ.get("GAPIC_PYTHON_ASYNC", "true") == "true":
@@ -101,7 +117,7 @@ async def test_async_unary_stream_reader(async_echo):
         content = 'The hail in Wales falls mainly on the snails.'
         call = await async_echo.expand({
             'content': content,
-        }, metadata=metadata)
+        }, metadata=_METADATA)
 
         # Consume the response and ensure it matches what we expect.
         # with pytest.raises(exceptions.NotFound) as exc:
@@ -111,14 +127,14 @@ async def test_async_unary_stream_reader(async_echo):
         assert ground_truth == 'snails.'
 
         trailing_metadata = await call.trailing_metadata()
-        assert trailing_metadata == metadata
+        assert _METADATA[0] in trailing_metadata.items()
 
     @pytest.mark.asyncio
     async def test_async_unary_stream_async_generator(async_echo):
         content = 'The hail in Wales falls mainly on the snails.'
         call = await async_echo.expand({
             'content': content,
-        }, metadata=metadata)
+        }, metadata=_METADATA)
 
         # Consume the response and ensure it matches what we expect.
         # with pytest.raises(exceptions.NotFound) as exc:
@@ -129,7 +145,7 @@ async def test_async_unary_stream_async_generator(async_echo):
         assert ground_truth == 'snails.'
 
         trailing_metadata = await call.trailing_metadata()
-        assert trailing_metadata == metadata
+        assert _METADATA[0] in trailing_metadata.items()
 
     @pytest.mark.asyncio
     async def test_async_stream_unary_iterable(async_echo):
@@ -171,7 +187,7 @@ async def test_async_stream_unary_passing_dict(async_echo):
 
     @pytest.mark.asyncio
     async def test_async_stream_stream_reader_writier(async_echo):
-        call = await async_echo.chat(metadata=metadata)
+        call = await async_echo.chat(metadata=_METADATA)
         await call.write(showcase.EchoRequest(content="hello"))
         await call.write(showcase.EchoRequest(content="world!"))
         await call.done_writing()
@@ -183,7 +199,7 @@ async def test_async_stream_stream_reader_writier(async_echo):
         assert contents == ['hello', 'world!']
 
         trailing_metadata = await call.trailing_metadata()
-        assert trailing_metadata == metadata
+        assert _METADATA[0] in trailing_metadata.items()
 
     @pytest.mark.asyncio
     async def test_async_stream_stream_async_generator(async_echo):
@@ -192,7 +208,7 @@ async def async_generator():
             yield showcase.EchoRequest(content="hello")
             yield showcase.EchoRequest(content="world!")
 
-        call = await async_echo.chat(async_generator(), metadata=metadata)
+        call = await async_echo.chat(async_generator(), metadata=_METADATA)
 
         contents = []
         async for response in call:
@@ -200,12 +216,12 @@ async def async_generator():
         assert contents == ['hello', 'world!']
 
         trailing_metadata = await call.trailing_metadata()
-        assert trailing_metadata == metadata
+        assert _METADATA[0] in trailing_metadata.items()
 
     @pytest.mark.asyncio
     async def test_async_stream_stream_passing_dict(async_echo):
         requests = [{'content': 'hello'}, {'content': 'world!'}]
-        call = await async_echo.chat(iter(requests), metadata=metadata)
+        call = await async_echo.chat(iter(requests), metadata=_METADATA)
 
         contents = []
         async for response in call:
@@ -213,4 +229,4 @@ async def test_async_stream_stream_passing_dict(async_echo):
         assert contents == ['hello', 'world!']
 
         trailing_metadata = await call.trailing_metadata()
-        assert trailing_metadata == metadata
+        assert _METADATA[0] in trailing_metadata.items()
diff --git a/tests/unit/schema/wrappers/test_service.py b/tests/unit/schema/wrappers/test_service.py
index 2664167c72..98a5ce1f3e 100644
--- a/tests/unit/schema/wrappers/test_service.py
+++ b/tests/unit/schema/wrappers/test_service.py
@@ -71,6 +71,16 @@ def test_service_host():
     assert service.host == 'thingdoer.googleapis.com'
 
 
+def test_service_api_version_not_specified():
+    service = make_service(host='thingdoer.googleapis.com')
+    assert not service.version
+
+
+def test_service_api_version_exists():
+    service = make_service(host='thingdoer.googleapis.com', version="goose")
+    assert service.version == "goose"
+
+
 def test_service_no_host():
     service = make_service()
     assert not service.host