CtrlK
BlogDocsLog inGet started
Tessl Logo

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

telemetry-logging.mddocs/

Telemetry & Logging

Telemetry and logging functionality for monitoring bot performance, usage analytics, and debugging. Includes telemetry clients, logging middleware, and transcript storage.

Capabilities

BotTelemetryClient

Abstract base class for telemetry clients that defines the interface for tracking bot events, exceptions, and custom metrics for monitoring and analytics.

class BotTelemetryClient:
    def track_event(self, name: str, properties: dict = None, measurements: dict = None):
        """
        Track custom telemetry events.
        
        Args:
            name (str): Event name
            properties (dict, optional): Event properties
            measurements (dict, optional): Event measurements
        """
    
    def track_exception(self, exception, properties: dict = None, measurements: dict = None):
        """
        Track exceptions for monitoring.
        
        Args:
            exception (Exception): Exception to track
            properties (dict, optional): Additional properties
            measurements (dict, optional): Additional measurements
        """
    
    def track_dependency(self, name: str, data: str, type_name: str, target: str, duration: int, success: bool, result_code: str = None, properties: dict = None, measurements: dict = None):
        """Track dependency calls."""
    
    def track_trace(self, name: str, properties: dict = None, severity=None):
        """Track trace messages."""
    
    def track_page_view(self, name: str, url: str = None, duration: int = None, properties: dict = None, measurements: dict = None):
        """Track page views."""
    
    def flush(self):
        """Flush telemetry data."""

NullTelemetryClient

No-operation telemetry client that provides a default implementation for scenarios where telemetry is not needed or configured.

class NullTelemetryClient(BotTelemetryClient):
    def track_event(self, name: str, properties: dict = None, measurements: dict = None):
        """No-op track event implementation."""
    
    def track_exception(self, exception, properties: dict = None, measurements: dict = None):
        """No-op track exception implementation."""
    
    def track_dependency(self, name: str, data: str, type_name: str, target: str, duration: int, success: bool, result_code: str = None, properties: dict = None, measurements: dict = None):
        """No-op track dependency implementation."""
    
    def track_trace(self, name: str, properties: dict = None, severity=None):
        """No-op track trace implementation."""
    
    def track_page_view(self, name: str, url: str = None, duration: int = None, properties: dict = None, measurements: dict = None):
        """No-op track page view implementation."""
    
    def flush(self):
        """No-op flush implementation."""

TranscriptLogger

Abstract base class for transcript logging that defines the interface for storing conversation transcripts for audit and debugging purposes.

class TranscriptLogger:
    async def log_activity(self, activity):
        """
        Log activity to transcript store.
        
        Args:
            activity (Activity): Activity to log
        """
    
    async def get_transcript_activities(self, channel_id: str, conversation_id: str, continuation_token: str = None, start_date = None):
        """
        Get transcript activities from store.
        
        Args:
            channel_id (str): Channel ID
            conversation_id (str): Conversation ID
            continuation_token (str, optional): Continuation token for paging
            start_date (optional): Start date filter
            
        Returns:
            PagedResult: Paged transcript activities
        """
    
    async def list_transcripts(self, channel_id: str, continuation_token: str = None):
        """
        List available transcripts.
        
        Args:
            channel_id (str): Channel ID
            continuation_token (str, optional): Continuation token for paging
            
        Returns:
            PagedResult: Paged transcript summaries
        """
    
    async def delete_transcript(self, channel_id: str, conversation_id: str):
        """
        Delete transcript.
        
        Args:
            channel_id (str): Channel ID
            conversation_id (str): Conversation ID
        """

MemoryTranscriptStore

In-memory implementation of transcript logger for development and testing scenarios where persistent transcript storage is not required.

class MemoryTranscriptStore(TranscriptLogger):
    def __init__(self):
        """Initialize memory transcript store."""
    
    async def log_activity(self, activity):
        """Log activity to memory store."""
    
    async def get_transcript_activities(self, channel_id: str, conversation_id: str, continuation_token: str = None, start_date = None):
        """Get transcript activities from memory."""
    
    async def list_transcripts(self, channel_id: str, continuation_token: str = None):
        """List transcripts from memory."""
    
    async def delete_transcript(self, channel_id: str, conversation_id: str):
        """Delete transcript from memory."""

Severity Enum

Enumeration of telemetry severity levels for categorizing telemetry events and traces.

class Severity:
    VERBOSE = 0
    INFORMATION = 1
    WARNING = 2
    ERROR = 3
    CRITICAL = 4

Telemetry Constants

Constants used throughout the telemetry system for standard property names and event types.

class TelemetryConstants:
    ACTIVITY_ID_PROPERTY = "activityId"
    ACTIVITY_TYPE_PROPERTY = "activityType"
    CHANNEL_ID_PROPERTY = "channelId"
    CONVERSATION_ID_PROPERTY = "conversationId"
    CONVERSATION_NAME_PROPERTY = "conversationName"
    DIALOG_ID_PROPERTY = "dialogId"
    FROM_ID_PROPERTY = "fromId"
    FROM_NAME_PROPERTY = "fromName"
    LOCALE_PROPERTY = "locale"
    RECIPIENT_ID_PROPERTY = "recipientId"
    RECIPIENT_NAME_PROPERTY = "recipientName"
    REPLY_ACTIVITY_ID_PROPERTY = "replyActivityId"
    TEXT_PROPERTY = "text"
    SPEAK_PROPERTY = "speak"
    USER_ID_PROPERTY = "userId"

Usage Examples

Basic Telemetry Setup

from botbuilder.core import ActivityHandler, TurnContext, BotTelemetryClient

