""" Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT License. """ import base64 import json import logging import warnings from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Awaitable, Callable, Generic, Optional, TypeGuard, TypeVar from microsoft_teams.api import ( Account, ActivityBase, ActivityParams, ApiClient, CardAction, CardActionType, ConversationReference, CreateConversationParams, GetBotSignInResourceParams, GetUserTokenParams, JsonWebToken, MessageActivity, MessageActivityInput, SentActivity, SignOutUserParams, TokenExchangeResource, TokenExchangeState, TokenPostResource, ) from microsoft_teams.api.auth.cloud_environment import PUBLIC, CloudEnvironment from microsoft_teams.api.models.attachment.card_attachment import ( OAuthCardAttachment, card_attachment, ) from microsoft_teams.api.models.oauth import OAuthCard from microsoft_teams.cards import AdaptiveCard from microsoft_teams.common import Storage from microsoft_teams.common.experimental import ExperimentalWarning, experimental from microsoft_teams.common.http.client_token import Token from ..activity_sender import ActivitySender from ..utils import create_graph_client if TYPE_CHECKING: from msgraph.graph_service_client import GraphServiceClient T = TypeVar("T", bound=ActivityBase, contravariant=True) logger = logging.getLogger(__name__) @dataclass class SignInOptions: """Options for the signin method.""" oauth_card_text: str = "Please Sign In..." sign_in_button_text: str = "Sign In" connection_name: Optional[str] = None override_sign_in_activity: Optional[ Callable[ [ Optional[TokenExchangeResource], Optional[TokenPostResource], Optional[str], ], ActivityParams, ] ] = None DEFAULT_SIGNIN_OPTIONS = SignInOptions() class ActivityContext(Generic[T]): """Context object passed to activity handlers with middleware support.""" def __init__( self, activity: T, app_id: str, storage: Storage[str, Any], api: ApiClient, user_token: Optional[str], conversation_ref: ConversationReference, is_signed_in: bool, connection_name: str, activity_sender: ActivitySender, app_token: Token, cloud: CloudEnvironment = PUBLIC, ): self.activity = activity self.app_id = app_id self.logger = logger self.conversation_ref = conversation_ref self.storage = storage self.api = api self.user_token = user_token self.connection_name = connection_name self.is_signed_in = is_signed_in self.cloud = cloud self._activity_sender = activity_sender self._app_token = app_token self.stream = activity_sender.create_stream(conversation_ref) self._next_handler: Optional[Callable[[], Awaitable[None]]] = None # Initialize graph clients as None - they'll be created lazily self._user_graph: Optional["GraphServiceClient"] = None self._app_graph: Optional["GraphServiceClient"] = None @property def user_graph(self) -> "GraphServiceClient": """ Get a Microsoft Graph client configured with the user's token. Raises: ValueError: If the user is not signed in or doesn't have a valid token. RuntimeError: If the graph client cannot be created. ImportError: If the graph dependencies are not installed. """ if not self.is_signed_in: raise ValueError("User must be signed in to access Graph client") if not self.user_token: raise ValueError("No user token available for Graph client") if self._user_graph is None: try: user_token = JsonWebToken(self.user_token) self._user_graph = create_graph_client(user_token, cloud=self.cloud) except ImportError: raise except Exception as e: self.logger.error(f"Failed to create user graph client: {e}") raise RuntimeError(f"Failed to create user graph client: {e}") from e return self._user_graph @property def app_graph(self) -> "GraphServiceClient": """ Get a Microsoft Graph client configured with the app's token. This client can be used for app-only operations that don't require user context. Raises: ValueError: If no app token is available. RuntimeError: If the graph client cannot be created. ImportError: If the graph dependencies are not installed. """ if self._app_graph is None: try: self._app_graph = create_graph_client(self._app_token, cloud=self.cloud) except ImportError: raise except Exception as e: self.logger.error(f"Failed to create app graph client: {e}") raise RuntimeError(f"Failed to create app graph client: {e}") from e return self._app_graph async def send( self, message: str | ActivityParams | AdaptiveCard, conversation_ref: Optional[ConversationReference] = None, ) -> SentActivity: """Send a message in the current conversation without quoting. In channels, sends to the current thread. In scopes that do not support threading (group chat, meetings), sends as a normal message. To send with a visual quote of the inbound message, use :meth:`reply`. Args: message: The message to send, can be a string, ActivityParams, or AdaptiveCard conversation_ref: Optional conversation reference to send to a different conversation or thread """ if isinstance(message, str): activity = MessageActivityInput(text=message) elif isinstance(message, AdaptiveCard): activity = MessageActivityInput().add_card(message) else: activity = message if self._should_outbound_be_auto_targeted(activity, conversation_ref): self._apply_targeted_recipient(activity) self._add_targeted_message_info_entity(activity) ref = conversation_ref or self.conversation_ref res = await self._activity_sender.send(activity, ref) return res async def reply(self, input: str | ActivityParams) -> SentActivity: """Send a message in the current conversation with a visual quote of the inbound message. In channels, sends to the current thread with a quoted reply. In other scopes, sends with a quoted reply. To send without quoting, use :meth:`send`. """ if self.activity.id: return await self.quote(self.activity.id, input) activity = MessageActivityInput(text=input) if isinstance(input, str) else input return await self.send(activity) @experimental("ExperimentalTeamsQuotedReplies") async def quote(self, message_id: str, input: str | ActivityParams) -> SentActivity: """ Send a message to the conversation with a quoted message reference prepended to the text. Teams renders the quoted message as a preview bubble above the response text. Args: message_id: The ID of the message to quote input: The response text or activity — a quote placeholder for message_id will be prepended to its text Returns: The sent activity .. warning:: Coming Soon This API is coming soon and may change in the future. Diagnostic: ExperimentalTeamsQuotedReplies """ activity = MessageActivityInput(text=input) if isinstance(input, str) else input if isinstance(activity, MessageActivityInput): activity.prepend_quote(message_id) return await self.send(activity) async def next(self) -> None: """Call the next middleware in the chain.""" if self._next_handler: await self._next_handler() def set_next(self, handler: Callable[[], Awaitable[None]]) -> None: """Set the next handler in the middleware chain.""" self._next_handler = handler def _incoming_targeted_sender(self) -> Optional[Account]: if not isinstance(self.activity, MessageActivity): return None if self.activity.recipient.is_targeted is not True: return None return self.activity.from_ def _should_outbound_be_auto_targeted( self, activity: ActivityParams, conversation_ref: Optional[ConversationReference] = None, ) -> bool: if not isinstance(activity, MessageActivityInput): return False if self._incoming_targeted_sender() is None: return False if not self._is_same_conversation(conversation_ref): return False return not activity.id and activity.recipient is None def _is_same_conversation(self, conversation_ref: Optional[ConversationReference] = None) -> bool: if conversation_ref is None: return True return conversation_ref.conversation.id == self.conversation_ref.conversation.id def _apply_targeted_recipient(self, activity: ActivityParams) -> None: sender = self._incoming_targeted_sender() if sender is None: return recipient = sender.model_copy() recipient.is_targeted = True activity.recipient = recipient def _is_targeted_outbound(self, activity: ActivityParams) -> TypeGuard[MessageActivityInput]: return ( isinstance(activity, MessageActivityInput) and activity.recipient is not None and activity.recipient.is_targeted is True ) def _add_targeted_message_info_entity(self, activity_params: ActivityParams) -> None: """Auto-populate targetedMessageInfo entity when replying to a targeted message. In the reactive flow, the SDK reads the incoming targeted message ID and attaches the entity automatically so the developer doesn't need to. Skips if the developer already attached a targetedMessageInfo entity. """ if self._incoming_targeted_sender() is None: return if not self._is_targeted_outbound(activity_params): return with warnings.catch_warnings(): warnings.simplefilter("ignore", ExperimentalWarning) activity_params.add_targeted_message_info(self.activity.id) async def sign_in(self, options: Optional[SignInOptions] = None) -> Optional[str]: """ Initiate a sign-in flow for the user. Args: options: Optional signin options to customize the flow Returns: The token if already available, otherwise None after sending OAuth card """ signin_opts = options or DEFAULT_SIGNIN_OPTIONS oauth_card_text = signin_opts.oauth_card_text sign_in_button_text = signin_opts.sign_in_button_text connection_name = signin_opts.connection_name or self.connection_name try: # Try to get existing token token_params = GetUserTokenParams( channel_id=self.activity.channel_id, user_id=self.activity.from_.id, connection_name=connection_name, ) res = await self.api.users.token.get(token_params) return res.token except Exception: # Token not available, continue with OAuth flow pass # Create token exchange state token_exchange_state = TokenExchangeState( connection_name=connection_name, conversation=self.conversation_ref, ms_app_id=self.app_id, ) # Check if this is a group conversation # if it's a group conversation, then we create a 1:1 conversation with the user # and send the OAuth card there since group oauth currently isn't released. conversation_id = self.conversation_ref.conversation.id if self.activity.conversation.is_group: one_on_one_conversation = await self.api.conversations.create( CreateConversationParams( tenant_id=self.activity.conversation.tenant_id, members=[self.activity.from_], ) ) conversation_id = one_on_one_conversation.id await self.send(MessageActivityInput(text=oauth_card_text)) # Encode state state = base64.b64encode(json.dumps(token_exchange_state.model_dump()).encode()).decode() # Get sign-in resource resource_params = GetBotSignInResourceParams(state=state) resource = await self.api.bots.sign_in.get_resource(resource_params) payload = MessageActivityInput(recipient=self.activity.from_).add_attachments( card_attachment( attachment=OAuthCardAttachment( content=OAuthCard( text=oauth_card_text, connection_name=connection_name, token_exchange_resource=resource.token_exchange_resource, token_post_resource=resource.token_post_resource, buttons=[ CardAction( type=CardActionType.SIGN_IN, title=sign_in_button_text, value=resource.sign_in_link, ) ], ) ), ) ) self.conversation_ref.conversation.id = conversation_id await self.send(payload, self.conversation_ref) return None async def sign_out(self) -> None: """ Sign out the user by clearing their token. This method will remove the user's token from the storage. """ try: sign_out_params = SignOutUserParams( channel_id=self.activity.channel_id, user_id=self.activity.from_.id, connection_name=self.connection_name, ) await self.api.users.token.sign_out(sign_out_params) self.logger.debug(f"User {self.activity.from_.id} signed out successfully.") except Exception as e: self.logger.error(f"Failed to sign out user: {e}")