diff --git a/src/azure-cli-core/azure/cli/core/_debug.py b/src/azure-cli-core/azure/cli/core/_debug.py
index d8b7358f985..e66b5b1c386 100644
--- a/src/azure-cli-core/azure/cli/core/_debug.py
+++ b/src/azure-cli-core/azure/cli/core/_debug.py
@@ -31,6 +31,22 @@ def change_ssl_cert_verification(client):
     return client
 
 
+def change_ssl_cert_verification_track2():
+    client_kwargs = {}
+    if should_disable_connection_verify():
+        logger.warning("Connection verification disabled by environment variable %s",
+                       DISABLE_VERIFY_VARIABLE_NAME)
+        os.environ[ADAL_PYTHON_SSL_NO_VERIFY] = '1'
+        client_kwargs['connection_verify'] = False
+    elif REQUESTS_CA_BUNDLE in os.environ:
+        ca_bundle_file = os.environ[REQUESTS_CA_BUNDLE]
+        if not os.path.isfile(ca_bundle_file):
+            raise CLIError('REQUESTS_CA_BUNDLE environment variable is specified with an invalid file path')
+        logger.debug("Using CA bundle file at '%s'.", ca_bundle_file)
+        client_kwargs['connection_verify'] = ca_bundle_file
+    return client_kwargs
+
+
 def allow_debug_adal_connection():
     if should_disable_connection_verify():
         os.environ[ADAL_PYTHON_SSL_NO_VERIFY] = '1'
diff --git a/src/azure-cli-core/azure/cli/core/adal_authentication.py b/src/azure-cli-core/azure/cli/core/adal_authentication.py
index e1609fce77c..9b9d0db6db9 100644
--- a/src/azure-cli-core/azure/cli/core/adal_authentication.py
+++ b/src/azure-cli-core/azure/cli/core/adal_authentication.py
@@ -3,11 +3,12 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+import time
 import requests
 import adal
 
 from msrest.authentication import Authentication
-
+from azure.core.credentials import AccessToken
 from azure.cli.core.util import in_cloud_console
 
 from knack.util import CLIError
@@ -19,11 +20,10 @@ def __init__(self, token_retriever, external_tenant_token_retriever=None):
         self._token_retriever = token_retriever
         self._external_tenant_token_retriever = external_tenant_token_retriever
 
-    def signed_session(self, session=None):  # pylint: disable=arguments-differ
-        session = session or super(AdalAuthentication, self).signed_session()
+    def _get_token(self):
         external_tenant_tokens = None
         try:
-            scheme, token, _ = self._token_retriever()
+            scheme, token, full_token = self._token_retriever()
             if self._external_tenant_token_retriever:
                 external_tenant_tokens = self._external_tenant_token_retriever()
         except CLIError as err:
@@ -55,6 +55,20 @@ def signed_session(self, session=None):  # pylint: disable=arguments-differ
         except requests.exceptions.ConnectionError as err:
             raise CLIError('Please ensure you have network connection. Error detail: ' + str(err))
 
+        return scheme, token, full_token, external_tenant_tokens
+
+    # This method is exposed for Azure Core.
+    def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
+        _, token, full_token, _ = self._get_token()
+
+        return AccessToken(token, int(full_token['expiresIn'] + time.time()))
+
+    # This method is exposed for msrest.
+    def signed_session(self, session=None):  # pylint: disable=arguments-differ
+        session = session or super(AdalAuthentication, self).signed_session()
+
+        scheme, token, _, external_tenant_tokens = self._get_token()
+
         header = "{} {}".format(scheme, token)
         session.headers['Authorization'] = header
         if external_tenant_tokens:
diff --git a/src/azure-cli-core/azure/cli/core/commands/__init__.py b/src/azure-cli-core/azure/cli/core/commands/__init__.py
index 4a336d9cbf3..b3db2856378 100644
--- a/src/azure-cli-core/azure/cli/core/commands/__init__.py
+++ b/src/azure-cli-core/azure/cli/core/commands/__init__.py
@@ -421,15 +421,21 @@ def _get_operation():
         return _get_operation()
 
 
-def cached_put(cmd_obj, operation, parameters, *args, **kwargs):
-
+def cached_put(cmd_obj, operation, parameters, *args, setter_arg_name='parameters', **kwargs):
+    """
+    setter_arg_name: The name of the argument in the setter which corresponds to the object being updated.
+    In track2, unknown kwargs will raise, so we should not pass 'parameters" for operation when the name of the argument
+    in the setter which corresponds to the object being updated is not 'parameters'.
+    """
     def _put_operation():
         result = None
         if args:
             extended_args = args + (parameters,)
             result = operation(*extended_args)
         elif kwargs is not None:
