I have a fairly simple copilot bot with a python-fastapi azure bot service in azure added as a skill doing passthrough of messages. When I use the send_activities method on the turn_context object, messages come through fine on Omnichannel's live chat widget that they provide for a chat stream.
However, when I try to send an event, such as:
transfer_event = {
"type": "event",
"name": "initiateHandoff",
"from": {
"id": "1234",
"name": "me"
}
}
response = await turn_context.send_activities([transfer_event_activity])
The event never makes it to copilot to trigger a topic. There are no logged errors or anything.
In the demo website, it uses directline channel and I can use the conversationId to send events. The conversation ids in the live chat widget and sms, however, do not work as they look like they're coming from ACS.
If anyone has any ideas, it would be greatly appreciated.
I discovered, by looking at the C# examples, that Omnichannel does not accept event activities. Instead, it wants you to format a message with channelData to tell it that it is an event. You also have to include the role property in the from property, which I couldn't find documented anywhere else.
To solve this, I created a middleware that is trigger on turns that come from omnichannel:
from botbuilder.core import Middleware, TurnContext
from botbuilder.schema import Activity, ActivityTypes, ChannelAccount
import json
from typing import Dict, Any, List, Callable
import logging
logger = logging.getLogger(__name__)
class OmnichannelMiddleware(Middleware):
"""Middleware to handle Omnichannel-specific message transformations."""
TAGS = "tags"
DELIVERY_MODE = "deliveryMode"
def _transform_activity(self, activity: Activity) -> Activity:
"""Transform an activity for Omnichannel compatibility."""
if activity.type == ActivityTypes.message:
channel_data = activity.channel_data or {}
channel_data[self.DELIVERY_MODE] = "bridged"
activity.channel_data = channel_data
elif activity.type == ActivityTypes.event and activity.name == "handoff.initiate":
handoff_context = activity.value
# Construct Omnichannel command
command = {
"type": "Escalate",
"context": handoff_context.get("Context", {})
}
# Convert to message type for Omnichannel
activity.type = ActivityTypes.message
activity.text = handoff_context.get("MessageToAgent", "")
activity.channel_data = self._build_command_channel_data(
command,
activity.channel_data
)
activity.value = None
activity.from_property = ChannelAccount(
role="bot"
)
elif activity.type == ActivityTypes.end_of_conversation:
command = {
"type": "EndConversation"
}
activity.type = ActivityTypes.message
activity.text = "End of conversation"
activity.channel_data = self._build_command_channel_data(
command,
activity.channel_data
)
return activity
async def on_turn(self, turn_context: TurnContext, next: Callable):
"""Process incoming and outgoing activities."""
# Intercept outgoing activities
original_send_activities = turn_context.send_activities
async def send_activities_handler(activities: List[Activity]):
modified_activities = [self._transform_activity(activity) for activity in activities]
return await original_send_activities(modified_activities)
# Replace the send_activities handler
turn_context.send_activities = send_activities_handler
# Continue processing
if next:
return await next()
def _build_command_channel_data(self, command: Dict[str, Any], channel_data: Dict[str, Any] = None) -> Dict[str, Any]:
"""Build channel data with command information."""
channel_data = channel_data or {}
channel_data[self.TAGS] = json.dumps(command)
return channel_data
Then I create a handoff event as described in the docs:
transfer_event = Activity(
type=ActivityTypes.event,
name="handoff.initiate",
value={
"Context": {
"BotHandoffContext": "Escalation from Ray",
},
"MessageToAgent": messageToAgent
}
)
turn_context.send_activity(transfer_event)