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

Middleware

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

Middleware pipeline for implementing cross-cutting concerns like logging, telemetry, typing indicators, and automatic state saving. Middleware components can inspect and modify activities as they flow through the bot.

Capabilities

Middleware Base Class

Abstract base class that defines the interface for all middleware components in the Bot Framework pipeline, providing the foundation for implementing cross-cutting concerns.

class Middleware:
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        """
        Process the turn. Must be implemented by derived classes.
        
        Args:
            turn_context (TurnContext): Current turn context
            next_middleware: Function to call the next middleware in pipeline
        """

MiddlewareSet

Collection that manages middleware components and provides the pipeline execution mechanism for processing activities through registered middleware.

class MiddlewareSet:
    def __init__(self):
        """Initialize empty middleware set."""
    
    def use(self, middleware):
        """
        Add middleware to the set.
        
        Args:
            middleware (Middleware): Middleware to add
            
        Returns:
            MiddlewareSet: Self for method chaining
        """
    
    async def receive_activity_with_status(self, turn_context: TurnContext, callback):
        """
        Execute middleware pipeline with status handling.
        
        Args:
            turn_context (TurnContext): Current turn context
            callback: Final callback to execute
            
        Returns:
            int: HTTP status code
        """
    
    async def receive_activity(self, turn_context: TurnContext, callback):
        """
        Execute middleware pipeline.
        
        Args:
            turn_context (TurnContext): Current turn context
            callback: Final callback to execute
        """

AutoSaveStateMiddleware

Automatically saves bot state after each turn completes, ensuring state changes are persisted without requiring manual save operations in bot logic.

class AutoSaveStateMiddleware(Middleware):
    def __init__(self, *bot_states):
        """
        Initialize auto-save state middleware.
        
        Args:
            *bot_states: Variable number of BotState objects to auto-save
        """
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        """
        Execute middleware and auto-save states.
        
        Args:
            turn_context (TurnContext): Current turn context
            next_middleware: Next middleware function
        """

ShowTypingMiddleware

Shows typing indicator to users while the bot is processing their message, providing better user experience by indicating bot activity.

class ShowTypingMiddleware(Middleware):
    def __init__(self, delay: float = 0.5, period: float = 2.0):
        """
        Initialize show typing middleware.
        
        Args:
            delay (float): Delay before showing typing indicator (seconds)
            period (float): How often to send typing indicator (seconds)
        """
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        """
        Execute middleware with typing indicators.
        
        Args:
            turn_context (TurnContext): Current turn context
            next_middleware: Next middleware function
        """

TelemetryLoggerMiddleware

Logs telemetry data for bot activities, enabling monitoring, analytics, and debugging of bot interactions and performance.

class TelemetryLoggerMiddleware(Middleware):
    def __init__(self, telemetry_client, log_personal_information: bool = False):
        """
        Initialize telemetry logger middleware.
        
        Args:
            telemetry_client (BotTelemetryClient): Telemetry client
            log_personal_information (bool): Whether to log PII
        """
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        """
        Execute middleware with telemetry logging.
        
        Args:
            turn_context (TurnContext): Current turn context
            next_middleware: Next middleware function
        """
    
    async def on_receive_activity(self, activity):
        """
        Log received activity.
        
        Args:
            activity (Activity): Received activity
        """
    
    async def on_send_activity(self, activity):
        """
        Log sent activity.
        
        Args:
            activity (Activity): Sent activity
        """
    
    async def on_update_activity(self, activity):
        """
        Log updated activity.
        
        Args:
            activity (Activity): Updated activity
        """
    
    async def on_delete_activity(self, activity):
        """
        Log deleted activity.
        
        Args:
            activity (Activity): Deleted activity
        """

RegisterClassMiddleware

Registers classes for dependency injection in the turn context, enabling access to services and utilities throughout the middleware pipeline and bot logic.

class RegisterClassMiddleware(Middleware):
    def __init__(self, classes_to_register: dict):
        """
        Initialize register class middleware.
        
        Args:
            classes_to_register (dict): Dictionary of class name to class instance
        """
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        """
        Execute middleware with class registration.
        
        Args:
            turn_context (TurnContext): Current turn context
            next_middleware: Next middleware function
        """

TranscriptLoggerMiddleware

Logs complete conversation transcripts for audit, compliance, and debugging purposes, storing the full conversation history.

class TranscriptLoggerMiddleware(Middleware):
    def __init__(self, transcript_logger):
        """
        Initialize transcript logger middleware.
        
        Args:
            transcript_logger (TranscriptLogger): Transcript logger implementation
        """
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        """
        Execute middleware with transcript logging.
        
        Args:
            turn_context (TurnContext): Current turn context
            next_middleware: Next middleware function
        """

AnonymousReceiveMiddleware

Wrapper that allows using anonymous functions or lambdas as middleware, providing a simple way to add custom middleware logic.

class AnonymousReceiveMiddleware(Middleware):
    def __init__(self, anonymous_method):
        """
        Initialize anonymous middleware.
        
        Args:
            anonymous_method: Function to execute as middleware
        """
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        """
        Execute anonymous middleware function.
        
        Args:
            turn_context (TurnContext): Current turn context
            next_middleware: Next middleware function
        """

Usage Examples

Basic Middleware Setup

from botbuilder.core import (
    BotFrameworkAdapter, MiddlewareSet, AutoSaveStateMiddleware,
    ShowTypingMiddleware, ConversationState, UserState, MemoryStorage
)

# Create adapter
adapter = BotFrameworkAdapter(settings)

# Create states
memory_storage = MemoryStorage()
conversation_state = ConversationState(memory_storage)
user_state = UserState(memory_storage)

