Prompt Preview Support

examples/targeted-messages/src/main.py

@@ -4,9 +4,9 @@
 """
 
 import asyncio
+from datetime import datetime, timezone
 
-from microsoft_teams.api import Account, MessageActivity, MessageActivityInput
-from microsoft_teams.api.activities.typing import TypingActivityInput
+from microsoft_teams.api import MessageActivity, MessageActivityInput
 from microsoft_teams.apps import ActivityContext, App
 
 """
@@ -23,112 +23,93 @@
 @app.on_message
 async def handle_message(ctx: ActivityContext[MessageActivity]):
     """Handle message activities."""
-    await ctx.reply(TypingActivityInput())
+    activity = ctx.activity
+    text = (activity.text or "").lower()
 
-    text = (ctx.activity.text or "").lower()
+    print(f"[MESSAGE] Received: {text}")
 
-    # ============================================
-    # Test targeted SEND (create)
-    # ============================================
-    if "test send" in text:
-        members = await ctx.api.conversations.members(ctx.activity.conversation.id).get_all()
+    if "test update" in text:
+        # UPDATE: Send a targeted message, then update it after 3 seconds
+        conversation_id = activity.conversation.id
 
-        for member in members:
-            print(f"Member: {member.name} - {member.id}")
+        targeted_message = MessageActivityInput(
+            text="📝 This message will be **updated** in 3 seconds..."
+        ).with_recipient(activity.from_, is_targeted=True)
 
-            targeted_message = MessageActivityInput(
-                text="🔒 [SEND] This is a targeted message - only YOU can see this!"
-            ).with_recipient(Account(id=member.id, name=member.name), is_targeted=True)
+        result = await ctx.send(targeted_message)
 
-            result = await ctx.send(targeted_message)
-        print("[SEND] Sent targeted message")
-        return
-
-    # ============================================
-    # Test targeted REPLY
-    # ============================================
-    if "test reply" in text:
-        targeted_reply = MessageActivityInput(text="🔒 [REPLY] Targeted reply - only YOU can see this!").with_recipient(
-            ctx.activity.from_, is_targeted=True
-        )
+        if result.id:
+            message_id = result.id
 
-        result = await ctx.reply(targeted_reply)
-        print(f"Targeted REPLY result: {result}")
-        return
+            async def update_after_delay():
+                await asyncio.sleep(3)
+                try:
+                    timestamp = datetime.now(timezone.utc).strftime("%H:%M:%S")
+                    updated_message = MessageActivityInput(
+                        text=f"✏️ **Updated!** This message was modified at {timestamp}"
+                    )
+                    updated_message.id = message_id
 
-    # ============================================
-    # Test targeted UPDATE
-    # ============================================
-    if "test update" in text:
-        # First send a targeted message
-        targeted_message = MessageActivityInput(text="🔒 [UPDATE] Original targeted message...").with_recipient(
-            ctx.activity.from_, is_targeted=True
-        )
+                    await ctx.api.conversations.activities(conversation_id).update_targeted(message_id, updated_message)
+                    print("[UPDATE] Updated targeted message")
+                except Exception as err:
+                    print(f"[UPDATE] Error: {err}")
+
+            asyncio.create_task(update_after_delay())
+
+        print("[UPDATE] Scheduled update in 3 seconds")
+
+    elif "test delete" in text:
+        # DELETE: Send a targeted message, then delete it after 3 seconds
+        conversation_id = activity.conversation.id
 
-        result = await ctx.send(targeted_message)
-        print(f"Initial targeted message ID: {result.id}")
-
-        # Wait then update
-        async def update_after_delay():
-            await asyncio.sleep(3)
-            try:
-                # For targeted updates, do not include recipient in the payload.
-                updated_message = MessageActivityInput(
-                    text="🔒 [UPDATE] ✅ UPDATED targeted message! (only you see this)"
-                )
-                updated_message.id = result.id
-
-                await ctx.api.conversations.activities(ctx.activity.conversation.id).update_targeted(
-                    result.id, updated_message
-                )
-                print("Targeted UPDATE completed")
-            except Exception as err:
-                print(f"Targeted UPDATE error: {err}")
-
-        asyncio.create_task(update_after_delay())
-        return
-
-    # ============================================
-    # Test targeted DELETE
-    # ============================================
-    if "test delete" in text:
-        # First send a targeted message
         targeted_message = MessageActivityInput(
-            text="🔒 [DELETE] This targeted message will be DELETED in 5 seconds..."
-        ).with_recipient(ctx.activity.from_, is_targeted=True)
+            text="🗑️ This message will be **deleted** in 3 seconds..."
+        ).with_recipient(activity.from_, is_targeted=True)
 
         result = await ctx.send(targeted_message)
