diff --git a/kopf/__init__.py b/kopf/__init__.py
index 49b0abe1..8e51bee6 100644
--- a/kopf/__init__.py
+++ b/kopf/__init__.py
@@ -150,6 +150,7 @@
 from kopf._core.intents.piggybacking import (
     login_via_pykube,
     login_via_client,
+    login_via_async_client,
     login_with_kubeconfig,
     login_with_service_account,
 )
@@ -184,6 +185,7 @@
     'configure', 'LogFormat',
     'login_via_pykube',
     'login_via_client',
+    'login_via_async_client',
     'login_with_kubeconfig',
     'login_with_service_account',
     'LoginError',
diff --git a/kopf/_core/intents/piggybacking.py b/kopf/_core/intents/piggybacking.py
index 882f78e2..79f117e9 100644
--- a/kopf/_core/intents/piggybacking.py
+++ b/kopf/_core/intents/piggybacking.py
@@ -20,6 +20,7 @@
 
 # Keep as constants to make them patchable. Higher priority is more preferred.
 PRIORITY_OF_CLIENT: int = 10
+PRIORITY_OF_ASYNC_CLIENT: int = 15
 PRIORITY_OF_PYKUBE: int = 20
 
 # Rudimentary logins are added only if the clients are absent, so the priorities can overlap.
@@ -28,6 +29,10 @@
 
 
 def has_client() -> bool:
+    return has_sync_client() or has_async_client()
+
+
+def has_sync_client() -> bool:
     try:
         import kubernetes
     except ImportError:
@@ -36,6 +41,15 @@ def has_client() -> bool:
         return True
 
 
+def has_async_client() -> bool:
+    try:
+        import kubernetes_asyncio
+    except ImportError:
+        return False
+    else:
+        return True
+
+
 def has_pykube() -> bool:
     try:
         import pykube
@@ -102,6 +116,63 @@ def login_via_client(
     )
 
 
+# This is basically a copy of `login_via_client` changed to use the
+# kubernetes_asyncio client library.
+async def login_via_async_client(
+        *,
+        logger: typedefs.Logger,
+        **_: Any,
+) -> Optional[credentials.ConnectionInfo]:
+
+    # Keep imports in the function, as module imports are mocked in some tests.
+    try:
+        import kubernetes_asyncio.config
+    except ImportError:
+        return None
+
+    try:
+        kubernetes_asyncio.config.load_incluster_config()  # cluster env vars
+        logger.debug("Async client is configured in cluster with service account.")
+    except kubernetes_asyncio.config.ConfigException as e1:
+        try:
+            await kubernetes_asyncio.config.load_kube_config()  # developer's config files
+            logger.debug("Async client is configured via kubeconfig file.")
+        except kubernetes_asyncio.config.ConfigException as e2:
+            raise credentials.LoginError("Cannot authenticate the async client library "
+                                         "neither in-cluster, nor via kubeconfig.")
+
+    # We do not even try to understand how it works and why. Just load it, and extract the results.
+    # For kubernetes client >= 12.0.0 use the new 'get_default_copy' method
+    if callable(getattr(kubernetes_asyncio.client.Configuration, 'get_default_copy', None)):
+        config = kubernetes_asyncio.client.Configuration.get_default_copy()
+    else:
+        config = kubernetes_asyncio.client.Configuration()
+
+    # For auth-providers, this method is monkey-patched with the auth-provider's one.
+    # We need the actual auth-provider's token, so we call it instead of accessing api_key.
+    # Other keys (token, tokenFile) also end up being retrieved via this method.
+    header: Optional[str] = config.get_api_key_with_prefix('BearerToken')
+    parts: Sequence[str] = header.split(' ', 1) if header else []
+    scheme, token = ((None, None) if len(parts) == 0 else
+                     (None, parts[0]) if len(parts) == 1 else
+                     (parts[0], parts[1]))  # RFC-7235, Appendix C.
+
+    # Interpret the config object for our own minimalistic credentials.
+    # Note: kubernetes client has no concept of a "current" context's namespace.
+    return credentials.ConnectionInfo(
+        server=config.host,
+        ca_path=config.ssl_ca_cert,  # can be a temporary file
+        insecure=not config.verify_ssl,
+        username=config.username or None,  # an empty string when not defined
+        password=config.password or None,  # an empty string when not defined
+        scheme=scheme,
+        token=token,
+        certificate_path=config.cert_file,  # can be a temporary file
+        private_key_path=config.key_file,  # can be a temporary file
+        priority=PRIORITY_OF_ASYNC_CLIENT,
+    )
+
+
 def login_via_pykube(
         *,
         logger: typedefs.Logger,
diff --git a/kopf/_core/intents/registries.py b/kopf/_core/intents/registries.py
index df0efb19..15f0989e 100644
--- a/kopf/_core/intents/registries.py
+++ b/kopf/_core/intents/registries.py
@@ -279,14 +279,24 @@ def __init__(self) -> None:
                 _fallback=True,
             ))
         if piggybacking.has_client():
-            self._activities.append(handlers.ActivityHandler(
-                id=ids.HandlerId('login_via_client'),
-                fn=piggybacking.login_via_client,
-                activity=causes.Activity.AUTHENTICATION,
-                errors=execution.ErrorsMode.IGNORED,
-                param=None, timeout=None, retries=None, backoff=None,
-                _fallback=True,
-            ))
+            if piggybacking.has_sync_client():
+                self._activities.append(handlers.ActivityHandler(
+                    id=ids.HandlerId('login_via_client'),
+                    fn=piggybacking.login_via_client,
+                    activity=causes.Activity.AUTHENTICATION,
+                    errors=execution.ErrorsMode.IGNORED,
+                    param=None, timeout=None, retries=None, backoff=None,
+                    _fallback=True,
+                ))
+            elif piggybacking.has_async_client():
+                self._activities.append(handlers.ActivityHandler(
+                    id=ids.HandlerId('login_via_async_client'),
+                    fn=piggybacking.login_via_async_client,
+                    activity=causes.Activity.AUTHENTICATION,
+                    errors=execution.ErrorsMode.IGNORED,
+                    param=None, timeout=None, retries=None, backoff=None,
+                    _fallback=True,
+                ))
 
         # As a last resort, fall back to rudimentary logins if no advanced ones are available.
         thirdparties_present = piggybacking.has_pykube() or piggybacking.has_client()