# Add middleware to adapter
adapter.use(ShowTypingMiddleware(delay=0.5, period=2.0))
adapter.use(AutoSaveStateMiddleware(conversation_state, user_state))

Custom Middleware Implementation

from botbuilder.core import Middleware, TurnContext, MessageFactory

class LoggingMiddleware(Middleware):
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        # Log incoming activity
        print(f"Incoming: {turn_context.activity.type} - {turn_context.activity.text}")
        
        # Continue pipeline
        await next_middleware()
        
        # Log after processing
        print(f"Completed processing for turn")

class ProfanityFilterMiddleware(Middleware):
    def __init__(self, bad_words: list):
        self.bad_words = [word.lower() for word in bad_words]
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        if turn_context.activity.type == "message":
            text = turn_context.activity.text.lower()
            
            # Check for profanity
            if any(bad_word in text for bad_word in self.bad_words):
                await turn_context.send_activity(
                    MessageFactory.text("Please keep the conversation respectful.")
                )
                return  # Don't continue pipeline
        
        # Continue pipeline
        await next_middleware()

# Usage
adapter.use(LoggingMiddleware())
adapter.use(ProfanityFilterMiddleware(["badword1", "badword2"]))

Telemetry Middleware Setup

from botbuilder.core import TelemetryLoggerMiddleware, NullTelemetryClient

# Create telemetry client (use actual implementation in production)
telemetry_client = NullTelemetryClient()

# Create telemetry middleware
telemetry_middleware = TelemetryLoggerMiddleware(
    telemetry_client,
    log_personal_information=False  # Set to False for privacy
)

# Add to adapter
adapter.use(telemetry_middleware)

Anonymous Middleware

from botbuilder.core import AnonymousReceiveMiddleware

# Simple logging middleware using lambda
logging_middleware = AnonymousReceiveMiddleware(
    lambda turn_context, next_middleware: self.log_and_continue(turn_context, next_middleware)
)

async def log_and_continue(self, turn_context, next_middleware):
    print(f"Processing message: {turn_context.activity.text}")
    await next_middleware()
    print("Message processed")

adapter.use(logging_middleware)

# Or inline with async lambda (Python 3.5+)
adapter.use(AnonymousReceiveMiddleware(
    lambda turn_context, next_middleware: self.simple_log(turn_context, next_middleware)
))

Middleware with Service Registration

from botbuilder.core import RegisterClassMiddleware

class CustomService:
    def __init__(self):
        self.data = {}
    
    def get_data(self, key):
        return self.data.get(key)
    
    def set_data(self, key, value):
        self.data[key] = value

# Register service
custom_service = CustomService()
service_middleware = RegisterClassMiddleware({
    "CustomService": custom_service
})

adapter.use(service_middleware)

# Access in bot
class ServiceBot(ActivityHandler):
    async def on_message_activity(self, turn_context: TurnContext):
        # Get service from turn context
        custom_service = turn_context.services.get("CustomService")
        
        if custom_service:
            custom_service.set_data("last_message", turn_context.activity.text)
            await turn_context.send_activity(
                MessageFactory.text("Message saved to service!")
            )

Conditional Middleware

class ConditionalMiddleware(Middleware):
    def __init__(self, condition_func, target_middleware):
        self.condition_func = condition_func
        self.target_middleware = target_middleware
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        # Check condition
        if self.condition_func(turn_context):
            # Execute target middleware
            await self.target_middleware.on_turn(turn_context, next_middleware)
        else:
            # Skip target middleware
            await next_middleware()

# Usage - only show typing for long messages
def is_long_message(turn_context):
    return (turn_context.activity.type == "message" and 
            len(turn_context.activity.text or "") > 50)

conditional_typing = ConditionalMiddleware(
    is_long_message,
    ShowTypingMiddleware(delay=0.2, period=1.0)
)

adapter.use(conditional_typing)

Error Handling Middleware

class ErrorHandlingMiddleware(Middleware):
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        try:
            await next_middleware()
        except Exception as e:
            print(f"Error in bot: {e}")
            
            # Send error message to user
            await turn_context.send_activity(
                MessageFactory.text("Sorry, something went wrong. Please try again.")
            )
            
            # Log error for debugging
            # In production, you might want to send this to a logging service

# Add error handling as first middleware
adapter.use(ErrorHandlingMiddleware())

Performance Monitoring Middleware

import time
from botbuilder.core import Middleware

class PerformanceMiddleware(Middleware):
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        start_time = time.time()
        
        try:
            await next_middleware()
        finally:
            duration = time.time() - start_time
            print(f"Turn completed in {duration:.2f} seconds")
            
            # Log slow turns
            if duration > 2.0:
                print(f"SLOW TURN: {duration:.2f}s for message: {turn_context.activity.text}")

adapter.use(PerformanceMiddleware())

Middleware Ordering

# Order matters! Middleware executes in registration order
# Error handling should be first
adapter.use(ErrorHandlingMiddleware())

# Performance monitoring 
adapter.use(PerformanceMiddleware())

# Typing indicator
adapter.use(ShowTypingMiddleware())

# Service registration
adapter.use(RegisterClassMiddleware(services))

# Telemetry logging
adapter.use(TelemetryLoggerMiddleware(telemetry_client))

# Auto-save state (should be last)
adapter.use(AutoSaveStateMiddleware(conversation_state, user_state))

Types

class NextDelegate:
    """Delegate for calling next middleware in pipeline."""
    async def __call__(self):
        """Execute next middleware."""
        pass

class ReceiveActivityDelegate:
    """Delegate for receiving activities."""
    async def __call__(self, turn_context: TurnContext):
        """Process turn context."""
        pass

Install with Tessl CLI