tessl/pypi-botbuilder-core
Microsoft Bot Framework Bot Builder core functionality for building conversational AI bots and chatbots in Python.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
turn-context.mddocs/
Turn Context & Activities
Turn context management and activity manipulation utilities. TurnContext represents the context of a single conversational turn and provides methods for sending responses, managing turn state, and working with activities.
Capabilities
TurnContext
Central context object for a single conversational turn that encapsulates the current activity, provides methods for sending responses, and manages turn-specific state and services.
class TurnContext:
def __init__(self, adapter, activity):
"""
Initialize turn context.
Args:
adapter (BotAdapter): The bot adapter
activity (Activity): The incoming activity
"""
self.adapter = adapter
self.activity = activity
self.responded = False
self.services = {}
self.turn_state = {}
async def send_activity(self, activity_or_text, speak: str = None, input_hint=None):
"""
Send a single activity response.
Args:
activity_or_text: Activity object or text string
speak (str, optional): Speech text for voice channels
input_hint (optional): Input hint for the client
Returns:
ResourceResponse: Response from sending the activity
"""
async def send_activities(self, activities):
"""
Send multiple activities in sequence.
Args:
activities (list): List of Activity objects to send
Returns:
list: List of ResourceResponse objects
"""
async def update_activity(self, activity):
"""
Update an existing activity.
Args:
activity (Activity): Activity with updated content
Returns:
ResourceResponse: Response from the update operation
"""
async def delete_activity(self, id_or_reference):
"""
Delete an activity.
Args:
id_or_reference: Activity ID string or ConversationReference
"""
@property
def responded(self):
"""
Check if a response has been sent during this turn.
Returns:
bool: True if a response was sent
"""
@property
def turn_state(self):
"""
Get turn-specific state dictionary.
Returns:
dict: Turn state storage
"""
@property
def services(self):
"""
Get services dictionary for dependency injection.
Returns:
dict: Services container
"""
def get(self, key: str):
"""
Get a service from the services collection.
Args:
key (str): Service key
Returns:
object: Service instance or None
"""
def has(self, key: str):
"""
Check if a service exists in the collection.
Args:
key (str): Service key
Returns:
bool: True if service exists
"""
def set(self, key: str, value):
"""
Set a service in the services collection.
Args:
key (str): Service key
value (object): Service instance
"""
def on_send_activities(self, handler):
"""
Register handler for send activities events.
Args:
handler: Event handler function
Returns:
TurnContext: Self for chaining
"""
def on_update_activity(self, handler):
"""
Register handler for update activity events.
Args:
handler: Event handler function
Returns:
TurnContext: Self for chaining
"""
def on_delete_activity(self, handler):
"""
Register handler for delete activity events.
Args:
handler: Event handler function
Returns:
TurnContext: Self for chaining
"""
@staticmethod
def get_conversation_reference(activity):
"""
Get conversation reference from an activity.
Args:
activity (Activity): Activity to extract reference from
Returns:
ConversationReference: Conversation reference
"""
@staticmethod
def apply_conversation_reference(activity, reference, is_incoming=False):
"""
Apply conversation reference to an activity.
Args:
activity (Activity): Activity to modify
reference (ConversationReference): Reference to apply
is_incoming (bool): Whether this is an incoming activity
Returns:
Activity: Modified activity
"""
@staticmethod
def remove_recipient_mention(activity):
"""
Remove @mentions of the recipient (bot) from activity text.
Args:
activity (Activity): Activity to process
Returns:
str: Text with recipient mentions removed
"""
@staticmethod
def get_mentions(activity):
"""
Get all @mentions from an activity.
Args:
activity (Activity): Activity to analyze
Returns:
list: List of Mention objects
"""Activity Properties and Methods
Core properties and methods available on the Activity object that represents messages and other interactions in the Bot Framework.
class Activity:
"""Represents a Bot Framework activity."""
# Core properties
type: str # Activity type (message, invoke, event, etc.)
id: str # Unique activity ID
timestamp: str # When activity was created
local_timestamp: str # Local timestamp
service_url: str # Service URL for the channel
channel_id: str # Channel identifier
from_property: ChannelAccount # Who sent the activity
conversation: ConversationAccount # Conversation context
recipient: ChannelAccount # Who receives the activity
# Message-specific properties
text: str # Text content of message
speak: str # Speech text for voice channels
input_hint: str # Input hint for client
summary: str # Summary of the activity
# Rich content
attachments: list # List of Attachment objects
entities: list # List of Entity objects
channel_data: object # Channel-specific data
# Interaction properties
action: str # Action being performed
reply_to_id: str # ID of activity this replies to
value: object # Value for invoke/event activities
name: str # Name for invoke/event activities
relate_to: ConversationReference # Related conversation
# Group conversation properties
members_added: list # Members added to conversation
members_removed: list # Members removed from conversation
reactions_added: list # Reactions added to message
reactions_removed: list # Reactions removed from message
# Suggested actions
suggested_actions: SuggestedActions # Suggested actions for user
def as_message_activity():
"""Convert to message activity."""
def as_contact_relation_update_activity():
"""Convert to contact relation update activity."""
def as_installation_update_activity():
"""Convert to installation update activity."""
def as_message_reaction_activity():
"""Convert to message reaction activity."""
def as_event_activity():
"""Convert to event activity."""
def as_invoke_activity():
"""Convert to invoke activity."""
def as_handoff_activity():
"""Convert to handoff activity."""
def create_reply(text: str = "", locale: str = ""):
"""
Create a reply to this activity.
Args:
text (str): Reply text
locale (str): Locale for the reply
Returns:
Activity: Reply activity
"""
def get_conversation_reference():
"""
Get conversation reference for this activity.
Returns:
ConversationReference: Conversation reference
"""
def apply_conversation_reference(reference, is_incoming: bool = False):
"""
Apply conversation reference to this activity.
Args:
reference (ConversationReference): Reference to apply
is_incoming (bool): Whether this is incoming
"""
def create_trace(name: str, value: object = None, value_type: str = None, label: str = None):
"""
Create a trace activity.
Args:
name (str): Trace name
value (object): Trace value
value_type (str): Value type
label (str): Trace label
Returns:
Activity: Trace activity
"""Conversation Reference Utilities
Utilities for working with conversation references that enable proactive messaging and conversation tracking.
def get_conversation_reference(activity):
"""
Extract conversation reference from activity.
Args:
activity (Activity): Source activity
Returns:
ConversationReference: Conversation reference
"""
def apply_conversation_reference(activity, reference, is_incoming=False):
"""
Apply conversation reference to activity.
Args:
activity (Activity): Target activity
reference (ConversationReference): Reference to apply
is_incoming (bool): Whether activity is incoming
Returns:
Activity: Modified activity
"""
def get_reply_conversation_reference(activity, reply):
"""
Get conversation reference for a reply.
Args:
activity (Activity): Original activity
reply (ResourceResponse): Reply response
Returns:
ConversationReference: Reply conversation reference
"""Usage Examples
Basic Turn Context Usage
from botbuilder.core import ActivityHandler, TurnContext, MessageFactory
class BasicBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
# Access the incoming activity
user_message = turn_context.activity.text
user_name = turn_context.activity.from_property.name
# Send a simple response
reply = f"Hello {user_name}, you said: {user_message}"
await turn_context.send_activity(MessageFactory.text(reply))
# Check if we've responded
print(f"Responded: {turn_context.has_responded()}")Sending Multiple Activities
async def on_message_activity(self, turn_context: TurnContext):
activities = [
MessageFactory.text("First message"),
MessageFactory.text("Second message"),
MessageFactory.text("Third message")
]
# Send all activities at once
responses = await turn_context.send_activities(activities)
print(f"Sent {len(responses)} activities")Working with Conversation References
async def on_message_activity(self, turn_context: TurnContext):
# Get conversation reference for proactive messaging
conv_ref = TurnContext.get_conversation_reference(turn_context.activity)
# Store conversation reference for later use
await self.store_conversation_reference(conv_ref)
await turn_context.send_activity(MessageFactory.text("I'll remember this conversation!"))
async def send_proactive_message(self, stored_conv_ref, message):
async def proactive_callback(proactive_turn_context):
await proactive_turn_context.send_activity(MessageFactory.text(message))
await self.adapter.continue_conversation(
stored_conv_ref,
proactive_callback
)Updating and Deleting Activities
async def on_message_activity(self, turn_context: TurnContext):
# Send initial message
response = await turn_context.send_activity(MessageFactory.text("Processing..."))
# Simulate some work
await asyncio.sleep(2)
# Update the message
updated_activity = MessageFactory.text("Processing complete!")
updated_activity.id = response.id
await turn_context.update_activity(updated_activity)
# Later, delete the message if needed
# await turn_context.delete_activity(response.id)Handling Mentions
async def on_message_activity(self, turn_context: TurnContext):
# Remove bot mentions from text
clean_text = TurnContext.remove_recipient_mention(turn_context.activity)
# Get all mentions in the activity
mentions = TurnContext.get_mentions(turn_context.activity)
if mentions:
mention_info = f"Found {len(mentions)} mentions"
await turn_context.send_activity(MessageFactory.text(mention_info))
# Process the clean text
reply = f"You said: {clean_text.strip()}"
await turn_context.send_activity(MessageFactory.text(reply))Turn State Management
async def on_message_activity(self, turn_context: TurnContext):
# Store data in turn state
turn_context.turn_state["user_message_count"] = turn_context.turn_state.get("user_message_count", 0) + 1
# Access turn services
if "custom_service" not in turn_context.services:
turn_context.services["custom_service"] = CustomService()
service = turn_context.services["custom_service"]
result = await service.process(turn_context.activity.text)
await turn_context.send_activity(MessageFactory.text(f"Processed: {result}"))Types
class ConversationReference:
"""Reference to a conversation."""
activity_id: str
user: ChannelAccount
bot: ChannelAccount
conversation: ConversationAccount
channel_id: str
service_url: str
locale: str
class ResourceResponse:
"""Response from activity operations."""
id: str
class ChannelAccount:
"""Channel account information."""
id: str
name: str
aad_object_id: str
role: str
class ConversationAccount:
"""Conversation information."""
is_group: bool
conversation_type: str
id: str
name: str
aad_object_id: str
role: str
tenant_id: str
class Attachment:
"""File or media attachment."""
content_type: str
content_url: str
content: object
name: str
thumbnail_url: str
class Entity:
"""Named entity in activity."""
type: str
class SuggestedActions:
"""Suggested actions for user."""
to: list
actions: list
class Mention:
"""@mention in activity."""
mentioned: ChannelAccount
text: str
type: strInstall with Tessl CLI
npx tessl i tessl/pypi-botbuilder-core