-        print(f"Targeted message to delete, ID: {result.id}")
-
-        # Wait then delete using the targeted API
-        async def delete_after_delay():
-            await asyncio.sleep(5)
-            try:
-                await ctx.api.conversations.activities(ctx.activity.conversation.id).delete_targeted(result.id)
-                print("Targeted DELETE completed")
-            except Exception as err:
-                print(f"Targeted DELETE error: {err}")
-
-        asyncio.create_task(delete_after_delay())
-        return
-
-    # ============================================
-    # Help / Default
-    # ============================================
-    if "help" in text:
-        await ctx.reply(
-            "**Targeted Messages Test Bot**\n\n"
+
+        if result.id:
+            message_id = result.id
+
+            async def delete_after_delay():
+                await asyncio.sleep(3)
+                try:
+                    await ctx.api.conversations.activities(conversation_id).delete_targeted(message_id)
+                    print("[DELETE] Deleted targeted message")
+                except Exception as err:
+                    print(f"[DELETE] Error: {err}")
+
+            asyncio.create_task(delete_after_delay())
+
+        print("[DELETE] Scheduled delete in 3 seconds")
+
+    elif "test public" in text:
+        # PUBLIC: Send a public message visible to everyone in the chat.
+        await ctx.send(MessageActivityInput(text="📋 Here is the public result — everyone can see this!"))
+        print("[PUBLIC] Sent public message")
+
+    elif "test send" in text:
+        # SEND: Send a targeted message visible only to the sender.
+        targeted_message = MessageActivityInput(
+            text="👋 This is a **targeted message** — only YOU can see this!"
+        ).with_recipient(activity.from_, is_targeted=True)
+        await ctx.send(targeted_message)
+        print("[SEND] Sent targeted message")
+
+    elif "help" in text:
+        await ctx.send(
+            "**🎯 Targeted Messages Demo**\n\n"
             "**Commands:**\n"
-            "- `test send` - Send a targeted message\n"
-            "- `test reply` - Reply with a targeted message\n"
-            "- `test update` - Send then update a targeted message\n"
-            "- `test delete` - Send then delete a targeted message\n\n"
-            "💡 *Test in a group chat to verify others don't see targeted messages!*"
+            "- `test send` - Send a targeted message (only visible to you)\n"
+            "- `test update` - Send a targeted message, then update it after 3 seconds\n"
+            "- `test delete` - Send a targeted message, then delete it after 3 seconds\n"
+            "- `test public` - Send a public reply (visible to all)\n\n"
+            "_Targeted messages are only visible to you, even in group chats!_"
         )
-        return
 
-    # Default
-    await ctx.reply('Say "help" for available commands.')
+    else:
+        await ctx.send(f"You said: '{activity.text}'\n\nType `help` to see available commands.")
 
 
 if __name__ == "__main__":

packages/api/src/microsoft_teams/api/activities/message/message.py

@@ -35,6 +35,7 @@
     Entity,
     Image,
     MessageEntity,
+    TargetedMessageInfoEntity,
 )
 from ..utils import StripMentionsTextOptions, strip_mentions_text
 
@@ -481,6 +482,36 @@ def add_quote(self, message_id: str, text: str | None = None) -> Self:
             self.add_text(f" {text}")
         return self
 
+    @experimental("ExperimentalTeamsTargeted")
+    def add_targeted_message_info(self, message_id: str) -> Self:
+        """Add a targetedMessageInfo entity for prompt preview.
+
+        If an entity with type ``"targetedMessageInfo"`` already exists,
+        it is not added again (one prompt preview per message).
+
+        When adding the entity, any ``quotedReply`` entities and matching
+        ``<quoted messageId="..."/>`` placeholder text are removed to avoid
+        collision with prompt preview.
+
+        Args:
+            message_id: The message ID of the targeted message.
+
+        Returns:
+            Self for method chaining
+        """
+        has_entity = any(isinstance(e, TargetedMessageInfoEntity) for e in (self.entities or []))
+
+        # Always strip quotedReply artifacts to avoid collision with prompt preview,
+        # if the developer already attached a targetedMessageInfo entity.
+        if self.entities is not None:
+            self.entities = [e for e in self.entities if getattr(e, "type", None) != "quotedReply"]
+        if self.text is not None:
+            self.text = self.text.replace(f'<quoted messageId="{message_id}"/>', "").strip()
+
+        if not has_entity:
+            self.add_entity(TargetedMessageInfoEntity(message_id=message_id))
+        return self
+
     def with_recipient(self, value: Account, is_targeted: Optional[bool] = None) -> Self:
         """
         Set the recipient.

packages/api/src/microsoft_teams/api/models/entity/__init__.py

@@ -21,6 +21,7 @@
 from .quoted_reply_entity import QuotedReplyData, QuotedReplyEntity
 from .sensitive_usage_entity import SensitiveUsage, SensitiveUsageEntity, SensitiveUsagePattern
 from .stream_info_entity import StreamInfoEntity
+from .targeted_message_info_entity import TargetedMessageInfoEntity
 
 __all__ = [
     "AIMessageEntity",
@@ -41,5 +42,6 @@
     "SensitiveUsage",
     "SensitiveUsagePattern",
     "StreamInfoEntity",
+    "TargetedMessageInfoEntity",
     "Entity",
 ]

packages/api/src/microsoft_teams/api/models/entity/entity.py

@@ -14,6 +14,7 @@
 from .quoted_reply_entity import QuotedReplyEntity
 from .sensitive_usage_entity import SensitiveUsageEntity
 from .stream_info_entity import StreamInfoEntity
+from .targeted_message_info_entity import TargetedMessageInfoEntity
 
 Entity = Union[
     ClientInfoEntity,
@@ -25,4 +26,5 @@
     SensitiveUsageEntity,
     ProductInfoEntity,
     QuotedReplyEntity,
+    TargetedMessageInfoEntity,
 ]

packages/api/src/microsoft_teams/api/models/entity/targeted_message_info_entity.py

@@ -0,0 +1,26 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from typing import Literal
+
+from microsoft_teams.common.experimental import experimental
+
+from ..custom_base_model import CustomBaseModel
+
+
+@experimental("ExperimentalTeamsTargeted")
+class TargetedMessageInfoEntity(CustomBaseModel):
+    """Entity containing targeted message information for prompt preview.
+
+    .. warning:: Preview
+        This class is in preview and may change in the future.
+        Diagnostic: ExperimentalTeamsTargeted
+    """
+
+    type: Literal["targetedMessageInfo"] = "targetedMessageInfo"
+    "Type identifier for targeted message info"
+
+    message_id: str
+    "The ID of the targeted message this activity is replying to"