diff --git a/changelog.md b/changelog.md
index e20fdd34..efc828db 100644
--- a/changelog.md
+++ b/changelog.md
@@ -6,6 +6,7 @@
 ## New Models & APIs
 
 - **MSAL Token Credential**: Added `MsalTokenCredential`, an Azure Core-compatible asynchronous token credential backed by MSAL, for authenticating Azure SDK clients that accept an `AsyncTokenCredential`.
+- **Authorization Token Credentials**: Added `AsyncTokenCredential` access through `AccessTokenProviderBase` and `Authorization`, including user-token refresh and exchange flows.
 
 ---
 
diff --git a/libraries/microsoft-agents-authentication-entra-auth-sidecar/microsoft_agents/authentication/entra_auth_sidecar/sidecar_auth.py b/libraries/microsoft-agents-authentication-entra-auth-sidecar/microsoft_agents/authentication/entra_auth_sidecar/sidecar_auth.py
index 89e13c0c..827381e8 100644
--- a/libraries/microsoft-agents-authentication-entra-auth-sidecar/microsoft_agents/authentication/entra_auth_sidecar/sidecar_auth.py
+++ b/libraries/microsoft-agents-authentication-entra-auth-sidecar/microsoft_agents/authentication/entra_auth_sidecar/sidecar_auth.py
@@ -12,6 +12,8 @@
     AgentAuthConfiguration,
 )
 
+from azure.core.credentials_async import AsyncTokenCredential
+
 from ._models import SidecarConnectionSettings, SidecarRequestOptions
 from ._token_expiry import SidecarTokenExpiry
 from .errors.error_resources import SidecarAuthErrorResources as _Errors
@@ -111,6 +113,21 @@ async def get_access_token(
         )
         return await self._get_cached_token(self._service_name, options)
 
+    def get_token_credential(self) -> AsyncTokenCredential:
+        """Gets the token credential for the access token provider.
+
+        :return: The token credential.
+        :rtype: AsyncTokenCredential
+        """
+        from .sidecar_token_credential import SidecarTokenCredential
+
+        return SidecarTokenCredential(self._configuration, provider=self)
+
+    async def acquire_token_on_behalf_of(
+        self, scopes: list[str], user_assertion: str
+    ) -> str:
+        raise NotImplementedError("acquire_token_on_behalf_of is not implemented.")
+
     async def get_agentic_application_token(
         self, tenant_id: str, agent_app_instance_id: str
     ) -> str | None:
diff --git a/libraries/microsoft-agents-authentication-entra-auth-sidecar/microsoft_agents/authentication/entra_auth_sidecar/sidecar_token_credential.py b/libraries/microsoft-agents-authentication-entra-auth-sidecar/microsoft_agents/authentication/entra_auth_sidecar/sidecar_token_credential.py
new file mode 100644
index 00000000..88d8fdb0
--- /dev/null
+++ b/libraries/microsoft-agents-authentication-entra-auth-sidecar/microsoft_agents/authentication/entra_auth_sidecar/sidecar_token_credential.py
@@ -0,0 +1,66 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import logging
+
+from azure.core.credentials import AccessToken
+from azure.core.credentials_async import AsyncTokenCredential
+
+from microsoft_agents.hosting.core import AgentAuthConfiguration
+
+from .sidecar_auth import SidecarAuth
+from ._token_expiry import SidecarTokenExpiry
+
+logger = logging.getLogger(__name__)
+
+
+def _get_resource(scope: str) -> str:
+    """Extracts the resource by removing a trailing '/.default' from the scope.
+
+    :param scope: The scope string.
+    :return: The extracted resource string.
+    :rtype: str
+    """
+    return scope.removesuffix("/.default")
+
+
+class SidecarTokenCredential(AsyncTokenCredential):
+    """Provides an asynchronous Azure Core token credential using the Sidecar."""
+
+    def __init__(
+        self,
+        config: AgentAuthConfiguration,
+        *,
+        provider: SidecarAuth | None = None,
+    ):
+        """Initializes the SidecarTokenCredential with the given configuration.
+
+        :param config: The agent authentication configuration.
+        :type config: :class:`microsoft_agents.hosting.core.AgentAuthConfiguration`
+        """
+        self._config = config
+        self._provider: SidecarAuth | None = provider
+
+    async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
+        """Acquire an access token for the specified scopes.
+
+        :param scopes: The scopes for which the access token is requested.
+        :param kwargs: Additional keyword arguments.
+
+        :return: The acquired access token.
+        :rtype: AccessToken
+        """
+
+        logger.debug("get_token scope=%s", scopes)
+
+        if not scopes:
+            raise ValueError("At least one scope must be provided.")
+
+        if not self._provider:
+            self._provider = SidecarAuth(self._config)
+
+        resource = _get_resource(scopes[0])
+
+        token = await self._provider.get_access_token(resource, list(scopes))
+
+        return AccessToken(token=token, expires_on=SidecarTokenExpiry.resolve(token))
diff --git a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py
index 74edf7c1..17ad6d9e 100644
--- a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py
+++ b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py
@@ -17,7 +17,10 @@
     SystemAssignedManagedIdentity,
     TokenCache,
 )