-            result = operation(parameters=parameters, **kwargs)
+            kwargs[setter_arg_name] = parameters
+            result = operation(**kwargs)
+            del kwargs[setter_arg_name]
         return result
 
     # early out if the command does not use the cache
@@ -1067,7 +1073,8 @@ def _is_paged(obj):
             and not isinstance(obj, list) \
             and not isinstance(obj, dict):
         from msrest.paging import Paged
-        return isinstance(obj, Paged)
+        from azure.core.paging import ItemPaged as AzureCorePaged
+        return isinstance(obj, (AzureCorePaged, Paged))
     return False
 
 
diff --git a/src/azure-cli-core/azure/cli/core/commands/arm.py b/src/azure-cli-core/azure/cli/core/commands/arm.py
index 336d7a9b304..a33c70e0e99 100644
--- a/src/azure-cli-core/azure/cli/core/commands/arm.py
+++ b/src/azure-cli-core/azure/cli/core/commands/arm.py
@@ -566,7 +566,7 @@ def handler(args):  # pylint: disable=too-many-branches,too-many-statements
         if setter_arg_name == 'parameters':
             result = cached_put(cmd, setter, **setterargs)
         else:
-            result = cached_put(cmd, setter, setterargs[setter_arg_name], **setterargs)
+            result = cached_put(cmd, setter, setterargs[setter_arg_name], setter_arg_name=setter_arg_name, **setterargs)
 
         if supports_no_wait and no_wait_enabled:
             return None
diff --git a/src/azure-cli-core/azure/cli/core/commands/client_factory.py b/src/azure-cli-core/azure/cli/core/commands/client_factory.py
index 478096580c8..5c77e89889c 100644
--- a/src/azure-cli-core/azure/cli/core/commands/client_factory.py
+++ b/src/azure-cli-core/azure/cli/core/commands/client_factory.py
@@ -7,7 +7,7 @@
 from azure.cli.core.extension import EXTENSIONS_MOD_PREFIX
 from azure.cli.core.profiles._shared import get_client_class, SDKProfile
 from azure.cli.core.profiles import ResourceType, CustomResourceType, get_api_version, get_sdk
-from azure.cli.core.util import get_az_user_agent
+from azure.cli.core.util import get_az_user_agent, is_track2
 
 from knack.log import get_logger
 from knack.util import CLIError
@@ -102,6 +102,34 @@ def configure_common_settings(cli_ctx, client):
     client.config.generate_client_request_id = 'x-ms-client-request-id' not in cli_ctx.data['headers']
 
 
