diff --git a/tools/run_tests/xds_k8s_test_driver/framework/xds_url_map_testcase.py b/tools/run_tests/xds_k8s_test_driver/framework/xds_url_map_testcase.py
index a3212e42177..8eb363b4da4 100644
--- a/tools/run_tests/xds_k8s_test_driver/framework/xds_url_map_testcase.py
+++ b/tools/run_tests/xds_k8s_test_driver/framework/xds_url_map_testcase.py
@@ -18,6 +18,7 @@ from dataclasses import dataclass
 import datetime
 import json
 import os
+import re
 import sys
 import time
 from typing import Any, Iterable, Mapping, Optional, Tuple, Union
@@ -59,6 +60,10 @@ JsonType = Any
 RpcTypeUnaryCall = 'UNARY_CALL'
 RpcTypeEmptyCall = 'EMPTY_CALL'
 
+# All client languages
+_CLIENT_LANGUAGES = ('cpp', 'java', 'go', 'python')
+_SERVER_LANGUAGES = _CLIENT_LANGUAGES
+
 
 def _split_camel(s: str, delimiter: str = '-') -> str:
     """Turn camel case name to snake-case-like name."""
@@ -66,6 +71,10 @@ def _split_camel(s: str, delimiter: str = '-') -> str:
                    for c in s).lstrip(delimiter)
 
 
+def _get_lang(image_name: str) -> str:
+    return re.search(r'/(\w+)-(client|server):', image_name).group(1)
+
+
 class DumpedXdsConfig(dict):
     """A convenience class to check xDS config.
 
@@ -221,6 +230,24 @@ class XdsUrlMapTestCase(absltest.TestCase, metaclass=_MetaXdsUrlMapTestCase):
     - rpc_distribution_validate: Validates if the routing behavior is correct
     """
 
