diff --git a/sdk/keyvault/azure-keyvault-administration/CHANGELOG.md b/sdk/keyvault/azure-keyvault-administration/CHANGELOG.md
index 908b367c0d05..1862da613195 100644
--- a/sdk/keyvault/azure-keyvault-administration/CHANGELOG.md
+++ b/sdk/keyvault/azure-keyvault-administration/CHANGELOG.md
@@ -5,6 +5,12 @@
 ### Features Added
 
 - Added support for service API version `2025-07-01` [#46716](https://github.com/Azure/azure-sdk-for-python/pull/46716)
+- Added `KeyVaultEkmClient` (sync and async) for managing Managed HSM External Key Manager
+  (EKM) connections via the `2026-01-01-preview` API. The new client exposes
+  `get_ekm_connection`, `create_ekm_connection`, `update_ekm_connection`,
+  `delete_ekm_connection`, `get_ekm_certificate`, and `check_ekm_connection`.
+- Added `KeyVaultEkmConnection`, `KeyVaultEkmProxyClientCertificateInfo`, and
+  `KeyVaultEkmProxyInfo` models supporting the EKM client.
 
 ### Breaking Changes
 
diff --git a/sdk/keyvault/azure-keyvault-administration/README.md b/sdk/keyvault/azure-keyvault-administration/README.md
index d63f617a3cc4..048e1ee94dcb 100644
--- a/sdk/keyvault/azure-keyvault-administration/README.md
+++ b/sdk/keyvault/azure-keyvault-administration/README.md
@@ -42,7 +42,7 @@ authentication as demonstrated below.
 * An existing [Key Vault Managed HSM][managed_hsm]. If you need to create one, you can do so using the Azure CLI by following the steps in [this document][managed_hsm_cli].
 
 ### Authenticate the client
-In order to interact with the Azure Key Vault service, you will need an instance of either a [KeyVaultAccessControlClient](#create-a-keyvaultaccesscontrolclient) or [KeyVaultBackupClient](#create-a-keyvaultbackupclient), as well as a **vault url** (which you may see as "DNS Name" in the Azure Portal) and a credential object. This document demonstrates using a [DefaultAzureCredential][default_cred_ref], which is appropriate for most scenarios, including local development and production environments. We recommend using a [managed identity][managed_identity] for authentication in production environments.
+In order to interact with the Azure Key Vault service, you will need an instance of either a [KeyVaultAccessControlClient](#create-a-keyvaultaccesscontrolclient), [KeyVaultBackupClient](#create-a-keyvaultbackupclient), [KeyVaultSettingsClient](#create-a-keyvaultsettingsclient), or [KeyVaultEkmClient](#create-a-keyvaultekmclient) as well as a **vault url** (which you may see as "DNS Name" in the Azure Portal) and a credential object. This document demonstrates using a [DefaultAzureCredential][default_cred_ref], which is appropriate for most scenarios, including local development and production environments. We recommend using a [managed identity][managed_identity] for authentication in production environments.
 
 See [azure-identity][azure_identity] documentation for more information about other methods of authentication and their corresponding credential types.
 
@@ -109,6 +109,24 @@ client = KeyVaultSettingsClient(vault_url=MANAGED_HSM_URL, credential=credential
 
 > **NOTE:** For an asynchronous client, import `azure.keyvault.administration.aio`'s `KeyVaultSettingsClient` instead.
 
+#### Create a KeyVaultEkmClient
+After configuring your environment for the [DefaultAzureCredential][default_cred_ref] to use a suitable method of authentication, you can do the following to create an EKM client (replacing the value of `vault_url` with your Managed HSM's URL):
+
+<!-- SNIPPET:ekm_operations.create_a_ekm_client -->
+
+```python
+from azure.identity import DefaultAzureCredential
+from azure.keyvault.administration import KeyVaultEkmClient
+
+MANAGED_HSM_URL = os.environ["MANAGED_HSM_URL"]
+credential = DefaultAzureCredential()
+client = KeyVaultEkmClient(vault_url=MANAGED_HSM_URL, credential=credential)
+```
+
+<!-- END SNIPPET -->
+
+> **NOTE:** For an asynchronous client, import `azure.keyvault.administration.aio`'s `KeyVaultEkmClient` instead.
+
 ## Key concepts
 
 ### Role definition
@@ -141,6 +159,14 @@ A restore operation represents a long-running operation for both a full key and
 
 A `KeyVaultSettingsClient` manages Managed HSM account settings.
 
+### KeyVaultEkmClient
+
+A `KeyVaultEkmClient` manages the Managed HSM's External Key Manager (EKM) connection.
+
+### EKM Connection
+
+An EKM connection represents the connection of an Azure Managed HSM resource with an external HSM.
+
 ## Examples
 This section contains code snippets covering common tasks:
 * Access control
@@ -193,7 +219,8 @@ role_definition = client.set_role_definition(scope=scope, role_name=role_name, p
 ```python
 new_permissions = [
     KeyVaultPermission(
-        data_actions=[KeyVaultDataAction.READ_HSM_KEY], not_data_actions=[KeyVaultDataAction.CREATE_HSM_KEY]
+        data_actions=[KeyVaultDataAction.READ_HSM_KEY],
+        not_data_actions=[KeyVaultDataAction.CREATE_HSM_KEY]
     )
 ]
 unique_definition_name = role_definition.name
@@ -413,6 +440,7 @@ Several samples are available in the Azure SDK for Python GitHub repository. The
 - [Create/update/delete role definitions and role assignments][access_control_operations_sample] ([async version][access_control_operations_async_sample])
 - [Full backup and restore][backup_operations_sample] ([async version][backup_operations_async_sample])
 - [List and update Key Vault settings][settings_operations_sample] ([async version][settings_operations_async_sample])
+- [Retrieve and manage EKM connections][ekm_operations_sample] ([async version][ekm_operations_async_sample])
 
 ###  Additional documentation
 For more extensive documentation on Azure Key Vault, see the [API reference documentation][reference_docs].
@@ -474,6 +502,8 @@ contact opencode@microsoft.com with any additional questions or comments.
 [sas_docs]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/storage/azure-storage-blob/README.md#types-of-credentials
 [settings_operations_sample]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/keyvault/azure-keyvault-administration/samples/settings_operations.py
 [settings_operations_async_sample]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/keyvault/azure-keyvault-administration/samples/settings_operations_async.py
+[ekm_operations_sample]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/keyvault/azure-keyvault-administration/samples/ekm_operations.py
+[ekm_operations_async_sample]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/keyvault/azure-keyvault-administration/samples/ekm_operations_async.py
 [storage_blob]: https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/storage/azure-storage-blob/README.md
 [storage_explorer]: https://learn.microsoft.com/azure/vs-azure-tools-storage-manage-with-storage-explorer
 
diff --git a/sdk/keyvault/azure-keyvault-administration/_metadata.json b/sdk/keyvault/azure-keyvault-administration/_metadata.json
index 0a2924fbf51d..61c17df73dc9 100644
--- a/sdk/keyvault/azure-keyvault-administration/_metadata.json
+++ b/sdk/keyvault/azure-keyvault-administration/_metadata.json
@@ -1,6 +1,6 @@
 {
-  "apiVersion": "2025-07-01",
+  "apiVersion": "2026-01-01-preview",
   "apiVersions": {
-    "KeyVault": "2025-07-01"
+    "KeyVault": "2026-01-01-preview"
   }
 }
\ No newline at end of file
diff --git a/sdk/keyvault/azure-keyvault-administration/apiview-properties.json b/sdk/keyvault/azure-keyvault-administration/apiview-properties.json
index 320d39e0f4fe..0089041338e7 100644
--- a/sdk/keyvault/azure-keyvault-administration/apiview-properties.json
+++ b/sdk/keyvault/azure-keyvault-administration/apiview-properties.json
@@ -1,6 +1,9 @@
 {
     "CrossLanguagePackageId": "KeyVault",
     "CrossLanguageDefinitionId": {
+        "azure.keyvault.administration._generated.models.EkmConnection": "KeyVault.EkmConnection",
+        "azure.keyvault.administration._generated.models.EkmProxyClientCertificateInfo": "KeyVault.EkmProxyClientCertificateInfo",
+        "azure.keyvault.administration._generated.models.EkmProxyInfo": "KeyVault.EkmProxyInfo",
         "azure.keyvault.administration._generated.models.FullBackupOperation": "KeyVault.FullBackupOperation",
         "azure.keyvault.administration._generated.models.FullBackupOperationError": "KeyVault.FullBackupOperation.error.anonymous",
         "azure.keyvault.administration._generated.models.KeyVaultError": "KeyVaultError",
@@ -65,6 +68,18 @@
         "azure.keyvault.administration._generated.KeyVaultClient.get_setting": "KeyVault.getSetting",
         "azure.keyvault.administration._generated.aio.KeyVaultClient.get_setting": "KeyVault.getSetting",
         "azure.keyvault.administration._generated.KeyVaultClient.get_settings": "KeyVault.getSettings",
-        "azure.keyvault.administration._generated.aio.KeyVaultClient.get_settings": "KeyVault.getSettings"
+        "azure.keyvault.administration._generated.aio.KeyVaultClient.get_settings": "KeyVault.getSettings",
+        "azure.keyvault.administration._generated.KeyVaultClient.get_ekm_connection": "KeyVault.getEkmConnection",
+        "azure.keyvault.administration._generated.aio.KeyVaultClient.get_ekm_connection": "KeyVault.getEkmConnection",
+        "azure.keyvault.administration._generated.KeyVaultClient.get_ekm_certificate": "KeyVault.getEkmCertificate",
+        "azure.keyvault.administration._generated.aio.KeyVaultClient.get_ekm_certificate": "KeyVault.getEkmCertificate",
+        "azure.keyvault.administration._generated.KeyVaultClient.check_ekm_connection": "KeyVault.checkEkmConnection",
+        "azure.keyvault.administration._generated.aio.KeyVaultClient.check_ekm_connection": "KeyVault.checkEkmConnection",
+        "azure.keyvault.administration._generated.KeyVaultClient.create_ekm_connection": "KeyVault.createEkmConnection",
+        "azure.keyvault.administration._generated.aio.KeyVaultClient.create_ekm_connection": "KeyVault.createEkmConnection",
+        "azure.keyvault.administration._generated.KeyVaultClient.update_ekm_connection": "KeyVault.updateEkmConnection",
+        "azure.keyvault.administration._generated.aio.KeyVaultClient.update_ekm_connection": "KeyVault.updateEkmConnection",
+        "azure.keyvault.administration._generated.KeyVaultClient.delete_ekm_connection": "KeyVault.deleteEkmConnection",
+        "azure.keyvault.administration._generated.aio.KeyVaultClient.delete_ekm_connection": "KeyVault.deleteEkmConnection"
     }
 }
\ No newline at end of file
diff --git a/sdk/keyvault/azure-keyvault-administration/assets.json b/sdk/keyvault/azure-keyvault-administration/assets.json
index 36e1355ab6ab..1703f83831e9 100644
--- a/sdk/keyvault/azure-keyvault-administration/assets.json
+++ b/sdk/keyvault/azure-keyvault-administration/assets.json
@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/keyvault/azure-keyvault-administration",
-  "Tag": "python/keyvault/azure-keyvault-administration_007a803c2c"
+  "Tag": "python/keyvault/azure-keyvault-administration_a1cfcc06db"
 }
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/__init__.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/__init__.py
index 5e6b3233b362..71a878885286 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/__init__.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/__init__.py
@@ -4,10 +4,14 @@
 # ------------------------------------
 from ._access_control_client import KeyVaultAccessControlClient
 from ._backup_client import KeyVaultBackupClient
-from ._enums import KeyVaultRoleScope, KeyVaultDataAction, KeyVaultSettingType
+from ._ekm_client import KeyVaultEkmClient
+from ._enums import KeyVaultDataAction, KeyVaultRoleScope, KeyVaultSettingType
 from ._internal.client_base import ApiVersion
 from ._models import (
     KeyVaultBackupResult,
+    KeyVaultEkmConnection,
+    KeyVaultEkmProxyClientCertificateInfo,
+    KeyVaultEkmProxyInfo,
     KeyVaultPermission,
     KeyVaultRoleAssignment,
     KeyVaultRoleAssignmentProperties,
@@ -16,13 +20,16 @@
 )
 from ._settings_client import KeyVaultSettingsClient
 
-
 __all__ = [
     "ApiVersion",
-    "KeyVaultBackupResult",
     "KeyVaultAccessControlClient",
     "KeyVaultBackupClient",
+    "KeyVaultBackupResult",
     "KeyVaultDataAction",
+    "KeyVaultEkmClient",
+    "KeyVaultEkmConnection",
+    "KeyVaultEkmProxyClientCertificateInfo",
+    "KeyVaultEkmProxyInfo",
     "KeyVaultPermission",
     "KeyVaultRoleAssignment",
     "KeyVaultRoleAssignmentProperties",
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_ekm_client.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_ekm_client.py
new file mode 100644
index 000000000000..c3809d8f5506
--- /dev/null
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_ekm_client.py
@@ -0,0 +1,114 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+from typing import Any
+
+from azure.core.tracing.decorator import distributed_trace
+
+from ._internal import KeyVaultClientBase
+from ._models import KeyVaultEkmConnection, KeyVaultEkmProxyClientCertificateInfo, KeyVaultEkmProxyInfo
+
+
+class KeyVaultEkmClient(KeyVaultClientBase):
+    """Provides methods to manage Managed HSM External Key Manager (EKM) connections.
+
+    :param str vault_url: URL of the vault on which the client will operate. This is also called the vault's "DNS Name".
+        You should validate that this URL references a valid Key Vault or Managed HSM resource.
+        See https://aka.ms/azsdk/blog/vault-uri for details.
+    :param credential: An object which can provide an access token for the vault, such as a credential from
+        :mod:`azure.identity`
+    :type credential: ~azure.core.credentials.TokenCredential
+
+    :keyword api_version: Version of the service API to use. EKM operations require service API version
+        ``2026-01-01-preview`` or later.
+    :paramtype api_version: ~azure.keyvault.administration.ApiVersion or str
+    :keyword bool verify_challenge_resource: Whether to verify the authentication challenge resource matches the Key
+        Vault or Managed HSM domain. Defaults to True.
+    """
+
+    # pylint:disable=protected-access
+
+    @distributed_trace
+    def get_ekm_connection(self, **kwargs: Any) -> KeyVaultEkmConnection:
+        """Gets the configured EKM connection.
+
+        :returns: The configured EKM connection.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = self._client.get_ekm_connection(**kwargs)
+        return KeyVaultEkmConnection._from_generated(result)
+
+    @distributed_trace
+    def create_ekm_connection(self, connection: KeyVaultEkmConnection, **kwargs: Any) -> KeyVaultEkmConnection:
+        """Creates the EKM connection.
+
+        If an EKM connection already exists, this operation fails.
+
+        :param connection: The EKM connection to create.
+        :type connection: ~azure.keyvault.administration.KeyVaultEkmConnection
+
+        :returns: The created EKM connection.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = self._client.create_ekm_connection(ekm_connection=connection._to_generated(), **kwargs)
+        return KeyVaultEkmConnection._from_generated(result)
+
+    @distributed_trace
+    def update_ekm_connection(self, connection: KeyVaultEkmConnection, **kwargs: Any) -> KeyVaultEkmConnection:
+        """Updates the existing EKM connection.
+
+        If no EKM connection exists, this operation fails.
+
+        :param connection: The EKM connection to update.
+        :type connection: ~azure.keyvault.administration.KeyVaultEkmConnection
+
+        :returns: The updated EKM connection.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = self._client.update_ekm_connection(ekm_connection=connection._to_generated(), **kwargs)
+        return KeyVaultEkmConnection._from_generated(result)
+
+    @distributed_trace
+    def delete_ekm_connection(  # pylint:disable=bad-option-value,delete-operation-wrong-return-type
+        self, **kwargs: Any
+    ) -> KeyVaultEkmConnection:
+        """Deletes the existing EKM connection.
+
+        If no EKM connection exists, this operation fails.
+
+        :returns: The deleted EKM connection.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = self._client.delete_ekm_connection(**kwargs)
+        return KeyVaultEkmConnection._from_generated(result)
+
+    @distributed_trace
+    def get_ekm_certificate(self, **kwargs: Any) -> KeyVaultEkmProxyClientCertificateInfo:
+        """Gets the EKM proxy client certificate information used to authenticate to the EKM proxy.
+
+        :returns: The EKM proxy client certificate information.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmProxyClientCertificateInfo
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = self._client.get_ekm_certificate(**kwargs)
+        return KeyVaultEkmProxyClientCertificateInfo._from_generated(result)
+
+    @distributed_trace
+    def check_ekm_connection(self, **kwargs: Any) -> KeyVaultEkmProxyInfo:
+        """Checks the EKM connection by pinging the EKM proxy.
+
+        :returns: Information about the EKM proxy returned by the connectivity check.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmProxyInfo
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = self._client.check_ekm_connection(**kwargs)
+        return KeyVaultEkmProxyInfo._from_generated(result)
+
+    def __enter__(self) -> "KeyVaultEkmClient":
+        self._client.__enter__()
+        return self
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/_client.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/_client.py
index 0a83e4c8f786..c23540ae6210 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/_client.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/_client.py
@@ -37,9 +37,9 @@ class KeyVaultClient(_KeyVaultClientOperationsMixin):
     :type vault_base_url: str
     :param credential: Credential used to authenticate requests to the service. Required.
     :type credential: ~azure.core.credentials.TokenCredential
-    :keyword api_version: The API version to use for this operation. Known values are "2025-07-01".
-     Default value is "2025-07-01". Note that overriding this default value may result in
-     unsupported behavior.
+    :keyword api_version: The API version to use for this operation. Known values are
+     "2026-01-01-preview". Default value is "2026-01-01-preview". Note that overriding this default
+     value may result in unsupported behavior.
     :paramtype api_version: str
     :keyword int polling_interval: Default waiting time between two polls for LRO operations if no
      Retry-After header is present.
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/_configuration.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/_configuration.py
index 1e7442eedff6..822c28a36f60 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/_configuration.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/_configuration.py
@@ -27,14 +27,14 @@ class KeyVaultClientConfiguration:  # pylint: disable=too-many-instance-attribut
     :type vault_base_url: str
     :param credential: Credential used to authenticate requests to the service. Required.
     :type credential: ~azure.core.credentials.TokenCredential
-    :keyword api_version: The API version to use for this operation. Known values are "2025-07-01".
-     Default value is "2025-07-01". Note that overriding this default value may result in
-     unsupported behavior.
+    :keyword api_version: The API version to use for this operation. Known values are
+     "2026-01-01-preview". Default value is "2026-01-01-preview". Note that overriding this default
+     value may result in unsupported behavior.
     :paramtype api_version: str
     """
 
     def __init__(self, vault_base_url: str, credential: "TokenCredential", **kwargs: Any) -> None:
-        api_version: str = kwargs.pop("api_version", "2025-07-01")
+        api_version: str = kwargs.pop("api_version", "2026-01-01-preview")
 
         if vault_base_url is None:
             raise ValueError("Parameter 'vault_base_url' must not be None.")
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/_client.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/_client.py
index de369772c92e..0bf5dd9453f6 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/_client.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/_client.py
@@ -37,9 +37,9 @@ class KeyVaultClient(_KeyVaultClientOperationsMixin):
     :type vault_base_url: str
     :param credential: Credential used to authenticate requests to the service. Required.
     :type credential: ~azure.core.credentials_async.AsyncTokenCredential
-    :keyword api_version: The API version to use for this operation. Known values are "2025-07-01".
-     Default value is "2025-07-01". Note that overriding this default value may result in
-     unsupported behavior.
+    :keyword api_version: The API version to use for this operation. Known values are
+     "2026-01-01-preview". Default value is "2026-01-01-preview". Note that overriding this default
+     value may result in unsupported behavior.
     :paramtype api_version: str
     :keyword int polling_interval: Default waiting time between two polls for LRO operations if no
      Retry-After header is present.
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/_configuration.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/_configuration.py
index 8048741ec595..ad7e59af7e43 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/_configuration.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/_configuration.py
@@ -27,14 +27,14 @@ class KeyVaultClientConfiguration:  # pylint: disable=too-many-instance-attribut
     :type vault_base_url: str
     :param credential: Credential used to authenticate requests to the service. Required.
     :type credential: ~azure.core.credentials_async.AsyncTokenCredential
-    :keyword api_version: The API version to use for this operation. Known values are "2025-07-01".
-     Default value is "2025-07-01". Note that overriding this default value may result in
-     unsupported behavior.
+    :keyword api_version: The API version to use for this operation. Known values are
+     "2026-01-01-preview". Default value is "2026-01-01-preview". Note that overriding this default
+     value may result in unsupported behavior.
     :paramtype api_version: str
     """
 
     def __init__(self, vault_base_url: str, credential: "AsyncTokenCredential", **kwargs: Any) -> None:
-        api_version: str = kwargs.pop("api_version", "2025-07-01")
+        api_version: str = kwargs.pop("api_version", "2026-01-01-preview")
 
         if vault_base_url is None:
             raise ValueError("Parameter 'vault_base_url' must not be None.")
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/operations/_operations.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/operations/_operations.py
index fb53df6115bc..0f3f2d17a504 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/operations/_operations.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/aio/operations/_operations.py
@@ -38,9 +38,14 @@
 from ..._utils.utils import ClientMixinABC
 from ..._validation import api_version_validation
 from ...operations._operations import (
+    build_key_vault_check_ekm_connection_request,
+    build_key_vault_create_ekm_connection_request,
+    build_key_vault_delete_ekm_connection_request,
     build_key_vault_full_backup_request,
     build_key_vault_full_backup_status_request,
     build_key_vault_full_restore_operation_request,
+    build_key_vault_get_ekm_certificate_request,
+    build_key_vault_get_ekm_connection_request,
     build_key_vault_get_setting_request,
     build_key_vault_get_settings_request,
     build_key_vault_pre_full_backup_request,
@@ -48,6 +53,7 @@
     build_key_vault_restore_status_request,
     build_key_vault_selective_key_restore_operation_request,
     build_key_vault_selective_key_restore_status_request,
+    build_key_vault_update_ekm_connection_request,
     build_key_vault_update_setting_request,
     build_role_assignments_create_request,
     build_role_assignments_delete_request,
@@ -934,7 +940,7 @@ async def get_next(next_link=None):
         return AsyncItemPaged(get_next, extract_data)
 
 
-class _KeyVaultClientOperationsMixin(
+class _KeyVaultClientOperationsMixin(  # pylint: disable=too-many-public-methods
     ClientMixinABC[AsyncPipelineClient[HttpRequest, AsyncHttpResponse], KeyVaultClientConfiguration]
 ):
 
@@ -1221,7 +1227,7 @@ def get_long_running_output(pipeline_response):
     @api_version_validation(
         method_added_on="7.6-preview.2",
         params_added_on={"7.6-preview.2": ["api_version", "content_type", "accept"]},
-        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01"],
+        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01", "2026-01-01-preview"],
     )
     async def _pre_full_backup_initial(
         self,
@@ -1362,7 +1368,7 @@ async def begin_pre_full_backup(
     @api_version_validation(
         method_added_on="7.6-preview.2",
         params_added_on={"7.6-preview.2": ["api_version", "content_type", "accept"]},
-        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01"],
+        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01", "2026-01-01-preview"],
     )
     async def begin_pre_full_backup(
         self,
@@ -1726,7 +1732,7 @@ def get_long_running_output(pipeline_response):
     @api_version_validation(
         method_added_on="7.6-preview.2",
         params_added_on={"7.6-preview.2": ["api_version", "content_type", "accept"]},
-        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01"],
+        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01", "2026-01-01-preview"],
     )
     async def _pre_full_restore_operation_initial(
         self,
@@ -1867,7 +1873,7 @@ async def begin_pre_full_restore_operation(
     @api_version_validation(
         method_added_on="7.6-preview.2",
         params_added_on={"7.6-preview.2": ["api_version", "content_type", "accept"]},
-        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01"],
+        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01", "2026-01-01-preview"],
     )
     async def begin_pre_full_restore_operation(
         self,
@@ -2539,3 +2545,582 @@ async def get_settings(self, **kwargs: Any) -> _models.SettingsListResult:
             return cls(pipeline_response, deserialized, {})  # type: ignore
 
         return deserialized  # type: ignore
+
+    @distributed_trace_async
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    async def get_ekm_connection(self, **kwargs: Any) -> _models.EkmConnection:
+        """Gets the EKM connection.
+
+        The External Key Manager (EKM) Get operation returns EKM connection. This operation requires
+        ekm/read permission.
+
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[_models.EkmConnection] = kwargs.pop("cls", None)
+
+        _request = build_key_vault_get_ekm_connection_request(
+            api_version=self._config.api_version,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # type: ignore # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    await response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmConnection, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace_async
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    async def get_ekm_certificate(self, **kwargs: Any) -> _models.EkmProxyClientCertificateInfo:
+        """Gets the EKM proxy client certificate.
+
+        The External Key Manager (EKM) Certificate Get operation returns Proxy client certificate. This
+        operation requires ekm/read permission.
+
+        :return: EkmProxyClientCertificateInfo. The EkmProxyClientCertificateInfo is compatible with
+         MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmProxyClientCertificateInfo
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[_models.EkmProxyClientCertificateInfo] = kwargs.pop("cls", None)
+
+        _request = build_key_vault_get_ekm_certificate_request(
+            api_version=self._config.api_version,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # type: ignore # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    await response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmProxyClientCertificateInfo, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace_async
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    async def check_ekm_connection(self, **kwargs: Any) -> _models.EkmProxyInfo:
+        """Checks the connectivity and authentication with the EKM proxy.
+
+        The External Key Manager (EKM) Check operation checks the connectivity and authentication with
+        the EKM proxy. This operation requires ekm/read permission.
+
+        :return: EkmProxyInfo. The EkmProxyInfo is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmProxyInfo
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[_models.EkmProxyInfo] = kwargs.pop("cls", None)
+
+        _request = build_key_vault_check_ekm_connection_request(
+            api_version=self._config.api_version,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # type: ignore # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    await response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmProxyInfo, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @overload
+    async def create_ekm_connection(
+        self, ekm_connection: _models.EkmConnection, *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Creates the EKM connection.
+
+        The External Key Manager (EKM) sets up the EKM connection. If the EKM connection already
+        exists, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to create. Required.
+        :type ekm_connection: ~azure.keyvault.administration._generated.models.EkmConnection
+        :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @overload
+    async def create_ekm_connection(
+        self, ekm_connection: JSON, *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Creates the EKM connection.
+
+        The External Key Manager (EKM) sets up the EKM connection. If the EKM connection already
+        exists, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to create. Required.
+        :type ekm_connection: JSON
+        :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @overload
+    async def create_ekm_connection(
+        self, ekm_connection: IO[bytes], *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Creates the EKM connection.
+
+        The External Key Manager (EKM) sets up the EKM connection. If the EKM connection already
+        exists, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to create. Required.
+        :type ekm_connection: IO[bytes]
+        :keyword content_type: Body Parameter content-type. Content type parameter for binary body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @distributed_trace_async
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "content_type", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    async def create_ekm_connection(
+        self, ekm_connection: Union[_models.EkmConnection, JSON, IO[bytes]], **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Creates the EKM connection.
+
+        The External Key Manager (EKM) sets up the EKM connection. If the EKM connection already
+        exists, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to create. Is one of the following types:
+         EkmConnection, JSON, IO[bytes] Required.
+        :type ekm_connection: ~azure.keyvault.administration._generated.models.EkmConnection or JSON or
+         IO[bytes]
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+        _params = kwargs.pop("params", {}) or {}
+
+        content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
+        cls: ClsType[_models.EkmConnection] = kwargs.pop("cls", None)
+
+        content_type = content_type or "application/json"
+        _content = None
+        if isinstance(ekm_connection, (IOBase, bytes)):
+            _content = ekm_connection
+        else:
+            _content = json.dumps(ekm_connection, cls=SdkJSONEncoder, exclude_readonly=True)  # type: ignore
+
+        _request = build_key_vault_create_ekm_connection_request(
+            content_type=content_type,
+            api_version=self._config.api_version,
+            content=_content,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # type: ignore # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    await response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmConnection, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @overload
+    async def update_ekm_connection(
+        self, ekm_connection: _models.EkmConnection, *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Updates the EKM connection.
+
+        The External Key Manager (EKM) updates the existing EKM connection. If the EKM connection does
+        not exist, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to update. Required.
+        :type ekm_connection: ~azure.keyvault.administration._generated.models.EkmConnection
+        :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @overload
+    async def update_ekm_connection(
+        self, ekm_connection: JSON, *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Updates the EKM connection.
+
+        The External Key Manager (EKM) updates the existing EKM connection. If the EKM connection does
+        not exist, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to update. Required.
+        :type ekm_connection: JSON
+        :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @overload
+    async def update_ekm_connection(
+        self, ekm_connection: IO[bytes], *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Updates the EKM connection.
+
+        The External Key Manager (EKM) updates the existing EKM connection. If the EKM connection does
+        not exist, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to update. Required.
+        :type ekm_connection: IO[bytes]
+        :keyword content_type: Body Parameter content-type. Content type parameter for binary body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @distributed_trace_async
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "content_type", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    async def update_ekm_connection(
+        self, ekm_connection: Union[_models.EkmConnection, JSON, IO[bytes]], **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Updates the EKM connection.
+
+        The External Key Manager (EKM) updates the existing EKM connection. If the EKM connection does
+        not exist, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to update. Is one of the following types:
+         EkmConnection, JSON, IO[bytes] Required.
+        :type ekm_connection: ~azure.keyvault.administration._generated.models.EkmConnection or JSON or
+         IO[bytes]
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+        _params = kwargs.pop("params", {}) or {}
+
+        content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
+        cls: ClsType[_models.EkmConnection] = kwargs.pop("cls", None)
+
+        content_type = content_type or "application/json"
+        _content = None
+        if isinstance(ekm_connection, (IOBase, bytes)):
+            _content = ekm_connection
+        else:
+            _content = json.dumps(ekm_connection, cls=SdkJSONEncoder, exclude_readonly=True)  # type: ignore
+
+        _request = build_key_vault_update_ekm_connection_request(
+            content_type=content_type,
+            api_version=self._config.api_version,
+            content=_content,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # type: ignore # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    await response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmConnection, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace_async
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    async def delete_ekm_connection(self, **kwargs: Any) -> _models.EkmConnection:
+        """Deletes the EKM connection.
+
+        The External Key Manager (EKM) deletes the existing EKM connection. If the EKM connection does
+        not already exists, this operation fails. This operation requires ekm/delete permission.
+
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[_models.EkmConnection] = kwargs.pop("cls", None)
+
+        _request = build_key_vault_delete_ekm_connection_request(
+            api_version=self._config.api_version,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # type: ignore # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    await response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmConnection, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/models/__init__.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/models/__init__.py
index 753cb6201cea..e514fe596eb4 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/models/__init__.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/models/__init__.py
@@ -14,6 +14,9 @@
 
 
 from ._models import (  # type: ignore
+    EkmConnection,
+    EkmProxyClientCertificateInfo,
+    EkmProxyInfo,
     FullBackupOperation,
     FullBackupOperationError,
     KeyVaultError,
@@ -50,6 +53,9 @@
 from ._patch import patch_sdk as _patch_sdk
 
 __all__ = [
+    "EkmConnection",
+    "EkmProxyClientCertificateInfo",
+    "EkmProxyInfo",
     "FullBackupOperation",
     "FullBackupOperationError",
     "KeyVaultError",
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/models/_models.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/models/_models.py
index d2e0d1bed8bf..1ee73a8048d5 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/models/_models.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/models/_models.py
@@ -1,3 +1,4 @@
+# pylint: disable=too-many-lines
 # coding=utf-8
 # --------------------------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
@@ -16,6 +17,126 @@
     from .. import models as _models
 
 
+class EkmConnection(_Model):
+    """A EkmConnection model object.
+
+    :ivar host: EKM proxy FQDN (Fully Qualified Domain Name). Only allowed characters are a-z, A-Z,
+     0-9, hyphen (-), dot (.), and colon (:). Required.
+    :vartype host: str
+    :ivar path_prefix: Optional path prefix for the EKM proxy (if any).
+    :vartype path_prefix: str
+    :ivar server_ca_certificates: The root CA certificate chain that issued the proxy server's
+     certificate. An array of certificates in the certificate chain, each in DER format and base64
+     encoded. Required.
+    :vartype server_ca_certificates: list[bytes]
+    :ivar server_subject_common_name: The subject common name of the server certificate of EKM
+     Proxy.
+    :vartype server_subject_common_name: str
+    """
+
+    host: str = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    """EKM proxy FQDN (Fully Qualified Domain Name). Only allowed characters are a-z, A-Z, 0-9, hyphen
+     (-), dot (.), and colon (:). Required."""
+    path_prefix: Optional[str] = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    """Optional path prefix for the EKM proxy (if any)."""
+    server_ca_certificates: list[bytes] = rest_field(
+        visibility=["read", "create", "update", "delete", "query"], format="base64"
+    )
+    """The root CA certificate chain that issued the proxy server's certificate. An array of
+     certificates in the certificate chain, each in DER format and base64 encoded. Required."""
+    server_subject_common_name: Optional[str] = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    """The subject common name of the server certificate of EKM Proxy."""
+
+    @overload
+    def __init__(
+        self,
+        *,
+        host: str,
+        server_ca_certificates: list[bytes],
+        path_prefix: Optional[str] = None,
+        server_subject_common_name: Optional[str] = None,
+    ) -> None: ...
+
+    @overload
+    def __init__(self, mapping: Mapping[str, Any]) -> None:
+        """
+        :param mapping: raw JSON to initialize the model.
+        :type mapping: Mapping[str, Any]
+        """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+
+
+class EkmProxyClientCertificateInfo(_Model):
+    """EKM proxy client certificate information.
+
+    :ivar ca_certificates: The client root CA certificate chain to authenticate to the EKM proxy.
+     An array of certificates in the certificate chain, each in DER format and base64 encoded.
+     Required.
+    :vartype ca_certificates: list[bytes]
+    :ivar subject_common_name: The subject common name of the client certificate used to
+     authenticate to the EKM proxy. Required.
+    :vartype subject_common_name: str
+    """
+
+    ca_certificates: list[bytes] = rest_field(visibility=["read"], format="base64")
+    """The client root CA certificate chain to authenticate to the EKM proxy. An array of certificates
+     in the certificate chain, each in DER format and base64 encoded. Required."""
+    subject_common_name: str = rest_field(visibility=["read"])
+    """The subject common name of the client certificate used to authenticate to the EKM proxy.
+     Required."""
+
+
+class EkmProxyInfo(_Model):
+    """EKM proxy information.
+
+    :ivar api_version: The highest version of proxy interface API supported by the EKM Proxy.
+     Required.
+    :vartype api_version: str
+    :ivar proxy_vendor: The name of the proxy vendor. Required.
+    :vartype proxy_vendor: str
+    :ivar proxy_name: The name of the proxy product and its version. Required.
+    :vartype proxy_name: str
+    :ivar ekm_vendor: The name of the EKM vendor. Required.
+    :vartype ekm_vendor: str
+    :ivar ekm_product: The name of the EKM product and its version. Required.
+    :vartype ekm_product: str
+    """
+
+    api_version: str = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    """The highest version of proxy interface API supported by the EKM Proxy. Required."""
+    proxy_vendor: str = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    """The name of the proxy vendor. Required."""
+    proxy_name: str = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    """The name of the proxy product and its version. Required."""
+    ekm_vendor: str = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    """The name of the EKM vendor. Required."""
+    ekm_product: str = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    """The name of the EKM product and its version. Required."""
+
+    @overload
+    def __init__(
+        self,
+        *,
+        api_version: str,
+        proxy_vendor: str,
+        proxy_name: str,
+        ekm_vendor: str,
+        ekm_product: str,
+    ) -> None: ...
+
+    @overload
+    def __init__(self, mapping: Mapping[str, Any]) -> None:
+        """
+        :param mapping: raw JSON to initialize the model.
+        :type mapping: Mapping[str, Any]
+        """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+
+
 class FullBackupOperation(_Model):
     """Full backup operation.
 
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/operations/_operations.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/operations/_operations.py
index 32511186d843..b4e5169cb2f4 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/operations/_operations.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_generated/operations/_operations.py
@@ -51,7 +51,7 @@ def build_role_definitions_delete_request(scope: str, role_definition_name: str,
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -79,7 +79,7 @@ def build_role_definitions_create_or_update_request(  # pylint: disable=name-too
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
     content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -106,7 +106,7 @@ def build_role_definitions_get_request(scope: str, role_definition_name: str, **
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -131,7 +131,7 @@ def build_role_definitions_list_request(scope: str, *, filter: Optional[str] = N
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -157,7 +157,7 @@ def build_role_assignments_delete_request(scope: str, role_assignment_name: str,
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -183,7 +183,7 @@ def build_role_assignments_create_request(scope: str, role_assignment_name: str,
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
     content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -210,7 +210,7 @@ def build_role_assignments_get_request(scope: str, role_assignment_name: str, **
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -237,7 +237,7 @@ def build_role_assignments_list_for_scope_request(  # pylint: disable=name-too-l
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -265,7 +265,7 @@ def build_key_vault_full_backup_status_request(  # pylint: disable=name-too-long
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -290,7 +290,7 @@ def build_key_vault_full_backup_request(**kwargs: Any) -> HttpRequest:
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
     content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -312,7 +312,7 @@ def build_key_vault_pre_full_backup_request(**kwargs: Any) -> HttpRequest:
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
     content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -333,7 +333,7 @@ def build_key_vault_restore_status_request(job_id: str, **kwargs: Any) -> HttpRe
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -358,7 +358,7 @@ def build_key_vault_full_restore_operation_request(**kwargs: Any) -> HttpRequest
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
     content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -380,7 +380,7 @@ def build_key_vault_pre_full_restore_operation_request(**kwargs: Any) -> HttpReq
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
     content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -403,7 +403,7 @@ def build_key_vault_selective_key_restore_status_request(  # pylint: disable=nam
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -430,7 +430,7 @@ def build_key_vault_selective_key_restore_operation_request(  # pylint: disable=
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
     content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -457,7 +457,7 @@ def build_key_vault_update_setting_request(setting_name: str, **kwargs: Any) ->
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
     content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -483,7 +483,7 @@ def build_key_vault_get_setting_request(setting_name: str, **kwargs: Any) -> Htt
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -507,7 +507,7 @@ def build_key_vault_get_settings_request(**kwargs: Any) -> HttpRequest:
     _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
     _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
 
-    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-07-01"))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
     accept = _headers.pop("Accept", "application/json")
 
     # Construct URL
@@ -522,6 +522,126 @@ def build_key_vault_get_settings_request(**kwargs: Any) -> HttpRequest:
     return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
 
 
+def build_key_vault_get_ekm_connection_request(**kwargs: Any) -> HttpRequest:  # pylint: disable=name-too-long
+    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
+
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
+    accept = _headers.pop("Accept", "application/json")
+
+    # Construct URL
+    _url = "/ekm"
+
+    # Construct parameters
+    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
+
+    # Construct headers
+    _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
+
+    return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
+
+
+def build_key_vault_get_ekm_certificate_request(**kwargs: Any) -> HttpRequest:  # pylint: disable=name-too-long
+    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
+
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
+    accept = _headers.pop("Accept", "application/json")
+
+    # Construct URL
+    _url = "/ekm/certificate"
+
+    # Construct parameters
+    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
+
+    # Construct headers
+    _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
+
+    return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
+
+
+def build_key_vault_check_ekm_connection_request(**kwargs: Any) -> HttpRequest:  # pylint: disable=name-too-long
+    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
+
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
+    accept = _headers.pop("Accept", "application/json")
+
+    # Construct URL
+    _url = "/ekm/check"
+
+    # Construct parameters
+    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
+
+    # Construct headers
+    _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
+
+    return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, **kwargs)
+
+
+def build_key_vault_create_ekm_connection_request(**kwargs: Any) -> HttpRequest:  # pylint: disable=name-too-long
+    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
+
+    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
+    accept = _headers.pop("Accept", "application/json")
+
+    # Construct URL
+    _url = "/ekm/create"
+
+    # Construct parameters
+    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
+
+    # Construct headers
+    if content_type is not None:
+        _headers["Content-Type"] = _SERIALIZER.header("content_type", content_type, "str")
+    _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
+
+    return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, **kwargs)
+
+
+def build_key_vault_update_ekm_connection_request(**kwargs: Any) -> HttpRequest:  # pylint: disable=name-too-long
+    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
+
+    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
+    accept = _headers.pop("Accept", "application/json")
+
+    # Construct URL
+    _url = "/ekm"
+
+    # Construct parameters
+    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
+
+    # Construct headers
+    if content_type is not None:
+        _headers["Content-Type"] = _SERIALIZER.header("content_type", content_type, "str")
+    _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
+
+    return HttpRequest(method="PATCH", url=_url, params=_params, headers=_headers, **kwargs)
+
+
+def build_key_vault_delete_ekm_connection_request(**kwargs: Any) -> HttpRequest:  # pylint: disable=name-too-long
+    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})
+
+    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2026-01-01-preview"))
+    accept = _headers.pop("Accept", "application/json")
+
+    # Construct URL
+    _url = "/ekm"
+
+    # Construct parameters
+    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
+
+    # Construct headers
+    _headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
+
+    return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs)
+
+
 class RoleDefinitionsOperations:
     """
     .. warning::
@@ -1388,7 +1508,7 @@ def get_next(next_link=None):
         return ItemPaged(get_next, extract_data)
 
 
-class _KeyVaultClientOperationsMixin(
+class _KeyVaultClientOperationsMixin(  # pylint: disable=too-many-public-methods
     ClientMixinABC[PipelineClient[HttpRequest, HttpResponse], KeyVaultClientConfiguration]
 ):
 
@@ -1674,7 +1794,7 @@ def get_long_running_output(pipeline_response):
     @api_version_validation(
         method_added_on="7.6-preview.2",
         params_added_on={"7.6-preview.2": ["api_version", "content_type", "accept"]},
-        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01"],
+        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01", "2026-01-01-preview"],
     )
     def _pre_full_backup_initial(
         self,
@@ -1815,7 +1935,7 @@ def begin_pre_full_backup(
     @api_version_validation(
         method_added_on="7.6-preview.2",
         params_added_on={"7.6-preview.2": ["api_version", "content_type", "accept"]},
-        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01"],
+        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01", "2026-01-01-preview"],
     )
     def begin_pre_full_backup(
         self,
@@ -2177,7 +2297,7 @@ def get_long_running_output(pipeline_response):
     @api_version_validation(
         method_added_on="7.6-preview.2",
         params_added_on={"7.6-preview.2": ["api_version", "content_type", "accept"]},
-        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01"],
+        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01", "2026-01-01-preview"],
     )
     def _pre_full_restore_operation_initial(
         self,
@@ -2318,7 +2438,7 @@ def begin_pre_full_restore_operation(
     @api_version_validation(
         method_added_on="7.6-preview.2",
         params_added_on={"7.6-preview.2": ["api_version", "content_type", "accept"]},
-        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01"],
+        api_versions_list=["7.6-preview.2", "7.6", "2025-06-01-preview", "2025-07-01", "2026-01-01-preview"],
     )
     def begin_pre_full_restore_operation(
         self,
@@ -2988,3 +3108,582 @@ def get_settings(self, **kwargs: Any) -> _models.SettingsListResult:
             return cls(pipeline_response, deserialized, {})  # type: ignore
 
         return deserialized  # type: ignore
+
+    @distributed_trace
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    def get_ekm_connection(self, **kwargs: Any) -> _models.EkmConnection:
+        """Gets the EKM connection.
+
+        The External Key Manager (EKM) Get operation returns EKM connection. This operation requires
+        ekm/read permission.
+
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[_models.EkmConnection] = kwargs.pop("cls", None)
+
+        _request = build_key_vault_get_ekm_connection_request(
+            api_version=self._config.api_version,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmConnection, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    def get_ekm_certificate(self, **kwargs: Any) -> _models.EkmProxyClientCertificateInfo:
+        """Gets the EKM proxy client certificate.
+
+        The External Key Manager (EKM) Certificate Get operation returns Proxy client certificate. This
+        operation requires ekm/read permission.
+
+        :return: EkmProxyClientCertificateInfo. The EkmProxyClientCertificateInfo is compatible with
+         MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmProxyClientCertificateInfo
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[_models.EkmProxyClientCertificateInfo] = kwargs.pop("cls", None)
+
+        _request = build_key_vault_get_ekm_certificate_request(
+            api_version=self._config.api_version,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmProxyClientCertificateInfo, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    def check_ekm_connection(self, **kwargs: Any) -> _models.EkmProxyInfo:
+        """Checks the connectivity and authentication with the EKM proxy.
+
+        The External Key Manager (EKM) Check operation checks the connectivity and authentication with
+        the EKM proxy. This operation requires ekm/read permission.
+
+        :return: EkmProxyInfo. The EkmProxyInfo is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmProxyInfo
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[_models.EkmProxyInfo] = kwargs.pop("cls", None)
+
+        _request = build_key_vault_check_ekm_connection_request(
+            api_version=self._config.api_version,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmProxyInfo, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @overload
+    def create_ekm_connection(
+        self, ekm_connection: _models.EkmConnection, *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Creates the EKM connection.
+
+        The External Key Manager (EKM) sets up the EKM connection. If the EKM connection already
+        exists, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to create. Required.
+        :type ekm_connection: ~azure.keyvault.administration._generated.models.EkmConnection
+        :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @overload
+    def create_ekm_connection(
+        self, ekm_connection: JSON, *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Creates the EKM connection.
+
+        The External Key Manager (EKM) sets up the EKM connection. If the EKM connection already
+        exists, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to create. Required.
+        :type ekm_connection: JSON
+        :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @overload
+    def create_ekm_connection(
+        self, ekm_connection: IO[bytes], *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Creates the EKM connection.
+
+        The External Key Manager (EKM) sets up the EKM connection. If the EKM connection already
+        exists, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to create. Required.
+        :type ekm_connection: IO[bytes]
+        :keyword content_type: Body Parameter content-type. Content type parameter for binary body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @distributed_trace
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "content_type", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    def create_ekm_connection(
+        self, ekm_connection: Union[_models.EkmConnection, JSON, IO[bytes]], **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Creates the EKM connection.
+
+        The External Key Manager (EKM) sets up the EKM connection. If the EKM connection already
+        exists, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to create. Is one of the following types:
+         EkmConnection, JSON, IO[bytes] Required.
+        :type ekm_connection: ~azure.keyvault.administration._generated.models.EkmConnection or JSON or
+         IO[bytes]
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+        _params = kwargs.pop("params", {}) or {}
+
+        content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
+        cls: ClsType[_models.EkmConnection] = kwargs.pop("cls", None)
+
+        content_type = content_type or "application/json"
+        _content = None
+        if isinstance(ekm_connection, (IOBase, bytes)):
+            _content = ekm_connection
+        else:
+            _content = json.dumps(ekm_connection, cls=SdkJSONEncoder, exclude_readonly=True)  # type: ignore
+
+        _request = build_key_vault_create_ekm_connection_request(
+            content_type=content_type,
+            api_version=self._config.api_version,
+            content=_content,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmConnection, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @overload
+    def update_ekm_connection(
+        self, ekm_connection: _models.EkmConnection, *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Updates the EKM connection.
+
+        The External Key Manager (EKM) updates the existing EKM connection. If the EKM connection does
+        not exist, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to update. Required.
+        :type ekm_connection: ~azure.keyvault.administration._generated.models.EkmConnection
+        :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @overload
+    def update_ekm_connection(
+        self, ekm_connection: JSON, *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Updates the EKM connection.
+
+        The External Key Manager (EKM) updates the existing EKM connection. If the EKM connection does
+        not exist, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to update. Required.
+        :type ekm_connection: JSON
+        :keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @overload
+    def update_ekm_connection(
+        self, ekm_connection: IO[bytes], *, content_type: str = "application/json", **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Updates the EKM connection.
+
+        The External Key Manager (EKM) updates the existing EKM connection. If the EKM connection does
+        not exist, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to update. Required.
+        :type ekm_connection: IO[bytes]
+        :keyword content_type: Body Parameter content-type. Content type parameter for binary body.
+         Default value is "application/json".
+        :paramtype content_type: str
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+
+    @distributed_trace
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "content_type", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    def update_ekm_connection(
+        self, ekm_connection: Union[_models.EkmConnection, JSON, IO[bytes]], **kwargs: Any
+    ) -> _models.EkmConnection:
+        """Updates the EKM connection.
+
+        The External Key Manager (EKM) updates the existing EKM connection. If the EKM connection does
+        not exist, this operation fails. This operation requires ekm/write permission.
+
+        :param ekm_connection: The ekmConnection to update. Is one of the following types:
+         EkmConnection, JSON, IO[bytes] Required.
+        :type ekm_connection: ~azure.keyvault.administration._generated.models.EkmConnection or JSON or
+         IO[bytes]
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
+        _params = kwargs.pop("params", {}) or {}
+
+        content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
+        cls: ClsType[_models.EkmConnection] = kwargs.pop("cls", None)
+
+        content_type = content_type or "application/json"
+        _content = None
+        if isinstance(ekm_connection, (IOBase, bytes)):
+            _content = ekm_connection
+        else:
+            _content = json.dumps(ekm_connection, cls=SdkJSONEncoder, exclude_readonly=True)  # type: ignore
+
+        _request = build_key_vault_update_ekm_connection_request(
+            content_type=content_type,
+            api_version=self._config.api_version,
+            content=_content,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmConnection, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace
+    @api_version_validation(
+        method_added_on="2026-01-01-preview",
+        params_added_on={"2026-01-01-preview": ["api_version", "accept"]},
+        api_versions_list=["2026-01-01-preview"],
+    )
+    def delete_ekm_connection(self, **kwargs: Any) -> _models.EkmConnection:
+        """Deletes the EKM connection.
+
+        The External Key Manager (EKM) deletes the existing EKM connection. If the EKM connection does
+        not already exists, this operation fails. This operation requires ekm/delete permission.
+
+        :return: EkmConnection. The EkmConnection is compatible with MutableMapping
+        :rtype: ~azure.keyvault.administration._generated.models.EkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[_models.EkmConnection] = kwargs.pop("cls", None)
+
+        _request = build_key_vault_delete_ekm_connection_request(
+            api_version=self._config.api_version,
+            headers=_headers,
+            params=_params,
+        )
+        path_format_arguments = {
+            "vaultBaseUrl": self._serialize.url(
+                "self._config.vault_base_url", self._config.vault_base_url, "str", skip_quote=True
+            ),
+        }
+        _request.url = self._client.format_url(_request.url, **path_format_arguments)
+
+        _decompress = kwargs.pop("decompress", True)
+        _stream = kwargs.pop("stream", False)
+        pipeline_response: PipelineResponse = self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            if _stream:
+                try:
+                    response.read()  # Load the body in memory and close the socket
+                except (StreamConsumedError, StreamClosedError):
+                    pass
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = _failsafe_deserialize(
+                _models.KeyVaultError,
+                response,
+            )
+            raise HttpResponseError(response=response, model=error)
+
+        if _stream:
+            deserialized = response.iter_bytes() if _decompress else response.iter_raw()
+        else:
+            deserialized = _deserialize(_models.EkmConnection, response.json())
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_internal/client_base.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_internal/client_base.py
index 0ed49a8c55b3..dcf10c5d27df 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_internal/client_base.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_internal/client_base.py
@@ -24,6 +24,7 @@ class ApiVersion(str, Enum, metaclass=CaseInsensitiveEnumMeta):
     """Key Vault API versions supported by this package"""
 
     #: this is the default version
+    V2026_01_01_PREVIEW = "2026-01-01-preview"
     V2025_07_01 = "2025-07-01"
     V7_6 = "7.6"
     V7_5 = "7.5"
@@ -32,7 +33,7 @@ class ApiVersion(str, Enum, metaclass=CaseInsensitiveEnumMeta):
     V7_2 = "7.2"
 
 
-DEFAULT_VERSION = ApiVersion.V2025_07_01
+DEFAULT_VERSION = ApiVersion.V2026_01_01_PREVIEW
 
 _SERIALIZER = Serializer()
 _SERIALIZER.client_side_validation = False
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_models.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_models.py
index d92c517b7510..ee2c1d8e5f59 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_models.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/_models.py
@@ -2,12 +2,15 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 from azure.core.rest import HttpResponse
 
 from ._enums import KeyVaultSettingType
 from ._generated.models import (
+    EkmConnection,
+    EkmProxyClientCertificateInfo,
+    EkmProxyInfo,
     FullBackupOperation,
     Permission,
     RoleAssignment,
@@ -234,3 +237,129 @@ def getboolean(self) -> bool:
     def _from_generated(cls, setting: Setting) -> "KeyVaultSetting":
         setting_type = KeyVaultSettingType.BOOLEAN if setting.type == "boolean" else setting.type
         return cls(name=setting.name, value=setting.value, setting_type=setting_type)
+
+
+class KeyVaultEkmConnection(object):
+    """An External Key Manager (EKM) connection.
+
+    :param str host: EKM proxy FQDN (Fully Qualified Domain Name). Only allowed characters are
+        ``a-z``, ``A-Z``, ``0-9``, hyphen (``-``), dot (``.``), and colon (``:``).
+    :param server_ca_certificates: The root CA certificate chain that issued the proxy server's
+        certificate. An array of certificates in the certificate chain, each in DER format and
+        base64 encoded.
+    :type server_ca_certificates: list[bytes]
+
+    :keyword str path_prefix: Optional path prefix for the EKM proxy (if any).
+    :keyword str server_subject_common_name: The subject common name of the server certificate of
+        the EKM proxy.
+
+    :ivar str host: EKM proxy FQDN.
+    :ivar list[bytes] server_ca_certificates: The root CA certificate chain that issued the proxy
+        server's certificate.
+    :ivar path_prefix: Optional path prefix for the EKM proxy (if any).
+    :vartype path_prefix: str or None
+    :ivar server_subject_common_name: The subject common name of the server certificate of the EKM
+        proxy.
+    :vartype server_subject_common_name: str or None
+    """
+
+    def __init__(
+        self,
+        host: str,
+        server_ca_certificates: List[bytes],
+        *,
+        path_prefix: Optional[str] = None,
+        server_subject_common_name: Optional[str] = None,
+    ) -> None:
+        self.host = host
+        self.server_ca_certificates = server_ca_certificates
+        self.path_prefix = path_prefix
+        self.server_subject_common_name = server_subject_common_name
+
+    def __repr__(self) -> str:
+        return f"KeyVaultEkmConnection<{self.host}>"
+
+    @classmethod
+    def _from_generated(cls, connection: EkmConnection) -> "KeyVaultEkmConnection":
+        return cls(
+            host=connection.host,
+            server_ca_certificates=connection.server_ca_certificates,
+            path_prefix=connection.path_prefix,
+            server_subject_common_name=connection.server_subject_common_name,
+        )
+
+    def _to_generated(self) -> EkmConnection:
+        return EkmConnection(
+            host=self.host,
+            server_ca_certificates=self.server_ca_certificates,
+            path_prefix=self.path_prefix,
+            server_subject_common_name=self.server_subject_common_name,
+        )
+
+
+class KeyVaultEkmProxyClientCertificateInfo(object):
+    """EKM proxy client certificate information.
+
+    :ivar ca_certificates: The client root CA certificate chain to authenticate to the EKM proxy.
+        An array of certificates in the certificate chain, each in DER format and base64 encoded.
+    :vartype ca_certificates: list[bytes] or None
+    :ivar subject_common_name: The subject common name of the client certificate used to
+        authenticate to the EKM proxy.
+    :vartype subject_common_name: str or None
+    """
+
+    def __init__(self, **kwargs: Any) -> None:
+        self.ca_certificates: Optional[List[bytes]] = kwargs.get("ca_certificates")
+        self.subject_common_name: Optional[str] = kwargs.get("subject_common_name")
+
+    def __repr__(self) -> str:
+        return f"KeyVaultEkmProxyClientCertificateInfo<{self.subject_common_name}>"
+
+    @classmethod
+    def _from_generated(
+        cls, certificate_info: EkmProxyClientCertificateInfo
+    ) -> "KeyVaultEkmProxyClientCertificateInfo":
+        return cls(
+            ca_certificates=certificate_info.ca_certificates,
+            subject_common_name=certificate_info.subject_common_name,
+        )
+
+
+class KeyVaultEkmProxyInfo(object):
+    """EKM proxy information returned when checking an EKM connection.
+
+    :ivar api_version: The highest version of proxy interface API supported by the EKM proxy.
+    :vartype api_version: str or None
+    :ivar proxy_vendor: The name of the proxy vendor.
+    :vartype proxy_vendor: str or None
+    :ivar proxy_name: The name of the proxy product and its version.
+    :vartype proxy_name: str or None
+    :ivar ekm_vendor: The name of the EKM vendor.
+    :vartype ekm_vendor: str or None
+    :ivar ekm_product: The name of the EKM product and its version.
+    :vartype ekm_product: str or None
+    """
+
+    def __init__(self, **kwargs: Any) -> None:
+        self.api_version: Optional[str] = kwargs.get("api_version")
+        self.proxy_vendor: Optional[str] = kwargs.get("proxy_vendor")
+        self.proxy_name: Optional[str] = kwargs.get("proxy_name")
+        self.ekm_vendor: Optional[str] = kwargs.get("ekm_vendor")
+        self.ekm_product: Optional[str] = kwargs.get("ekm_product")
+
+    def __repr__(self) -> str:
+        return (
+            f"KeyVaultEkmProxyInfo(api_version={self.api_version}, "
+            f"proxy_vendor={self.proxy_vendor}, proxy_name={self.proxy_name}, "
+            f"ekm_vendor={self.ekm_vendor}, ekm_product={self.ekm_product})"
+        )
+
+    @classmethod
+    def _from_generated(cls, proxy_info: EkmProxyInfo) -> "KeyVaultEkmProxyInfo":
+        return cls(
+            api_version=proxy_info.api_version,
+            proxy_vendor=proxy_info.proxy_vendor,
+            proxy_name=proxy_info.proxy_name,
+            ekm_vendor=proxy_info.ekm_vendor,
+            ekm_product=proxy_info.ekm_product,
+        )
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/aio/__init__.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/aio/__init__.py
index d1f1bd6d374a..fe26a5ed276e 100644
--- a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/aio/__init__.py
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/aio/__init__.py
@@ -4,6 +4,12 @@
 # ------------------------------------
 from ._access_control_client import KeyVaultAccessControlClient
 from ._backup_client import KeyVaultBackupClient
+from ._ekm_client import KeyVaultEkmClient
 from ._settings_client import KeyVaultSettingsClient
 
-__all__ = ["KeyVaultAccessControlClient", "KeyVaultBackupClient", "KeyVaultSettingsClient"]
+__all__ = [
+    "KeyVaultAccessControlClient",
+    "KeyVaultBackupClient",
+    "KeyVaultEkmClient",
+    "KeyVaultSettingsClient",
+]
diff --git a/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/aio/_ekm_client.py b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/aio/_ekm_client.py
new file mode 100644
index 000000000000..66b093c9fc13
--- /dev/null
+++ b/sdk/keyvault/azure-keyvault-administration/azure/keyvault/administration/aio/_ekm_client.py
@@ -0,0 +1,114 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+from typing import Any
+
+from azure.core.tracing.decorator_async import distributed_trace_async
+
+from .._internal import AsyncKeyVaultClientBase
+from .._models import KeyVaultEkmConnection, KeyVaultEkmProxyClientCertificateInfo, KeyVaultEkmProxyInfo
+
+
+class KeyVaultEkmClient(AsyncKeyVaultClientBase):
+    """Provides methods to manage Managed HSM External Key Manager (EKM) connections.
+
+    :param str vault_url: URL of the vault on which the client will operate. This is also called the vault's "DNS Name".
+        You should validate that this URL references a valid Key Vault or Managed HSM resource.
+        See https://aka.ms/azsdk/blog/vault-uri for details.
+    :param credential: An object which can provide an access token for the vault, such as a credential from
+        :mod:`azure.identity.aio`
+    :type credential: ~azure.core.credentials_async.AsyncTokenCredential
+
+    :keyword api_version: Version of the service API to use. EKM operations require service API version
+        ``2026-01-01-preview`` or later.
+    :paramtype api_version: ~azure.keyvault.administration.ApiVersion or str
+    :keyword bool verify_challenge_resource: Whether to verify the authentication challenge resource matches the Key
+        Vault or Managed HSM domain. Defaults to True.
+    """
+
+    # pylint:disable=protected-access
+
+    @distributed_trace_async
+    async def get_ekm_connection(self, **kwargs: Any) -> KeyVaultEkmConnection:
+        """Gets the configured EKM connection.
+
+        :returns: The configured EKM connection.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = await self._client.get_ekm_connection(**kwargs)
+        return KeyVaultEkmConnection._from_generated(result)
+
+    @distributed_trace_async
+    async def create_ekm_connection(self, connection: KeyVaultEkmConnection, **kwargs: Any) -> KeyVaultEkmConnection:
+        """Creates the EKM connection.
+
+        If an EKM connection already exists, this operation fails.
+
+        :param connection: The EKM connection to create.
+        :type connection: ~azure.keyvault.administration.KeyVaultEkmConnection
+
+        :returns: The created EKM connection.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = await self._client.create_ekm_connection(ekm_connection=connection._to_generated(), **kwargs)
+        return KeyVaultEkmConnection._from_generated(result)
+
+    @distributed_trace_async
+    async def update_ekm_connection(self, connection: KeyVaultEkmConnection, **kwargs: Any) -> KeyVaultEkmConnection:
+        """Updates the existing EKM connection.
+
+        If no EKM connection exists, this operation fails.
+
+        :param connection: The EKM connection to update.
+        :type connection: ~azure.keyvault.administration.KeyVaultEkmConnection
+
+        :returns: The updated EKM connection.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = await self._client.update_ekm_connection(ekm_connection=connection._to_generated(), **kwargs)
+        return KeyVaultEkmConnection._from_generated(result)
+
+    @distributed_trace_async
+    async def delete_ekm_connection(  # pylint:disable=bad-option-value,delete-operation-wrong-return-type
+        self, **kwargs: Any
+    ) -> KeyVaultEkmConnection:
+        """Deletes the existing EKM connection.
+
+        If no EKM connection exists, this operation fails.
+
+        :returns: The deleted EKM connection.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmConnection
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = await self._client.delete_ekm_connection(**kwargs)
+        return KeyVaultEkmConnection._from_generated(result)
+
+    @distributed_trace_async
+    async def get_ekm_certificate(self, **kwargs: Any) -> KeyVaultEkmProxyClientCertificateInfo:
+        """Gets the EKM proxy client certificate information used to authenticate to the EKM proxy.
+
+        :returns: The EKM proxy client certificate information.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmProxyClientCertificateInfo
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = await self._client.get_ekm_certificate(**kwargs)
+        return KeyVaultEkmProxyClientCertificateInfo._from_generated(result)
+
+    @distributed_trace_async
+    async def check_ekm_connection(self, **kwargs: Any) -> KeyVaultEkmProxyInfo:
+        """Checks the EKM connection by pinging the EKM proxy.
+
+        :returns: Information about the EKM proxy returned by the connectivity check.
+        :rtype: ~azure.keyvault.administration.KeyVaultEkmProxyInfo
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        result = await self._client.check_ekm_connection(**kwargs)
+        return KeyVaultEkmProxyInfo._from_generated(result)
+
+    async def __aenter__(self) -> "KeyVaultEkmClient":
+        await self._client.__aenter__()
+        return self
diff --git a/sdk/keyvault/azure-keyvault-administration/samples/ekm_operations.py b/sdk/keyvault/azure-keyvault-administration/samples/ekm_operations.py
new file mode 100644
index 000000000000..faac37252f6e
--- /dev/null
+++ b/sdk/keyvault/azure-keyvault-administration/samples/ekm_operations.py
@@ -0,0 +1,114 @@
+# pylint: disable=line-too-long,useless-suppression
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+
+# ----------------------------------------------------------------------------------------------------------
+# Prerequisites:
+# 1. A managed HSM (https://learn.microsoft.com/azure/key-vault/managed-hsm/quick-create-cli)
+#
+# 2. azure-keyvault-administration and azure-identity libraries (pip install these)
+#
+# 3. Set environment variable MANAGED_HSM_URL with the URL of your managed HSM, EKM_PROXY_HOST with the host of your EKM proxy,
+#    and CA_CERTIFICATE with the proxy server's certificate in DER format and base64 encoded.
+#
+# 4. Set up your environment to use azure-identity's DefaultAzureCredential. For more information about how to configure
+#    the DefaultAzureCredential, refer to https://aka.ms/azsdk/python/identity/docs#azure.identity.DefaultAzureCredential
+#
+# 5. An EKM (External Key Manager) proxy reachable from the Managed HSM. Replace the placeholder host
+#    and certificate bytes below with values that match your EKM proxy deployment.
+#
+# ----------------------------------------------------------------------------------------------------------
+# Sample - demonstrates Managed HSM External Key Manager (EKM) connection management
+#
+# 1. Create an EKM connection (create_ekm_connection)
+#
+# 2. Read the EKM connection (get_ekm_connection)
+#
+# 3. Get the EKM proxy client certificate (get_ekm_certificate)
+#
+# 4. Check an EKM connection (check_ekm_connection)
+#
+# 5. Update the EKM connection (update_ekm_connection)
+#
+# 6. Delete the EKM connection (delete_ekm_connection)
+# ----------------------------------------------------------------------------------------------------------
+
+# Instantiate an EKM client that will be used to call the service.
+# Here we use the DefaultAzureCredential, but any azure-identity credential can be used.
+# [START create_a_ekm_client]
+import os
+import base64
+from azure.identity import DefaultAzureCredential
+from azure.keyvault.administration import KeyVaultEkmClient, KeyVaultEkmConnection
+
+MANAGED_HSM_URL = os.environ["MANAGED_HSM_URL"]
+EKM_PROXY_HOST = os.environ["EKM_PROXY_HOST"]
+CA_CERTIFICATE = os.environ["CA_CERTIFICATE"]
+CA_CERTIFICATES = [CA_CERTIFICATE]
+credential = DefaultAzureCredential()
+client = KeyVaultEkmClient(vault_url=MANAGED_HSM_URL, credential=credential)
+# [END create_a_ekm_client]
+
+# First, let's create an EKM connection
+print("\n.. Create EKM connection")
+# [START create_ekm_connection]
+ekm_connection = KeyVaultEkmConnection(
+    host=EKM_PROXY_HOST,
+    server_ca_certificates=[base64.b64decode(cert) for cert in CA_CERTIFICATES],
+    path_prefix="/api/v1",
+)
+created_ekm_connection = client.create_ekm_connection(connection=ekm_connection)
+print(f"EKM connection created with host: {created_ekm_connection.host}")
+# [END create_ekm_connection]
+
+# Let's get the EKM connection we just created
+print("\n.. Get EKM connection")
+# [START get_ekm_connection]
+retrieved_ekm_connection = client.get_ekm_connection()
+print("Retrieved EKM connection with:")
+print(f"\tHost: {retrieved_ekm_connection.host}")
+print(f"\tPath prefix: {retrieved_ekm_connection.path_prefix}")
+print(
+    f"\tServer subject common name: {retrieved_ekm_connection.server_subject_common_name}"
+)
+# [END get_ekm_connection]
+
+# Get the EKM certificate
+print("\n.. Get EKM certificate")
+# [START get_ekm_certificate]
+ekm_certificate = client.get_ekm_certificate()
+print(f"EKM certificate retrieved with subject: {ekm_certificate.subject_common_name}")
+# [END get_ekm_certificate]
+
+# Check the EKM connection status
+print("\n.. Check EKM connection")
+# [START check_ekm_connection]
+connection_status = client.check_ekm_connection()
+print("EKM connection status:")
+print(f"\tAPI Version: {connection_status.api_version}")
+print(f"\tProxy Vendor: {connection_status.proxy_vendor}")
+print(f"\tProxy Name: {connection_status.proxy_name}")
+print(f"\tEKM Vendor: {connection_status.ekm_vendor}")
+print(f"\tEKM Product: {connection_status.ekm_product}")
+# [END check_ekm_connection]
+
+# Update the EKM connection
+print("\n.. Update EKM connection")
+# [START update_ekm_connection]
+updated_ekm_connection = KeyVaultEkmConnection(
+    host="ekm-proxy-updated.contoso.com",
+    server_ca_certificates=[base64.b64decode(cert) for cert in CA_CERTIFICATES],
+    path_prefix="/api/v2",
+)
+result = client.update_ekm_connection(connection=updated_ekm_connection)
+print(f"EKM connection updated with host: {result.host}")
+# [END update_ekm_connection]
+
+# Finally, let's delete the EKM connection
+print("\n.. Delete EKM connection")
+# [START delete_ekm_connection]
+deleted_ekm_connection = client.delete_ekm_connection()
+print("EKM connection deleted successfully")
+# [END delete_ekm_connection]
diff --git a/sdk/keyvault/azure-keyvault-administration/samples/ekm_operations_async.py b/sdk/keyvault/azure-keyvault-administration/samples/ekm_operations_async.py
new file mode 100644
index 000000000000..c4c868850340
--- /dev/null
+++ b/sdk/keyvault/azure-keyvault-administration/samples/ekm_operations_async.py
@@ -0,0 +1,129 @@
+# pylint: disable=line-too-long,useless-suppression
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+
+# ----------------------------------------------------------------------------------------------------------
+# Prerequisites:
+# 1. A managed HSM (https://learn.microsoft.com/azure/key-vault/managed-hsm/quick-create-cli)
+#
+# 2. azure-keyvault-administration and azure-identity libraries (pip install these)
+#
+# 3. Set environment variable MANAGED_HSM_URL with the URL of your managed HSM, EKM_PROXY_HOST with the host of your EKM proxy,
+#    and CA_CERTIFICATE with the proxy server's certificate in DER format and base64 encoded.
+#
+# 4. Set up your environment to use azure-identity's DefaultAzureCredential. For more information about how to configure
+#    the DefaultAzureCredential, refer to https://aka.ms/azsdk/python/identity/docs#azure.identity.DefaultAzureCredential
+#
+# 5. An EKM (External Key Manager) proxy reachable from the Managed HSM. Replace the placeholder host
+#    and certificate bytes below with values that match your EKM proxy deployment.
+#
+# ----------------------------------------------------------------------------------------------------------
+# Sample - demonstrates Managed HSM External Key Manager (EKM) connection management
+#
+# 1. Create an EKM connection (create_ekm_connection)
+#
+# 2. Read the EKM connection (get_ekm_connection)
+#
+# 3. Get the EKM proxy client certificate (get_ekm_certificate)
+#
+# 4. Check an EKM connection (check_ekm_connection)
+#
+# 5. Update the EKM connection (update_ekm_connection)
+#
+# 6. Delete the EKM connection (delete_ekm_connection)
+# ----------------------------------------------------------------------------------------------------------
+
+# Instantiate an EKM client that will be used to call the service.
+# Here we use the DefaultAzureCredential, but any azure-identity credential can be used.
+# [START create_a_ekm_client]
+import os
+import base64
+import asyncio
+from azure.identity.aio import DefaultAzureCredential
+from azure.keyvault.administration import KeyVaultEkmConnection
+from azure.keyvault.administration.aio import KeyVaultEkmClient
+
+
+async def run_sample():
+    MANAGED_HSM_URL = os.environ["MANAGED_HSM_URL"]
+    EKM_PROXY_HOST = os.environ["EKM_PROXY_HOST"]
+    CA_CERTIFICATE = os.environ["CA_CERTIFICATE"]
+    CA_CERTIFICATES = [CA_CERTIFICATE]
+    credential = DefaultAzureCredential()
+    client = KeyVaultEkmClient(vault_url=MANAGED_HSM_URL, credential=credential)
+    # [END create_a_ekm_client]
+
+    # First, let's create an EKM connection
+    print("\n.. Create EKM connection")
+    # [START create_ekm_connection]
+    ekm_connection = KeyVaultEkmConnection(
+        host=EKM_PROXY_HOST,
+        server_ca_certificates=[base64.b64decode(cert) for cert in CA_CERTIFICATES],
+        path_prefix="/api/v1",
+    )
+    created_ekm_connection = await client.create_ekm_connection(
+        connection=ekm_connection
+    )
+    print(f"EKM connection created with host: {created_ekm_connection.host}")
+    # [END create_ekm_connection]
+
+    # Let's get the EKM connection we just created
+    print("\n.. Get EKM connection")
+    # [START get_ekm_connection]
+    retrieved_ekm_connection = await client.get_ekm_connection()
+    print("Retrieved EKM connection with:")
+    print(f"\tHost: {retrieved_ekm_connection.host}")
+    print(f"\tPath prefix: {retrieved_ekm_connection.path_prefix}")
+    print(
+        f"\tServer subject common name: {retrieved_ekm_connection.server_subject_common_name}"
+    )
+    # [END get_ekm_connection]
+
+    # Get the EKM certificate
+    print("\n.. Get EKM certificate")
+    # [START get_ekm_certificate]
+    ekm_certificate = await client.get_ekm_certificate()
+    print(
+        f"EKM certificate retrieved with subject: {ekm_certificate.subject_common_name}"
+    )
+    # [END get_ekm_certificate]
+
+    # Check the EKM connection status
+    print("\n.. Check EKM connection")
+    # [START check_ekm_connection]
+    connection_status = await client.check_ekm_connection()
+    print("EKM connection status:")
+    print(f"\tAPI Version: {connection_status.api_version}")
+    print(f"\tProxy Vendor: {connection_status.proxy_vendor}")
+    print(f"\tProxy Name: {connection_status.proxy_name}")
+    print(f"\tEKM Vendor: {connection_status.ekm_vendor}")
+    print(f"\tEKM Product: {connection_status.ekm_product}")
+    # [END check_ekm_connection]
+
+    # Update the EKM connection
+    print("\n.. Update EKM connection")
+    # [START update_ekm_connection]
+    updated_ekm_connection = KeyVaultEkmConnection(
+        host="ekm-proxy-updated.contoso.com",
+        server_ca_certificates=[base64.b64decode(cert) for cert in CA_CERTIFICATES],
+        path_prefix="/api/v2",
+    )
+    result = await client.update_ekm_connection(connection=updated_ekm_connection)
+    print(f"EKM connection updated with host: {result.host}")
+    # [END update_ekm_connection]
+
+    # Finally, let's delete the EKM connection
+    print("\n.. Delete EKM connection")
+    # [START delete_ekm_connection]
+    deleted_ekm_connection = await client.delete_ekm_connection()
+    print("EKM connection deleted successfully")
+    # [END delete_ekm_connection]
+
+    await client.close()
+    await credential.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(run_sample())
diff --git a/sdk/keyvault/azure-keyvault-administration/tests/_async_test_case.py b/sdk/keyvault/azure-keyvault-administration/tests/_async_test_case.py
index ab9fac0ebe9d..f1b99ca8271c 100644
--- a/sdk/keyvault/azure-keyvault-administration/tests/_async_test_case.py
+++ b/sdk/keyvault/azure-keyvault-administration/tests/_async_test_case.py
@@ -14,6 +14,8 @@ def __init__(self, **kwargs) -> None:
         hsm_playback_url = "https://managedhsmvaultname.managedhsm.azure.net"
         container_playback_uri = "https://storagename.blob.core.windows.net/container"
         playback_sas_token = "fake-sas"
+        playback_ekm_host = "fake-ekm-host"
+        playback_ekm_certificate = "fake-server-ca-certificate"
 
         if self.is_live:
             hsm = os.environ.get("AZURE_MANAGEDHSM_URL")
@@ -23,11 +25,15 @@ def __init__(self, **kwargs) -> None:
             self.container_uri = f"{storage_url.rstrip('/')}/{container_name}"
 
             self.sas_token = os.environ.get("BLOB_STORAGE_SAS_TOKEN")
+            self.ekm_host = os.environ.get("EKM_PROXY_HOST")
+            self.ekm_certificate = os.environ.get("EKM_SERVER_CA_CERTIFICATE")
 
         else:
             self.managed_hsm_url = hsm_playback_url
             self.container_uri = container_playback_uri
             self.sas_token = playback_sas_token
+            self.ekm_host = playback_ekm_host
+            self.ekm_certificate = playback_ekm_certificate
 
         use_pwsh = os.environ.get("AZURE_TEST_USE_PWSH_AUTH", "false")
         use_cli = os.environ.get("AZURE_TEST_USE_CLI_AUTH", "false")
@@ -133,3 +139,25 @@ def create_access_control_client(self, **kwargs):
         return self.create_client_from_credential(
             KeyVaultSettingsClient, credential=credential, vault_url=self.managed_hsm_url, **kwargs
         )
+    
+
+class KeyVaultEkmClientPreparer(BaseClientPreparer):
+    def __call__(self, fn):
+        async def _preparer(test_class, api_version, **kwargs):
+            self._skip_if_not_configured(api_version)
+            kwargs["ekm_host"] = self.ekm_host
+            kwargs["ekm_certificate"] = self.ekm_certificate
+            client = self.create_ekm_client(api_version=api_version, **kwargs)
+
+            async with client:
+                await fn(test_class, client, **kwargs)
+
+        return _preparer
+
+    def create_ekm_client(self, **kwargs):
+        from azure.keyvault.administration.aio import KeyVaultEkmClient
+
+        credential = self.get_credential(KeyVaultEkmClient, is_async=True)
+        return self.create_client_from_credential(
+            KeyVaultEkmClient, credential=credential, vault_url=self.managed_hsm_url, **kwargs
+        )
diff --git a/sdk/keyvault/azure-keyvault-administration/tests/_test_case.py b/sdk/keyvault/azure-keyvault-administration/tests/_test_case.py
index fbeca4fd8ec7..e0bc2d83805f 100644
--- a/sdk/keyvault/azure-keyvault-administration/tests/_test_case.py
+++ b/sdk/keyvault/azure-keyvault-administration/tests/_test_case.py
@@ -16,6 +16,8 @@ def __init__(self, **kwargs) -> None:
         hsm_playback_url = "https://managedhsmvaultname.managedhsm.azure.net"
         container_playback_uri = "https://storagename.blob.core.windows.net/container"
         playback_sas_token = "fake-sas"
+        playback_ekm_host = "fake-ekm-host"
+        playback_server_ca_certificate = "fake-server-ca-certificate"
 
         if self.is_live:
             hsm = os.environ.get("AZURE_MANAGEDHSM_URL")
@@ -25,11 +27,15 @@ def __init__(self, **kwargs) -> None:
             self.container_uri = f"{storage_url.rstrip('/')}/{container_name}"
 
             self.sas_token = os.environ.get("BLOB_STORAGE_SAS_TOKEN")
+            self.ekm_host = os.environ.get("EKM_PROXY_HOST")
+            self.ekm_certificate = os.environ.get("EKM_SERVER_CA_CERTIFICATE")
 
         else:
             self.managed_hsm_url = hsm_playback_url
             self.container_uri = container_playback_uri
             self.sas_token = playback_sas_token
+            self.ekm_host = playback_ekm_host
+            self.ekm_certificate = playback_server_ca_certificate
 
         use_pwsh = os.environ.get("AZURE_TEST_USE_PWSH_AUTH", "false")
         use_cli = os.environ.get("AZURE_TEST_USE_CLI_AUTH", "false")
@@ -149,6 +155,31 @@ def create_settings_client(self, **kwargs):
         )
 
 
+class KeyVaultEkmClientPreparer(BaseClientPreparer):
+    def __init__(self, **kwargs) -> None:
+        super().__init__(**kwargs)
+
+    def __call__(self, fn):
+        def _preparer(test_class, api_version, **kwargs):
+            self._skip_if_not_configured(api_version)
+            kwargs["ekm_host"] = self.ekm_host
+            kwargs["ekm_certificate"] = self.ekm_certificate
+            client = self.create_ekm_client(api_version=api_version, **kwargs)
+
+            with client:
+                fn(test_class, client, **kwargs)
+
+        return _preparer
+
+    def create_ekm_client(self, **kwargs):
+        from azure.keyvault.administration import KeyVaultEkmClient
+
+        credential = self.get_credential(KeyVaultEkmClient)
+        return self.create_client_from_credential(
+            KeyVaultEkmClient, credential=credential, vault_url=self.managed_hsm_url, **kwargs
+        )
+
+
 def get_decorator(**kwargs):
     """returns a test decorator for test parameterization"""
     versions = kwargs.pop("api_versions", None) or ApiVersion
diff --git a/sdk/keyvault/azure-keyvault-administration/tests/conftest.py b/sdk/keyvault/azure-keyvault-administration/tests/conftest.py
index 872fdb9d38f7..f1a9e3464618 100644
--- a/sdk/keyvault/azure-keyvault-administration/tests/conftest.py
+++ b/sdk/keyvault/azure-keyvault-administration/tests/conftest.py
@@ -32,6 +32,8 @@ def add_sanitizers(test_proxy):
     storage_url = os.environ.get("BLOB_STORAGE_URL", "https://Sanitized.blob.core.windows.net")
     client_id = os.environ.get("KEYVAULT_CLIENT_ID", "service-principal-id")
     sas_token = os.environ.get("BLOB_STORAGE_SAS_TOKEN", "fake-sas")
+    ekm_host = os.environ.get("EKM_PROXY_HOST", "fake-ekm-host")
+    ekm_certificate = os.environ.get("EKM_SERVER_CA_CERTIFICATE", "fake-server-ca-certificate")
 
     add_general_string_sanitizer(target=azure_keyvault_url, value="https://Sanitized.vault.azure.net")
     add_general_string_sanitizer(target=keyvault_tenant_id, value="00000000-0000-0000-0000-000000000000")
@@ -40,6 +42,8 @@ def add_sanitizers(test_proxy):
     add_general_string_sanitizer(target=azure_attestation_uri, value="https://Sanitized.azurewebsites.net")
     add_general_string_sanitizer(target=storage_url, value="https://Sanitized.blob.core.windows.net")
     add_general_string_sanitizer(target=sas_token, value="fake-sas")
+    add_general_string_sanitizer(target=ekm_host, value="fake-ekm-host")
+    add_general_string_sanitizer(target=ekm_certificate, value="fake-server-ca-certificate")
     add_general_string_sanitizer(target=client_id, value="service-principal-id")
     # Sanitize API versions of `azure-keyvault-keys` requests
     add_uri_regex_sanitizer(
diff --git a/sdk/keyvault/azure-keyvault-administration/tests/test_ekm_client.py b/sdk/keyvault/azure-keyvault-administration/tests/test_ekm_client.py
new file mode 100644
index 000000000000..1bf4cccd8c18
--- /dev/null
+++ b/sdk/keyvault/azure-keyvault-administration/tests/test_ekm_client.py
@@ -0,0 +1,101 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+
+import base64
+import pytest
+
+from azure.core.exceptions import HttpResponseError
+from azure.keyvault.administration import KeyVaultEkmClient, KeyVaultEkmConnection
+from azure.keyvault.administration._internal.client_base import DEFAULT_VERSION
+
+from devtools_testutils import recorded_by_proxy
+
+from _shared.test_case import KeyVaultTestCase
+from _test_case import KeyVaultEkmClientPreparer, get_decorator
+
+only_latest = get_decorator(api_versions=[DEFAULT_VERSION])
+
+# Note: These tests require an EKM connection to be established with an EKM Sample Proxy.
+
+
+class TestEkm(KeyVaultTestCase):
+    @pytest.mark.live_test_only
+    @pytest.mark.parametrize("api_version", only_latest)
+    @KeyVaultEkmClientPreparer()
+    @recorded_by_proxy
+    def test_ekm_connection(self, client: KeyVaultEkmClient, **kwargs):
+        ekm_host = kwargs.pop("ekm_host")
+        server_ca_certificate = kwargs.pop("ekm_certificate")
+        if not server_ca_certificate or not ekm_host:
+            pytest.skip(
+                "EKM CA certificate is required for live tests. Please set the EKM_PROXY_HOST and EKM_SERVER_CA_CERTIFICATE environment variables."
+            )
+
+        # Cleanup
+        try:
+            client.delete_ekm_connection()
+        except HttpResponseError:
+            pass
+
+        # Create an EKM connection
+        ekm_connection = KeyVaultEkmConnection(
+            host=ekm_host,
+            server_ca_certificates=[base64.b64decode(server_ca_certificate)],
+            path_prefix="/api/v1",
+        )
+        created_ekm_connection = client.create_ekm_connection(connection=ekm_connection)
+        assert created_ekm_connection is not None
+        assert created_ekm_connection.host == ekm_host
+        assert created_ekm_connection.server_ca_certificates is not None
+        assert len(created_ekm_connection.server_ca_certificates) == 1
+        assert created_ekm_connection.path_prefix == ekm_connection.path_prefix
+        assert created_ekm_connection.server_subject_common_name == ekm_connection.server_subject_common_name
+
+        # Get the EKM connection
+        retrieved_ekm_connection = client.get_ekm_connection()
+        assert retrieved_ekm_connection is not None
+        assert retrieved_ekm_connection.host == ekm_host
+        assert retrieved_ekm_connection.server_ca_certificates is not None
+        assert len(retrieved_ekm_connection.server_ca_certificates) == 1
+        assert retrieved_ekm_connection.path_prefix == ekm_connection.path_prefix
+        assert retrieved_ekm_connection.server_subject_common_name == created_ekm_connection.server_subject_common_name
+
+        # Get the EKM certificate
+        ekm_certificate = client.get_ekm_certificate()
+        assert ekm_certificate is not None
+        assert ekm_certificate.ca_certificates is not None
+        assert len(ekm_certificate.ca_certificates) == 1
+
+        # Check the EKM connection status
+        connection_status = client.check_ekm_connection()
+        assert connection_status is not None
+        assert connection_status.api_version is not None
+        assert connection_status.proxy_vendor is not None
+        assert connection_status.proxy_name is not None
+        assert connection_status.ekm_vendor is not None
+        assert connection_status.ekm_product is not None
+
+        # Update the EKM connection
+        updated_ekm_connection = KeyVaultEkmConnection(
+            host=ekm_host,
+            server_ca_certificates=[base64.b64decode(server_ca_certificate)],
+            path_prefix="/api/v1",
+        )
+        result = client.update_ekm_connection(connection=updated_ekm_connection)
+        assert result is not None
+        assert result.host == updated_ekm_connection.host
+        assert result.server_ca_certificates is not None
+        assert len(result.server_ca_certificates) == 1
+        assert result.path_prefix == updated_ekm_connection.path_prefix
+        assert result.server_subject_common_name == updated_ekm_connection.server_subject_common_name
+
+        # Delete the EKM connection
+        result = client.delete_ekm_connection()
+        assert result is not None
+        assert result.host == updated_ekm_connection.host
+        assert result.server_ca_certificates is not None
+        assert len(result.server_ca_certificates) == 1
+        assert result.path_prefix == updated_ekm_connection.path_prefix
+        assert result.server_subject_common_name == updated_ekm_connection.server_subject_common_name
diff --git a/sdk/keyvault/azure-keyvault-administration/tests/test_ekm_client_async.py b/sdk/keyvault/azure-keyvault-administration/tests/test_ekm_client_async.py
new file mode 100644
index 000000000000..702b316ef57a
--- /dev/null
+++ b/sdk/keyvault/azure-keyvault-administration/tests/test_ekm_client_async.py
@@ -0,0 +1,104 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+
+import base64
+import pytest
+
+from azure.core.exceptions import HttpResponseError
+from azure.keyvault.administration import KeyVaultEkmConnection
+from azure.keyvault.administration.aio import KeyVaultEkmClient
+from azure.keyvault.administration._internal.client_base import DEFAULT_VERSION
+
+from devtools_testutils.aio import recorded_by_proxy_async
+
+from _async_test_case import KeyVaultEkmClientPreparer
+from _test_case import get_decorator
+from _shared.test_case_async import KeyVaultTestCase
+
+only_latest = get_decorator(api_versions=[DEFAULT_VERSION])
+
+# Note: These tests require an EKM connection to be established with an EKM Sample Proxy.
+
+
+class TestEkm(KeyVaultTestCase):
+    @pytest.mark.asyncio
+    @pytest.mark.live_test_only
+    @pytest.mark.parametrize("api_version", only_latest)
+    @KeyVaultEkmClientPreparer()
+    @recorded_by_proxy_async
+    async def test_ekm_connection(self, client: KeyVaultEkmClient, **kwargs):
+        ekm_host = kwargs.pop("ekm_host")
+        server_ca_certificate = kwargs.pop("ekm_certificate")
+        if not server_ca_certificate or not ekm_host:
+            pytest.skip(
+                "EKM CA certificate is required for live tests. Please set the EKM_PROXY_HOST and EKM_SERVER_CA_CERTIFICATE environment variables."
+            )
+
+        # Cleanup
+        try:
+            await client.delete_ekm_connection()
+        except HttpResponseError:
+            pass
+
+        # Create an EKM connection
+        ekm_connection = KeyVaultEkmConnection(
+            host=ekm_host,
+            server_ca_certificates=[base64.b64decode(server_ca_certificate)],
+            path_prefix="/api/v1",
+        )
+        created_ekm_connection = await client.create_ekm_connection(connection=ekm_connection)
+        assert created_ekm_connection is not None
+        assert created_ekm_connection.host == ekm_host
+        assert created_ekm_connection.server_ca_certificates is not None
+        assert len(created_ekm_connection.server_ca_certificates) == 1
+        assert created_ekm_connection.path_prefix == ekm_connection.path_prefix
+        assert created_ekm_connection.server_subject_common_name == ekm_connection.server_subject_common_name
+
+        # Get the EKM connection
+        retrieved_ekm_connection = await client.get_ekm_connection()
+        assert retrieved_ekm_connection is not None
+        assert retrieved_ekm_connection.host == ekm_host
+        assert retrieved_ekm_connection.server_ca_certificates is not None
+        assert len(retrieved_ekm_connection.server_ca_certificates) == 1
+        assert retrieved_ekm_connection.path_prefix == ekm_connection.path_prefix
+        assert retrieved_ekm_connection.server_subject_common_name == created_ekm_connection.server_subject_common_name
+
+        # Get the EKM certificate
+        ekm_certificate = await client.get_ekm_certificate()
+        assert ekm_certificate is not None
+        assert ekm_certificate.ca_certificates is not None
+        assert len(ekm_certificate.ca_certificates) == 1
+
+        # Check the EKM connection status
+        connection_status = await client.check_ekm_connection()
+        assert connection_status is not None
+        assert connection_status.api_version is not None
+        assert connection_status.proxy_vendor is not None
+        assert connection_status.proxy_name is not None
+        assert connection_status.ekm_vendor is not None
+        assert connection_status.ekm_product is not None
+
+        # Update the EKM connection
+        updated_ekm_connection = KeyVaultEkmConnection(
+            host=ekm_host,
+            server_ca_certificates=[base64.b64decode(server_ca_certificate)],
+            path_prefix="/api/v1",
+        )
+        result = await client.update_ekm_connection(connection=updated_ekm_connection)
+        assert result is not None
+        assert result.host == updated_ekm_connection.host
+        assert result.server_ca_certificates is not None
+        assert len(result.server_ca_certificates) == 1
+        assert result.path_prefix == updated_ekm_connection.path_prefix
+        assert result.server_subject_common_name == updated_ekm_connection.server_subject_common_name
+
+        # Delete the EKM connection
+        result = await client.delete_ekm_connection()
+        assert result is not None
+        assert result.host == updated_ekm_connection.host
+        assert result.server_ca_certificates is not None
+        assert len(result.server_ca_certificates) == 1
+        assert result.path_prefix == updated_ekm_connection.path_prefix
+        assert result.server_subject_common_name == updated_ekm_connection.server_subject_common_name
diff --git a/sdk/keyvault/azure-keyvault-administration/tsp-location.yaml b/sdk/keyvault/azure-keyvault-administration/tsp-location.yaml
index 26fe3ba24c63..658ae5523886 100644
--- a/sdk/keyvault/azure-keyvault-administration/tsp-location.yaml
+++ b/sdk/keyvault/azure-keyvault-administration/tsp-location.yaml
@@ -1,3 +1,3 @@
 directory: specification/keyvault/data-plane/Administration
-commit: f6bd06be22baf3a18504ffef0f590230850953e5
+commit: 0aea5f5666b9d631d37074a2f9c9bd8ba0dd1eff
 repo: Azure/azure-rest-api-specs