+
 from azure.core.credentials import AccessToken
+from azure.core.credentials_async import AsyncTokenCredential
+
 from requests import Session
 
 from microsoft_agents.activity._utils import _DeferredString
@@ -86,6 +89,16 @@ async def get_access_token(
         access_token = await self._get_access_token(resource_url, scopes, force_refresh)
         return access_token.token
 
+    def get_token_credential(self) -> AsyncTokenCredential:
+        """Gets the token credential for the access token provider.
+
+        :return: The token credential.
+        :rtype: AsyncTokenCredential
+        """
+        from .msal_token_credential import MsalTokenCredential
+
+        return MsalTokenCredential(self._msal_configuration, provider=self)
+
     async def _get_access_token(
         self, resource_url: str, scopes: list[str], force_refresh: bool = False
     ) -> AccessToken:
diff --git a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_token_credential.py b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_token_credential.py
index 933eebd0..7cd4b586 100644
--- a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_token_credential.py
+++ b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_token_credential.py
@@ -26,14 +26,19 @@ def _get_resource(scope: str) -> str:
 class MsalTokenCredential(AsyncTokenCredential):
     """Provides an asynchronous Azure Core token credential using MSAL."""
 
-    def __init__(self, config: AgentAuthConfiguration):
+    def __init__(
+        self,
+        config: AgentAuthConfiguration,
+        *,
+        provider: MsalAuth | None = None,
+    ):
         """Initializes the MsalTokenCredential with the given configuration.
 
         :param config: The agent authentication configuration.
         :type config: :class:`microsoft_agents.hosting.core.AgentAuthConfiguration`
         """
         self._config = config
-        self._provider: MsalAuth | None = None
+        self._provider: MsalAuth | None = provider
 
     async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
         """Acquire an access token for the specified scopes.
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/oauth/authorization.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/oauth/authorization.py
index bd40288b..a8702fdb 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/oauth/authorization.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/oauth/authorization.py
@@ -7,6 +7,8 @@
 from typing import Optional, Callable, Awaitable, cast
 from dataclasses import dataclass
 
+from azure.core.credentials_async import AsyncTokenCredential
+
 from microsoft_agents.activity import (
     Activity,
     Channels,
@@ -16,6 +18,10 @@
 )
 from microsoft_agents.activity.activity_types import ActivityTypes
 
+from microsoft_agents.hosting.core.authorization._helpers import (
+    _CallableTokenCredential,
+)
+
 from ...turn_context import TurnContext
 from ...storage import Storage
 from ...authorization import Connections
@@ -60,7 +66,7 @@ def __init__(
         self,
         storage: Storage,
         connection_manager: Connections,
-        auth_handlers: Optional[dict[str, AuthHandler]] = None,
+        auth_handlers: dict[str, AuthHandler] | None = None,
         auto_sign_in: bool = False,
         use_cache: bool = False,
         **kwargs,
@@ -86,10 +92,10 @@ def __init__(
         self._connection_manager = connection_manager
 
         self._sign_in_success_handler: Optional[
-            Callable[[TurnContext, TurnState, Optional[str]], Awaitable[None]]
+            Callable[[TurnContext, TurnState, str | None], Awaitable[None]]
         ] = None
         self._sign_in_failure_handler: Optional[
-            Callable[[TurnContext, TurnState, Optional[str]], Awaitable[None]]
+            Callable[[TurnContext, TurnState, str | None], Awaitable[None]]
         ] = None
 
         self._handlers = {}
@@ -165,13 +171,13 @@ def _sign_in_state_key(context: TurnContext) -> str:
         """
         return f"auth:_SignInState:{context.activity.channel_id}:{context.activity.from_property.id}"
 
-    async def _load_sign_in_state(self, context: TurnContext) -> Optional[_SignInState]:
+    async def _load_sign_in_state(self, context: TurnContext) -> _SignInState | None:
         """Load the sign-in state from storage for the given context.
 
         :param context: The turn context for the current turn of conversation.
         :type context: :class:`microsoft_agents.hosting.core.turn_context.TurnContext`
         :return: The sign-in state if found, None otherwise.
-        :rtype: Optional[:class:`microsoft_agents.hosting.core.app.oauth._sign_in_state._SignInState`]
+        :rtype: :class:`microsoft_agents.hosting.core.app.oauth._sign_in_state._SignInState` | None
         """
         key = self._sign_in_state_key(context)
         return (await self._storage.read([key], target_cls=_SignInState)).get(key)
@@ -205,9 +211,9 @@ def _cache_key(context: TurnContext, handler_id: str) -> str:
     @staticmethod
     def _get_cached_token(
         context: TurnContext, handler_id: str
-    ) -> Optional[TokenResponse]:
+    ) -> TokenResponse | None:
         key = Authorization._cache_key(context, handler_id)
