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

Overview

Eval results

Files

State Management

Name: tessl/pypi-botbuilder-core
Author: tessl

Persistent state management across conversation turns with support for different scopes (user, conversation, private conversation) and custom storage implementations. Includes state property accessors and automatic state persistence.

Capabilities

BotState Base Class

Abstract base class that provides the foundation for all state management in the Bot Framework, implementing property management, loading, and saving state data with storage backend integration.

class BotState:
    def __init__(self, storage, context_service_key: str):
        """
        Initialize bot state.
        
        Args:
            storage (Storage): Storage implementation
            context_service_key (str): Key for storing state in turn context
        """
    
    def create_property(self, name: str):
        """
        Create a state property accessor.
        
        Args:
            name (str): Name of the property
            
        Returns:
            StatePropertyAccessor: Property accessor
        """
    
    async def load(self, turn_context: TurnContext, force: bool = False):
        """
        Load state from storage.
        
        Args:
            turn_context (TurnContext): Current turn context
            force (bool): Force reload even if already loaded
        """
    
    async def save_changes(self, turn_context: TurnContext, force: bool = False):
        """
        Save state changes to storage.
        
        Args:
            turn_context (TurnContext): Current turn context
            force (bool): Force save even if no changes detected
        """
    
    async def clear_state(self, turn_context: TurnContext):
        """
        Clear current state.
        
        Args:
            turn_context (TurnContext): Current turn context
        """
    
    def get_storage_key(self, turn_context: TurnContext):
        """
        Get storage key for this state. Must be implemented by derived classes.
        
        Args:
            turn_context (TurnContext): Current turn context
            
        Returns:
            str: Storage key
        """
    
    async def delete(self, turn_context: TurnContext):
        """
        Delete state from storage.
        
        Args:
            turn_context (TurnContext): Current turn context
        """

ConversationState

Manages conversation-scoped state that persists across turns within a single conversation. Data is shared among all participants in the conversation.

class ConversationState(BotState):
    def __init__(self, storage):
        """
        Initialize conversation state.
        
        Args:
            storage (Storage): Storage implementation for persistence
        """
    
    def get_storage_key(self, turn_context: TurnContext):
        """
        Get storage key based on conversation ID.
        
        Args:
            turn_context (TurnContext): Current turn context
            
        Returns:
            str: Conversation-scoped storage key
        """

UserState

Manages user-scoped state that persists across conversations for a specific user. Data follows the user regardless of which conversation they're in.

class UserState(BotState):
    def __init__(self, storage):
        """
        Initialize user state.
        
        Args:
            storage (Storage): Storage implementation for persistence
        """
    
    def get_storage_key(self, turn_context: TurnContext):
        """
        Get storage key based on user ID.
        
        Args:
            turn_context (TurnContext): Current turn context
            
        Returns:
            str: User-scoped storage key
        """

PrivateConversationState

Manages private conversation state that is scoped to both user and conversation. Provides user-specific state within a particular conversation context.

class PrivateConversationState(BotState):
    def __init__(self, storage):
        """
        Initialize private conversation state.
        
        Args:
            storage (Storage): Storage implementation for persistence
        """
    
    def get_storage_key(self, turn_context: TurnContext):
        """
        Get storage key based on user and conversation ID.
        
        Args:
            turn_context (TurnContext): Current turn context
            
        Returns:
            str: Private conversation-scoped storage key
        """

StatePropertyAccessor

Provides type-safe access to individual properties within bot state, handling default values, property getting/setting, and deletion.

class StatePropertyAccessor:
    def __init__(self, state: BotState, name: str):
        """
        Initialize property accessor.
        
        Args:
            state (BotState): Parent state object
            name (str): Property name
        """
    
    async def get(self, turn_context: TurnContext, default_value_or_factory=None):
        """
        Get property value.
        
        Args:
            turn_context (TurnContext): Current turn context
            default_value_or_factory: Default value or factory function
            
        Returns:
            object: Property value or default
        """
    
    async def set(self, turn_context: TurnContext, value):
        """
        Set property value.
        
        Args:
            turn_context (TurnContext): Current turn context
            value: Value to set
        """
    
    async def delete(self, turn_context: TurnContext):
        """
        Delete property.
        
        Args:
            turn_context (TurnContext): Current turn context
        """