class TelemetryBot(ActivityHandler):
    def __init__(self, telemetry_client: BotTelemetryClient):
        self.telemetry_client = telemetry_client
    
    async def on_message_activity(self, turn_context: TurnContext):
        # Track message received event
        self.telemetry_client.track_event(
            "MessageReceived",
            properties={
                "conversationId": turn_context.activity.conversation.id,
                "userId": turn_context.activity.from_property.id,
                "text": turn_context.activity.text[:100]  # Truncate for privacy
            }
        )
        
        try:
            # Process message
            response = await self.process_message(turn_context.activity.text)
            await turn_context.send_activity(MessageFactory.text(response))
            
            # Track successful response
            self.telemetry_client.track_event(
                "MessageProcessed",
                properties={
                    "conversationId": turn_context.activity.conversation.id,
                    "success": "true"
                }
            )
            
        except Exception as e:
            # Track exceptions
            self.telemetry_client.track_exception(
                e,
                properties={
                    "conversationId": turn_context.activity.conversation.id,
                    "userId": turn_context.activity.from_property.id
                }
            )
            await turn_context.send_activity(MessageFactory.text("Sorry, an error occurred"))
    
    async def process_message(self, text: str):
        # Simulate some processing
        import time
        start_time = time.time()
        
        # Track dependency call (e.g., to external API)
        try:
            result = await self.call_external_api(text)
            duration = int((time.time() - start_time) * 1000)
            
            self.telemetry_client.track_dependency(
                name="ExternalAPI",
                data=f"ProcessText: {text[:50]}",
                type_name="HTTP",
                target="api.example.com",
                duration=duration,
                success=True,
                result_code="200"
            )
            
            return result
            
        except Exception as e:
            duration = int((time.time() - start_time) * 1000)
            
            self.telemetry_client.track_dependency(
                name="ExternalAPI",
                data=f"ProcessText: {text[:50]}",
                type_name="HTTP", 
                target="api.example.com",
                duration=duration,
                success=False,
                result_code="500"
            )
            raise

Transcript Logging

from botbuilder.core import MemoryTranscriptStore, TranscriptLoggerMiddleware

class TranscriptBot(ActivityHandler):
    def __init__(self):
        # Create transcript store
        self.transcript_store = MemoryTranscriptStore()
        
        # Add transcript logging middleware to adapter
        self.adapter.use(TranscriptLoggerMiddleware(self.transcript_store))
    
    async def on_message_activity(self, turn_context: TurnContext):
        # Regular bot processing - transcript logging happens automatically
        await turn_context.send_activity(MessageFactory.text(f"You said: {turn_context.activity.text}"))
    
    async def get_conversation_history(self, channel_id: str, conversation_id: str):
        """Get conversation transcript."""
        transcript = await self.transcript_store.get_transcript_activities(
            channel_id, 
            conversation_id
        )
        return transcript.items

Custom Telemetry Client

import logging

class CustomTelemetryClient(BotTelemetryClient):
    def __init__(self, logger=None):
        self.logger = logger or logging.getLogger(__name__)
    
    def track_event(self, name: str, properties: dict = None, measurements: dict = None):
        """Custom event tracking implementation."""
        log_data = {
            "event": name,
            "properties": properties or {},
            "measurements": measurements or {}
        }
        self.logger.info(f"Event: {log_data}")
        
        # Send to your analytics service
        # await self.send_to_analytics_service(log_data)
    
    def track_exception(self, exception, properties: dict = None, measurements: dict = None):
        """Custom exception tracking."""
        log_data = {
            "exception": str(exception),
            "type": type(exception).__name__,
            "properties": properties or {}
        }
        self.logger.error(f"Exception: {log_data}")
        
        # Send to your error tracking service
        # await self.send_to_error_service(log_data)
    
    def track_trace(self, name: str, properties: dict = None, severity=None):
        """Custom trace tracking."""
        level = logging.INFO
        if severity == Severity.WARNING:
            level = logging.WARNING
        elif severity == Severity.ERROR:
            level = logging.ERROR
        elif severity == Severity.CRITICAL:
            level = logging.CRITICAL
        
        self.logger.log(level, f"Trace: {name}", extra=properties or {})
    
    def flush(self):
        """Flush telemetry data."""
        # Implement flushing logic for your telemetry backend
        pass

Performance Metrics

import time
from botbuilder.core import Middleware

class PerformanceTelemetryMiddleware(Middleware):
    def __init__(self, telemetry_client: BotTelemetryClient):
        self.telemetry_client = telemetry_client
    
    async def on_turn(self, turn_context: TurnContext, next_middleware):
        start_time = time.time()
        
        try:
            await next_middleware()
            
            # Track successful turn
            duration = (time.time() - start_time) * 1000
            self.telemetry_client.track_event(
                "TurnCompleted",
                properties={
                    "conversationId": turn_context.activity.conversation.id,
                    "activityType": turn_context.activity.type
                },
                measurements={
                    "durationMs": duration
                }
            )
            
        except Exception as e:
            # Track failed turn
            duration = (time.time() - start_time) * 1000
            self.telemetry_client.track_event(
                "TurnFailed",
                properties={
                    "conversationId": turn_context.activity.conversation.id,
                    "error": str(e)
                },
                measurements={
                    "durationMs": duration
                }
            )
            raise

Types

class PagedResult:
    """Paged result for transcript queries."""
    def __init__(self):
        self.items: list = []
        self.continuation_token: str = None

class TranscriptInfo:
    """Information about a transcript."""
    def __init__(self):
        self.channel_id: str = None
        self.id: str = None
        self.created: str = None

Install with Tessl CLI

npx tessl i tessl/pypi-botbuilder-core

docs

activity-handling.md

bot-adapters.md

index.md

message-factories.md

middleware.md

oauth-authentication.md

state-management.md

storage.md

telemetry-logging.md

testing-utilities.md

turn-context.md

tile.json