+def configure_common_settings_track2(cli_ctx):
+    client_kwargs = {}
+
+    client_kwargs.update(_debug.change_ssl_cert_verification_track2())
+
+    client_kwargs['logging_enable'] = True
+    client_kwargs['user_agent'] = get_az_user_agent()
+
+    try:
+        command_ext_name = cli_ctx.data['command_extension_name']
+        if command_ext_name:
+            client_kwargs['user_agent'] += "CliExtension/{}".format(command_ext_name)
+    except KeyError:
+        pass
+
+    headers = dict(cli_ctx.data['headers'])
+    command_name_suffix = ';completer-request' if cli_ctx.data['completer_active'] else ''
+    headers['CommandName'] = "{}{}".format(cli_ctx.data['command'], command_name_suffix)
+    if cli_ctx.data.get('safe_params'):
+        headers['ParameterSetName'] = ' '.join(cli_ctx.data['safe_params'])
+    client_kwargs['headers'] = headers
+
+    if 'x-ms-client-request-id' in cli_ctx.data['headers']:
+        client_kwargs['request_id'] = cli_ctx.data['headers']['x-ms-client-request-id']
+
+    return client_kwargs
+
+
 def _get_mgmt_service_client(cli_ctx,
                              client_type,
                              subscription_bound=True,
@@ -131,12 +159,16 @@ def _get_mgmt_service_client(cli_ctx,
     if kwargs:
         client_kwargs.update(kwargs)
 
+    if is_track2(client_type):
+        client_kwargs.update(configure_common_settings_track2(cli_ctx))
+
     if subscription_bound:
         client = client_type(cred, subscription_id, **client_kwargs)
     else:
         client = client_type(cred, **client_kwargs)
 
-    configure_common_settings(cli_ctx, client)
+    if not is_track2(client):
+        configure_common_settings(cli_ctx, client)
 
     return client, subscription_id
 
diff --git a/src/azure-cli-core/azure/cli/core/util.py b/src/azure-cli-core/azure/cli/core/util.py
index 1be9fc6e7ed..61e571410c6 100644
--- a/src/azure-cli-core/azure/cli/core/util.py
+++ b/src/azure-cli-core/azure/cli/core/util.py
@@ -16,9 +16,12 @@
 import logging
 
 from six.moves.urllib.request import urlopen  # pylint: disable=import-error
+
+from azure.common import AzureException
+from azure.core.exceptions import AzureError
 from knack.log import get_logger
 from knack.util import CLIError, to_snake_case
-from azure.common import AzureException
+from inspect import getfullargspec as get_arg_spec
 
 logger = get_logger(__name__)
 
@@ -51,7 +54,7 @@ def handle_exception(ex):  # pylint: disable=too-many-return-statements
             logger.error("To learn more about --query, please visit: "
                          "https://docs.microsoft.com/cli/azure/query-azure-cli?view=azure-cli-latest")
             return 1
-        if isinstance(ex, (CLIError, CloudError, AzureException)):
+        if isinstance(ex, (CLIError, CloudError, AzureException, AzureError)):
             logger.error(ex.args[0])
             try:
                 for detail in ex.args[0].error.details:
@@ -365,6 +368,14 @@ def get_arg_list(op):
         return sig.args
 
 
+def is_track2(client_class):
+    """ IS this client a autorestv3/track2 one?.
+    Could be refined later if necessary.
+    """
+    args = get_arg_spec(client_class.__init__).args
+    return "credential" in args
+
+
 DISABLE_VERIFY_VARIABLE_NAME = "AZURE_CLI_DISABLE_CONNECTION_VERIFICATION"
 
 
@@ -376,7 +387,8 @@ def should_disable_connection_verify():
 def poller_classes():
     from msrestazure.azure_operation import AzureOperationPoller
     from msrest.polling.poller import LROPoller
-    return (AzureOperationPoller, LROPoller)
+    from azure.core.polling import LROPoller as AzureCoreLROPoller
+    return (AzureOperationPoller, LROPoller, AzureCoreLROPoller)
 
 
 def augment_no_wait_handler_args(no_wait_enabled, handler, handler_args):
@@ -394,7 +406,7 @@ def augment_no_wait_handler_args(no_wait_enabled, handler, handler_args):
 
 def sdk_no_wait(no_wait, func, *args, **kwargs):
     if no_wait:
-        kwargs.update({'raw': True, 'polling': False})
+        kwargs.update({'polling': False})
     return func(*args, **kwargs)
 
 
diff --git a/src/azure-cli-core/setup.py b/src/azure-cli-core/setup.py
index 3b7772641fd..ccef20817fc 100644
--- a/src/azure-cli-core/setup.py
+++ b/src/azure-cli-core/setup.py
@@ -67,6 +67,7 @@
     'six~=1.12',
     'wheel==0.30.0',
     'azure-mgmt-resource==8.0.1',
+    'azure-mgmt-core==1.0.0'
 ]
 
 TESTS_REQUIRE = [
diff --git a/src/azure-cli/azure/cli/command_modules/advisor/custom.py b/src/azure-cli/azure/cli/command_modules/advisor/custom.py
index ca3fddac678..f1d475a0508 100644
--- a/src/azure-cli/azure/cli/command_modules/advisor/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/advisor/custom.py
@@ -4,7 +4,6 @@
 # --------------------------------------------------------------------------------------------
 
 import uuid
-from azure.cli.core.util import sdk_no_wait
 
 
 def list_recommendations(client, ids=None, resource_group_name=None,
@@ -132,7 +131,7 @@ def _parse_recommendation_uri(recommendation_uri):
 def _generate_recommendations(client):
     from msrestazure.azure_exceptions import CloudError
 
-    response = sdk_no_wait(True, client.generate)
+    response = client.generate(raw=True)
     location = response.headers['Location']
     operation_id = _parse_operation_id(location)
 
diff --git a/src/azure-cli/azure/cli/command_modules/backup/custom.py b/src/azure-cli/azure/cli/command_modules/backup/custom.py
index 39a8f1dec9c..84dcda6f312 100644
--- a/src/azure-cli/azure/cli/command_modules/backup/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/backup/custom.py
@@ -22,7 +22,7 @@
     JobStatus, ILRRequestResource, IaasVMILRRegistrationRequest, BackupResourceConfig, BackupResourceConfigResource, \
     BackupResourceVaultConfig, BackupResourceVaultConfigResource, DiskExclusionProperties, ExtendedProperties
 
-from azure.cli.core.util import CLIError, sdk_no_wait
+from azure.cli.core.util import CLIError
 from azure.cli.command_modules.backup._client_factory import (
     vaults_cf, backup_protected_items_cf, protection_policies_cf, virtual_machines_cf, recovery_points_cf,
     protection_containers_cf, backup_protectable_items_cf, resources_cf, backup_operation_statuses_cf,
@@ -255,8 +255,8 @@ def enable_protection_for_vm(cmd, client, resource_group_name, vault_name, vm, p
     vm_item = ProtectedItemResource(properties=vm_item_properties)
 
     # Trigger enable protection and wait for completion
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, vm_item)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, vm_item, raw=True)
     return _track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -291,8 +291,8 @@ def update_protection_for_vm(cmd, client, resource_group_name, vault_name, item,
     vm_item = ProtectedItemResource(properties=vm_item_properties)
 
     # Trigger enable protection and wait for completion
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, vm_item)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, vm_item, raw=True)
     return _track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -354,8 +354,8 @@ def update_policy_for_item(cmd, client, resource_group_name, vault_name, item, p
     vm_item = ProtectedItemResource(properties=vm_item_properties)
 
     # Update policy
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, vm_item)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, vm_item, raw=True)
     return _track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -366,9 +366,8 @@ def backup_now(cmd, client, resource_group_name, vault_name, item, retain_until)
     trigger_backup_request = _get_backup_request(item.properties.workload_type, retain_until)
 
     # Trigger backup
-    result = sdk_no_wait(True, client.trigger,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri,
-                         trigger_backup_request)
+    result = client.trigger(vault_name, resource_group_name, fabric_name,
+                            container_uri, item_uri, trigger_backup_request, raw=True)
     return _track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -505,9 +504,9 @@ def restore_disks(cmd, client, resource_group_name, vault_name, container_name,
     trigger_restore_request = RestoreRequestResource(properties=trigger_restore_properties)
 
     # Trigger restore
-    result = sdk_no_wait(True, client.trigger,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, rp_name,
-                         trigger_restore_request)
+    result = client.trigger(vault_name, resource_group_name, fabric_name,
+                            container_uri, item_uri, rp_name,
+                            trigger_restore_request, raw=True)
     return _track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -532,9 +531,8 @@ def restore_files_mount_rp(cmd, client, resource_group_name, vault_name, contain
     if recovery_point.properties.is_instant_ilr_session_active:
         recovery_point.properties.renew_existing_registration = True
 
-    result = sdk_no_wait(True, client.provision,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, rp_name,
-                         file_restore_request)
+    result = client.provision(vault_name, resource_group_name, fabric_name,
+                              container_uri, item_uri, rp_name, file_restore_request, raw=True)
 
     client_scripts = _track_backup_ilr(cmd.cli_ctx, result, vault_name, resource_group_name)
 
@@ -557,8 +555,8 @@ def restore_files_unmount_rp(cmd, client, resource_group_name, vault_name, conta
                                                          container_uri, item_uri, rp_name)
 
     if recovery_point.properties.is_instant_ilr_session_active:
-        result = sdk_no_wait(True, client.revoke,
-                             vault_name, resource_group_name, fabric_name, container_uri, item_uri, rp_name)
+        result = client.revoke(vault_name, resource_group_name, fabric_name,
+                               container_uri, item_uri, rp_name, raw=True)
         _track_backup_operation(cmd.cli_ctx, resource_group_name, result, vault_name)
 
 
@@ -569,14 +567,14 @@ def disable_protection(cmd, client, resource_group_name, vault_name, item, delet
 
     # Trigger disable protection and wait for completion
     if delete_backup_data:
-        result = sdk_no_wait(True, client.delete,
-                             vault_name, resource_group_name, fabric_name, container_uri, item_uri)
+        result = client.delete(vault_name, resource_group_name, fabric_name,
+                               container_uri, item_uri, raw=True)
         return _track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
     vm_item = _get_disable_protection_request(item)
 
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, vm_item)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, vm_item, raw=True)
     return _track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -585,8 +583,8 @@ def undelete_protection(cmd, client, resource_group_name, vault_name, item):
     item_uri = _get_protected_item_uri_from_id(item.id)
 
     vm_item = _get_disable_protection_request(item, True)
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, vm_item)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, vm_item, raw=True)
     return _track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -656,8 +654,7 @@ def _get_protectable_item_for_vm(cli_ctx, vault_name, vault_rg, vm_name, vm_rg):
     protectable_item = _try_get_protectable_item_for_vm(cli_ctx, vault_name, vault_rg, vm_name, vm_rg)
     if protectable_item is None:
         # Protectable item not found. Trigger discovery.
