diff --git a/sdk/identity/azure-identity/CHANGELOG.md b/sdk/identity/azure-identity/CHANGELOG.md
index 6d5ea8f50ac1..f70a19de4d9a 100644
--- a/sdk/identity/azure-identity/CHANGELOG.md
+++ b/sdk/identity/azure-identity/CHANGELOG.md
@@ -14,6 +14,9 @@
 
 ### Bugs Fixed
 
+- Credential types correctly implement `azure-core`'s `TokenCredential` protocol.
+  ([#25175](https://github.com/Azure/azure-sdk-for-python/issues/25175))
+
 ### Other Changes
 
 ## 1.14.0b2 (2023-07-11)
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/application.py b/sdk/identity/azure-identity/azure/identity/_credentials/application.py
index 42d46835f228..28f64c376329 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/application.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/application.py
@@ -4,7 +4,7 @@
 # ------------------------------------
 import logging
 import os
-from typing import Any
+from typing import Any, Optional
 
 from azure.core.credentials import AccessToken
 from .chained import ChainedTokenCredential
@@ -63,7 +63,9 @@ def __init__(self, **kwargs: Any) -> None:
             ManagedIdentityCredential(client_id=managed_identity_client_id, **kwargs),
         )
 
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -71,16 +73,20 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
+        :keyword str tenant_id: optional tenant to include in the token request.
+
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
         :raises ~azure.core.exceptions.ClientAuthenticationError: authentication failed. The exception has a
             `message` attribute listing each authentication attempt and its error message.
         """
         if self._successful_credential:
-            token = self._successful_credential.get_token(*scopes, **kwargs)
+            token = self._successful_credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             _LOGGER.info(
                 "%s acquired a token from %s", self.__class__.__name__, self._successful_credential.__class__.__name__
             )
             return token
 
-        return super(AzureApplicationCredential, self).get_token(*scopes, **kwargs)
+        return super(AzureApplicationCredential, self).get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/authorization_code.py b/sdk/identity/azure-identity/azure/identity/_credentials/authorization_code.py
index ba0220353b37..1895fcf2a628 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/authorization_code.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/authorization_code.py
@@ -61,7 +61,9 @@ def close(self) -> None:
         """Close the credential's transport session."""
         self.__exit__()
 
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -73,6 +75,8 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -82,7 +86,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
           ``response`` attribute.
         """
         # pylint:disable=useless-super-delegation
-        return super(AuthorizationCodeCredential, self).get_token(*scopes, **kwargs)
+        return super(AuthorizationCodeCredential, self).get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
 
     def _acquire_token_silently(self, *scopes: str, **kwargs) -> Optional[AccessToken]:
         return self._client.get_cached_access_token(scopes, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/azd_cli.py b/sdk/identity/azure-identity/azure/identity/_credentials/azd_cli.py
index 9cf4f244fe2a..4d1d5ff9e740 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/azd_cli.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/azd_cli.py
@@ -91,7 +91,13 @@ def close(self) -> None:
         """Calling this method is unnecessary."""
 
     @log_get_token("AzureDeveloperCliCredential")
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self,
+        *scopes: str,
+        claims: Optional[str] = None,  # pylint:disable=unused-argument
+        tenant_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients. Applications calling this method directly must
@@ -100,6 +106,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: not used by this credential; any value provided will be ignored.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -117,7 +124,10 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         commandString = " --scope ".join(scopes)
         command = COMMAND_LINE.format(commandString)
         tenant = resolve_tenant(
-            default_tenant=self.tenant_id, additionally_allowed_tenants=self._additionally_allowed_tenants, **kwargs
+            default_tenant=self.tenant_id,
+            tenant_id=tenant_id,
+            additionally_allowed_tenants=self._additionally_allowed_tenants,
+            **kwargs,
         )
         if tenant:
             command += " --tenant-id " + tenant
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/azure_cli.py b/sdk/identity/azure-identity/azure/identity/_credentials/azure_cli.py
index 530f5c1cd0c1..2cdd7f1a3a32 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/azure_cli.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/azure_cli.py
@@ -69,7 +69,13 @@ def close(self) -> None:
         """Calling this method is unnecessary."""
 
     @log_get_token("AzureCliCredential")
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self,
+        *scopes: str,
+        claims: Optional[str] = None,  # pylint:disable=unused-argument
+        tenant_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients. Applications calling this method directly must
@@ -78,6 +84,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: not used by this credential; any value provided will be ignored.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -91,7 +98,10 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         resource = _scopes_to_resource(*scopes)
         command = COMMAND_LINE.format(resource)
         tenant = resolve_tenant(
-            default_tenant=self.tenant_id, additionally_allowed_tenants=self._additionally_allowed_tenants, **kwargs
+            default_tenant=self.tenant_id,
+            tenant_id=tenant_id,
+            additionally_allowed_tenants=self._additionally_allowed_tenants,
+            **kwargs,
         )
         if tenant:
             command += " --tenant " + tenant
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/azure_powershell.py b/sdk/identity/azure-identity/azure/identity/_credentials/azure_powershell.py
index d671a0ff3cc5..4007ef006c3f 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/azure_powershell.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/azure_powershell.py
@@ -6,7 +6,7 @@
 import logging
 import subprocess
 import sys
-from typing import List, Tuple, Optional, Any
+from typing import Any, List, Tuple, Optional
 
 from azure.core.credentials import AccessToken
 from azure.core.exceptions import ClientAuthenticationError
@@ -83,7 +83,13 @@ def close(self) -> None:
         """Calling this method is unnecessary."""
 
     @log_get_token("AzurePowerShellCredential")
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self,
+        *scopes: str,
+        claims: Optional[str] = None,  # pylint:disable=unused-argument
+        tenant_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients. Applications calling this method directly must
@@ -92,6 +98,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: not used by this credential; any value provided will be ignored.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -103,7 +110,10 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
           receive an access token
         """
         tenant_id = resolve_tenant(
-            default_tenant=self.tenant_id, additionally_allowed_tenants=self._additionally_allowed_tenants, **kwargs
+            default_tenant=self.tenant_id,
+            tenant_id=tenant_id,
+            additionally_allowed_tenants=self._additionally_allowed_tenants,
+            **kwargs,
         )
         command_line = get_command_line(scopes, tenant_id)
         output = run_command_line(command_line, self._process_timeout)
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/chained.py b/sdk/identity/azure-identity/azure/identity/_credentials/chained.py
index 5a6a2046629a..8c2e41844c68 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/chained.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/chained.py
@@ -69,7 +69,9 @@ def close(self) -> None:
         """Close the transport session of each credential in the chain."""
         self.__exit__()
 
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:  # pylint:disable=unused-argument
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request a token from each chained credential, in order, returning the first token received.
 
         This method is called automatically by Azure SDK clients.
@@ -77,6 +79,9 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:  # pylint:disab
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
+        :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
@@ -86,7 +91,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:  # pylint:disab
         history = []
         for credential in self.credentials:
             try:
-                token = credential.get_token(*scopes, **kwargs)
+                token = credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
                 _LOGGER.info("%s acquired a token from %s", self.__class__.__name__, credential.__class__.__name__)
                 self._successful_credential = credential
                 return token
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/default.py b/sdk/identity/azure-identity/azure/identity/_credentials/default.py
index 8e188d028f33..b58ee999ef59 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/default.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/default.py
@@ -4,7 +4,7 @@
 # ------------------------------------
 import logging
 import os
-from typing import List, TYPE_CHECKING, Any, cast
+from typing import List, TYPE_CHECKING, Any, Optional, cast
 
 from azure.core.credentials import AccessToken
 from .._constants import EnvironmentVariables
@@ -195,7 +195,9 @@ def __init__(self, **kwargs: Any) -> None:  # pylint: disable=too-many-statement
 
         super(DefaultAzureCredential, self).__init__(*credentials)
 
-    def get_token(self, *scopes: str, **kwargs) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -203,6 +205,8 @@ def get_token(self, *scopes: str, **kwargs) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -212,12 +216,12 @@ def get_token(self, *scopes: str, **kwargs) -> AccessToken:
           `message` attribute listing each authentication attempt and its error message.
         """
         if self._successful_credential:
-            token = self._successful_credential.get_token(*scopes, **kwargs)
+            token = self._successful_credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             _LOGGER.info(
                 "%s acquired a token from %s", self.__class__.__name__, self._successful_credential.__class__.__name__
             )
             return token
         within_dac.set(True)
-        token = super(DefaultAzureCredential, self).get_token(*scopes, **kwargs)
+        token = super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
         within_dac.set(False)
         return token
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/environment.py b/sdk/identity/azure-identity/azure/identity/_credentials/environment.py
index abd4675c9526..fb7e6b88083a 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/environment.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/environment.py
@@ -120,7 +120,9 @@ def close(self) -> None:
         self.__exit__()
 
     @log_get_token("EnvironmentCredential")
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -128,6 +130,8 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -142,4 +146,4 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
                 "this issue."
             )
             raise CredentialUnavailableError(message=message)
-        return self._credential.get_token(*scopes, **kwargs)
+        return self._credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/managed_identity.py b/sdk/identity/azure-identity/azure/identity/_credentials/managed_identity.py
index 52d0a00cc538..400a7f7a4487 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/managed_identity.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/managed_identity.py
@@ -108,7 +108,9 @@ def close(self) -> None:
         self.__exit__()
 
     @log_get_token("ManagedIdentityCredential")
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -117,6 +119,9 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
 
+        :keyword str claims: not used by this credential; any value provided will be ignored.
+        :keyword str tenant_id: not used by this credential; any value provided will be ignored.
+
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
         :raises ~azure.identity.CredentialUnavailableError: managed identity isn't available in the hosting environment
@@ -129,4 +134,4 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
                 "Visit https://aka.ms/azsdk/python/identity/managedidentitycredential/troubleshoot to "
                 "troubleshoot this issue."
             )
-        return self._credential.get_token(*scopes, **kwargs)
+        return self._credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/shared_cache.py b/sdk/identity/azure-identity/azure/identity/_credentials/shared_cache.py
index 3e78c9a42bac..ecf7d1fcb1ac 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/shared_cache.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/shared_cache.py
@@ -52,7 +52,9 @@ def close(self) -> None:
         self.__exit__()
 
     @log_get_token("SharedTokenCacheCredential")
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Get an access token for `scopes` from the shared cache.
 
         If no access token is cached, attempt to acquire one using a cached refresh token.
@@ -64,8 +66,10 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
         :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
             claims challenge following an authorization failure
+        :keyword str tenant_id: not used by this credential; any value provided will be ignored.
         :keyword bool enable_cae: indicates whether to enable Continuous Access Evaluation (CAE) for the requested
             token. Defaults to False.
+
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
         :raises ~azure.identity.CredentialUnavailableError: the cache is unavailable or contains insufficient user
@@ -73,7 +77,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :raises ~azure.core.exceptions.ClientAuthenticationError: authentication failed. The error's ``message``
             attribute gives a reason.
         """
-        return self._credential.get_token(*scopes, **kwargs)
+        return self._credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
 
     @staticmethod
     def supported() -> bool:
@@ -97,7 +101,9 @@ def __exit__(self, *args):
         if self._client:
             self._client.__exit__(*args)
 
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         if not scopes:
             raise ValueError("'get_token' requires at least one scope")
 
@@ -123,7 +129,9 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
 
         # try each refresh token, returning the first access token acquired
         for refresh_token in self._get_refresh_tokens(account, is_cae=is_cae):
-            token = self._client.obtain_token_by_refresh_token(scopes, refresh_token, **kwargs)
+            token = self._client.obtain_token_by_refresh_token(
+                scopes, refresh_token, claims=claims, tenant_id=tenant_id, **kwargs
+            )
             return token
 
         raise CredentialUnavailableError(message=NO_TOKEN.format(account.get("username")))
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/silent.py b/sdk/identity/azure-identity/azure/identity/_credentials/silent.py
index 04dc6a42b940..ec15d0435cfa 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/silent.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/silent.py
@@ -54,7 +54,9 @@ def __enter__(self):
     def __exit__(self, *args):
         self._client.__exit__(*args)
 
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         if not scopes:
             raise ValueError('"get_token" requires at least one scope')
 
@@ -70,7 +72,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
                     raise CredentialUnavailableError(message="Shared token cache unavailable")
                 raise ClientAuthenticationError(message="Shared token cache unavailable")
 
-        return self._acquire_token_silent(*scopes, **kwargs)
+        return self._acquire_token_silent(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
 
     def _initialize_cache(self, is_cae: bool = False) -> Optional[TokenCache]:
 
diff --git a/sdk/identity/azure-identity/azure/identity/_credentials/vscode.py b/sdk/identity/azure-identity/azure/identity/_credentials/vscode.py
index f73b7832e972..92c9e6a70905 100644
--- a/sdk/identity/azure-identity/azure/identity/_credentials/vscode.py
+++ b/sdk/identity/azure-identity/azure/identity/_credentials/vscode.py
@@ -140,7 +140,9 @@ def close(self) -> None:
         self.__exit__()
 
     @log_get_token("VSCodeCredential")
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes` as the user currently signed in to Visual Studio Code.
 
         This method is called automatically by Azure SDK clients.
@@ -148,6 +150,9 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
+        :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
@@ -163,11 +168,11 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
             raise CredentialUnavailableError(message=error_message)
         if within_dac.get():
             try:
-                token = super(VisualStudioCodeCredential, self).get_token(*scopes, **kwargs)
+                token = super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
                 return token
             except ClientAuthenticationError as ex:
                 raise CredentialUnavailableError(message=ex.message) from ex
-        return super(VisualStudioCodeCredential, self).get_token(*scopes, **kwargs)
+        return super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
 
     def _acquire_token_silently(self, *scopes: str, **kwargs: Any) -> Optional[AccessToken]:
         self._client = cast(AadClient, self._client)
diff --git a/sdk/identity/azure-identity/azure/identity/_internal/get_token_mixin.py b/sdk/identity/azure-identity/azure/identity/_internal/get_token_mixin.py
index ebfdd2effc35..a3dc76ce507b 100644
--- a/sdk/identity/azure-identity/azure/identity/_internal/get_token_mixin.py
+++ b/sdk/identity/azure-identity/azure/identity/_internal/get_token_mixin.py
@@ -53,7 +53,9 @@ def _should_refresh(self, token: AccessToken) -> bool:
             return False
         return True
 
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -61,9 +63,12 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
         :keyword bool enable_cae: indicates whether to enable Continuous Access Evaluation (CAE) for the requested
             token. Defaults to False.
+
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
         :raises CredentialUnavailableError: the credential is unable to attempt authentication because it lacks
@@ -75,14 +80,14 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
             raise ValueError('"get_token" requires at least one scope')
 
         try:
-            token = self._acquire_token_silently(*scopes, **kwargs)
+            token = self._acquire_token_silently(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             if not token:
                 self._last_request_time = int(time.time())
-                token = self._request_token(*scopes, **kwargs)
+                token = self._request_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             elif self._should_refresh(token):
                 try:
                     self._last_request_time = int(time.time())
-                    token = self._request_token(*scopes, **kwargs)
+                    token = self._request_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
                 except Exception:  # pylint:disable=broad-except
                     pass
             _LOGGER.log(
diff --git a/sdk/identity/azure-identity/azure/identity/_internal/interactive.py b/sdk/identity/azure-identity/azure/identity/_internal/interactive.py
index 868e86a4a22a..f01a85f397de 100644
--- a/sdk/identity/azure-identity/azure/identity/_internal/interactive.py
+++ b/sdk/identity/azure-identity/azure/identity/_internal/interactive.py
@@ -111,7 +111,9 @@ def __init__(
         else:
             super(InteractiveCredential, self).__init__(**kwargs)
 
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -120,7 +122,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
         :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
-          claims challenge following an authorization failure
+            claims challenge following an authorization failure
         :keyword str tenant_id: optional tenant to include in the token request.
         :keyword bool enable_cae: indicates whether to enable Continuous Access Evaluation (CAE) for the requested
             token. Defaults to False.
@@ -140,7 +142,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
 
         allow_prompt = kwargs.pop("_allow_prompt", not self._disable_automatic_authentication)
         try:
-            token = self._acquire_token_silent(*scopes, **kwargs)
+            token = self._acquire_token_silent(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             _LOGGER.info("%s.get_token succeeded", self.__class__.__name__)
             return token
         except Exception as ex:  # pylint:disable=broad-except
@@ -157,7 +159,7 @@ def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         now = int(time.time())
 
         try:
-            result = self._request_token(*scopes, **kwargs)
+            result = self._request_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             if "access_token" not in result:
                 message = "Authentication failed: {}".format(result.get("error_description") or result.get("error"))
                 response = self._client.get_error_response(result)
diff --git a/sdk/identity/azure-identity/azure/identity/_internal/managed_identity_base.py b/sdk/identity/azure-identity/azure/identity/_internal/managed_identity_base.py
index 33b8b2fd36f6..554122530f83 100644
--- a/sdk/identity/azure-identity/azure/identity/_internal/managed_identity_base.py
+++ b/sdk/identity/azure-identity/azure/identity/_internal/managed_identity_base.py
@@ -39,10 +39,12 @@ def __exit__(self, *args):
     def close(self) -> None:
         self.__exit__()
 
-    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         if not self._client:
             raise CredentialUnavailableError(message=self.get_unavailable_message())
-        return super(ManagedIdentityBase, self).get_token(*scopes, **kwargs)
+        return super(ManagedIdentityBase, self).get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
 
     def _acquire_token_silently(self, *scopes: str, **kwargs: Any) -> Optional[AccessToken]:
         # casting because mypy can't determine that these methods are called
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/application.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/application.py
index e902364586b4..3d38ec1c5482 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/application.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/application.py
@@ -62,7 +62,9 @@ def __init__(
             ManagedIdentityCredential(client_id=managed_identity_client_id, **kwargs),
         )
 
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Asynchronously request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -70,6 +72,9 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
+        :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
@@ -77,10 +82,10 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
             `message` attribute listing each authentication attempt and its error message.
         """
         if self._successful_credential:
-            token = await self._successful_credential.get_token(*scopes, **kwargs)
+            token = await self._successful_credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             _LOGGER.info(
                 "%s acquired a token from %s", self.__class__.__name__, self._successful_credential.__class__.__name__
             )
             return token
 
-        return await super().get_token(*scopes, **kwargs)
+        return await super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/authorization_code.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/authorization_code.py
index 4f7b30601f9e..ee9e394ae470 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/authorization_code.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/authorization_code.py
@@ -68,7 +68,9 @@ def __init__(
         self._redirect_uri = redirect_uri
         super().__init__()
 
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -80,6 +82,8 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -88,7 +92,7 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
           attribute gives a reason. Any error response from Azure Active Directory is available as the error's
           ``response`` attribute.
         """
-        return await super().get_token(*scopes, **kwargs)
+        return await super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
 
     async def _acquire_token_silently(self, *scopes: str, **kwargs: Any) -> Optional[AccessToken]:
         return self._client.get_cached_access_token(scopes, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/azd_cli.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/azd_cli.py
index b03ea546660e..b1c229edf617 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/azd_cli.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/azd_cli.py
@@ -79,7 +79,13 @@ def __init__(
         self._process_timeout = process_timeout
 
     @log_get_token_async
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self,
+        *scopes: str,
+        claims: Optional[str] = None,  # pylint:disable=unused-argument
+        tenant_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients. Applications calling this method directly must
@@ -88,6 +94,7 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: not used by this credential; any value provided will be ignored.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -98,7 +105,7 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         """
         # only ProactorEventLoop supports subprocesses on Windows (and it isn't the default loop on Python < 3.8)
         if sys.platform.startswith("win") and not isinstance(asyncio.get_event_loop(), asyncio.ProactorEventLoop):
-            return _SyncAzureDeveloperCliCredential().get_token(*scopes, **kwargs)
+            return _SyncAzureDeveloperCliCredential().get_token(*scopes, tenant_id=tenant_id, **kwargs)
 
         if not scopes:
             raise ValueError("Missing scope in request. \n")
@@ -106,7 +113,10 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         commandString = " --scope ".join(scopes)
         command = COMMAND_LINE.format(commandString)
         tenant = resolve_tenant(
-            default_tenant=self.tenant_id, additionally_allowed_tenants=self._additionally_allowed_tenants, **kwargs
+            default_tenant=self.tenant_id,
+            tenant_id=tenant_id,
+            additionally_allowed_tenants=self._additionally_allowed_tenants,
+            **kwargs,
         )
 
         if tenant:
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_cli.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_cli.py
index 649ada95a965..c767e8840009 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_cli.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_cli.py
@@ -6,7 +6,7 @@
 import os
 import shutil
 import sys
-from typing import List, Any, Optional
+from typing import Any, List, Optional
 
 from azure.core.exceptions import ClientAuthenticationError
 from azure.core.credentials import AccessToken
@@ -60,7 +60,13 @@ def __init__(
         self._process_timeout = process_timeout
 
     @log_get_token_async
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self,
+        *scopes: str,
+        claims: Optional[str] = None,  # pylint:disable=unused-argument
+        tenant_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients. Applications calling this method directly must
@@ -69,6 +75,7 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: not used by this credential; any value provided will be ignored.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -79,12 +86,15 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         """
         # only ProactorEventLoop supports subprocesses on Windows (and it isn't the default loop on Python < 3.8)
         if sys.platform.startswith("win") and not isinstance(asyncio.get_event_loop(), asyncio.ProactorEventLoop):
-            return _SyncAzureCliCredential().get_token(*scopes, **kwargs)
+            return _SyncAzureCliCredential().get_token(*scopes, tenant_id=tenant_id, **kwargs)
 
         resource = _scopes_to_resource(*scopes)
         command = COMMAND_LINE.format(resource)
         tenant = resolve_tenant(
-            default_tenant=self.tenant_id, additionally_allowed_tenants=self._additionally_allowed_tenants, **kwargs
+            default_tenant=self.tenant_id,
+            tenant_id=tenant_id,
+            additionally_allowed_tenants=self._additionally_allowed_tenants,
+            **kwargs,
         )
 
         if tenant:
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_powershell.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_powershell.py
index c704c9614de6..59560f26b3fb 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_powershell.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_powershell.py
@@ -4,7 +4,7 @@
 # ------------------------------------
 import asyncio
 import sys
-from typing import cast, List, Any, Optional
+from typing import Any, cast, List, Optional
 from azure.core.credentials import AccessToken
 
 from .._internal import AsyncContextManager
@@ -54,7 +54,13 @@ def __init__(
         self._process_timeout = process_timeout
 
     @log_get_token_async
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self,
+        *scopes: str,
+        claims: Optional[str] = None,  # pylint:disable=unused-argument
+        tenant_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients. Applications calling this method directly must
@@ -63,6 +69,7 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: not used by this credential; any value provided will be ignored.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -74,10 +81,13 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         """
         # only ProactorEventLoop supports subprocesses on Windows (and it isn't the default loop on Python < 3.8)
         if sys.platform.startswith("win") and not isinstance(asyncio.get_event_loop(), asyncio.ProactorEventLoop):
-            return _SyncCredential().get_token(*scopes, **kwargs)
+            return _SyncCredential().get_token(*scopes, tenant_id=tenant_id, **kwargs)
 
         tenant_id = resolve_tenant(
-            default_tenant=self.tenant_id, additionally_allowed_tenants=self._additionally_allowed_tenants, **kwargs
+            default_tenant=self.tenant_id,
+            tenant_id=tenant_id,
+            additionally_allowed_tenants=self._additionally_allowed_tenants,
+            **kwargs,
         )
         command_line = get_command_line(scopes, tenant_id)
         output = await run_command_line(command_line, self._process_timeout)
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/chained.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/chained.py
index 8547d5e134f2..9c11f66e2538 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/chained.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/chained.py
@@ -4,7 +4,7 @@
 # ------------------------------------
 import asyncio
 import logging
-from typing import Optional, TYPE_CHECKING, Any
+from typing import Any, Optional, TYPE_CHECKING
 
 from azure.core.exceptions import ClientAuthenticationError
 from azure.core.credentials import AccessToken
@@ -50,7 +50,9 @@ async def close(self) -> None:
 
         await asyncio.gather(*(credential.close() for credential in self.credentials))
 
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Asynchronously request a token from each credential, in order, returning the first token received.
 
         If no credential provides a token, raises :class:`azure.core.exceptions.ClientAuthenticationError`
@@ -61,6 +63,9 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
+        :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
@@ -70,7 +75,7 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         history = []
         for credential in self.credentials:
             try:
-                token = await credential.get_token(*scopes, **kwargs)
+                token = await credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
                 _LOGGER.info("%s acquired a token from %s", self.__class__.__name__, credential.__class__.__name__)
                 self._successful_credential = credential
                 return token
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/default.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/default.py
index bb539650d7db..a78887c2db2c 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/default.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/default.py
@@ -4,7 +4,7 @@
 # ------------------------------------
 import logging
 import os
-from typing import List, TYPE_CHECKING, Any, cast
+from typing import List, Optional, TYPE_CHECKING, Any, cast
 
 from azure.core.credentials import AccessToken
 from ..._constants import EnvironmentVariables
@@ -176,7 +176,9 @@ def __init__(self, **kwargs: Any) -> None:
 
         super().__init__(*credentials)
 
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Asynchronously request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -184,6 +186,8 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -192,8 +196,8 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
           `message` attribute listing each authentication attempt and its error message.
         """
         if self._successful_credential:
-            return await self._successful_credential.get_token(*scopes, **kwargs)
+            return await self._successful_credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
         within_dac.set(True)
-        token = await super().get_token(*scopes, **kwargs)
+        token = await super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
         within_dac.set(False)
         return token
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/environment.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/environment.py
index 750af4040325..1451e858f199 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/environment.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/environment.py
@@ -93,7 +93,9 @@ async def close(self) -> None:
             await self._credential.__aexit__()
 
     @log_get_token_async
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Asynchronously request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -101,6 +103,8 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -114,4 +118,4 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
                 "this issue."
             )
             raise CredentialUnavailableError(message=message)
-        return await self._credential.get_token(*scopes, **kwargs)
+        return await self._credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/managed_identity.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/managed_identity.py
index 1e4d7a507e5c..9bf257e4bc11 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/managed_identity.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/managed_identity.py
@@ -114,7 +114,9 @@ async def close(self) -> None:
             await self._credential.close()
 
     @log_get_token_async
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Asynchronously request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -122,6 +124,8 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: not used by this credential; any value provided will be ignored.
+        :keyword str tenant_id: not used by this credential; any value provided will be ignored.
 
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
@@ -134,4 +138,4 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
                 "Visit https://aka.ms/azsdk/python/identity/managedidentitycredential/troubleshoot to "
                 "troubleshoot this issue."
             )
-        return await self._credential.get_token(*scopes, **kwargs)
+        return await self._credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/shared_cache.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/shared_cache.py
index 04c3cd3ffaad..4284d3267c06 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/shared_cache.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/shared_cache.py
@@ -2,7 +2,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
-from typing import Any
+from typing import Any, Optional
 from azure.core.credentials import AccessToken
 from ..._internal.aad_client import AadClientBase
 from ... import CredentialUnavailableError
@@ -42,7 +42,9 @@ async def close(self) -> None:
             await self._client.__aexit__()  # type: ignore
 
     @log_get_token_async
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:  # pylint:disable=unused-argument
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Get an access token for `scopes` from the shared cache.
 
         If no access token is cached, attempt to acquire one using a cached refresh token.
@@ -52,9 +54,12 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:  # pylint
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
         :keyword bool enable_cae: indicates whether to enable Continuous Access Evaluation (CAE) for the requested
             token. Defaults to False.
+
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
         :raises ~azure.identity.CredentialUnavailableError: the cache is unavailable or contains insufficient user
@@ -88,7 +93,9 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:  # pylint
 
         # try each refresh token, returning the first access token acquired
         for refresh_token in self._get_refresh_tokens(account, is_cae=is_cae):
-            token = await self._client.obtain_token_by_refresh_token(scopes, refresh_token, **kwargs)
+            token = await self._client.obtain_token_by_refresh_token(
+                scopes, refresh_token, claims=claims, tenant_id=tenant_id, **kwargs
+            )
             return token
 
         raise CredentialUnavailableError(message=NO_TOKEN.format(account.get("username")))
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_credentials/vscode.py b/sdk/identity/azure-identity/azure/identity/aio/_credentials/vscode.py
index ddd1d0e012ec..2a3d45f1b574 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_credentials/vscode.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_credentials/vscode.py
@@ -47,7 +47,9 @@ async def close(self) -> None:
             await self._client.__aexit__()
 
     @log_get_token_async
-    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes` as the user currently signed in to Visual Studio Code.
 
         This method is called automatically by Azure SDK clients.
@@ -55,6 +57,8 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
@@ -73,11 +77,11 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
             raise CredentialUnavailableError("Initialization failed")
         if within_dac.get():
             try:
-                token = await super().get_token(*scopes, **kwargs)
+                token = await super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
                 return token
             except ClientAuthenticationError as ex:
                 raise CredentialUnavailableError(message=ex.message) from ex
-        return await super().get_token(*scopes, **kwargs)
+        return await super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
 
     async def _acquire_token_silently(self, *scopes: str, **kwargs: Any) -> Optional[AccessToken]:
         self._client = cast(AadClient, self._client)
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_internal/get_token_mixin.py b/sdk/identity/azure-identity/azure/identity/aio/_internal/get_token_mixin.py
index 6128113454ab..162e6a51da57 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_internal/get_token_mixin.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_internal/get_token_mixin.py
@@ -5,7 +5,7 @@
 import abc
 import logging
 import time
-from typing import Optional
+from typing import Any, Optional
 
 from azure.core.credentials import AccessToken
 from ..._constants import DEFAULT_REFRESH_OFFSET, DEFAULT_TOKEN_REFRESH_RETRY_DELAY
@@ -53,7 +53,9 @@ def _should_refresh(self, token: AccessToken) -> bool:
             return False
         return True
 
-    async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         """Request an access token for `scopes`.
 
         This method is called automatically by Azure SDK clients.
@@ -61,9 +63,12 @@ async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
         :param str scopes: desired scopes for the access token. This method requires at least one scope.
             For more information about scopes, see
             https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str claims: additional claims required in the token, such as those returned in a resource provider's
+            claims challenge following an authorization failure.
         :keyword str tenant_id: optional tenant to include in the token request.
         :keyword bool enable_cae: indicates whether to enable Continuous Access Evaluation (CAE) for the requested
             token. Defaults to False.
+
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
         :raises CredentialUnavailableError: the credential is unable to attempt authentication because it lacks
@@ -75,14 +80,14 @@ async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
             raise ValueError('"get_token" requires at least one scope')
 
         try:
-            token = await self._acquire_token_silently(*scopes, **kwargs)
+            token = await self._acquire_token_silently(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             if not token:
                 self._last_request_time = int(time.time())
-                token = await self._request_token(*scopes, **kwargs)
+                token = await self._request_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
             elif self._should_refresh(token):
                 try:
                     self._last_request_time = int(time.time())
-                    token = await self._request_token(*scopes, **kwargs)
+                    token = await self._request_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
                 except Exception:  # pylint:disable=broad-except
                     pass
             _LOGGER.log(
diff --git a/sdk/identity/azure-identity/azure/identity/aio/_internal/managed_identity_base.py b/sdk/identity/azure-identity/azure/identity/aio/_internal/managed_identity_base.py
index d0a81349d898..22d5119e19bd 100644
--- a/sdk/identity/azure-identity/azure/identity/aio/_internal/managed_identity_base.py
+++ b/sdk/identity/azure-identity/azure/identity/aio/_internal/managed_identity_base.py
@@ -3,7 +3,7 @@
 # Licensed under the MIT License.
 # ------------------------------------
 import abc
-from typing import cast, Optional
+from typing import Any, cast, Optional
 
 from azure.core.credentials import AccessToken
 from . import AsyncContextManager
@@ -39,10 +39,12 @@ async def __aexit__(self, *args):
     async def close(self) -> None:
         await self.__aexit__()
 
-    async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
+    async def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
+    ) -> AccessToken:
         if not self._client:
             raise CredentialUnavailableError(message=self.get_unavailable_message())
-        return await super().get_token(*scopes, **kwargs)
+        return await super().get_token(*scopes, claims=claims, tenant_id=tenant_id, **kwargs)
 
     async def _acquire_token_silently(self, *scopes: str, **kwargs) -> Optional[AccessToken]:
         # casting because mypy can't determine that these methods are called
diff --git a/sdk/identity/azure-identity/samples/custom_credentials.py b/sdk/identity/azure-identity/samples/custom_credentials.py
index 2787c2454214..6177a4dbf0a7 100644
--- a/sdk/identity/azure-identity/samples/custom_credentials.py
+++ b/sdk/identity/azure-identity/samples/custom_credentials.py
@@ -5,34 +5,31 @@
 """Demonstrates custom credential implementations using existing access tokens and an MSAL client"""
 
 import time
-from typing import TYPE_CHECKING
+from typing import Optional, Union
 
 from azure.core.credentials import AccessToken
 from azure.identity import AuthenticationRequiredError, AzureAuthorityHosts
 import msal
 
-if TYPE_CHECKING:
-    from typing import Any, Union
-
 
 class StaticTokenCredential(object):
-    """Authenticates with a previously acquired access token
+    """Authenticates with a previously-acquired access token
 
     Note that an access token is valid only for certain resources and eventually expires. This credential is therefore
     quite limited. An application using it must ensure the token is valid and contains all claims required by any
     service client given an instance of this credential.
     """
 
-    def __init__(self, access_token):
-        # type: (Union[str, AccessToken]) -> None
+    def __init__(self, access_token: Union[str, AccessToken]) -> None:
         if isinstance(access_token, AccessToken):
             self._token = access_token
         else:
             # setting expires_on in the past causes Azure SDK clients to call get_token every time they need a token
             self._token = AccessToken(token=access_token, expires_on=0)
 
-    def get_token(self, *scopes, **kwargs):
-        # type: (*str, **Any) -> AccessToken
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs
+    ) -> AccessToken:
         """get_token is the only method a credential must implement"""
 
         return self._token
@@ -41,18 +38,18 @@ def get_token(self, *scopes, **kwargs):
 class MsalTokenCredential(object):
     """Uses an MSAL client directly to obtain access tokens with an interactive flow."""
 
-    def __init__(self, tenant_id, client_id):
-        # type: (str, str) -> None
+    def __init__(self, tenant_id: str, client_id: str) -> None:
         self._app = msal.PublicClientApplication(
             client_id=client_id, authority="https://{}/{}".format(AzureAuthorityHosts.AZURE_PUBLIC_CLOUD, tenant_id)
         )
 
-    def get_token(self, *scopes, **kwargs):
-        # type: (*str, **Any) -> AccessToken
+    def get_token(
+        self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs
+    ) -> AccessToken:
         """get_token is the only method a credential must implement"""
 
         now = int(time.time())
-        result = self._app.acquire_token_interactive(list(scopes), **kwargs)
+        result = self._app.acquire_token_interactive(list(scopes), claims=claims, tenant_id=tenant_id, **kwargs)
 
         try:
             return AccessToken(result["access_token"], now + int(result["expires_in"]))
diff --git a/sdk/identity/azure-identity/tests/test_application_credential.py b/sdk/identity/azure-identity/tests/test_application_credential.py
index 49235c7fd576..689bc044e095 100644
--- a/sdk/identity/azure-identity/tests/test_application_credential.py
+++ b/sdk/identity/azure-identity/tests/test_application_credential.py
@@ -22,7 +22,10 @@
 def test_get_token():
     expected_token = "***"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant_id = parsed.path.split("/")[1]
         if "/oauth2/v2.0/token" in request.url:
diff --git a/sdk/identity/azure-identity/tests/test_application_credential_async.py b/sdk/identity/azure-identity/tests/test_application_credential_async.py
index ce4b3c8e192c..f3cfab729897 100644
--- a/sdk/identity/azure-identity/tests/test_application_credential_async.py
+++ b/sdk/identity/azure-identity/tests/test_application_credential_async.py
@@ -79,7 +79,10 @@ def test_initialization(mock_credential, expect_argument):
 async def test_get_token():
     expected_token = "***"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token=expected_token))
 
     with patch.dict("os.environ", {var: "..." for var in EnvironmentVariables.CLIENT_SECRET_VARS}, clear=True):