BotStateSet

Collection of bot states that can be managed together, providing convenient methods for loading and saving multiple state objects.

class BotStateSet:
    def __init__(self, *bot_states):
        """
        Initialize bot state set.
        
        Args:
            *bot_states: Variable number of BotState objects
        """
    
    def add(self, bot_state: BotState):
        """
        Add a bot state to the set.
        
        Args:
            bot_state (BotState): State to add
        """
    
    async def load_all(self, turn_context: TurnContext, force: bool = False):
        """
        Load all states in the set.
        
        Args:
            turn_context (TurnContext): Current turn context
            force (bool): Force reload even if already loaded
        """
    
    async def save_all_changes(self, turn_context: TurnContext, force: bool = False):
        """
        Save changes for all states in the set.
        
        Args:
            turn_context (TurnContext): Current turn context
            force (bool): Force save even if no changes detected
        """

StatePropertyInfo

Provides information about state properties for introspection and debugging purposes.

class StatePropertyInfo:
    def __init__(self, name: str, type_name: str = None):
        """
        Initialize property info.
        
        Args:
            name (str): Property name
            type_name (str, optional): Type name for the property
        """
        self.name = name
        self.type_name = type_name

Usage Examples

Basic State Setup

from botbuilder.core import (
    ActivityHandler, TurnContext, ConversationState, 
    UserState, MemoryStorage, MessageFactory
)

class StateBot(ActivityHandler):
    def __init__(self):
        # Create storage
        memory_storage = MemoryStorage()
        
        # Create state objects
        self.conversation_state = ConversationState(memory_storage)
        self.user_state = UserState(memory_storage)
        
        # Create property accessors
        self.user_profile_accessor = self.user_state.create_property("UserProfile")
        self.conversation_data_accessor = self.conversation_state.create_property("ConversationData")
    
    async def on_message_activity(self, turn_context: TurnContext):
        # Get user profile
        user_profile = await self.user_profile_accessor.get(
            turn_context, 
            lambda: {"name": None, "message_count": 0}
        )
        
        # Get conversation data
        conversation_data = await self.conversation_data_accessor.get(
            turn_context,
            lambda: {"turn_count": 0}
        )
        
        # Update data
        user_profile["message_count"] += 1
        conversation_data["turn_count"] += 1
        
        # Save changes
        await self.conversation_state.save_changes(turn_context)
        await self.user_state.save_changes(turn_context)
        
        # Send response
        reply = f"User messages: {user_profile['message_count']}, Turn: {conversation_data['turn_count']}"
        await turn_context.send_activity(MessageFactory.text(reply))

Complex State Management

class UserProfile:
    def __init__(self):
        self.name = None
        self.age = None
        self.preferences = {}
        self.last_activity = None

class ConversationData:
    def __init__(self):
        self.topic = None
        self.turn_count = 0
        self.participants = []

