Add agentic auth provider infrastructure

examples/agent365/src/main.py

@@ -7,7 +7,7 @@
 import os
 
 from dotenv import load_dotenv
-from microsoft_teams.api import ClientCredentials
+from microsoft_teams.api import AgenticIdentity, ClientCredentials
 from microsoft_teams.apps.token_manager import AGENT_BOT_API_SCOPE, TokenManager
 
 
@@ -26,8 +26,7 @@ async def main():
     blueprint_client_id = get_required_env("AGENT365_BLUEPRINT_CLIENT_ID")
     blueprint_client_secret = get_required_env("AGENT365_BLUEPRINT_CLIENT_SECRET")
     agentic_app_id = get_required_env("AGENT365_AGENTIC_APP_ID")
-    agentic_user_id = os.getenv("AGENT365_AGENTIC_USER_ID")
-    agentic_user_upn = os.getenv("AGENT365_AGENTIC_USER_UPN")
+    agentic_user_id = get_required_env("AGENT365_AGENTIC_USER_ID")
     scope = os.getenv("AGENT365_SCOPE", AGENT_BOT_API_SCOPE)
 
     credentials = ClientCredentials(
@@ -38,11 +37,8 @@ async def main():
     token_manager = TokenManager(credentials=credentials)
 
     token = await token_manager.get_agentic_token(
-        tenant_id,
-        agentic_app_id,
         scope,
-        agentic_user_id=agentic_user_id,
-        agentic_user_upn=agentic_user_upn,
+        AgenticIdentity(agentic_app_id=agentic_app_id, agentic_user_id=agentic_user_id, tenant_id=tenant_id),
     )
 
     print(f"Acquired agent user token for {scope}")

packages/api/src/microsoft_teams/api/auth/__init__.py

@@ -10,11 +10,13 @@
 )
 from .cloud_environment import from_name as config_from_cloud_name
 from .credentials import (
+    BasicTokenProvider,
     ClientCredentials,
     Credentials,
     FederatedIdentityCredentials,
     ManagedIdentityCredentials,
     TokenCredentials,
+    TokenProvider,
 )
 from .json_web_token import JsonWebToken, JsonWebTokenPayload
 from .token import TokenProtocol
@@ -23,12 +25,14 @@
     "CallerIds",
     "CallerType",
     "CloudEnvironment",
+    "BasicTokenProvider",
     "ClientCredentials",
     "config_from_cloud_name",
     "Credentials",
     "FederatedIdentityCredentials",
     "ManagedIdentityCredentials",
     "TokenCredentials",
+    "TokenProvider",
     "TokenProtocol",
     "JsonWebToken",
     "JsonWebTokenPayload",

packages/api/src/microsoft_teams/api/auth/cloud_environment.py

@@ -21,6 +21,8 @@ class CloudEnvironment:
     """The default multi-tenant login tenant (e.g. "botframework.com")."""
     bot_scope: str
     """The Bot Framework OAuth scope (e.g. "https://api.botframework.com/.default")."""
+    agentic_bot_scope: str
+    """The Teams Bot API scope for Agent ID user-token calls."""
     token_service_url: str
     """The Bot Framework token service base URL (e.g. "https://token.botframework.com")."""
     openid_metadata_url: str
@@ -35,6 +37,7 @@ class CloudEnvironment:
     login_endpoint="https://login.microsoftonline.com",
     login_tenant="botframework.com",
     bot_scope="https://api.botframework.com/.default",
+    agentic_bot_scope="https://botapi.skype.com/.default",
     token_service_url="https://token.botframework.com",
     openid_metadata_url="https://login.botframework.com/v1/.well-known/openidconfiguration",
     token_issuer="https://api.botframework.com",
@@ -46,6 +49,8 @@ class CloudEnvironment:
     login_endpoint="https://login.microsoftonline.us",
     login_tenant="MicrosoftServices.onmicrosoft.us",
     bot_scope="https://api.botframework.us/.default",
+    # TODO: confirm Agent ID Bot API scope for this cloud before enabling sovereign Agent ID scenarios.
+    agentic_bot_scope="https://botapi.skype.com/.default",
     token_service_url="https://tokengcch.botframework.azure.us",
     openid_metadata_url="https://login.botframework.azure.us/v1/.well-known/openidconfiguration",
     token_issuer="https://api.botframework.us",