diff --git a/sdk/identity/azure-identity/tests/test_auth_code.py b/sdk/identity/azure-identity/tests/test_auth_code.py
index 6ffcbec0b0a0..e3ba13bf2971 100644
--- a/sdk/identity/azure-identity/tests/test_auth_code.py
+++ b/sdk/identity/azure-identity/tests/test_auth_code.py
@@ -30,7 +30,10 @@ def test_no_scopes():
 def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock())
 
-    def send(*_, **__):
+    def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     credential = AuthorizationCodeCredential(
@@ -142,7 +145,10 @@ def test_multitenant_authentication():
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         assert tenant in (first_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
@@ -175,7 +181,10 @@ def test_multitenant_authentication_not_allowed():
     expected_tenant = "expected-tenant"
     expected_token = "***"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         token = expected_token if tenant == expected_tenant else expected_token * 2
diff --git a/sdk/identity/azure-identity/tests/test_auth_code_async.py b/sdk/identity/azure-identity/tests/test_auth_code_async.py
index baf786b9a9b5..8bd8946d4205 100644
--- a/sdk/identity/azure-identity/tests/test_auth_code_async.py
+++ b/sdk/identity/azure-identity/tests/test_auth_code_async.py
@@ -30,7 +30,10 @@ async def test_no_scopes():
 async def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock())
 
-    async def send(*_, **__):
+    async def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     credential = AuthorizationCodeCredential(
@@ -166,7 +169,10 @@ async def test_multitenant_authentication():
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         assert tenant in (first_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
@@ -199,7 +205,10 @@ async def test_multitenant_authentication_not_allowed():
     expected_tenant = "expected-tenant"
     expected_token = "***"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         token = expected_token if tenant == expected_tenant else expected_token * 2
diff --git a/sdk/identity/azure-identity/tests/test_certificate_credential.py b/sdk/identity/azure-identity/tests/test_certificate_credential.py
index 9f9674196234..d75fe489d06d 100644
--- a/sdk/identity/azure-identity/tests/test_certificate_credential.py
+++ b/sdk/identity/azure-identity/tests/test_certificate_credential.py
@@ -299,7 +299,10 @@ def test_token_cache_persistent(cert_path, cert_password):
 
     access_token = "foo token"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         if "/oauth2/v2.0/token" not in parsed.path:
@@ -337,7 +340,10 @@ def test_token_cache_memory(cert_path, cert_password):
     """The credential should default to in-memory cache if no persistence options are provided."""
     access_token = "foo token"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         if "/oauth2/v2.0/token" not in parsed.path:
@@ -427,7 +433,10 @@ def test_multitenant_authentication(cert_path, cert_password):
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         assert tenant in (first_tenant, second_tenant, "common"), 'unexpected tenant "{}"'.format(tenant)
@@ -464,7 +473,10 @@ def test_multitenant_authentication_backcompat(cert_path, cert_password):
     expected_tenant = "expected-tenant"
     expected_token = "***"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         if "/oauth2/v2.0/token" not in parsed.path:
             return get_discovery_response("https://{}/{}".format(parsed.netloc, expected_tenant))
diff --git a/sdk/identity/azure-identity/tests/test_certificate_credential_async.py b/sdk/identity/azure-identity/tests/test_certificate_credential_async.py
index c0a5d6dd5863..e93990d92cfe 100644
--- a/sdk/identity/azure-identity/tests/test_certificate_credential_async.py
+++ b/sdk/identity/azure-identity/tests/test_certificate_credential_async.py
@@ -76,7 +76,10 @@ async def test_context_manager():
 async def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock())
 
-    async def send(*_, **__):
+    async def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     credential = CertificateCredential(
@@ -351,7 +354,10 @@ async def test_multitenant_authentication(cert_path, cert_password):
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         assert tenant in (first_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
@@ -386,7 +392,10 @@ async def test_multitenant_authentication_backcompat(cert_path, cert_password):
     expected_tenant = "expected-tenant"
     expected_token = "***"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         token = expected_token if tenant == expected_tenant else expected_token * 2
diff --git a/sdk/identity/azure-identity/tests/test_chained_credential.py b/sdk/identity/azure-identity/tests/test_chained_credential.py
index db6da5dc80c7..2d74ef9538e3 100644
--- a/sdk/identity/azure-identity/tests/test_chained_credential.py
+++ b/sdk/identity/azure-identity/tests/test_chained_credential.py
@@ -102,7 +102,7 @@ def test_raises_for_unexpected_error():
 
 def test_returns_first_token():
     expected_token = Mock()
-    first_credential = Mock(get_token=lambda _: expected_token)
+    first_credential = Mock(get_token=lambda _, **__: expected_token)
     second_credential = Mock(get_token=Mock())
 
     aggregate = ChainedTokenCredential(first_credential, second_credential)
diff --git a/sdk/identity/azure-identity/tests/test_chained_token_credential_async.py b/sdk/identity/azure-identity/tests/test_chained_token_credential_async.py
index 1b751be946e1..df04deebb1b5 100644
--- a/sdk/identity/azure-identity/tests/test_chained_token_credential_async.py
+++ b/sdk/identity/azure-identity/tests/test_chained_token_credential_async.py
@@ -56,14 +56,14 @@ async def test_credential_chain_error_message():
 
 @pytest.mark.asyncio
 async def test_chain_attempts_all_credentials():
-    async def credential_unavailable(message="it didn't work"):
+    async def credential_unavailable(message="it didn't work", **_):
         raise CredentialUnavailableError(message)
 
     expected_token = AccessToken("expected_token", 0)
     credentials = [
         Mock(get_token=Mock(wraps=credential_unavailable)),
         Mock(get_token=Mock(wraps=credential_unavailable)),
-        Mock(get_token=wrap_in_future(lambda _: expected_token)),
+        Mock(get_token=wrap_in_future(lambda _, **__: expected_token)),
     ]
 
     token = await ChainedTokenCredential(*credentials).get_token("scope")
@@ -77,7 +77,7 @@ async def credential_unavailable(message="it didn't work"):
 async def test_chain_raises_for_unexpected_error():
     """the chain should not continue after an unexpected error (i.e. anything but CredentialUnavailableError)"""
 
-    async def credential_unavailable(message="it didn't work"):
+    async def credential_unavailable(message="it didn't work", **_):
         raise CredentialUnavailableError(message)
 
     expected_message = "it can't be done"
@@ -85,7 +85,7 @@ async def credential_unavailable(message="it didn't work"):
     credentials = [
         Mock(get_token=Mock(wraps=credential_unavailable)),
         Mock(get_token=Mock(side_effect=ValueError(expected_message))),
-        Mock(get_token=Mock(wraps=wrap_in_future(lambda _: AccessToken("**", 42)))),
+        Mock(get_token=Mock(wraps=wrap_in_future(lambda _, **__: AccessToken("**", 42)))),
     ]
 
     with pytest.raises(ClientAuthenticationError) as ex:
@@ -98,7 +98,7 @@ async def credential_unavailable(message="it didn't work"):
 @pytest.mark.asyncio
 async def test_returns_first_token():
     expected_token = Mock()
-    first_credential = Mock(get_token=wrap_in_future(lambda _: expected_token))
+    first_credential = Mock(get_token=wrap_in_future(lambda _, **__: expected_token))
     second_credential = Mock(get_token=Mock())
 
     aggregate = ChainedTokenCredential(first_credential, second_credential)
diff --git a/sdk/identity/azure-identity/tests/test_client_secret_credential.py b/sdk/identity/azure-identity/tests/test_client_secret_credential.py
index 75074f5f1cab..5a2c3023767b 100644
--- a/sdk/identity/azure-identity/tests/test_client_secret_credential.py
+++ b/sdk/identity/azure-identity/tests/test_client_secret_credential.py
@@ -153,7 +153,10 @@ def test_token_cache_persistent():
 
     access_token = "foo token"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         if "/oauth2/v2.0/token" not in parsed.path:
@@ -188,7 +191,10 @@ def test_token_cache_memory():
     """The credential should default to in-memory cache if no persistence options are provided."""
     access_token = "foo token"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         if "/oauth2/v2.0/token" not in parsed.path:
@@ -264,7 +270,9 @@ def test_multitenant_authentication():
     second_token = first_token * 2
 
     def send(request, **kwargs):
-        assert "tenant_id" not in kwargs, "tenant_id kwarg shouldn't get passed to send method"
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
 
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
@@ -310,7 +318,10 @@ def test_multitenant_authentication_not_allowed():
     expected_tenant = "expected-tenant"
     expected_token = "***"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         if "/oauth2/v2.0/token" not in parsed.path:
             return get_discovery_response("https://{}/{}".format(parsed.netloc, expected_tenant))
diff --git a/sdk/identity/azure-identity/tests/test_client_secret_credential_async.py b/sdk/identity/azure-identity/tests/test_client_secret_credential_async.py
index 68c6ba161a46..d84105807014 100644
--- a/sdk/identity/azure-identity/tests/test_client_secret_credential_async.py
+++ b/sdk/identity/azure-identity/tests/test_client_secret_credential_async.py
@@ -67,7 +67,10 @@ async def test_context_manager():
 async def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock())
 
-    async def send(*_, **__):
+    async def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     credential = ClientSecretCredential(
@@ -312,7 +315,9 @@ async def test_multitenant_authentication():
     second_token = first_token * 2
 
     async def send(request, **kwargs):
-        assert "tenant_id" not in kwargs, "tenant_id kwarg shouldn't get passed to send method"
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
 
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
@@ -360,7 +365,10 @@ async def test_multitenant_authentication_not_allowed():
     expected_tenant = "expected-tenant"
     expected_token = "***"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         token = expected_token if tenant == expected_tenant else expected_token * 2
diff --git a/sdk/identity/azure-identity/tests/test_get_token_mixin.py b/sdk/identity/azure-identity/tests/test_get_token_mixin.py
index 62931fdc082a..0deee7ec8a9d 100644
--- a/sdk/identity/azure-identity/tests/test_get_token_mixin.py
+++ b/sdk/identity/azure-identity/tests/test_get_token_mixin.py
@@ -40,8 +40,8 @@ def test_no_cached_token():
     credential = MockCredential()
     token = credential.get_token(SCOPE)
 
-    credential.acquire_token_silently.assert_called_once_with(SCOPE)
-    credential.request_token.assert_called_once_with(SCOPE)
+    credential.acquire_token_silently.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
+    credential.request_token.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
     assert token.token == MockCredential.NEW_TOKEN.token
 
 
@@ -61,7 +61,7 @@ def test_token_acquisition_failure():
         with pytest.raises(Exception):
             credential.get_token(SCOPE)
         assert credential.request_token.call_count == i + 1
-        credential.request_token.assert_called_with(SCOPE)
+        credential.request_token.assert_called_with(SCOPE, claims=None, tenant_id=None)
 
 
 def test_expired_token():
@@ -71,8 +71,8 @@ def test_expired_token():
     credential = MockCredential(cached_token=AccessToken(CACHED_TOKEN, now - 1))
     token = credential.get_token(SCOPE)
 
-    credential.acquire_token_silently.assert_called_once_with(SCOPE)
-    credential.request_token.assert_called_once_with(SCOPE)
+    credential.acquire_token_silently.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
+    credential.request_token.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
     assert token.token == MockCredential.NEW_TOKEN.token
 
 
@@ -82,7 +82,7 @@ def test_cached_token_outside_refresh_window():
     credential = MockCredential(cached_token=AccessToken(CACHED_TOKEN, time.time() + DEFAULT_REFRESH_OFFSET + 1))
     token = credential.get_token(SCOPE)
 
-    credential.acquire_token_silently.assert_called_once_with(SCOPE)
+    credential.acquire_token_silently.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
     assert credential.request_token.call_count == 0
     assert token.token == CACHED_TOKEN
 
@@ -93,8 +93,8 @@ def test_cached_token_within_refresh_window():
     credential = MockCredential(cached_token=AccessToken(CACHED_TOKEN, time.time() + DEFAULT_REFRESH_OFFSET - 1))
     token = credential.get_token(SCOPE)
 
-    credential.acquire_token_silently.assert_called_once_with(SCOPE)
-    credential.request_token.assert_called_once_with(SCOPE)
+    credential.acquire_token_silently.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
+    credential.request_token.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
     assert token.token == MockCredential.NEW_TOKEN.token
 
 
@@ -109,5 +109,5 @@ def test_retry_delay():
     for i in range(4):
         token = credential.get_token(SCOPE)
         assert token.token == CACHED_TOKEN
-        credential.acquire_token_silently.assert_called_with(SCOPE)
-        credential.request_token.assert_called_once_with(SCOPE)
+        credential.acquire_token_silently.assert_called_with(SCOPE, claims=None, tenant_id=None)
+        credential.request_token.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
diff --git a/sdk/identity/azure-identity/tests/test_get_token_mixin_async.py b/sdk/identity/azure-identity/tests/test_get_token_mixin_async.py
index 4fcc3098d1e1..1b84608dfb30 100644
--- a/sdk/identity/azure-identity/tests/test_get_token_mixin_async.py
+++ b/sdk/identity/azure-identity/tests/test_get_token_mixin_async.py
@@ -43,8 +43,8 @@ async def test_no_cached_token():
     credential = MockCredential()
     token = await credential.get_token(SCOPE)
 
-    credential.acquire_token_silently.assert_called_once_with(SCOPE)
-    credential.request_token.assert_called_once_with(SCOPE)
+    credential.acquire_token_silently.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
+    credential.request_token.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
     assert token.token == MockCredential.NEW_TOKEN.token
 
 
@@ -64,7 +64,7 @@ async def test_token_acquisition_failure():
         with pytest.raises(Exception):
             await credential.get_token(SCOPE)
         assert credential.request_token.call_count == i + 1
-        credential.request_token.assert_called_with(SCOPE)
+        credential.request_token.assert_called_with(SCOPE, claims=None, tenant_id=None)
 
 
 async def test_expired_token():
@@ -74,8 +74,8 @@ async def test_expired_token():
     credential = MockCredential(cached_token=AccessToken(CACHED_TOKEN, now - 1))
     token = await credential.get_token(SCOPE)
 
-    credential.acquire_token_silently.assert_called_once_with(SCOPE)
-    credential.request_token.assert_called_once_with(SCOPE)
+    credential.acquire_token_silently.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
+    credential.request_token.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
     assert token.token == MockCredential.NEW_TOKEN.token
 
 
@@ -85,7 +85,7 @@ async def test_cached_token_outside_refresh_window():
     credential = MockCredential(cached_token=AccessToken(CACHED_TOKEN, time.time() + DEFAULT_REFRESH_OFFSET + 1))
     token = await credential.get_token(SCOPE)
 
-    credential.acquire_token_silently.assert_called_once_with(SCOPE)
+    credential.acquire_token_silently.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
     assert credential.request_token.call_count == 0
     assert token.token == CACHED_TOKEN
 
@@ -96,8 +96,8 @@ async def test_cached_token_within_refresh_window():
     credential = MockCredential(cached_token=AccessToken(CACHED_TOKEN, time.time() + DEFAULT_REFRESH_OFFSET - 1))
     token = await credential.get_token(SCOPE)
 
-    credential.acquire_token_silently.assert_called_once_with(SCOPE)
-    credential.request_token.assert_called_once_with(SCOPE)
+    credential.acquire_token_silently.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
+    credential.request_token.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
     assert token.token == MockCredential.NEW_TOKEN.token
 
 
@@ -112,5 +112,5 @@ async def test_retry_delay():
     for i in range(4):
         token = await credential.get_token(SCOPE)
         assert token.token == CACHED_TOKEN
-        credential.acquire_token_silently.assert_called_with(SCOPE)
-        credential.request_token.assert_called_once_with(SCOPE)
+        credential.acquire_token_silently.assert_called_with(SCOPE, claims=None, tenant_id=None)
+        credential.request_token.assert_called_once_with(SCOPE, claims=None, tenant_id=None)
diff --git a/sdk/identity/azure-identity/tests/test_imds_credential.py b/sdk/identity/azure-identity/tests/test_imds_credential.py
index 7eec40eb9bae..2ecf0cc0bf8b 100644
--- a/sdk/identity/azure-identity/tests/test_imds_credential.py
+++ b/sdk/identity/azure-identity/tests/test_imds_credential.py
@@ -62,7 +62,10 @@ def test_unexpected_error():
 
     for code in range(401, 600):
 
-        def send(request, **_):
+        def send(request, **kwargs):
+            # ensure the `claims` and `tenant_id` kwargs from credential's `get_token` method don't make it to transport
+            assert "claims" not in kwargs
+            assert "tenant_id" not in kwargs
             if "resource" not in request.query:
                 # availability probe
                 return mock_response(status_code=400, json_payload={})
diff --git a/sdk/identity/azure-identity/tests/test_imds_credential_async.py b/sdk/identity/azure-identity/tests/test_imds_credential_async.py
index d8f13e62cfb9..fb631121da18 100644
--- a/sdk/identity/azure-identity/tests/test_imds_credential_async.py
+++ b/sdk/identity/azure-identity/tests/test_imds_credential_async.py
@@ -92,7 +92,10 @@ async def test_unexpected_error():
 
     for code in range(401, 600):
 
-        async def send(request, **_):
+        async def send(request, **kwargs):
+            # ensure the `claims` and `tenant_id` kwargs from credential's `get_token` method don't make it to transport
+            assert "claims" not in kwargs
+            assert "tenant_id" not in kwargs
             if "resource" not in request.query:
                 # availability probe
                 return mock_response(status_code=400, json_payload={})
diff --git a/sdk/identity/azure-identity/tests/test_interactive_credential.py b/sdk/identity/azure-identity/tests/test_interactive_credential.py
index 22c65e68eb05..30c6cd012505 100644
--- a/sdk/identity/azure-identity/tests/test_interactive_credential.py
+++ b/sdk/identity/azure-identity/tests/test_interactive_credential.py
@@ -157,7 +157,7 @@ def validate_scopes(*scopes, **_):
 def test_authenticate_default_scopes(authority, expected_scope):
     """when given no scopes, authenticate should default to the ARM scope appropriate for the configured authority"""
 
-    def validate_scopes(*scopes):
+    def validate_scopes(*scopes, **_):
         assert scopes == (expected_scope,)
         return REQUEST_TOKEN_RESULT
 
@@ -328,7 +328,10 @@ def request_token(*args, **kwargs):
             ),
         )
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert "/oauth2/v2.0/token" not in request.url, 'mock "request_token" should prevent sending a token request'
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
@@ -371,7 +374,10 @@ def request_token(*_, **__):
             ),
         )
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert "/oauth2/v2.0/token" not in request.url, 'mock "request_token" should prevent sending a token request'
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
diff --git a/sdk/identity/azure-identity/tests/test_managed_identity.py b/sdk/identity/azure-identity/tests/test_managed_identity.py
index 3c1a373a81e9..8f8aec2da44e 100644
--- a/sdk/identity/azure-identity/tests/test_managed_identity.py
+++ b/sdk/identity/azure-identity/tests/test_managed_identity.py
@@ -487,7 +487,10 @@ def test_app_service_2019_08_01():
     new_secret = "new-expected-secret"
     scope = "scope"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(new_endpoint)
         assert request.method == "GET"
         assert request.headers["X-IDENTITY-HEADER"] == new_secret