-        refresh_result = sdk_no_wait(True, protection_containers_client.refresh,
-                                     vault_name, vault_rg, fabric_name)
+        refresh_result = protection_containers_client.refresh(vault_name, vault_rg, fabric_name, raw=True)
         _track_refresh_operation(cli_ctx, refresh_result, vault_name, vault_rg)
     protectable_item = _try_get_protectable_item_for_vm(cli_ctx, vault_name, vault_rg, vm_name, vm_rg)
     return protectable_item
@@ -882,12 +879,12 @@ def _track_refresh_operation(cli_ctx, result, vault_name, resource_group):
     protection_container_refresh_operation_results_client = protection_container_refresh_operation_results_cf(cli_ctx)
 
     operation_id = _get_operation_id_from_header(result.response.headers['Location'])
-    result = sdk_no_wait(True, protection_container_refresh_operation_results_client.get,
-                         vault_name, resource_group, fabric_name, operation_id)
+    result = protection_container_refresh_operation_results_client.get(vault_name, resource_group,
+                                                                       fabric_name, operation_id, raw=True)
     while result.response.status_code == 202:
         time.sleep(1)
-        result = sdk_no_wait(True, protection_container_refresh_operation_results_client.get,
-                             vault_name, resource_group, fabric_name, operation_id)
+        result = protection_container_refresh_operation_results_client.get(vault_name, resource_group,
+                                                                           fabric_name, operation_id, raw=True)
 
 
 def _job_in_progress(job_status):