@@ -57,6 +62,8 @@ class CloudEnvironment:
     login_endpoint="https://login.microsoftonline.us",
     login_tenant="MicrosoftServices.onmicrosoft.us",
     bot_scope="https://api.botframework.us/.default",
+    # TODO: confirm Agent ID Bot API scope for this cloud before enabling sovereign Agent ID scenarios.
+    agentic_bot_scope="https://botapi.skype.com/.default",
     token_service_url="https://apiDoD.botframework.azure.us",
     openid_metadata_url="https://login.botframework.azure.us/v1/.well-known/openidconfiguration",
     token_issuer="https://api.botframework.us",
@@ -68,6 +75,8 @@ class CloudEnvironment:
     login_endpoint="https://login.partner.microsoftonline.cn",
     login_tenant="microsoftservices.partner.onmschina.cn",
     bot_scope="https://api.botframework.azure.cn/.default",
+    # TODO: confirm Agent ID Bot API scope for this cloud before enabling China cloud Agent ID scenarios.
+    agentic_bot_scope="https://botapi.skype.com/.default",
     token_service_url="https://token.botframework.azure.cn",
     openid_metadata_url="https://login.botframework.azure.cn/v1/.well-known/openidconfiguration",
     token_issuer="https://api.botframework.azure.cn",

packages/api/src/microsoft_teams/api/auth/credentials.py

@@ -3,9 +3,34 @@
 Licensed under the MIT License.
 """
 
-from typing import Awaitable, Callable, Literal, Optional, Union
+from typing import Awaitable, Callable, Literal, Optional, Protocol, TypeAlias, Union, runtime_checkable
 
-from ..models import CustomBaseModel
+from ..models import AgenticIdentity, CustomBaseModel
+
+TokenScope: TypeAlias = Union[str, list[str]]
+TokenResult: TypeAlias = Union[str, Awaitable[str]]
+BasicTokenProvider: TypeAlias = Callable[[TokenScope, Optional[str]], TokenResult]
+_PositionalAgenticTokenProvider: TypeAlias = Callable[
+    [TokenScope, Optional[str], Optional[AgenticIdentity]], TokenResult
+]
+
+
+@runtime_checkable
+class _KeywordAgenticTokenProvider(Protocol):
+    def __call__(
+        self,
+        scope: TokenScope,
+        tenant_id: Optional[str],
+        *,
+        agentic_identity: Optional[AgenticIdentity] = None,
+    ) -> TokenResult: ...
+
+
+TokenProvider: TypeAlias = Union[
+    BasicTokenProvider,
+    _PositionalAgenticTokenProvider,
+    _KeywordAgenticTokenProvider,
+]
 
 
 class ClientCredentials(CustomBaseModel):
@@ -36,8 +61,8 @@ class TokenCredentials(CustomBaseModel):
     """
     The tenant ID.
     """
-    # (scope: string | string[], tenantId?: string) => string | Promise<string>
-    token: Callable[[Union[str, list[str]], Optional[str]], Union[str, Awaitable[str]]]
+    # (scope: string | string[], tenantId?: string, agenticIdentity?: AgenticIdentity) => string | Promise<string>
+    token: TokenProvider
     """
     The token function.
     """

packages/api/src/microsoft_teams/api/clients/_auth_provider_interceptor.py

@@ -0,0 +1,54 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Awaitable, Protocol, cast
+
+from microsoft_teams.common import InterceptorRequestContext, resolve_token
+from microsoft_teams.common.http.client_token import StringLike
+
+from ..models.agentic_identity import AgenticIdentity
+
+
+class AuthProvider(Protocol):
+    def token(
+        self, *, scope: str | None = None, agentic_identity: AgenticIdentity | None = None
+    ) -> str | StringLike | None | Awaitable[str | StringLike | None]: ...
+
+
+AGENTIC_IDENTITY_EXTENSION = "microsoft_teams.agentic_identity"
+logger = logging.getLogger(__name__)
+
+
+class AuthProviderInterceptor:
+    """Adds an auth-provider token when a request has no Authorization header."""
+
+    def __init__(
+        self,
+        auth_provider: AuthProvider,
+        *,
+        default_agentic_identity: AgenticIdentity | None = None,
+    ) -> None:
+        self._auth_provider = auth_provider
+        self._default_agentic_identity = default_agentic_identity
+
+    async def request(self, ctx: InterceptorRequestContext) -> None:
+        if "Authorization" in ctx.request.headers:
+            return
+
+        request_agentic_identity = cast(AgenticIdentity | None, ctx.request.extensions.get(AGENTIC_IDENTITY_EXTENSION))
+        agentic_identity = request_agentic_identity or self._default_agentic_identity
+        token = await resolve_token(lambda: self._auth_provider.token(agentic_identity=agentic_identity))
+        if token is None:
+            return
+
+        token = token.strip()
+        if not token:
+            logger.warning("Auth provider returned an empty token; skipping Authorization header.")
+            return
+
+        ctx.request.headers["Authorization"] = f"Bearer {token}"

packages/api/src/microsoft_teams/api/clients/api_client.py

@@ -5,12 +5,15 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Optional, Union
+from typing import Optional, Union, cast
 
 from microsoft_teams.common import Client as HttpClient
-from microsoft_teams.common import ClientOptions
+from microsoft_teams.common import ClientOptions, Interceptor
 
-from .api_client_settings import ApiClientSettings
+from ..auth.cloud_environment import PUBLIC, CloudEnvironment
+from ..models import AgenticIdentity
+from ._auth_provider_interceptor import AuthProvider, AuthProviderInterceptor
+from .api_client_settings import ApiClientSettings, merge_api_client_settings
 from .base_client import BaseClient
 from .bot import BotClient
 from .conversation import ConversationClient
@@ -19,9 +22,6 @@
 from .team import TeamClient
 from .user import UserClient
 
-if TYPE_CHECKING:
-    from ..auth.cloud_environment import CloudEnvironment
-
 
 class ApiClient(BaseClient):
     """Unified client for Microsoft Teams API operations."""
@@ -32,6 +32,9 @@ def __init__(
         options: Optional[Union[HttpClient, ClientOptions]] = None,
         api_client_settings: Optional[ApiClientSettings] = None,
         cloud: Optional[CloudEnvironment] = None,
+        *,
+        auth_provider: Optional[AuthProvider] = None,
+        agentic_identity: Optional[AgenticIdentity] = None,
     ) -> None:
         """Initialize the unified Teams API client.
 