@@ -531,7 +534,10 @@ def test_app_service_2019_08_01_tenant_id():
     new_secret = "new-expected-secret"
     scope = "scope"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(new_endpoint)
         assert request.method == "GET"
         assert request.headers["X-IDENTITY-HEADER"] == new_secret
@@ -707,7 +713,10 @@ def test_client_id_none():
     expected_access_token = "****"
     scope = "scope"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert "client_id" not in request.query
         if request.data:
             assert "client_id" not in request.body  # Cloud Shell
@@ -779,7 +788,10 @@ def test_service_fabric():
     thumbprint = "SHA1HEX"
     scope = "scope"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(endpoint)
         assert request.method == "GET"
         assert request.headers["Secret"] == secret
@@ -816,7 +828,10 @@ def test_service_fabric_tenant_id():
     thumbprint = "SHA1HEX"
     scope = "scope"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(endpoint)
         assert request.method == "GET"
         assert request.headers["Secret"] == secret
diff --git a/sdk/identity/azure-identity/tests/test_managed_identity_async.py b/sdk/identity/azure-identity/tests/test_managed_identity_async.py
index c36050fe71c3..ff91e3816edb 100644
--- a/sdk/identity/azure-identity/tests/test_managed_identity_async.py
+++ b/sdk/identity/azure-identity/tests/test_managed_identity_async.py
@@ -451,7 +451,10 @@ async def test_app_service_2019_08_01():
     new_secret = "new-expected-secret"
     scope = "scope"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(new_endpoint)
         assert request.method == "GET"
         assert request.headers["X-IDENTITY-HEADER"] == new_secret