class AdvancedStateBot(ActivityHandler):
    def __init__(self):
        memory_storage = MemoryStorage()
        
        self.conversation_state = ConversationState(memory_storage)
        self.user_state = UserState(memory_storage)
        self.private_conversation_state = PrivateConversationState(memory_storage)
        
        # Create typed property accessors
        self.user_profile_accessor = self.user_state.create_property("UserProfile")
        self.conversation_data_accessor = self.conversation_state.create_property("ConversationData")
        self.private_data_accessor = self.private_conversation_state.create_property("PrivateData")
    
    async def on_message_activity(self, turn_context: TurnContext):
        # Get state data with default objects
        user_profile = await self.user_profile_accessor.get(turn_context, UserProfile)
        conversation_data = await self.conversation_data_accessor.get(turn_context, ConversationData)
        private_data = await self.private_data_accessor.get(turn_context, lambda: {"notes": []})
        
        # Update state based on message
        text = turn_context.activity.text.lower()
        
        if text.startswith("my name is"):
            user_profile.name = text[11:].strip()
            await turn_context.send_activity(MessageFactory.text(f"Nice to meet you, {user_profile.name}!"))
        
        elif text.startswith("topic:"):
            conversation_data.topic = text[6:].strip()
            await turn_context.send_activity(MessageFactory.text(f"Changed topic to: {conversation_data.topic}"))
        
        else:
            # Add private note
            private_data["notes"].append(f"Turn {conversation_data.turn_count}: {text}")
        
        # Update turn count
        conversation_data.turn_count += 1
        
        # Save all changes
        await self.user_state.save_changes(turn_context)
        await self.conversation_state.save_changes(turn_context)
        await self.private_conversation_state.save_changes(turn_context)

Using BotStateSet

class StateSetBot(ActivityHandler):
    def __init__(self):
        memory_storage = MemoryStorage()
        
        self.conversation_state = ConversationState(memory_storage)
        self.user_state = UserState(memory_storage)
        
        # Create state set for batch operations
        self.state_set = BotStateSet(self.conversation_state, self.user_state)
        
        self.user_name_accessor = self.user_state.create_property("UserName")
        self.turn_count_accessor = self.conversation_state.create_property("TurnCount")
    
    async def on_message_activity(self, turn_context: TurnContext):
        # Load all states at once
        await self.state_set.load_all(turn_context)
        
        # Work with state properties
        user_name = await self.user_name_accessor.get(turn_context, lambda: "Anonymous")
        turn_count = await self.turn_count_accessor.get(turn_context, lambda: 0)
        
        # Update data
        turn_count += 1
        await self.turn_count_accessor.set(turn_context, turn_count)
        
        if turn_context.activity.text.startswith("call me"):
            new_name = turn_context.activity.text[8:].strip()
            await self.user_name_accessor.set(turn_context, new_name)
            user_name = new_name
        
        # Save all states at once
        await self.state_set.save_all_changes(turn_context)
        
        reply = f"Hello {user_name}! This is turn #{turn_count}"
        await turn_context.send_activity(MessageFactory.text(reply))

State Clearing and Deletion

async def on_message_activity(self, turn_context: TurnContext):
    text = turn_context.activity.text.lower()
    
    if text == "reset user":
        # Clear user state
        await self.user_state.clear_state(turn_context)
        await self.user_state.save_changes(turn_context)
        await turn_context.send_activity(MessageFactory.text("User state cleared!"))
    
    elif text == "reset conversation":
        # Clear conversation state
        await self.conversation_state.clear_state(turn_context)
        await self.conversation_state.save_changes(turn_context)
        await turn_context.send_activity(MessageFactory.text("Conversation state cleared!"))
    
    elif text == "delete my data":
        # Delete specific property
        await self.user_profile_accessor.delete(turn_context)
        await self.user_state.save_changes(turn_context)
        await turn_context.send_activity(MessageFactory.text("User profile deleted!"))

Custom Default Value Factory

def create_default_user_profile():
    return {
        "name": "New User",
        "join_date": datetime.now().isoformat(),
        "settings": {
            "notifications": True,
            "theme": "light"
        }
    }

async def on_message_activity(self, turn_context: TurnContext):
    # Use factory function for complex default values
    user_profile = await self.user_profile_accessor.get(
        turn_context,
        create_default_user_profile
    )
    
    # Now user_profile is guaranteed to have the structure we need
    await turn_context.send_activity(
        MessageFactory.text(f"Welcome back, {user_profile['name']}!")
    )

Types

class PropertyManager:
    """Base class for managing properties."""
    pass

class StoreItem:
    """Base class for items stored in bot state."""
    def __init__(self):
        self.e_tag = "*"

Install with Tessl CLI