diff --git a/src/azure-cli/azure/cli/command_modules/backup/custom_afs.py b/src/azure-cli/azure/cli/command_modules/backup/custom_afs.py
index 72820e4acb5..6d47466d79c 100644
--- a/src/azure-cli/azure/cli/command_modules/backup/custom_afs.py
+++ b/src/azure-cli/azure/cli/command_modules/backup/custom_afs.py
@@ -14,7 +14,7 @@
     AzureFileShareBackupRequest, AzureFileshareProtectedItem, AzureFileShareRestoreRequest, \
     TargetAFSRestoreInfo, ProtectionState, ProtectionContainerResource, AzureStorageContainer
 
-from azure.cli.core.util import CLIError, sdk_no_wait
+from azure.cli.core.util import CLIError
 from azure.cli.command_modules.backup._client_factory import protection_containers_cf, protectable_containers_cf, \
     protection_policies_cf, backup_protection_containers_cf, backup_protectable_items_cf, \
     resources_cf
@@ -31,8 +31,8 @@ def enable_for_AzureFileShare(cmd, client, resource_group_name, vault_name, afs_
     filter_string = helper.get_filter_string({
         'backupManagementType': "AzureStorage"})
 
-    refresh_result = sdk_no_wait(True, protection_containers_client.refresh,
-                                 vault_name, resource_group_name, fabric_name, filter=filter_string)
+    refresh_result = protection_containers_client.refresh(vault_name, resource_group_name, fabric_name,
+                                                          filter=filter_string, raw=True)
     helper.track_refresh_operation(cmd.cli_ctx, refresh_result, vault_name, resource_group_name)
 
     # get registered storage accounts
@@ -54,8 +54,8 @@ def enable_for_AzureFileShare(cmd, client, resource_group_name, vault_name, afs_
                                            source_resource_id=storage_account.properties.container_id,
                                            workload_type="AzureFileShare")
         param = ProtectionContainerResource(properties=properties)
-        result = sdk_no_wait(True, protection_containers_client.register,
-                             vault_name, resource_group_name, fabric_name, storage_account.name, param)
+        result = protection_containers_client.register(vault_name, resource_group_name, fabric_name,
+                                                       storage_account.name, param, raw=True)
         helper.track_register_operation(cmd.cli_ctx, result, vault_name, resource_group_name, storage_account.name)
 
     policy = common.show_policy(protection_policies_cf(cmd.cli_ctx), resource_group_name, vault_name, policy_name)
@@ -71,8 +71,8 @@ def enable_for_AzureFileShare(cmd, client, resource_group_name, vault_name, afs_
     item_properties.source_resource_id = protectable_item.properties.parent_container_fabric_id
     item = ProtectedItemResource(properties=item_properties)
 
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, item)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, item, raw=True)
     return helper.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -81,9 +81,8 @@ def backup_now(cmd, client, resource_group_name, vault_name, item, retain_until)
     item_uri = helper.get_protected_item_uri_from_id(item.id)
     trigger_backup_request = _get_backup_request(retain_until)
 
-    result = sdk_no_wait(True, client.trigger,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri,
-                         trigger_backup_request)
+    result = client.trigger(vault_name, resource_group_name, fabric_name,
+                            container_uri, item_uri, trigger_backup_request, raw=True)
     return helper.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -102,8 +101,8 @@ def _get_protectable_item_for_afs(cli_ctx, vault_name, resource_group_name, afs_
         'workloadType': "AzureFileShare"})
 
     if protectable_item is None:
-        result = sdk_no_wait(True, protection_containers_client.inquire, vault_name, resource_group_name, fabric_name,
-                             storage_account.name, filter=filter_string)
+        result = protection_containers_client.inquire(vault_name, resource_group_name, fabric_name,
+                                                      storage_account.name, filter=filter_string, raw=True)
 
         helper.track_inquiry_operation(cli_ctx, result, vault_name, resource_group_name, storage_account.name)
 
@@ -179,8 +178,9 @@ def restore_AzureFileShare(cmd, client, resource_group_name, vault_name, rp_name
 
     trigger_restore_request = RestoreRequestResource(properties=afs_restore_request)
 
-    result = sdk_no_wait(True, client.trigger, vault_name, resource_group_name, fabric_name,
-                         container_uri, item_uri, rp_name, trigger_restore_request)
+    result = client.trigger(vault_name, resource_group_name, fabric_name,
+                            container_uri, item_uri, rp_name,
+                            trigger_restore_request, raw=True)
 
     return helper.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
@@ -222,8 +222,8 @@ def update_policy_for_item(cmd, client, resource_group_name, vault_name, item, p
     afs_item = ProtectedItemResource(properties=afs_item_properties)
 
     # Update policy
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, afs_item)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, afs_item, raw=True)
     return helper.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -235,8 +235,7 @@ def disable_protection(cmd, client, resource_group_name, vault_name, item,
 
     # Trigger disable protection and wait for completion
     if delete_backup_data:
-        result = sdk_no_wait(True, client.delete,
-                             vault_name, resource_group_name, fabric_name, container_uri, item_uri)
+        result = client.delete(vault_name, resource_group_name, fabric_name, container_uri, item_uri, raw=True)
         return helper.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
     afs_item_properties = AzureFileshareProtectedItem()
@@ -244,8 +243,8 @@ def disable_protection(cmd, client, resource_group_name, vault_name, item,
     afs_item_properties.protection_state = ProtectionState.protection_stopped
     afs_item_properties.source_resource_id = item.properties.source_resource_id
     afs_item = ProtectedItemResource(properties=afs_item_properties)
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, afs_item)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, afs_item, raw=True)
     return helper.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -298,8 +297,7 @@ def create_policy(client, resource_group_name, vault_name, name, policy):
 
 
 def unregister_afs_container(cmd, client, vault_name, resource_group_name, container_name):
-    result = sdk_no_wait(True, client.unregister,
-                         vault_name, resource_group_name, fabric_name, container_name)
+    result = client.unregister(vault_name, resource_group_name, fabric_name, container_name, raw=True)
     return helper.track_register_operation(cmd.cli_ctx, result, vault_name, resource_group_name, container_name)
 
 
diff --git a/src/azure-cli/azure/cli/command_modules/backup/custom_help.py b/src/azure-cli/azure/cli/command_modules/backup/custom_help.py
index 022e0e06f27..59d0330bc67 100644
--- a/src/azure-cli/azure/cli/command_modules/backup/custom_help.py
+++ b/src/azure-cli/azure/cli/command_modules/backup/custom_help.py
@@ -17,7 +17,7 @@
 
 from azure.mgmt.recoveryservicesbackup.models import OperationStatusValues, JobStatus
 
-from azure.cli.core.util import CLIError, sdk_no_wait
+from azure.cli.core.util import CLIError
 from azure.cli.command_modules.backup._client_factory import (
     job_details_cf, protection_container_refresh_operation_results_cf,
     backup_operation_statuses_cf, protection_container_operation_results_cf)
@@ -159,36 +159,42 @@ def track_refresh_operation(cli_ctx, result, vault_name, resource_group):
     protection_container_refresh_operation_results_client = protection_container_refresh_operation_results_cf(cli_ctx)
 
     operation_id = get_operation_id_from_header(result.response.headers['Location'])
-    result = sdk_no_wait(True, protection_container_refresh_operation_results_client.get,
-                         vault_name, resource_group, fabric_name, operation_id)
+    result = protection_container_refresh_operation_results_client.get(vault_name, resource_group,
+                                                                       fabric_name, operation_id,
+                                                                       raw=True)
     while result.response.status_code == 202:
         time.sleep(1)
-        result = sdk_no_wait(True, protection_container_refresh_operation_results_client.get,
-                             vault_name, resource_group, fabric_name, operation_id)
+        result = protection_container_refresh_operation_results_client.get(vault_name, resource_group,
+                                                                           fabric_name, operation_id,
+                                                                           raw=True)
 
 
 def track_register_operation(cli_ctx, result, vault_name, resource_group, container_name):
     protection_container_operation_results_client = protection_container_operation_results_cf(cli_ctx)
 
     operation_id = get_operation_id_from_header(result.response.headers['Location'])
