diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
index fba8615921af..d472c6fc6e2f 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2
@@ -50,122 +50,129 @@ except ImportError:  # pragma: NO COVER
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
 
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
 
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
 
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
+        )
 
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
 
 {% if has_auto_populated_fields %}
 
@@ -215,61 +222,64 @@ def setup_request_id(
 
 {% endif %}
 
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
-        )
-
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
 
 {% endblock %}
diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
index 737c5e34e7bb..c89737277701 100644
--- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
+++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
@@ -461,7 +461,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = {{ service.client_name }}._read_environment_variables()
         self._client_cert_source = {{ service.client_name }}._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe={{ service.client_name }}._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py
index 3939248a56c6..bfe8535d7000 100755
--- a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py
@@ -42,177 +42,187 @@ def should_use_client_cert():
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
-
-
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
+
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
+
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
         )
 
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
+
+
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
index 182cfc6017bf..5a50b2cae91b 100755
--- a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py
@@ -512,7 +512,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = AssetServiceClient._read_environment_variables()
         self._client_cert_source = AssetServiceClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=AssetServiceClient._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py
index 3939248a56c6..bfe8535d7000 100755
--- a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/_compat.py
@@ -42,177 +42,187 @@ def should_use_client_cert():
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
-
-
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
+
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
+
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
         )
 
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
+
+
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
index 4a3cb7bad6c3..71582db0730a 100755
--- a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py
@@ -449,7 +449,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = IAMCredentialsClient._read_environment_variables()
         self._client_cert_source = IAMCredentialsClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=IAMCredentialsClient._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py
index 3939248a56c6..bfe8535d7000 100755
--- a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/_compat.py
@@ -42,177 +42,187 @@ def should_use_client_cert():
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
-
-
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
+
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
+
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
         )
 
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
+
+
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py
index b015982530cc..c74702160f7d 100755
--- a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py
@@ -632,7 +632,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = EventarcClient._read_environment_variables()
         self._client_cert_source = EventarcClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=EventarcClient._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py
index 3939248a56c6..bfe8535d7000 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/_compat.py
@@ -42,177 +42,187 @@ def should_use_client_cert():
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
-
-
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
+
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
+
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
         )
 
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
+
+
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
index 36468f3d26c7..96befe769e23 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py
@@ -505,7 +505,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = ConfigServiceV2Client._read_environment_variables()
         self._client_cert_source = ConfigServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=ConfigServiceV2Client._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
index 6009409d3685..372a19816da2 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py
@@ -436,7 +436,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = LoggingServiceV2Client._read_environment_variables()
         self._client_cert_source = LoggingServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=LoggingServiceV2Client._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
index 1c353672fae1..ed4fe3142e5c 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py
@@ -437,7 +437,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = MetricsServiceV2Client._read_environment_variables()
         self._client_cert_source = MetricsServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=MetricsServiceV2Client._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py
index 3939248a56c6..bfe8535d7000 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/_compat.py
@@ -42,177 +42,187 @@ def should_use_client_cert():
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
-
-
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
+
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
+
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
         )
 
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
+
+
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py
index 0b3b7c49baba..b645d5b6ed26 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py
@@ -505,7 +505,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = BaseConfigServiceV2Client._read_environment_variables()
         self._client_cert_source = BaseConfigServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=BaseConfigServiceV2Client._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py
index 6009409d3685..372a19816da2 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py
@@ -436,7 +436,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = LoggingServiceV2Client._read_environment_variables()
         self._client_cert_source = LoggingServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=LoggingServiceV2Client._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py
index 7581350f8ec8..61416469b814 100755
--- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py
@@ -437,7 +437,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = BaseMetricsServiceV2Client._read_environment_variables()
         self._client_cert_source = BaseMetricsServiceV2Client._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=BaseMetricsServiceV2Client._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py
index 3939248a56c6..bfe8535d7000 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/_compat.py
@@ -42,177 +42,187 @@ def should_use_client_cert():
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
-
-
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
+
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
+
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
         )
 
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
+
+
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
index add3e3f67f59..985fdcffccaf 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py
@@ -477,7 +477,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = CloudRedisClient._read_environment_variables()
         self._client_cert_source = CloudRedisClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=CloudRedisClient._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py
