Microsoft Bot Framework Bot Builder core functionality for building conversational AI bots and chatbots in Python.
npx @tessl/cli install tessl/pypi-botbuilder-core@4.17.0Microsoft Bot Framework core functionality for building conversational AI bots and chatbots in Python. This package provides the foundational components for activity handling, conversation state management, middleware support, bot adapters, and turn context management, enabling developers to build sophisticated enterprise-grade conversational experiences.
pip install botbuilder-corefrom botbuilder.core import (
ActivityHandler, Bot, BotFrameworkAdapter, TurnContext,
MemoryStorage, ConversationState, UserState, MessageFactory, CardFactory
)Common patterns:
# Bot and activity handling
from botbuilder.core import Bot, ActivityHandler, TurnContext
# Adapters and settings
from botbuilder.core import BotFrameworkAdapter, BotFrameworkAdapterSettings
# State management
from botbuilder.core import ConversationState, UserState, MemoryStorage, BotState
# Message creation
from botbuilder.core import MessageFactory, CardFactory
# Middleware
from botbuilder.core import AutoSaveStateMiddleware, ShowTypingMiddleware
# OAuth and authentication
from botbuilder.core import ExtendedUserTokenProvider, UserTokenProvider
# Utilities
from botbuilder.core import BotAssert, ComponentRegistrationfrom botbuilder.core import ActivityHandler, TurnContext, MessageFactory
class EchoBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
# Echo back what the user said
reply_text = f"You said: {turn_context.activity.text}"
await turn_context.send_activity(MessageFactory.text(reply_text))
async def on_members_added_activity(self, members_added, turn_context: TurnContext):
for member in members_added:
if member.id != turn_context.activity.recipient.id:
await turn_context.send_activity(MessageFactory.text("Welcome!"))
# Usage with adapter
from botbuilder.core import BotFrameworkAdapter, BotFrameworkAdapterSettings
# Create adapter
settings = BotFrameworkAdapterSettings(app_id="", app_password="")
adapter = BotFrameworkAdapter(settings)
# Create bot
bot = EchoBot()
# Process incoming activities (typically in web framework)
async def process_request(req):
auth_header = req.headers.get("Authorization", "")
response = await adapter.process_activity(req, auth_header, bot.on_turn)
return responseThe Bot Framework SDK follows a layered architecture:
This design enables building scalable bots that handle complex conversation flows, integrate with various channels, and maintain context across multiple turns.
Abstract base class for all bots that defines the core contract for processing activities. All bot implementations must inherit from this class and implement the on_turn method.
class Bot:
"""Abstract base class for bots."""
async def on_turn(self, context: TurnContext):
"""Handle an incoming activity."""Core functionality for processing bot activities including messages, member additions, invokes, and other Bot Framework activity types. The ActivityHandler provides event-driven methods for handling different scenarios in conversational flows.
class ActivityHandler(Bot):
async def on_turn(self, turn_context: TurnContext):
"""Main entry point for processing activities."""
async def on_message_activity(self, turn_context: TurnContext):
"""Handle message activities from users."""
async def on_members_added_activity(self, members_added, turn_context: TurnContext):
"""Handle when members are added to the conversation."""
async def on_members_removed_activity(self, members_removed, turn_context: TurnContext):
"""Handle when members are removed from the conversation."""
async def on_invoke_activity(self, turn_context: TurnContext):
"""Handle invoke activities (typically from cards or skills)."""
async def on_message_reaction_activity(self, turn_context: TurnContext):
"""Handle message reaction activities."""
async def on_event_activity(self, turn_context: TurnContext):
"""Handle event activities."""Adapter classes that connect bot logic to various communication channels and handle the low-level details of activity processing, authentication, and response management. Includes the main BotFrameworkAdapter and specialized adapters.
class BotFrameworkAdapter:
def __init__(self, settings: BotFrameworkAdapterSettings):
"""Initialize adapter with Bot Framework settings."""
async def process_activity(self, req, auth_header: str, logic):
"""Process incoming activity from Bot Framework channels."""
async def send_activities(self, context: TurnContext, activities):
"""Send multiple activities to the channel."""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.
class TurnContext:
activity: Activity
responded: bool
async def send_activity(self, activity_or_text, speak=None, input_hint=None):
"""Send a single activity response."""
async def send_activities(self, activities):
"""Send multiple activities in sequence."""
async def update_activity(self, activity):
"""Update an existing activity."""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.
class ConversationState(BotState):
"""Manages conversation-scoped state."""
class UserState(BotState):
"""Manages user-scoped state."""
class BotState:
def create_property(self, name: str):
"""Create a state property accessor."""
async def load(self, turn_context: TurnContext, force: bool = False):
"""Load state from storage."""
async def save_changes(self, turn_context: TurnContext, force: bool = False):
"""Save state changes to storage."""Storage implementations for persisting bot state and data. Includes memory storage for development and testing, plus abstract base classes for implementing custom storage solutions.
class MemoryStorage(Storage):
"""In-memory storage implementation for development and testing."""
class Storage:
async def read(self, keys):
"""Read items from storage."""
async def write(self, changes):
"""Write items to storage."""
async def delete(self, keys):
"""Delete items from storage."""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.
class AutoSaveStateMiddleware(Middleware):
"""Automatically saves state after each turn."""
class ShowTypingMiddleware(Middleware):
"""Shows typing indicator during processing."""
class TelemetryLoggerMiddleware(Middleware):
"""Logs telemetry data for analytics."""Factory classes for creating rich message activities and card attachments. Simplifies the creation of formatted responses, cards, and other rich content types supported by the Bot Framework.
class MessageFactory:
@staticmethod
def text(text: str, speak: str = None, input_hint=None):
"""Create a text message activity."""
@staticmethod
def attachment(attachment):
"""Create a message with an attachment."""
class CardFactory:
@staticmethod
def adaptive_card(card):
"""Create an Adaptive Card attachment."""
@staticmethod
def hero_card(title: str, subtitle: str = None, text: str = None):
"""Create a Hero Card attachment."""OAuth token management and user authentication functionality. Includes token providers, authentication utilities, and integration with Bot Framework authentication services.
class ExtendedUserTokenProvider:
async def get_user_token(self, turn_context: TurnContext, connection_name: str):
"""Get OAuth token for user."""
async def sign_out_user(self, turn_context: TurnContext, connection_name: str):
"""Sign out user from OAuth provider."""Telemetry and logging functionality for monitoring bot performance, usage analytics, and debugging. Includes telemetry clients, logging middleware, and transcript storage.
class BotTelemetryClient:
def track_event(self, name: str, properties: dict = None):
"""Track custom telemetry events."""
def track_exception(self, exception, properties: dict = None):
"""Track exceptions for monitoring."""
class TranscriptLoggerMiddleware(Middleware):
"""Logs conversation transcripts."""Testing utilities for unit testing bots including test adapters, test flows, and assertion helpers. Enables comprehensive testing of bot logic without requiring actual Bot Framework channels.
class TestAdapter(BotAdapter):
"""Test adapter for unit testing bots."""
class TestFlow:
def test(self, user_says: str, expected: str):
"""Test a single turn of conversation."""
async def start_test(self):
"""Start a test conversation flow."""Cloud-based adapter implementations and advanced components for enterprise scenarios including cloud channel service handlers, component registration systems, and utility classes.
class CloudAdapterBase(BotAdapter):
"""Base class for cloud-based bot adapters."""
class CloudChannelServiceHandler:
"""Handler for cloud channel service operations."""
class ComponentRegistration:
"""Component registration system for dependency injection."""
@staticmethod
def get_components():
"""Get all registered components."""
@staticmethod
def add(component_registration):
"""Add a component registration."""
class BotAssert:
"""Assertion utilities for validating bot objects."""
@staticmethod
def activity_not_none(activity: Activity):
"""Assert activity is not None."""
@staticmethod
def context_not_none(turn_context: TurnContext):
"""Assert turn context is not None."""class Activity:
"""Represents a Bot Framework activity."""
type: str
text: str
from_property: ChannelAccount
recipient: ChannelAccount
conversation: ConversationAccount
class ChannelAccount:
"""Represents a channel account (user or bot)."""
id: str
name: str
class ConversationAccount:
"""Represents a conversation."""
id: str
name: str
is_group: bool
class ResourceResponse:
"""Response from activity operations."""
id: str
class InvokeResponse:
"""Response for invoke activities."""
status: int
body: object