-    result = sdk_no_wait(True, protection_container_operation_results_client.get,
-                         vault_name, resource_group, fabric_name, container_name, operation_id)
+    result = protection_container_operation_results_client.get(vault_name, resource_group,
+                                                               fabric_name, container_name,
+                                                               operation_id, raw=True)
     while result.response.status_code == 202:
         time.sleep(1)
-        result = sdk_no_wait(True, protection_container_operation_results_client.get,
-                             vault_name, resource_group, fabric_name, container_name, operation_id)
+        result = protection_container_operation_results_client.get(vault_name, resource_group,
+                                                                   fabric_name, container_name,
+                                                                   operation_id, raw=True)
 
 
 def track_inquiry_operation(cli_ctx, result, vault_name, resource_group, container_name):
     protection_container_operation_results_client = protection_container_operation_results_cf(cli_ctx)
 
     operation_id = get_operation_id_from_header(result.response.headers['Location'])
-    result = sdk_no_wait(True, protection_container_operation_results_client.get,
-                         vault_name, resource_group, fabric_name, container_name, operation_id)
+    result = protection_container_operation_results_client.get(vault_name, resource_group,
+                                                               fabric_name, container_name,
+                                                               operation_id, raw=True)
     while result.response.status_code == 202:
         time.sleep(1)
-        result = sdk_no_wait(True, protection_container_operation_results_client.get,
-                             vault_name, resource_group, fabric_name, container_name, operation_id)
+        result = protection_container_operation_results_client.get(vault_name, resource_group,
+                                                                   fabric_name, container_name,
+                                                                   operation_id, raw=True)
 
 
 def job_in_progress(job_status):
diff --git a/src/azure-cli/azure/cli/command_modules/backup/custom_wl.py b/src/azure-cli/azure/cli/command_modules/backup/custom_wl.py
index fd3d5e87275..fc6ebbca16d 100644
--- a/src/azure-cli/azure/cli/command_modules/backup/custom_wl.py
+++ b/src/azure-cli/azure/cli/command_modules/backup/custom_wl.py
@@ -20,7 +20,7 @@
     AzureWorkloadSAPHanaPointInTimeRestoreRequest, AzureWorkloadSQLPointInTimeRestoreRequest, \
     AzureVmWorkloadSAPHanaDatabaseProtectedItem, AzureVmWorkloadSQLDatabaseProtectedItem
 
-from azure.cli.core.util import CLIError, sdk_no_wait
+from azure.cli.core.util import CLIError
 from azure.cli.command_modules.backup._client_factory import backup_workload_items_cf, \
     protectable_containers_cf, backup_protection_containers_cf, backup_protected_items_cf
 