@@ -494,7 +497,10 @@ async def test_app_service_2019_08_01_tenant_id():
     new_secret = "new-expected-secret"
     scope = "scope"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(new_endpoint)
         assert request.method == "GET"
         assert request.headers["X-IDENTITY-HEADER"] == new_secret
@@ -592,7 +598,10 @@ async def test_client_id_none():
     expected_access_token = "****"
     scope = "scope"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert "client_id" not in request.query  # IMDS
         if request.data:
             assert "client_id" not in request.body  # Cloud Shell
@@ -742,7 +751,10 @@ async def test_service_fabric():
     thumbprint = "SHA1HEX"
     scope = "scope"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(endpoint)
         assert request.method == "GET"
         assert request.headers["Secret"] == secret
@@ -780,7 +792,10 @@ async def test_service_fabric_tenant_id():
     thumbprint = "SHA1HEX"
     scope = "scope"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(endpoint)
         assert request.method == "GET"
         assert request.headers["Secret"] == secret
diff --git a/sdk/identity/azure-identity/tests/test_obo.py b/sdk/identity/azure-identity/tests/test_obo.py
index 8256f103914e..4761dc40a50a 100644
--- a/sdk/identity/azure-identity/tests/test_obo.py
+++ b/sdk/identity/azure-identity/tests/test_obo.py
@@ -100,7 +100,10 @@ def test_multitenant_authentication():
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.headers["User-Agent"].startswith(USER_AGENT)
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
@@ -193,7 +196,10 @@ def test_no_scopes():
 def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock(), on_exception=lambda _: False)
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         if "/oauth2/v2.0/token" not in parsed.path:
diff --git a/sdk/identity/azure-identity/tests/test_obo_async.py b/sdk/identity/azure-identity/tests/test_obo_async.py
index adf4a3846d61..8c143d4b72e2 100644
--- a/sdk/identity/azure-identity/tests/test_obo_async.py
+++ b/sdk/identity/azure-identity/tests/test_obo_async.py
@@ -128,7 +128,10 @@ async def test_multitenant_authentication():
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.headers["User-Agent"].startswith(USER_AGENT)
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
@@ -174,7 +177,10 @@ async def test_authority(authority):
     expected_authority = "https://{}/{}".format(expected_netloc, tenant_id)
     expected_token = "***"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         assert request.url.startswith(expected_authority)
         return mock_response(json_payload=build_aad_response(access_token=expected_token))
 
