Authorization AsyncTokenCredential-returning variants

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.
 
 ---
 

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:

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))

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:

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.

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,26 +395,48 @@ 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.
 
         :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: 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