diff --git a/libraries/microsoft-agents-activity/microsoft_agents/activity/activity.py b/libraries/microsoft-agents-activity/microsoft_agents/activity/activity.py index d5f960e0..d9bc0132 100644 --- a/libraries/microsoft-agents-activity/microsoft_agents/activity/activity.py +++ b/libraries/microsoft-agents-activity/microsoft_agents/activity/activity.py @@ -6,7 +6,8 @@ import logging from copy import copy from datetime import datetime, timezone -from typing import Optional, Any +from typing import Optional, Any, cast, Annotated, TypeVar +from typing_extensions import Self from pydantic import ( Field, @@ -48,6 +49,8 @@ logger = logging.getLogger(__name__) +_EntityT = TypeVar("_EntityT", bound=Entity) + # TODO: A2A Agent 2 is responding with None as id, had to mark it as optional (investigate) class Activity(AgentsModel, _ChannelIdFieldMixin): @@ -157,7 +160,7 @@ class Activity(AgentsModel, _ChannelIdFieldMixin): local_timestamp: datetime = None local_timezone: NonEmptyString = None service_url: NonEmptyString = None - from_property: ChannelAccount = Field(None, alias="from") + from_property: Annotated[ChannelAccount, Field(alias="from")] = None conversation: ConversationAccount = None recipient: ChannelAccount = None text_format: NonEmptyString = None @@ -509,33 +512,36 @@ def create_reply(self, text: str | None = None, locale: str | None = None): .. remarks:: The new activity sets up routing information based on this activity. """ - return pick_model( - Activity, - type=ActivityTypes.message, - timestamp=datetime.now(timezone.utc), - from_property=SkipNone( - ChannelAccount.pick_properties(self.recipient, ["id", "name"]) - ), - recipient=SkipNone( - ChannelAccount.pick_properties(self.from_property, ["id", "name"]) - ), - reply_to_id=( - SkipNone(self.id) - if type != ActivityTypes.conversation_update - or self.channel_id not in ["directline", "webchat"] - else None + return cast( + Self, + pick_model( + self.__class__, + type=ActivityTypes.message, + timestamp=datetime.now(timezone.utc), + from_property=SkipNone( + ChannelAccount.pick_properties(self.recipient, ["id", "name"]) + ), + recipient=SkipNone( + ChannelAccount.pick_properties(self.from_property, ["id", "name"]) + ), + reply_to_id=( + SkipNone(self.id) + if self.type != ActivityTypes.conversation_update + or self.channel_id not in ["directline", "webchat"] + else None + ), + service_url=self.service_url, + channel_id=self.channel_id, + conversation=SkipNone( + ConversationAccount.pick_properties( + self.conversation, ["is_group", "id", "name"] + ) + ), + text=text if text else "", + locale=locale if locale else SkipNone(self.locale), + attachments=[], + entities=[], ), - service_url=self.service_url, - channel_id=self.channel_id, - conversation=SkipNone( - ConversationAccount.pick_properties( - self.conversation, ["is_group", "id", "name"] - ) - ), - text=text if text else "", - locale=locale if locale else SkipNone(self.locale), - attachments=[], - entities=[], ) def create_trace( @@ -558,33 +564,36 @@ def create_trace( if not value_type and value: value_type = type(value).__name__ - return pick_model( - Activity, - type=ActivityTypes.trace, - timestamp=datetime.now(timezone.utc), - from_property=SkipNone( - ChannelAccount.pick_properties(self.recipient, ["id", "name"]) - ), - recipient=SkipNone( - ChannelAccount.pick_properties(self.from_property, ["id", "name"]) + return cast( + Self, + pick_model( + self.__class__, + type=ActivityTypes.trace, + timestamp=datetime.now(timezone.utc), + from_property=SkipNone( + ChannelAccount.pick_properties(self.recipient, ["id", "name"]) + ), + recipient=SkipNone( + ChannelAccount.pick_properties(self.from_property, ["id", "name"]) + ), + reply_to_id=( + SkipNone(self.id) # preserve unset + if self.type != ActivityTypes.conversation_update + or self.channel_id not in ["directline", "webchat"] + else None + ), + service_url=self.service_url, + channel_id=self.channel_id, + conversation=SkipNone( + ConversationAccount.pick_properties( + self.conversation, ["is_group", "id", "name"] + ) + ), + name=SkipNone(name), + label=SkipNone(label), + value_type=SkipNone(value_type), + value=SkipNone(value), ), - reply_to_id=( - SkipNone(self.id) # preserve unset - if type != ActivityTypes.conversation_update - or self.channel_id not in ["directline", "webchat"] - else None - ), - service_url=self.service_url, - channel_id=self.channel_id, - conversation=SkipNone( - ConversationAccount.pick_properties( - self.conversation, ["is_group", "id", "name"] - ) - ), - name=SkipNone(name), - label=SkipNone(label), - value_type=SkipNone(value_type), - value=SkipNone(value), ).as_trace_activity() @staticmethod @@ -593,7 +602,7 @@ def create_trace_activity( value: object = None, value_type: str | None = None, label: str | None = None, - ): + ) -> Activity: """ Creates an instance of the :class:`microsoft_agents.activity.Activity` class as a TraceActivity object. @@ -607,13 +616,16 @@ def create_trace_activity( if not value_type and value: value_type = type(value).__name__ - return pick_model( + return cast( Activity, - type=ActivityTypes.trace, - name=name, - label=SkipNone(label), - value_type=SkipNone(value_type), - value=SkipNone(value), + pick_model( + Activity, + type=ActivityTypes.trace, + name=name, + label=SkipNone(label), + value_type=SkipNone(value_type), + value=SkipNone(value), + ), ) @staticmethod @@ -636,33 +648,71 @@ def get_conversation_reference( Composite values are split only on the first ``:``. :returns: A conversation reference for the conversation that contains this activity. """ - return pick_model( + return cast( ConversationReference, - activity_id=( - SkipNone(self.id) - if self.type != ActivityTypes.conversation_update - or self.channel_id not in ["directline", "webchat"] - else None - ), - user=copy(self.from_property), - agent=copy(self.recipient), - conversation=copy(self.conversation), - channel_id=( - self.channel_id.split(":", 1)[0] - if force_base_channel and self.channel_id is not None - else self.channel_id + pick_model( + ConversationReference, + activity_id=( + SkipNone(self.id) + if self.type != ActivityTypes.conversation_update + or self.channel_id not in ["directline", "webchat"] + else None + ), + user=copy(self.from_property), + agent=copy(self.recipient), + conversation=copy(self.conversation), + channel_id=( + self.channel_id.split(":", 1)[0] + if force_base_channel and self.channel_id is not None + else self.channel_id + ), + locale=self.locale, + service_url=self.service_url, ), - locale=self.locale, - service_url=self.service_url, ) + @staticmethod + def _convert_entity(raw_entity: Entity, entity_cls: type[_EntityT]) -> _EntityT: + """ + Converts an entity to a specific entity type. + + :param raw_entity: The entity to convert. + :param entity_cls: The class of the entity type to convert to. + :return: The converted entity of the specified type. + """ + if isinstance(raw_entity, entity_cls): + return raw_entity + return entity_cls.model_validate(raw_entity.model_dump()) + + @staticmethod + def _convert_entity_list( + raw_entities: list[Entity], entity_cls: type[_EntityT] + ) -> list[_EntityT]: + """ + Converts a list of entities to a list of a specific entity type. + + :param raw_entities: The list of entities to convert. + :param entity_cls: The class of the entity type to convert to. + :return: The list of converted entities of the specified type. + """ + + entities: list[_EntityT] = [] + for e in raw_entities: + entities.append(Activity._convert_entity(e, entity_cls)) + return entities + def get_product_info_entity(self) -> Optional[ProductInfo]: if not self.entities: return None target = EntityTypes.PRODUCT_INFO.lower() # validated entities can be Entity, and that prevents us from # making assumptions about the casing of the 'type' attribute - return next(filter(lambda e: e.type.lower() == target, self.entities), None) + raw_product_info = next( + filter(lambda e: e.type.lower() == target, self.entities), None + ) + if raw_product_info is None: + return None + return Activity._convert_entity(raw_product_info, ProductInfo) def get_mentions(self) -> list[Mention]: """ @@ -676,7 +726,11 @@ def get_mentions(self) -> list[Mention]: """ if not self.entities: return [] - return [x for x in self.entities if x.type.lower() == EntityTypes.MENTION] + raw_mentions = [ + x for x in self.entities if x.type.lower() == EntityTypes.MENTION + ] + + return Activity._convert_entity_list(raw_mentions, Mention) def get_reply_conversation_reference( self, reply: ResourceResponse diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py index 1368f624..2ce8e082 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py @@ -440,7 +440,7 @@ def __call(func: RouteHandler[StateT]) -> RouteHandler[StateT]: def conversation_update( self, - type: ConversationUpdateTypes, + type: ConversationUpdateTypes | str, *, auth_handlers: Optional[list[str]] = None, **kwargs, @@ -457,35 +457,37 @@ async def on_channel_created(context: TurnContext, state: TurnState): return True :param type: Conversation update category that must match the incoming activity. - :type type: microsoft_agents.activity.ConversationUpdateTypes + :type type: microsoft_agents.activity.ConversationUpdateTypes | str :param auth_handlers: Optional list of authorization handler IDs for the route. :type auth_handlers: Optional[list[str]] :param kwargs: Additional route configuration passed to :meth:`microsoft_agents.hosting.core.AgentApplication.add_route`. """ + update_type = type.value if isinstance(type, ConversationUpdateTypes) else type + def __selector(context: TurnContext): if context.activity.type != ActivityTypes.conversation_update: return False - if type == "membersAdded": + if update_type == "membersAdded": if isinstance(context.activity.members_added, list): return len(context.activity.members_added) > 0 return False - if type == "membersRemoved": + if update_type == "membersRemoved": if isinstance(context.activity.members_removed, list): return len(context.activity.members_removed) > 0 return False if isinstance(context.activity.channel_data, object): data = vars(context.activity.channel_data) - return data["event_type"] == type + return data["event_type"] == update_type return False def __call(func: RouteHandler[StateT]) -> RouteHandler[StateT]: logger.debug( - f"Registering conversation update handler for route handler {func.__name__} with type: {type} with auth handlers: {auth_handlers}" + f"Registering conversation update handler for route handler {func.__name__} with type: {update_type} with auth handlers: {auth_handlers}" ) self.add_route(__selector, func, auth_handlers=auth_handlers, **kwargs) return func @@ -494,7 +496,7 @@ def __call(func: RouteHandler[StateT]) -> RouteHandler[StateT]: def message_reaction( self, - type: MessageReactionTypes, + type: MessageReactionTypes | str, *, auth_handlers: Optional[list[str]] = None, **kwargs, @@ -511,22 +513,24 @@ async def on_reactions_added(context: TurnContext, state: TurnState): return True :param type: Reaction category that must match the incoming activity. - :type type: microsoft_agents.activity.MessageReactionTypes + :type type: microsoft_agents.activity.MessageReactionTypes | str :param auth_handlers: Optional list of authorization handler IDs for the route. :type auth_handlers: Optional[list[str]] :param kwargs: Additional route configuration passed to :meth:`microsoft_agents.hosting.core.AgentApplication.add_route`. """ + reaction_type = type.value if isinstance(type, MessageReactionTypes) else type + def __selector(context: TurnContext): if context.activity.type != ActivityTypes.message_reaction: return False - if type == "reactionsAdded": + if reaction_type == "reactionsAdded": if isinstance(context.activity.reactions_added, list): return len(context.activity.reactions_added) > 0 return False - if type == "reactionsRemoved": + if reaction_type == "reactionsRemoved": if isinstance(context.activity.reactions_removed, list): return len(context.activity.reactions_removed) > 0 return False @@ -535,7 +539,7 @@ def __selector(context: TurnContext): def __call(func: RouteHandler[StateT]) -> RouteHandler[StateT]: logger.debug( - f"Registering message reaction handler for route handler {func.__name__} with type: {type} with auth handlers: {auth_handlers}" + f"Registering message reaction handler for route handler {func.__name__} with type: {reaction_type} with auth handlers: {auth_handlers}" ) self.add_route(__selector, func, auth_handlers=auth_handlers, **kwargs) return func @@ -544,7 +548,7 @@ def __call(func: RouteHandler[StateT]) -> RouteHandler[StateT]: def message_update( self, - type: MessageUpdateTypes, + type: MessageUpdateTypes | str, *, auth_handlers: Optional[list[str]] = None, **kwargs, @@ -561,44 +565,46 @@ async def on_edit_message(context: TurnContext, state: TurnState): return True :param type: Message update category that must match the incoming activity. - :type type: microsoft_agents.activity.MessageUpdateTypes + :type type: microsoft_agents.activity.MessageUpdateTypes | str :param auth_handlers: Optional list of authorization handler IDs for the route. :type auth_handlers: Optional[list[str]] :param kwargs: Additional route configuration passed to :meth:`microsoft_agents.hosting.core.AgentApplication.add_route`. """ + update_type = type.value if isinstance(type, MessageUpdateTypes) else type + def __selector(context: TurnContext): - if type == "editMessage": + if update_type == "editMessage": if ( context.activity.type == ActivityTypes.message_update and isinstance(context.activity.channel_data, dict) ): data = context.activity.channel_data - return data["event_type"] == type + return data["event_type"] == update_type return False - if type == "softDeleteMessage": + if update_type == "softDeleteMessage": if ( context.activity.type == ActivityTypes.message_delete and isinstance(context.activity.channel_data, dict) ): data = context.activity.channel_data - return data["event_type"] == type + return data["event_type"] == update_type return False - if type == "undeleteMessage": + if update_type == "undeleteMessage": if ( context.activity.type == ActivityTypes.message_update and isinstance(context.activity.channel_data, dict) ): data = context.activity.channel_data - return data["event_type"] == type + return data["event_type"] == update_type return False return False def __call(func: RouteHandler[StateT]) -> RouteHandler[StateT]: logger.debug( - f"Registering message update handler for route handler {func.__name__} with type: {type} with auth handlers: {auth_handlers}" + f"Registering message update handler for route handler {func.__name__} with type: {update_type} with auth handlers: {auth_handlers}" ) self.add_route(__selector, func, auth_handlers=auth_handlers, **kwargs) return func diff --git a/tests/activity/test_activity.py b/tests/activity/test_activity.py index f44667dc..bec57f15 100644 --- a/tests/activity/test_activity.py +++ b/tests/activity/test_activity.py @@ -403,7 +403,7 @@ def test_get_mentions(self): mentions = activity.get_mentions() assert mentions == [ Mention(text="Hello"), - Entity(type="mention", text="Another mention"), + Mention(text="Another mention"), ] @pytest.mark.parametrize( @@ -418,8 +418,7 @@ def test_get_mentions(self): Entity(type="other"), Entity(type="mention", text="Another mention"), ], - Entity( - type="ProductInfo", + ProductInfo( id="product_123", ), ], @@ -442,7 +441,7 @@ def test_get_mentions(self): ), Entity(type="mention", text="Another mention"), ], - Entity(type="ProductInfo", id="product_123"), + ProductInfo(id="product_123"), ], [[], None], ],