-        return cast(Optional[TokenResponse], context.turn_state.get(key))
+        return cast(TokenResponse | None, context.turn_state.get(key))
 
     @staticmethod
     def _cache_token(
@@ -241,7 +247,7 @@ async def _start_or_continue_sign_in(
         self,
         context: TurnContext,
         state: TurnState,
-        auth_handler_id: Optional[str] = None,
+        auth_handler_id: str | None = None,
     ) -> _SignInResponse:
         """Start or continue the sign-in process for the user with the given auth handler.
 
@@ -303,14 +309,14 @@ async def _start_or_continue_sign_in(
         return sign_in_response
 
     async def sign_out(
-        self, context: TurnContext, auth_handler_id: Optional[str] = None
+        self, context: TurnContext, auth_handler_id: str | None = None
     ) -> None:
         """Attempts to sign out the user from a specified auth handler or the default handler.
 
         :param context: The turn context for the current turn of conversation.
         :type context: :class:`microsoft_agents.hosting.core.turn_context.TurnContext`
         :param auth_handler_id: The ID of the auth handler to sign out from. If None, sign out from all handlers.
-        :type auth_handler_id: Optional[str]
+        :type auth_handler_id: str | None
         :return: None
         """
         auth_handler_id = auth_handler_id or self._default_handler_id
@@ -372,7 +378,9 @@ async def _on_turn_auth_intercept(
         )
 
     async def get_token(
-        self, context: TurnContext, auth_handler_id: Optional[str] = None
+        self,
+        context: TurnContext,
+        auth_handler_id: str | None = None,
     ) -> TokenResponse:
         """Gets the token for a specific auth handler or the default handler.
 
@@ -387,12 +395,34 @@ async def get_token(
         """
         return await self.exchange_token(context, auth_handler_id=auth_handler_id)
 
+    def get_token_as_token_credential(
+        self,
+        context: TurnContext,
+        auth_handler_id: str | None = None,
+    ) -> AsyncTokenCredential:
+        """Gets the token as an AsyncTokenCredential for a specific auth handler or the default handler.
+
+        :param context: The context object for the current turn.
+        :type context: :class:`microsoft_agents.hosting.core.turn_context.TurnContext`
+        :param auth_handler_id: The ID of the auth handler to get the token for.
+        :type auth_handler_id: str | None
+        :return: An AsyncTokenCredential for the specified auth handler or the default handler.
+        :rtype: :class:`azure.core.credentials_async.AsyncTokenCredential`
+        """
+
+        async def func(*scopes: str, **_kwargs) -> TokenResponse:
+            return await self.exchange_token(
+                context, auth_handler_id=auth_handler_id, scopes=list(scopes)
+            )
+
+        return _CallableTokenCredential(func)
+
     async def exchange_token(
         self,
         context: TurnContext,
-        scopes: Optional[list[str]] = None,
-        auth_handler_id: Optional[str] = None,
-        exchange_connection: Optional[str] = None,
+        scopes: list[str] | None = None,
+        auth_handler_id: str | None = None,
+        exchange_connection: str | None = None,
     ) -> TokenResponse:
         """Exchanges or refreshes the token for a specific auth handler or the default handler.
 
@@ -400,13 +430,13 @@ async def exchange_token(
         :type context: :class:`microsoft_agents.hosting.core.turn_context.TurnContext`
         :param scopes: The scopes to request during the token exchange or refresh. Defaults
             to the list given in the AuthHandler configuration if None.
-        :type scopes: Optional[list[str]]
+        :type scopes: list[str] | None
         :param auth_handler_id: The ID of the auth handler to exchange or refresh the token for.
             If None, the default handler will be used.
-        :type auth_handler_id: Optional[str]
+        :type auth_handler_id: str | None
         :param exchange_connection: The name of the connection to use for token exchange. If None,
             the connection defined in the AuthHandler configuration will be used.
-        :type exchange_connection: Optional[str]
+        :type exchange_connection: str | None
         :return: The token response from the OAuth provider.
         :rtype: :class:`microsoft_agents.activity.TokenResponse`
         :raises ValueError: If the specified auth handler ID is not recognized or not configured.
@@ -441,26 +471,63 @@ async def exchange_token(
                 return res
         return TokenResponse()
 
+    def exchange_token_as_token_credential(
+        self,
+        context: TurnContext,
+        scopes: list[str] | None = None,
+        auth_handler_id: str | None = None,
+        exchange_connection: str | None = None,
+    ) -> AsyncTokenCredential:
+        """Gets a token credential that exchanges or refreshes the token for a specific auth handler or the default handler.
+
+        :param context: The context object for the current turn.
+        :type context: :class:`microsoft_agents.hosting.core.turn_context.TurnContext`
+        :param scopes: The scopes to request during the token exchange or refresh. Defaults
+            to the list given in the AuthHandler configuration if None.
+        :type scopes: list[str] | None
+        :param auth_handler_id: The ID of the auth handler to exchange or refresh the token for.
+            If None, the default handler will be used.
+        :type auth_handler_id: str | None
+        :param exchange_connection: The name of the connection to use for token exchange. If None,
+            the connection defined in the AuthHandler configuration will be used.
+        :type exchange_connection: str | None
+        :return: An instance of `AsyncTokenCredential`.
+        :rtype: :class:`azure.core.credentials_async.AsyncTokenCredential`
+        :raises ValueError: If the specified auth handler ID is not recognized or not configured.
+        """
+
+        async def func(*new_scopes: str, **_kwargs) -> TokenResponse:
+            prev_scopes: list[str] = scopes or []
+            all_scopes = list(dict.fromkeys([*prev_scopes, *new_scopes]))
+            return await self.exchange_token(
+                context,
+                scopes=all_scopes,
+                auth_handler_id=auth_handler_id,
+                exchange_connection=exchange_connection,
+            )
+
+        return _CallableTokenCredential(func)
+
     def on_sign_in_success(
         self,
-        handler: Callable[[TurnContext, TurnState, Optional[str]], Awaitable[None]],
+        handler: Callable[[TurnContext, TurnState, str | None], Awaitable[None]],
     ) -> None:
         """
         Sets a handler to be called when sign-in is successfully completed.
 
         :param handler: The handler function to call on successful sign-in.
-        :type handler: Callable[[:class:`microsoft_agents.hosting.core.turn_context.TurnContext`, :class:`microsoft_agents.hosting.core.app.state.turn_state.TurnState`, Optional[str]], Awaitable[None]]
+        :type handler: Callable[[:class:`microsoft_agents.hosting.core.turn_context.TurnContext`, :class:`microsoft_agents.hosting.core.app.state.turn_state.TurnState`, str | None], Awaitable[None]]
         """
         self._sign_in_success_handler = handler
 
     def on_sign_in_failure(
         self,
-        handler: Callable[[TurnContext, TurnState, Optional[str]], Awaitable[None]],
+        handler: Callable[[TurnContext, TurnState, str | None], Awaitable[None]],
     ) -> None:
         """
         Sets a handler to be called when sign-in fails.
 
         :param handler: The handler function to call on sign-in failure.
-        :type handler: Callable[[:class:`microsoft_agents.hosting.core.turn_context.TurnContext`, :class:`microsoft_agents.hosting.core.app.state.turn_state.TurnState`, Optional[str]], Awaitable[None]]
+        :type handler: Callable[[:class:`microsoft_agents.hosting.core.turn_context.TurnContext`, :class:`microsoft_agents.hosting.core.app.state.turn_state.TurnState`, str | None], Awaitable[None]]
         """
         self._sign_in_failure_handler = handler
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_helpers.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_helpers.py
new file mode 100644
index 00000000..bb3d0095
--- /dev/null
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_helpers.py
@@ -0,0 +1,65 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import datetime
+from typing import Protocol
+
+from azure.core.credentials_async import AsyncTokenCredential
+from azure.core.credentials import AccessToken
+
+from microsoft_agents.activity import TokenResponse
+
+
+class _TokenRetrieverProtocol(Protocol):
+    async def __call__(
+        self, *scopes: str, **kwargs
+    ) -> TokenResponse | AccessToken | None: ...
+
+
+def _access_token_from_token_response(token_response: TokenResponse) -> AccessToken:
+    """Convert a `TokenResponse` to an `AccessToken`.
+
+    :param token_response: An instance of `TokenResponse` to convert.
+    :return: An instance of `AccessToken`.
+    """
+    if not token_response:
+        raise ValueError("Failed to retrieve token")
+
+    expires_on: int = 0
+    if token_response.expiration:
+        exp = token_response.expiration
+        if exp.endswith("Z"):
+            exp = exp[:-1] + "+00:00"
+        dt = datetime.datetime.fromisoformat(exp)
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=datetime.timezone.utc)
+        else:
+            dt = dt.astimezone(datetime.timezone.utc)
+        expires_on = int(dt.timestamp())
+
+    return AccessToken(token=token_response.token, expires_on=expires_on)
+
+
+class _CallableTokenCredential(AsyncTokenCredential):
+    """A wrapper class that implements `AsyncTokenCredential` using a callable to retrieve tokens."""
+
+    def __init__(self, get_token_callable: _TokenRetrieverProtocol):
+        """Initialize the `_CallableTokenCredential` with a token retriever callable.
+
+        :param get_token_callable: A callable that retrieves tokens.
+        """
+        self._get_token_callable = get_token_callable
+
+    async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
+        """Get an access token using the provided callable.
+
+        :param scopes: The scopes for which the access token is requested.
+        :param kwargs: Additional keyword arguments to pass to the token retriever callable.
+        :return: An instance of `AccessToken`.
+        """
+        res = await self._get_token_callable(*scopes, **kwargs)
+        if not res:
+            raise ValueError("Failed to retrieve token")
+        if isinstance(res, AccessToken):
+            return res
+        return _access_token_from_token_response(res)
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/access_token_provider_base.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/access_token_provider_base.py
index 5cd29e29..250c9c99 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/access_token_provider_base.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/access_token_provider_base.py
@@ -4,6 +4,8 @@
 from typing import Protocol
 from abc import abstractmethod
 
+from azure.core.credentials_async import AsyncTokenCredential
+
 from .agent_auth_configuration import AgentAuthConfiguration
 
 
@@ -33,6 +35,15 @@ async def get_access_token(
         """
         pass
 
+    @abstractmethod
+    def get_token_credential(self) -> AsyncTokenCredential:
+        """
+        Get the token credential for the access token provider.
+
+        :return: The token credential.
+        """
+        pass
+
     async def acquire_token_on_behalf_of(
         self, scopes: list[str], user_assertion: str
     ) -> str:
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/anonymous_token_provider.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/anonymous_token_provider.py
index 801e8a9a..8be0fb2c 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/anonymous_token_provider.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/anonymous_token_provider.py
@@ -3,8 +3,14 @@
 
 from typing import Optional
 
+from azure.core.credentials import AccessToken
+from azure.core.credentials_async import AsyncTokenCredential
+
 from .access_token_provider_base import AccessTokenProviderBase
 from .agent_auth_configuration import AgentAuthConfiguration
+from ._helpers import _CallableTokenCredential
+
+_ANONYMOUS_TOKEN_EXPIRATION = 2**31 - 1
 
 
 class AnonymousTokenProvider(AccessTokenProviderBase):
@@ -26,6 +32,12 @@ async def get_access_token(
     ) -> str:
         return ""
 
+    def get_token_credential(self) -> AsyncTokenCredential:
+        async def get_token(*scopes: str, **kwargs) -> AccessToken:
+            return AccessToken("", _ANONYMOUS_TOKEN_EXPIRATION)
+
+        return _CallableTokenCredential(get_token)
+
     async def acquire_token_on_behalf_of(
         self, scopes: list[str], user_assertion: str
     ) -> str:
diff --git a/libraries/microsoft-agents-hosting-core/setup.py b/libraries/microsoft-agents-hosting-core/setup.py
index 6703d855..56079d3e 100644
--- a/libraries/microsoft-agents-hosting-core/setup.py
+++ b/libraries/microsoft-agents-hosting-core/setup.py
@@ -20,5 +20,6 @@
         "opentelemetry-sdk>=1.27.0",
         "aiohttp>=3.11.11",
         "yarl>=1.17.0,<2.0",
+        "azure-core",
     ],
 )
diff --git a/tests/_common/testing_objects/testing_token_provider.py b/tests/_common/testing_objects/testing_token_provider.py
index 9e3f1c1d..c3dcca12 100644
--- a/tests/_common/testing_objects/testing_token_provider.py
+++ b/tests/_common/testing_objects/testing_token_provider.py
@@ -2,6 +2,14 @@
     AccessTokenProviderBase,
     AgentAuthConfiguration,
 )
+from azure.core.credentials import AccessToken
+from azure.core.credentials_async import AsyncTokenCredential
+
+from microsoft_agents.hosting.core.authorization._helpers import (
+    _CallableTokenCredential,
+)
+
+_TEST_TOKEN_EXPIRATION = 2**31 - 1
 
 
 class TestingTokenProvider(AccessTokenProviderBase):
@@ -46,6 +54,12 @@ async def get_access_token(
         """
         return f"{self.name}-token"
 
+    def get_token_credential(self) -> AsyncTokenCredential:
+        async def get_token(*scopes: str, **kwargs) -> AccessToken:
+            return AccessToken(f"{self.name}-token", _TEST_TOKEN_EXPIRATION)
+
+        return _CallableTokenCredential(get_token)
+
     async def acquire_token_on_behalf_of(
         self, scopes: list[str], user_assertion: str
     ) -> str:
diff --git a/tests/hosting_core/app/_oauth/test_authorization.py b/tests/hosting_core/app/_oauth/test_authorization.py
index 13238e8f..50f603da 100644
--- a/tests/hosting_core/app/_oauth/test_authorization.py
+++ b/tests/hosting_core/app/_oauth/test_authorization.py
@@ -560,6 +560,37 @@ async def test_get_token(
         assert sign_in_state_eq(initial_state, final_state)
         assert context.turn_state == expected_turn_state
 
+    @pytest.mark.asyncio
+    async def test_get_token_as_token_credential(self, mocker, authorization, context):
+        token_response = TokenResponse(
+            token=DEFAULTS.token,
+            expiration="2030-01-01T00:00:00Z",
+        )
+        exchange_token = mocker.patch.object(
+            authorization,
+            "exchange_token",
+            new=mocker.AsyncMock(return_value=token_response),
+        )
+
+        credential = authorization.get_token_as_token_credential(
+            context,
+            auth_handler_id=DEFAULTS.auth_handler_id,
+        )
+        token = await credential.get_token(
+            "scope1",
+            "scope2",
+            claims="claims",
+            tenant_id="tenant",
+            enable_cae=True,
+        )
+
+        assert token.token == DEFAULTS.token
+        exchange_token.assert_awaited_once_with(
+            context,
+            auth_handler_id=DEFAULTS.auth_handler_id,
+            scopes=["scope1", "scope2"],
+        )
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         "initial_state, initial_cache, handler_id, refreshed, refresh_token",
@@ -636,6 +667,40 @@ async def test_exchange_token(
         assert sign_in_state_eq(initial_state, final_state)
         assert context.turn_state == expected_turn_state
 
+    @pytest.mark.asyncio
+    async def test_exchange_token_as_token_credential(
+        self, mocker, authorization, context
+    ):
+        token_response = TokenResponse(
+            token=DEFAULTS.token,
+            expiration="2030-01-01T00:00:00Z",
+        )
+        exchange_token = mocker.patch.object(
+            authorization,
+            "exchange_token",
+            new=mocker.AsyncMock(return_value=token_response),
+        )
+
+        credential = authorization.exchange_token_as_token_credential(
+            context,
+            scopes=["configured", "shared"],
+            auth_handler_id=DEFAULTS.auth_handler_id,
+            exchange_connection="connection",
+        )
+        token = await credential.get_token(
+            "shared",
+            "requested",
+            claims="claims",
+        )
+
+        assert token.token == DEFAULTS.token
+        exchange_token.assert_awaited_once_with(
+            context,
+            scopes=["configured", "shared", "requested"],
+            auth_handler_id=DEFAULTS.auth_handler_id,
+            exchange_connection="connection",
+        )
+
     @pytest.mark.asyncio
     async def test_on_turn_auth_intercept_no_intercept(
         self, storage, authorization, context
diff --git a/tests/hosting_core/authorization/test_anonymous_token_provider.py b/tests/hosting_core/authorization/test_anonymous_token_provider.py
new file mode 100644
index 00000000..53e602c6
--- /dev/null
+++ b/tests/hosting_core/authorization/test_anonymous_token_provider.py
@@ -0,0 +1,17 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import pytest
+from azure.core.credentials import AccessToken
+
+from microsoft_agents.hosting.core import AnonymousTokenProvider
+
+
+@pytest.mark.asyncio
+async def test_get_token_credential_is_synchronous():
+    provider = AnonymousTokenProvider()
+
+    credential = provider.get_token_credential()
+    token = await credential.get_token("scope")
+
+    assert token == AccessToken("", 2**31 - 1)
diff --git a/tests/hosting_core/authorization/test_helpers.py b/tests/hosting_core/authorization/test_helpers.py
new file mode 100644
index 00000000..5387dbaf
--- /dev/null
+++ b/tests/hosting_core/authorization/test_helpers.py
@@ -0,0 +1,82 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import pytest
+from azure.core.credentials import AccessToken
+
+from microsoft_agents.activity import TokenResponse
+from microsoft_agents.hosting.core.authorization._helpers import (
+    _CallableTokenCredential,
+    _access_token_from_token_response,
+)
+
+
+@pytest.mark.parametrize(
+    "expiration, expected",
+    [
+        (None, 0),
+        ("2030-01-01T00:00:00Z", 1893456000),
+        ("2030-01-01T00:00:00", 1893456000),
+        ("2030-01-01T02:00:00+02:00", 1893456000),
+    ],
+)
+def test_access_token_from_token_response_expiration(expiration, expected):
+    response = TokenResponse(
+        token="token",
+        **({"expiration": expiration} if expiration is not None else {}),
+    )
+
+    token = _access_token_from_token_response(response)
+
+    assert token == AccessToken("token", expected)
+
+
+def test_access_token_from_token_response_rejects_missing_token():
+    with pytest.raises(ValueError, match="Failed to retrieve token"):
+        _access_token_from_token_response(TokenResponse())
+
+
+def test_access_token_from_token_response_rejects_invalid_expiration():
+    with pytest.raises(ValueError):
+        _access_token_from_token_response(
+            TokenResponse(token="token", expiration="not-a-date")
+        )
+
+
+@pytest.mark.asyncio
+async def test_callable_token_credential_returns_access_token_unchanged(mocker):
+    expected = AccessToken("token", 123)
+    retriever = mocker.AsyncMock(return_value=expected)
+    credential = _CallableTokenCredential(retriever)
+
+    token = await credential.get_token("scope", claims="claims")
+
+    assert token is expected
+    retriever.assert_awaited_once_with("scope", claims="claims")
+
+
+@pytest.mark.asyncio
+async def test_callable_token_credential_converts_token_response():
+    async def retrieve_token(*scopes: str, **kwargs):
+        return TokenResponse(
+            token="token",
+            expiration="2030-01-01T02:00:00+02:00",
+        )
+
+    credential = _CallableTokenCredential(retrieve_token)
+
+    token = await credential.get_token("scope")
+
+    assert token == AccessToken("token", 1893456000)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("result", [None, TokenResponse()])
+async def test_callable_token_credential_rejects_missing_token(result):
+    async def retrieve_token(*scopes: str, **kwargs):
+        return result
+
+    credential = _CallableTokenCredential(retrieve_token)
+
+    with pytest.raises(ValueError, match="Failed to retrieve token"):
+        await credential.get_token("scope")
diff --git a/tests/hosting_core/test_connection_manager.py b/tests/hosting_core/test_connection_manager.py
index 9fb41f15..12d9455f 100644
--- a/tests/hosting_core/test_connection_manager.py
+++ b/tests/hosting_core/test_connection_manager.py
@@ -2,6 +2,8 @@
 # Licensed under the MIT License.
 
 import pytest
+from azure.core.credentials import AccessToken
+from azure.core.credentials_async import AsyncTokenCredential
 
 from microsoft_agents.activity import Activity, ChannelAccount, RoleTypes
 from microsoft_agents.hosting.core import (
@@ -10,6 +12,11 @@
     ClaimsIdentity,
     ConnectionManager,
 )
+from microsoft_agents.hosting.core.authorization._helpers import (
+    _CallableTokenCredential,
+)
+
+_TEST_TOKEN_EXPIRATION = 2**31 - 1
 
 
 class FakeProvider(AccessTokenProviderBase):
@@ -25,6 +32,22 @@ def configuration(self) -> AgentAuthConfiguration:
     async def get_access_token(self, resource_url, scopes, force_refresh=False):
         return "fake-token"
 
+    def get_token_credential(self) -> AsyncTokenCredential:
+        async def get_token(*scopes: str, **kwargs) -> AccessToken:
+            return AccessToken("fake-token", _TEST_TOKEN_EXPIRATION)
+
+        return _CallableTokenCredential(get_token)
+
+
+@pytest.mark.asyncio
+async def test_fake_provider_get_token_credential_is_synchronous():
+    provider = FakeProvider(AgentAuthConfiguration())
+
+    credential = provider.get_token_credential()
+    token = await credential.get_token("scope")
+
+    assert token == AccessToken("fake-token", _TEST_TOKEN_EXPIRATION)
+
 
 ENV_CONFIG = {
     "CONNECTIONS": {