index 3939248a56c6..bfe8535d7000 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/_compat.py
@@ -42,177 +42,187 @@ def should_use_client_cert():
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
-
-
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
-
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
-
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
-
-
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
+
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
+
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
+
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
         )
 
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
+
+
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py
index 1f6885be1999..7ba4a9e07f16 100755
--- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py
@@ -477,7 +477,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = CloudRedisClient._read_environment_variables()
         self._client_cert_source = CloudRedisClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=CloudRedisClient._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.
diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py
index ffa58f4f04eb..9605a29a513a 100755
--- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py
+++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/_compat.py
@@ -46,122 +46,129 @@ def should_use_client_cert():
             )
         return use_client_cert == "true"
 
-DEFAULT_UNIVERSE = "googleapis.com"
 
+try:
+    from google.api_core.universe import get_default_mtls_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Other URLs (including those that do not match these domain suffixes or
+        already contain '.mtls.') are passed through as-is.
+
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+
+        Returns:
+            Optional[str]: converted mTLS api endpoint.
+        """
+        if not api_endpoint or ".mtls." in api_endpoint.lower():
+            return api_endpoint
+
+        has_scheme = "://" in api_endpoint
+        if not has_scheme:
+            parsed = urlparse("//" + api_endpoint)
+        else:
+            parsed = urlparse(api_endpoint)
+
+        host = parsed.hostname
+        if not host:
+            return api_endpoint
+
+        port = f":{parsed.port}" if parsed.port else ""
+
+        lowered_host = host.lower()
+        suffix_sandbox = ".sandbox.googleapis.com"
+        suffix_google = ".googleapis.com"
+        if lowered_host.endswith(suffix_sandbox):
+            new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
+        elif lowered_host.endswith(suffix_google):
+            new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
+        else:
+            return api_endpoint
+
+        netloc = new_host + port
+        new_parsed = parsed._replace(netloc=netloc)
+
+        if not has_scheme:
+            return urlunparse(new_parsed)[2:]
+        else:
+            return urlunparse(new_parsed)
 
-def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
-    """Converts api endpoint to mTLS endpoint.
-
-    Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-    "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-    Other URLs (including those that do not match these domain suffixes or
-    already contain '.mtls.') are passed through as-is.
-
-    Args:
-        api_endpoint (Optional[str]): the api endpoint to convert.
-
-    Returns:
-        Optional[str]: converted mTLS api endpoint.
-    """
-    if not api_endpoint or ".mtls." in api_endpoint.lower():
-        return api_endpoint
-
-    has_scheme = "://" in api_endpoint
-    if not has_scheme:
-        parsed = urlparse("//" + api_endpoint)
-    else:
-        parsed = urlparse(api_endpoint)
-
-    host = parsed.hostname
-    if not host:
-        return api_endpoint
-
-    port = f":{parsed.port}" if parsed.port else ""
-
-    lowered_host = host.lower()
-    suffix_sandbox = ".sandbox.googleapis.com"
-    suffix_google = ".googleapis.com"
-    if lowered_host.endswith(suffix_sandbox):
-        new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com"
-    elif lowered_host.endswith(suffix_google):
-        new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com"
-    else:
-        return api_endpoint
-
-    netloc = new_host + port
-    new_parsed = parsed._replace(netloc=netloc)
-
-    if not has_scheme:
-        return urlunparse(new_parsed)[2:]
-    else:
-        return urlunparse(new_parsed)
-
-def get_api_endpoint(
-    api_override: Optional[str],
-    universe_domain: str,
-    default_universe: str,
-    default_mtls_endpoint: Optional[str],
-    default_endpoint_template: str,
-    use_mtls: bool,
-) -> str:
-    """Return the API endpoint used by the client.
-
-    Args:
-        api_override (Optional[str]): The API endpoint override. If specified,
-            this is always returned.
-        universe_domain (str): The universe domain used by the client.
-        default_universe (str): The default universe domain.
-        default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
-        default_endpoint_template (str): The default endpoint template containing
-            a placeholder `{UNIVERSE_DOMAIN}`.
-        use_mtls (bool): Whether to use the mTLS endpoint.
-
-    Returns:
-        str: The API endpoint to be used by the client.
-
-    Raises:
-        google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
-            not supported in the configured universe domain.
-        ValueError: If mTLS is requested but no mTLS endpoint is available.
-    """
-    if api_override is not None:
-        return api_override
-
-    if use_mtls:
-        if universe_domain.lower() != default_universe.lower():
-            raise MutualTLSChannelError(
-                f"mTLS is not supported in any universe other than {default_universe}."
-            )
-        if not default_mtls_endpoint:
-            raise ValueError("mTLS endpoint is not available.")
-        return default_mtls_endpoint
-    else:
-        return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
-
-def get_universe_domain(
-    *potential_universes: Optional[str],
-    default_universe: str = DEFAULT_UNIVERSE,
-) -> str:
-    """Return the universe domain used by the client.
-
-    Args:
-        *potential_universes (Optional[str]): Potential universe domains in order of preference.
-        default_universe (str): The default universe domain.
-
-    Returns:
-        str: The universe domain to be used by the client.
+try:
+    from google.api_core.universe import get_api_endpoint
+except ImportError:  # pragma: NO COVER
+    def get_api_endpoint(
+        api_override: Optional[str],
+        universe_domain: str,
+        default_universe: str,
+        default_mtls_endpoint: Optional[str],
+        default_endpoint_template: str,
+        use_mtls: bool,
+    ) -> str:
+        """Return the API endpoint used by the client.
+
+        Args:
+            api_override (Optional[str]): The API endpoint override. If specified,
+                this is always returned.
+            universe_domain (str): The universe domain used by the client.
+            default_universe (str): The default universe domain.
+            default_mtls_endpoint (Optional[str]): The default mTLS endpoint.
+            default_endpoint_template (str): The default endpoint template containing
+                a placeholder `{UNIVERSE_DOMAIN}`.
+            use_mtls (bool): Whether to use the mTLS endpoint.
+
+        Returns:
+            str: The API endpoint to be used by the client.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but
+                not supported in the configured universe domain.
+            ValueError: If mTLS is requested but no mTLS endpoint is available.
+        """
+        if api_override is not None:
+            return api_override
+
+        if use_mtls:
+            if universe_domain.lower() != default_universe.lower():
+                raise MutualTLSChannelError(
+                    f"mTLS is not supported in any universe other than {default_universe}."
+                )
+            if not default_mtls_endpoint:
+                raise ValueError("mTLS endpoint is not available.")
+            return default_mtls_endpoint
+        else:
+            return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)
 