@@ -115,8 +115,7 @@ def register_wl_container(cmd, client, vault_name, resource_group_name, workload
     param = ProtectionContainerResource(properties=properties)
 
     # Trigger register and wait for completion
-    result = sdk_no_wait(True, client.register,
-                         vault_name, resource_group_name, fabric_name, container_name, param)
+    result = client.register(vault_name, resource_group_name, fabric_name, container_name, param, raw=True)
     return cust_help.track_register_operation(cmd.cli_ctx, result, vault_name, resource_group_name, container_name)
 
 
@@ -151,8 +150,7 @@ def re_register_wl_container(cmd, client, vault_name, resource_group_name, workl
                                                         source_resource_id=source_resource_id)
     param = ProtectionContainerResource(properties=properties)
     # Trigger register and wait for completion
-    result = sdk_no_wait(True, client.register,
-                         vault_name, resource_group_name, fabric_name, container_name, param)
+    result = client.register(vault_name, resource_group_name, fabric_name, container_name, param, raw=True)
     return cust_help.track_register_operation(cmd.cli_ctx, result, vault_name, resource_group_name, container_name)
 
 
@@ -165,8 +163,7 @@ def unregister_wl_container(cmd, client, vault_name, resource_group_name, contai
             """)
 
     # Trigger unregister and wait for completion
-    result = sdk_no_wait(True, client.unregister,
-                         vault_name, resource_group_name, fabric_name, container_name)
+    result = client.unregister(vault_name, resource_group_name, fabric_name, container_name, raw=True)
     return cust_help.track_register_operation(cmd.cli_ctx, result, vault_name, resource_group_name, container_name)
 
 
@@ -186,8 +183,8 @@ def update_policy_for_item(cmd, client, resource_group_name, vault_name, item, p
     param = ProtectedItemResource(properties=item_properties)
 
     # Update policy
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, param)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, param, raw=True)
     return cust_help.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -327,8 +324,8 @@ def enable_protection_for_azure_wl(cmd, client, resource_group_name, vault_name,
     param = ProtectionContainerResource(properties=properties)
 
     # Trigger enable protection and wait for completion
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_name, item_name, param)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_name, item_name, param, raw=True)
     return cust_help.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -358,9 +355,8 @@ def backup_now(cmd, client, resource_group_name, vault_name, item, retain_until,
     param = BackupRequestResource(properties=properties)
 
     # Trigger backup and wait for completion
-    result = sdk_no_wait(True, client.trigger,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri,
-                         param)
+    result = client.trigger(vault_name, resource_group_name, fabric_name, container_uri,
+                            item_uri, param, raw=True)
     return cust_help.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -377,8 +373,7 @@ def disable_protection(cmd, client, resource_group_name, vault_name, item, delet
             """)
 
     if delete_backup_data:
-        result = sdk_no_wait(True, client.delete,
-                             vault_name, resource_group_name, fabric_name, container_uri, item_uri)
+        result = client.delete(vault_name, resource_group_name, fabric_name, container_uri, item_uri, raw=True)
         return cust_help.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
     properties = _get_protected_item_instance(backup_item_type)
@@ -387,8 +382,8 @@ def disable_protection(cmd, client, resource_group_name, vault_name, item, delet
     param = ProtectedItemResource(properties=properties)
 
     # Trigger disable protection and wait for completion
-    result = sdk_no_wait(True, client.create_or_update,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, param)
+    result = client.create_or_update(vault_name, resource_group_name, fabric_name,
+                                     container_uri, item_uri, param, raw=True)
     return cust_help.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
@@ -496,9 +491,8 @@ def restore_azure_wl(cmd, client, resource_group_name, vault_name, recovery_conf
         setattr(trigger_restore_properties, 'is_non_recoverable', False)
     trigger_restore_request = RestoreRequestResource(properties=trigger_restore_properties)
     # Trigger restore and wait for completion
-    result = sdk_no_wait(True, client.trigger,
-                         vault_name, resource_group_name, fabric_name, container_uri, item_uri, recovery_point_id,
-                         trigger_restore_request)
+    result = client.trigger(vault_name, resource_group_name, fabric_name, container_uri,
+                            item_uri, recovery_point_id, trigger_restore_request, raw=True)
     return cust_help.track_backup_job(cmd.cli_ctx, result, vault_name, resource_group_name)
 
 
diff --git a/src/azure-cli/requirements.py3.Darwin.txt b/src/azure-cli/requirements.py3.Darwin.txt
index 323f50bf6d4..cfa9ab85ea8 100644
--- a/src/azure-cli/requirements.py3.Darwin.txt
+++ b/src/azure-cli/requirements.py3.Darwin.txt
@@ -32,6 +32,7 @@ azure-mgmt-consumption==2.0.0
 azure-mgmt-containerinstance==1.5.0
 azure-mgmt-containerregistry==3.0.0rc11
 azure-mgmt-containerservice==9.0.1
+azure-mgmt-core==1.0.0
 azure-mgmt-cosmosdb==0.12.0
 azure-mgmt-datalake-analytics==0.2.1
 azure-mgmt-datalake-nspkg==3.0.1
diff --git a/src/azure-cli/requirements.py3.Linux.txt b/src/azure-cli/requirements.py3.Linux.txt
index 9c907bfa3da..92dd805a354 100644
--- a/src/azure-cli/requirements.py3.Linux.txt
+++ b/src/azure-cli/requirements.py3.Linux.txt
@@ -32,6 +32,7 @@ azure-mgmt-consumption==2.0.0
 azure-mgmt-containerinstance==1.5.0
 azure-mgmt-containerregistry==3.0.0rc11
 azure-mgmt-containerservice==9.0.1
+azure-mgmt-core==1.0.0
 azure-mgmt-cosmosdb==0.12.0
 azure-mgmt-datalake-analytics==0.2.1
 azure-mgmt-datalake-nspkg==3.0.1
diff --git a/src/azure-cli/requirements.py3.windows.txt b/src/azure-cli/requirements.py3.windows.txt
index a6b4b2594aa..acf09012a53 100644
--- a/src/azure-cli/requirements.py3.windows.txt
+++ b/src/azure-cli/requirements.py3.windows.txt
@@ -31,6 +31,7 @@ azure-mgmt-consumption==2.0.0
 azure-mgmt-containerinstance==1.5.0
 azure-mgmt-containerregistry==3.0.0rc11
 azure-mgmt-containerservice==9.0.1
+azure-mgmt-core==1.0.0
 azure-mgmt-cosmosdb==0.12.0
 azure-mgmt-datalake-analytics==0.2.1
 azure-mgmt-datalake-nspkg==3.0.1