@@ -203,7 +209,10 @@ async def send(request, **_):
 async def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock(), on_exception=lambda _: False)
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         if "/oauth2/v2.0/token" not in parsed.path:
@@ -235,7 +244,10 @@ async def test_refresh_token():
     refresh_token = "refresh-token"
     requests = 0
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         nonlocal requests
         assert requests < 3, "unexpected request"
         requests += 1
diff --git a/sdk/identity/azure-identity/tests/test_shared_cache_credential.py b/sdk/identity/azure-identity/tests/test_shared_cache_credential.py
index 281aecf48757..78b02142724c 100644
--- a/sdk/identity/azure-identity/tests/test_shared_cache_credential.py
+++ b/sdk/identity/azure-identity/tests/test_shared_cache_credential.py
@@ -103,7 +103,10 @@ def test_no_scopes():
 def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock())
 
-    def send(*_, **__):
+    def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     credential = SharedTokenCacheCredential(
@@ -585,7 +588,10 @@ def test_authority_environment_variable():
 def test_authentication_record_empty_cache():
     record = AuthenticationRecord("tenant-id", "client_id", "authority", "home_account_id", "username")
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         # expecting only MSAL discovery requests
         assert request.method == "GET"
         return get_discovery_response()
@@ -607,7 +613,10 @@ def test_authentication_record_no_match():
     username = "me"
     record = AuthenticationRecord(tenant_id, client_id, authority, home_account_id, username)
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         # expecting only MSAL discovery requests
         assert request.method == "GET"
         return get_discovery_response()
@@ -811,7 +820,10 @@ def mock_send(request, **_):
 def test_client_capabilities():
     """the credential should configure MSAL for capability CP1 only if enable_cae is passed."""
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         # expecting only the discovery requests triggered by creating an msal.PublicClientApplication
         # because the cache is empty--the credential shouldn't send a token request
         return get_discovery_response("https://localhost/tenant")
@@ -836,7 +848,10 @@ def send(request, **_):
 
 
 def test_within_dac_error():
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         # expecting only the discovery requests triggered by creating an msal.PublicClientApplication
         # because the cache is empty--the credential shouldn't send a token request
         return get_discovery_response("https://localhost/tenant")
@@ -880,7 +895,10 @@ def test_multitenant_authentication():
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant_id = parsed.path.split("/")[1]
         assert tenant_id in (default_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant_id)
@@ -925,7 +943,10 @@ def test_multitenant_authentication_auth_record():
     home_account_id = object_id + "." + default_tenant
     record = AuthenticationRecord(default_tenant, "client-id", authority, home_account_id, "user")
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant_id = parsed.path.split("/")[1]
         if "/oauth2/v2.0/token" not in request.url:
@@ -1000,7 +1021,10 @@ def test_multitenant_authentication_not_allowed():
     default_tenant = "organizations"
     expected_token = "***"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant_id = parsed.path.split("/")[1]
         assert tenant_id == default_tenant
diff --git a/sdk/identity/azure-identity/tests/test_shared_cache_credential_async.py b/sdk/identity/azure-identity/tests/test_shared_cache_credential_async.py
index df013eac426c..1982b98ccf7e 100644
--- a/sdk/identity/azure-identity/tests/test_shared_cache_credential_async.py
+++ b/sdk/identity/azure-identity/tests/test_shared_cache_credential_async.py
@@ -42,7 +42,10 @@ async def test_no_scopes():
 
 @pytest.mark.asyncio
 async def test_close():
-    async def send(*_, **__):
+    async def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     transport = AsyncMockTransport(send=send)
@@ -60,7 +63,10 @@ async def send(*_, **__):
 
 @pytest.mark.asyncio
 async def test_context_manager():
-    async def send(*_, **__):
+    async def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     transport = AsyncMockTransport(send=send)
@@ -102,7 +108,10 @@ async def test_context_manager_no_cache():
 async def test_policies_configurable():
     policy = Mock(spec_set=SansIOHTTPPolicy, on_request=Mock())
 
-    async def send(*_, **__):
+    async def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     credential = SharedTokenCacheCredential(
@@ -643,7 +652,10 @@ async def test_multitenant_authentication():
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant_id = parsed.path.split("/")[1]
         return mock_response(
@@ -681,7 +693,10 @@ async def test_multitenant_authentication_not_allowed():
     default_tenant = "organizations"
     expected_token = "***"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant_id = parsed.path.split("/")[1]
         assert tenant_id == default_tenant
diff --git a/sdk/identity/azure-identity/tests/test_vscode_credential.py b/sdk/identity/azure-identity/tests/test_vscode_credential.py
index 324223e1bae9..3cc495122afd 100644
--- a/sdk/identity/azure-identity/tests/test_vscode_credential.py
+++ b/sdk/identity/azure-identity/tests/test_vscode_credential.py
@@ -87,7 +87,10 @@ def test_no_scopes():
 def test_policies_configurable():
     policy = mock.Mock(spec_set=SansIOHTTPPolicy, on_request=mock.Mock())
 
-    def send(*_, **__):
+    def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     credential = get_credential(policies=[policy], transport=mock.Mock(send=send))
@@ -157,7 +160,9 @@ def test_redeem_token():
         credential = get_credential(_client=mock_client)
         token = credential.get_token("scope")
         assert token is expected_token
-        mock_client.obtain_token_by_refresh_token.assert_called_with(("scope",), expected_value)
+        mock_client.obtain_token_by_refresh_token.assert_called_with(
+            ("scope",), expected_value, claims=None, tenant_id=None
+        )
         assert mock_client.obtain_token_by_refresh_token.call_count == 1
 
 
@@ -283,7 +288,10 @@ def test_multitenant_authentication():
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         assert tenant in (first_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
@@ -313,7 +321,10 @@ def test_multitenant_authentication_not_allowed():
     expected_tenant = "expected-tenant"
     expected_token = "***"
 
-    def send(request, **_):
+    def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         token = expected_token if tenant == expected_tenant else expected_token * 2
diff --git a/sdk/identity/azure-identity/tests/test_vscode_credential_async.py b/sdk/identity/azure-identity/tests/test_vscode_credential_async.py
index ec4bbcdacfc7..c6bb7c8eef20 100644
--- a/sdk/identity/azure-identity/tests/test_vscode_credential_async.py
+++ b/sdk/identity/azure-identity/tests/test_vscode_credential_async.py
@@ -85,7 +85,10 @@ async def test_no_scopes():
 async def test_policies_configurable():
     policy = mock.Mock(spec_set=SansIOHTTPPolicy, on_request=mock.Mock())
 
-    async def send(*_, **__):
+    async def send(*_, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         return mock_response(json_payload=build_aad_response(access_token="**"))
 
     credential = get_credential(policies=[policy], transport=mock.Mock(send=send))
@@ -160,7 +163,7 @@ async def test_redeem_token():
         credential = get_credential(_client=mock_client)
         token = await credential.get_token("scope")
         assert token is expected_token
-        token_by_refresh_token.assert_called_with(("scope",), expected_value)
+        token_by_refresh_token.assert_called_with(("scope",), expected_value, claims=None, tenant_id=None)
 
 
 @pytest.mark.asyncio
@@ -274,7 +277,10 @@ async def test_multitenant_authentication():
     second_tenant = "second-tenant"
     second_token = first_token * 2
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         assert tenant in (first_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
@@ -305,7 +311,10 @@ async def test_multitenant_authentication_not_allowed():
     expected_tenant = "expected-tenant"
     expected_token = "***"
 
-    async def send(request, **_):
+    async def send(request, **kwargs):
+        # ensure the `claims` and `tenant_id` keywords from credential's `get_token` method don't make it to transport
+        assert "claims" not in kwargs
+        assert "tenant_id" not in kwargs
         parsed = urlparse(request.url)
         tenant = parsed.path.split("/")[1]
         token = expected_token if tenant == expected_tenant else expected_token * 2