-    Raises:
-        EmptyUniverseError: If the resolved universe domain is an empty string.
-    """
-    resolved = next(
-        (x.strip() for x in potential_universes if x is not None),
-        default_universe,
-    )
+try:
+    from google.api_core.universe import get_universe_domain
+except ImportError:  # pragma: NO COVER
+    def get_universe_domain(
+        *potential_universes: Optional[str],
+        default_universe: str,
+    ) -> str:
+        """Return the universe domain used by the client.
+
+        Args:
+            *potential_universes (Optional[str]): Potential universe domains in order of preference.
+            default_universe (str): The default universe domain.
+
+        Returns:
+            str: The universe domain to be used by the client.
+
+        Raises:
+            EmptyUniverseError: If the resolved universe domain is an empty string.
+        """
+        resolved = next(
+            (x.strip() for x in potential_universes if x is not None),
+            default_universe,
+        )
 
-    if not resolved:
-        raise EmptyUniverseError()
-    return resolved
+        if not resolved:
+            raise EmptyUniverseError()
+        return resolved
 
 
 def setup_request_id(
@@ -209,59 +216,62 @@ def setup_request_id(
             setattr(request, field_name, str(uuid.uuid4()))
 
 
-def transcode_request(
-    http_options: List[Dict[str, str]],
-    request: Any,
-    required_fields_default_values: Optional[Dict[str, Any]] = None,
-    rest_numeric_enums: bool = False,
-) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
-    """Transcodes a request into HTTP method, URI, body, and query parameters.
-
-    Args:
-        http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
-        request (Any): The protobuf or proto-plus request message.
-        required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
-            of required fields default values to merge into query parameters if missing.
-        rest_numeric_enums (bool): Whether to encode enums as integers.
-
-    Returns:
-        Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
-            - The raw transcoded request dictionary (containing keys like 'uri', 'method').
-            - The serialized request body JSON string, or None if no body.
-            - The query parameters dictionary.
-    """
-    if request is None:
-        raise TypeError("request cannot be None")
-
-    # Convert proto-plus message to its underlying protobuf message if needed
-    pb_request = getattr(request, "_pb", request)
-
-    transcoded_request = path_template.transcode(http_options, pb_request)
-
-    body_json = None
-    if transcoded_request.get("body") is not None:
-        body_json = json_format.MessageToJson(
-            transcoded_request["body"],
-            use_integers_for_enums=rest_numeric_enums,
-        )
-
-    query_params_json = {}
-    if transcoded_request.get("query_params") is not None:
-        query_params_json = json.loads(
-            json_format.MessageToJson(
-                transcoded_request["query_params"],
+try:
+    from google.api_core.rest_helpers import transcode_request # type: ignore
+except ImportError:  # pragma: NO COVER
+    def transcode_request(
+        http_options: List[Dict[str, str]],
+        request: Any,
+        required_fields_default_values: Optional[Dict[str, Any]] = None,
+        rest_numeric_enums: bool = False,
+    ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]:
+        """Transcodes a request into HTTP method, URI, body, and query parameters.
+
+        Args:
+            http_options (List[Dict[str, str]]): List of HTTP transcoding rules.
+            request (Any): The protobuf or proto-plus request message.
+            required_fields_default_values (Optional[Dict[str, Any]]): Dictionary
+                of required fields default values to merge into query parameters if missing.
+            rest_numeric_enums (bool): Whether to encode enums as integers.
+
+        Returns:
+            Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing:
+                - The raw transcoded request dictionary (containing keys like 'uri', 'method').
+                - The serialized request body JSON string, or None if no body.
+                - The query parameters dictionary.
+        """
+        if request is None:
+            raise TypeError("request cannot be None")
+
+        # Convert proto-plus message to its underlying protobuf message if needed
+        pb_request = getattr(request, "_pb", request)
+
+        transcoded_request = path_template.transcode(http_options, pb_request)
+
+        body_json = None
+        if transcoded_request.get("body") is not None:
+            body_json = json_format.MessageToJson(
+                transcoded_request["body"],
                 use_integers_for_enums=rest_numeric_enums,
             )
-        )
 
-    # If required_fields_default_values is provided, we merge default values for missing
-    # required fields into the query parameters.
-    if required_fields_default_values:
-        for k, v in required_fields_default_values.items():
-            if k not in query_params_json:
-                query_params_json[k] = v
+        query_params_json = {}
+        if transcoded_request.get("query_params") is not None:
+            query_params_json = json.loads(
+                json_format.MessageToJson(
+                    transcoded_request["query_params"],
+                    use_integers_for_enums=rest_numeric_enums,
+                )
+            )
+
+        # If required_fields_default_values is provided, we merge default values for missing
+        # required fields into the query parameters.
+        if required_fields_default_values:
+            for k, v in required_fields_default_values.items():
+                if k not in query_params_json:
+                    query_params_json[k] = v
 
-    if rest_numeric_enums:
-        query_params_json["$alt"] = "json;enum-encoding=int"
+        if rest_numeric_enums:
+            query_params_json["$alt"] = "json;enum-encoding=int"
 
-    return transcoded_request, body_json, query_params_json
+        return transcoded_request, body_json, query_params_json
diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py
index d8c6bd9009a7..ee8c6ac679a6 100755
--- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py
+++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py
@@ -473,7 +473,7 @@ def __init__(self, *,
 
         self._use_client_cert, self._use_mtls_endpoint, self._universe_domain_env = StorageBatchOperationsClient._read_environment_variables()
         self._client_cert_source = StorageBatchOperationsClient._get_client_cert_source(self._client_options.client_cert_source, self._use_client_cert)
-        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env)
+        self._universe_domain = get_universe_domain(universe_domain_opt, self._universe_domain_env, default_universe=StorageBatchOperationsClient._DEFAULT_UNIVERSE)
         self._api_endpoint: str = ""  # updated below, depending on `transport`
 
         # Initialize the universe domain validation.