+    @staticmethod
+    def supported_clients() -> Tuple[str]:
+        """Declare supported languages of clients.
+
+        Returns:
+          A tuple of strings contains the supported languages for this test.
+        """
+        return _CLIENT_LANGUAGES
+
+    @staticmethod
+    def supported_servers() -> Tuple[str]:
+        """Declare supported languages of servers.
+
+        Returns:
+          A tuple of strings contains the supported languages for this test.
+        """
+        return _SERVER_LANGUAGES
+
     @staticmethod
     def client_init_config(rpc: str, metadata: str) -> Tuple[str, str]:
         """Updates the initial RPC configs for this test case.
@@ -302,6 +329,22 @@ class XdsUrlMapTestCase(absltest.TestCase, metaclass=_MetaXdsUrlMapTestCase):
             # Create the GCP resource once before the first test start
             GcpResourceManager().setup(cls.test_case_classes)
         cls.started_test_cases.add(cls.__name__)
+        # NOTE(lidiz) a manual skip mechanism is needed because absl/flags
+        # cannot be used in the built-in test-skipping decorators. See the
+        # official FAQs:
+        # https://abseil.io/docs/python/guides/flags#faqs
+        client_lang = _get_lang(GcpResourceManager().client_image)
+        server_lang = _get_lang(GcpResourceManager().server_image)
+        if client_lang not in cls.supported_clients():
+            cls.skip_reason = (f'Unsupported client language {client_lang} '
+                               f'not in {cls.supported_clients()}')
+            return
+        elif server_lang not in cls.supported_servers():
+            cls.skip_reason = (f'Unsupported server language {server_lang} '
+                               f'not in {cls.supported_servers()}')
+            return
+        else:
+            cls.skip_reason = None
         # TODO(lidiz) concurrency is possible, pending multiple-instance change
         GcpResourceManager().test_client_runner.cleanup(force=True)
         # Start the client, and allow the test to override the initial RPC config.
@@ -316,7 +359,8 @@ class XdsUrlMapTestCase(absltest.TestCase, metaclass=_MetaXdsUrlMapTestCase):
 
     @classmethod
     def tearDownClass(cls):
-        GcpResourceManager().test_client_runner.cleanup(force=True)
+        if cls.skip_reason is not None:
+            GcpResourceManager().test_client_runner.cleanup(force=True)
         cls.finished_test_cases.add(cls.__name__)
         if cls.finished_test_cases == cls.test_case_names:
             # Tear down the GCP resource after all tests finished
@@ -355,6 +399,9 @@ class XdsUrlMapTestCase(absltest.TestCase, metaclass=_MetaXdsUrlMapTestCase):
             super().run(result)
 
     def test_client_config(self):
+        if self.skip_reason:
+            logging.info('Skipping: %s', self.skip_reason)
+            self.skipTest(self.skip_reason)
         self._last_xds_config_exception = None
         retryer = retryers.constant_retryer(
             wait_fixed=datetime.timedelta(
@@ -371,6 +418,9 @@ class XdsUrlMapTestCase(absltest.TestCase, metaclass=_MetaXdsUrlMapTestCase):
                     self._xds_json_config))
 
     def test_rpc_distribution(self):
+        if self.skip_reason:
+            logging.info('Skipping: %s', self.skip_reason)
+            self.skipTest(self.skip_reason)
         self.rpc_distribution_validate(self.test_client)
 
     @staticmethod
diff --git a/tools/run_tests/xds_k8s_test_driver/tests/url_map/affinity_test.py b/tools/run_tests/xds_k8s_test_driver/tests/url_map/affinity_test.py
index d8caf0c6088..fdbad45949d 100644
--- a/tools/run_tests/xds_k8s_test_driver/tests/url_map/affinity_test.py
+++ b/tools/run_tests/xds_k8s_test_driver/tests/url_map/affinity_test.py
@@ -53,11 +53,12 @@ _TEST_METADATA = (
 _ChannelzChannelState = grpc_channelz.ChannelState
 
 
-@absltest.skipUnless('cpp-client' in xds_k8s_flags.CLIENT_IMAGE.value or \
-                     'java-client' in xds_k8s_flags.CLIENT_IMAGE.value,
-                     'Affinity is currently only implemented in C++ and Java.')
 class TestHeaderBasedAffinity(xds_url_map_testcase.XdsUrlMapTestCase):
 
+    @staticmethod
+    def supported_clients() -> Tuple[str]:
+        return 'cpp', 'java'
+
     @staticmethod
     def client_init_config(rpc: str, metadata: str):
         # Config the init RPCs to send with the same set of metadata. Without
@@ -116,12 +117,13 @@ class TestHeaderBasedAffinity(xds_url_map_testcase.XdsUrlMapTestCase):
         )
 
 
-@absltest.skipUnless('cpp-client' in xds_k8s_flags.CLIENT_IMAGE.value or \
-                     'java-client' in xds_k8s_flags.CLIENT_IMAGE.value,
-                     'Affinity is currently only implemented in C++ and Java.')
 class TestHeaderBasedAffinityMultipleHeaders(
         xds_url_map_testcase.XdsUrlMapTestCase):
 
+    @staticmethod
+    def supported_clients() -> Tuple[str]:
+        return 'cpp', 'java'
+
     @staticmethod
     def client_init_config(rpc: str, metadata: str):
         # Config the init RPCs to send with the same set of metadata. Without
@@ -215,3 +217,6 @@ class TestHeaderBasedAffinityMultipleHeaders(
 # TODO: add more test cases
 # 1. based on the basic test, turn down the backend in use, then verify that all
 #    RPCs are sent to another backend
+
+if __name__ == '__main__':
+    absltest.main()
diff --git a/tools/run_tests/xds_k8s_test_driver/tests/url_map/retry_test.py b/tools/run_tests/xds_k8s_test_driver/tests/url_map/retry_test.py
index 8f61b10e1f9..ab0a2c6be5f 100644
--- a/tools/run_tests/xds_k8s_test_driver/tests/url_map/retry_test.py
+++ b/tools/run_tests/xds_k8s_test_driver/tests/url_map/retry_test.py
@@ -63,11 +63,12 @@ def _build_retry_route_rule(retryConditions, num_retries):
     }
 
 
-@absltest.skipUnless('cpp-client' in xds_k8s_flags.CLIENT_IMAGE.value or \
-                     'java-client' in xds_k8s_flags.CLIENT_IMAGE.value,
-                     'Xds-retry is currently only implemented in C++ and Java.')
 class TestRetryUpTo3AttemptsAndFail(xds_url_map_testcase.XdsUrlMapTestCase):
 
+    @staticmethod
+    def supported_clients() -> Tuple[str]:
+        return 'cpp', 'java'
+
     @staticmethod
     def url_map_change(
             host_rule: HostRule,
@@ -101,11 +102,12 @@ class TestRetryUpTo3AttemptsAndFail(xds_url_map_testcase.XdsUrlMapTestCase):
                                  tolerance=_NON_RANDOM_ERROR_TOLERANCE)
 
 
-@absltest.skipUnless('cpp-client' in xds_k8s_flags.CLIENT_IMAGE.value or \
-                     'java-client' in xds_k8s_flags.CLIENT_IMAGE.value,
-                     'Xds-retry is currently only implemented in C++ Java.')
 class TestRetryUpTo4AttemptsAndSucceed(xds_url_map_testcase.XdsUrlMapTestCase):
 
+    @staticmethod
+    def supported_clients() -> Tuple[str]:
+        return 'cpp', 'java'
+
     @staticmethod
     def url_map_change(
             host_rule: HostRule,
diff --git a/tools/run_tests/xds_k8s_test_driver/tests/url_map/timeout_test.py b/tools/run_tests/xds_k8s_test_driver/tests/url_map/timeout_test.py
index 4f639bd84d1..dd81306bef9 100644
--- a/tools/run_tests/xds_k8s_test_driver/tests/url_map/timeout_test.py
+++ b/tools/run_tests/xds_k8s_test_driver/tests/url_map/timeout_test.py
@@ -78,12 +78,14 @@ class _BaseXdsTimeOutTestCase(XdsUrlMapTestCase):
         raise NotImplementedError()
 
 
-# TODO(lidiz) either add support for rpc-behavior to other languages, or we
-# should always use Java server as backend.
-@absltest.skipUnless('java-server' in xds_k8s_flags.SERVER_IMAGE.value,
-                     'Only Java server supports the rpc-behavior metadata.')
 class TestTimeoutInRouteRule(_BaseXdsTimeOutTestCase):
 
+    @staticmethod
+    def supported_servers() -> Tuple[str]:
+        # TODO(lidiz) either add support for rpc-behavior to other languages, or we
+        # should always use Java server as backend.
+        return 'java',
+
     def rpc_distribution_validate(self, test_client: XdsTestClient):
         rpc_distribution = self.configure_and_send(
             test_client,
@@ -107,10 +109,12 @@ class TestTimeoutInRouteRule(_BaseXdsTimeOutTestCase):
             tolerance=_ERROR_TOLERANCE)
 
 
-@absltest.skipUnless('java-server' in xds_k8s_flags.SERVER_IMAGE.value,
-                     'Only Java server supports the rpc-behavior metadata.')
 class TestTimeoutInApplication(_BaseXdsTimeOutTestCase):
 
+    @staticmethod
+    def supported_servers() -> Tuple[str]:
+        return 'java',
+
     def rpc_distribution_validate(self, test_client: XdsTestClient):
         rpc_distribution = self.configure_and_send(
             test_client,