Need help intercepting outgoing messages and accessing chat history in Teams bot (python)

To achieve the tasks you're aiming for using the Teams AI Library in Python, you can indeed use decorators and built-in functionalities to intercept outgoing messages and access conversation history. 

1. To intercept outgoing messages, you can create a decorator to wrap the send_activity method. This decorator can modify the message before it is sent.

Example Code:

from botbuilder.core import TurnContext, ActivityHandler

from botbuilder.schema import Activity

# Decorator to intercept and modify outgoing messages

def intercept_outgoing_messages(func):

    async def wrapper(*args, **kwargs):

        context = args[1]

        if isinstance(context, TurnContext):

            for activity in context.activities:

                if activity.type == 'message':

                    # Modify the message here

                    activity.text = f"Modified: {activity.text}"

        return await func(*args, **kwargs)

    return wrapper

class MyBot(ActivityHandler):

    @intercept_outgoing_messages

    async def send_activity(self, turn_context: TurnContext, activity: Activity):

        await turn_context.send_activity(activity)

# Usage example

async def on_message_activity(turn_context: TurnContext):

    my_bot = MyBot()

    await my_bot.send_activity(turn_context, turn_context.activity)

 

2.To access the conversation history, you can use the TurnContext to retrieve previous messages. You can store messages in a state property and retrieve them as needed.

Example Code:

from botbuilder.core import TurnContext, ConversationState, MemoryStorage, ActivityHandler

from botbuilder.core.adapters import BotFrameworkAdapter

from botbuilder.schema import Activity

# Initialize memory storage and conversation state

memory_storage = MemoryStorage()

conversation_state = ConversationState(memory_storage)

class MyBot(ActivityHandler):

    def __init__(self, conversation_state: ConversationState):

        self.conversation_state = conversation_state

        self.conversation_data_accessor = conversation_state.create_property("ConversationData")

    async def on_message_activity(self, turn_context: TurnContext):

        # Retrieve conversation history

        conversation_data = await self.conversation_data_accessor.get(turn_context, {})

        if "history" not in conversation_data:

            conversation_data["history"] = []

        # Add current message to history

        conversation_data["history"].append(turn_context.activity.text)

        # Save updated conversation data

        await self.conversation_data_accessor.set(turn_context, conversation_data)

        await self.conversation_state.save_changes(turn_context)

        # Send a response

        await turn_context.send_activity(f"Message received: {turn_context.activity.text}")

    async def get_conversation_history(self, turn_context: TurnContext):

        conversation_data = await self.conversation_data_accessor.get(turn_context, {})

        return conversation_data.get("history", [])

# Usage example

async def on_message_activity(turn_context: TurnContext):

    my_bot = MyBot(conversation_state)

    await my_bot.on_message_activity(turn_context)

    history = await my_bot.get_conversation_history(turn_context)

    print(f"Conversation history: {history}")

# Initialize adapter and run the bot

adapter = BotFrameworkAdapter()

adapter.process_activity(on_message_activity)

Reference Document-
1.Core Capabilities of Teams AI Library - Teams | Microsoft Learn
2.Create a Teams AI Bot with RAG - Teams | Microsoft Learn
3.Build a basic AI Chatbot in Teams - Teams | Microsoft Learn

 

Thank you for your inquiry about your Teams app development issue!

We are checking the issue. We will get back to you shortly.