@@ -41,12 +44,17 @@ def __init__(
             api_client_settings: Optional API client settings.
             cloud: Optional cloud environment for sovereign cloud support.
         """
-        super().__init__(options, api_client_settings)
+        self._cloud = cloud or PUBLIC
+        merged_settings = merge_api_client_settings(api_client_settings, self._cloud)
+        super().__init__(options, merged_settings)
         self.service_url = service_url.rstrip("/")
+        self._auth_provider = auth_provider
+        self._default_agentic_identity = agentic_identity
+        self._apply_auth_provider_interceptor()
 
         # Initialize all client types
-        self.bots = BotClient(self._http, self._api_client_settings, cloud=cloud)
-        self.users = UserClient(self._http, self._api_client_settings)
+        self.bots = BotClient(self._http, self._api_client_settings, cloud=self._cloud)
+        self.users = UserClient(self._http, self._api_client_settings, cloud=self._cloud)
         self.conversations = ConversationClient(self.service_url, self._http, self._api_client_settings)
         self.teams = TeamClient(self.service_url, self._http, self._api_client_settings)
         self.meetings = MeetingClient(self.service_url, self._http, self._api_client_settings)
@@ -59,6 +67,23 @@ def reactions(self) -> ReactionClient:
             self._reactions = ReactionClient(self.service_url, self._http, self._api_client_settings)
         return self._reactions
 
+    def _apply_auth_provider_interceptor(self) -> None:
+        if self._auth_provider is None:
+            return
+
+        if any(isinstance(interceptor, AuthProviderInterceptor) for interceptor in self._http.interceptors):
+            return
+
+        self._http.use_interceptor(
+            cast(
+                Interceptor,
+                AuthProviderInterceptor(
+                    self._auth_provider,
+                    default_agentic_identity=self._default_agentic_identity,
+                ),
+            )
+        )
+
     @property
     def http(self) -> HttpClient:
         """Get the HTTP client instance."""
@@ -67,11 +92,12 @@ def http(self) -> HttpClient:
     @http.setter
     def http(self, value: HttpClient) -> None:
         """Set the HTTP client instance and propagate to all sub-clients."""
-        self.bots.http = value
-        self.conversations.http = value
-        self.users.http = value
-        self.teams.http = value
-        self.meetings.http = value
-        if self._reactions is not None:
-            self._reactions.http = value
         self._http = value
+        self._apply_auth_provider_interceptor()
+        self.bots.http = self._http
+        self.conversations.http = self._http
+        self.users.http = self._http
+        self.teams.http = self._http
+        self.meetings.http = self._http
+        if self._reactions is not None:
+            self._reactions.http = self._http

packages/api/src/microsoft_teams/api/clients/base_client.py

@@ -5,7 +5,7 @@
 
 from typing import Optional, Union
 
-from microsoft_teams.common.http import Client, ClientOptions
+from microsoft_teams.common import Client, ClientOptions
 
 from .api_client_settings import ApiClientSettings, merge_api_client_settings
 

packages/api/src/microsoft_teams/api/clients/bot/client.py

@@ -5,18 +5,16 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Optional, Union
+from typing import Optional, Union
 
 from microsoft_teams.common.http import Client, ClientOptions
 
-from ..api_client_settings import ApiClientSettings
+from ...auth.cloud_environment import PUBLIC, CloudEnvironment
+from ..api_client_settings import ApiClientSettings, merge_api_client_settings
 from ..base_client import BaseClient
 from .sign_in_client import BotSignInClient
 from .token_client import BotTokenClient
 
-if TYPE_CHECKING:
-    from ...auth.cloud_environment import CloudEnvironment
-
 
 class BotClient(BaseClient):
     """Client for managing bot operations."""
@@ -34,9 +32,11 @@ def __init__(
             api_client_settings: Optional API client settings.
             cloud: Optional cloud environment for sovereign cloud support.
         """
-        super().__init__(options, api_client_settings)
-        self.token = BotTokenClient(self.http, self._api_client_settings, cloud=cloud)
-        self.sign_in = BotSignInClient(self.http, self._api_client_settings)
+        self._cloud = cloud or PUBLIC
+        merged_settings = merge_api_client_settings(api_client_settings, self._cloud)
+        super().__init__(options, merged_settings)
+        self.token = BotTokenClient(self.http, self._api_client_settings, cloud=self._cloud)
+        self.sign_in = BotSignInClient(self.http, self._api_client_settings, cloud=self._cloud)
 
     @property
     def http(self) -> Client:

packages/api/src/microsoft_teams/api/clients/bot/sign_in_client.py

@@ -3,10 +3,13 @@
 Licensed under the MIT License.
 """
 
+from __future__ import annotations
+
 from typing import Optional, Union, cast
 
 from microsoft_teams.common.http import Client, ClientOptions
 
+from ...auth.cloud_environment import PUBLIC, CloudEnvironment
 from ...models import SignInUrlResponse
 from ..api_client_settings import ApiClientSettings, merge_api_client_settings
 from ..base_client import BaseClient
@@ -26,15 +29,17 @@ def __init__(
         self,
         options: Union[Client, ClientOptions, None] = None,
         api_client_settings: Optional[ApiClientSettings] = None,
+        cloud: Optional[CloudEnvironment] = None,
     ) -> None:
         """Initialize the bot sign-in client.
 
         Args:
             options: Optional Client or ClientOptions instance.
             api_client_settings: Optional API client settings.
         """
-        super().__init__(options)
-        self._api_client_settings = merge_api_client_settings(api_client_settings)
+        self._cloud = cloud or PUBLIC
+        merged_settings = merge_api_client_settings(api_client_settings, self._cloud)
+        super().__init__(options, merged_settings)
 
     async def get_url(self, params: GetBotSignInUrlParams) -> str:
         """Get a sign-in URL.