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
bot-adapters.mddocs/
Bot Adapters
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.
Capabilities
BotFrameworkAdapter
Main adapter for Bot Framework integration that handles activity processing, authentication, OAuth token management, and connector client creation for communicating with Bot Framework channels.
class BotFrameworkAdapter:
def __init__(self, settings: BotFrameworkAdapterSettings):
"""
Initialize adapter with Bot Framework settings.
Args:
settings (BotFrameworkAdapterSettings): Configuration settings
"""
async def process_activity(self, req, auth_header: str, logic):
"""
Process incoming activity from Bot Framework channels.
Args:
req: HTTP request containing the activity
auth_header (str): Authorization header from the request
logic: Bot logic function to execute
Returns:
Response object for the HTTP framework
"""
async def send_activities(self, context: TurnContext, activities):
"""
Send multiple activities to the channel.
Args:
context (TurnContext): Turn context
activities (list): List of Activity objects to send
Returns:
list: List of ResourceResponse objects
"""
async def update_activity(self, context: TurnContext, activity):
"""
Update an existing activity.
Args:
context (TurnContext): Turn context
activity (Activity): Activity to update
Returns:
ResourceResponse: Response from the update operation
"""
async def delete_activity(self, context: TurnContext, reference):
"""
Delete an activity.
Args:
context (TurnContext): Turn context
reference (ConversationReference): Reference to activity to delete
Returns:
ResourceResponse: Response from the delete operation
"""
async def continue_conversation(self, reference, callback, bot_id=None, claims_identity=None, audience=None):
"""
Continue a conversation with a proactive message.
Args:
reference (ConversationReference): Conversation reference
callback: Callback function to execute
bot_id (str, optional): Bot ID
claims_identity (ClaimsIdentity, optional): Claims identity
audience (str, optional): Audience for the conversation
"""
async def get_user_token(self, context: TurnContext, connection_name: str, magic_code: str = None, oauth_app_credentials=None):
"""
Get OAuth token for user.
Args:
context (TurnContext): Turn context
connection_name (str): OAuth connection name
magic_code (str, optional): Magic code from OAuth provider
oauth_app_credentials (optional): OAuth app credentials
Returns:
TokenResponse: Token response or None if no token
"""
async def sign_out_user(self, context: TurnContext, connection_name: str, user_id: str = None, oauth_app_credentials=None):
"""
Sign out user from OAuth provider.
Args:
context (TurnContext): Turn context
connection_name (str): OAuth connection name
user_id (str, optional): User ID to sign out
oauth_app_credentials (optional): OAuth app credentials
"""
async def get_oauth_sign_in_link(self, context: TurnContext, connection_name: str, oauth_app_credentials=None):
"""
Get OAuth sign-in link.
Args:
context (TurnContext): Turn context
connection_name (str): OAuth connection name
oauth_app_credentials (optional): OAuth app credentials
Returns:
str: Sign-in URL
"""
async def create_connector_client(self, service_url: str, identity=None, audience: str = None):
"""
Create connector client for communicating with Bot Framework.
Args:
service_url (str): Service URL for the channel
identity (optional): Claims identity
audience (str, optional): Audience for the client
Returns:
ConnectorClient: Configured connector client
"""
async def exchange_token(self, turn_context: TurnContext, connection_name: str, user_id: str, exchange_request):
"""
Exchange token with OAuth provider.
Args:
turn_context (TurnContext): Turn context
connection_name (str): OAuth connection name
user_id (str): User ID
exchange_request: Token exchange request
Returns:
TokenResponse: Exchange response
"""BotFrameworkAdapterSettings
Configuration settings for BotFrameworkAdapter including app credentials, OAuth endpoints, and channel authentication settings.
class BotFrameworkAdapterSettings:
def __init__(self, app_id: str = None, app_password: str = None, channel_auth_tenant: str = None,
oauth_endpoint: str = None, open_id_metadata: str = None, channel_service: str = None,
validate_authority: bool = True, to_channel_from_bot_login_url: str = None,
to_channel_from_bot_oauth_scope: str = None, to_bot_from_channel_token_issuer: str = None,
to_bot_from_channel_open_id_metadata_url: str = None, to_bot_from_emulator_open_id_metadata_url: str = None,
caller_id: str = None, authority: str = None):
"""
Initialize Bot Framework adapter settings.
Args:
app_id (str, optional): Microsoft App ID
app_password (str, optional): Microsoft App Password
channel_auth_tenant (str, optional): Channel auth tenant
oauth_endpoint (str, optional): OAuth endpoint URL
open_id_metadata (str, optional): OpenID metadata URL
channel_service (str, optional): Channel service URL
validate_authority (bool): Whether to validate authority
to_channel_from_bot_login_url (str, optional): Login URL for bot-to-channel
to_channel_from_bot_oauth_scope (str, optional): OAuth scope for bot-to-channel
to_bot_from_channel_token_issuer (str, optional): Token issuer for channel-to-bot
to_bot_from_channel_open_id_metadata_url (str, optional): OpenID metadata for channel-to-bot
to_bot_from_emulator_open_id_metadata_url (str, optional): OpenID metadata for emulator
caller_id (str, optional): Caller ID
authority (str, optional): Authority URL
"""BotAdapter Base Class
Abstract base class that defines the interface for all bot adapters, providing core functionality for activity processing, middleware management, and conversation handling.
class BotAdapter:
# Constants for turn context services
BOT_IDENTITY_KEY: str = "BotIdentity"
BOT_OAUTH_SCOPE_KEY: str = "BotOAuthScope"
BOT_CONNECTOR_CLIENT_KEY: str = "ConnectorClient"
async def send_activities(self, context: TurnContext, activities):
"""
Send multiple activities. Must be implemented by derived classes.
Args:
context (TurnContext): Turn context
activities (list): Activities to send
Returns:
list: List of ResourceResponse objects
"""
async def update_activity(self, context: TurnContext, activity):
"""
Update an activity. Must be implemented by derived classes.
Args:
context (TurnContext): Turn context
activity (Activity): Activity to update
Returns:
ResourceResponse: Update response
"""
async def delete_activity(self, context: TurnContext, reference):
"""
Delete an activity. Must be implemented by derived classes.
Args:
context (TurnContext): Turn context
reference (ConversationReference): Activity reference
"""
def use(self, middleware):
"""
Register middleware with the adapter.
Args:
middleware (Middleware): Middleware to register
Returns:
BotAdapter: Self for method chaining
"""
async def continue_conversation(self, reference, callback, bot_id=None, claims_identity=None, audience=None):
"""
Continue a conversation.
Args:
reference (ConversationReference): Conversation reference
callback: Callback function
bot_id (str, optional): Bot ID
claims_identity (optional): Claims identity
audience (str, optional): Audience
"""
async def run_pipeline(self, context: TurnContext, callback):
"""
Run the middleware pipeline.
Args:
context (TurnContext): Turn context
callback: Bot logic callback
"""CloudAdapterBase
Abstract base class for cloud-based adapters that provides additional functionality for cloud deployment scenarios and streaming connections.
class CloudAdapterBase(BotAdapter):
async def process_activity(self, req, auth_header: str, logic):
"""
Process incoming activity. Must be implemented by derived classes.
Args:
req: HTTP request
auth_header (str): Authorization header
logic: Bot logic function
"""
async def create_connector_client(self, service_url: str, identity=None, audience: str = None):
"""
Create connector client. Must be implemented by derived classes.
Args:
service_url (str): Service URL
identity (optional): Claims identity
audience (str, optional): Audience
Returns:
ConnectorClient: Connector client
"""Usage Examples
Basic Adapter Setup
from botbuilder.core import BotFrameworkAdapter, BotFrameworkAdapterSettings
from botbuilder.core import ActivityHandler
# Create adapter settings
settings = BotFrameworkAdapterSettings(
app_id="your-app-id",
app_password="your-app-password"
)
# Create adapter
adapter = BotFrameworkAdapter(settings)
# Create bot
class EchoBot(ActivityHandler):
async def on_message_activity(self, turn_context):
await turn_context.send_activity(f"You said: {turn_context.activity.text}")
bot = EchoBot()Processing Activities with Error Handling
async def process_request(req):
try:
# Extract auth header
auth_header = req.headers.get("Authorization", "")
# Process the activity
response = await adapter.process_activity(req, auth_header, bot.on_turn)
return response
except Exception as e:
print(f"Error processing activity: {e}")
return {"status": 500, "body": "Internal server error"}Middleware Registration
from botbuilder.core import AutoSaveStateMiddleware, ShowTypingMiddleware
# Add middleware to adapter
adapter.use(ShowTypingMiddleware(delay=0.5, period=2.0))
adapter.use(AutoSaveStateMiddleware([conversation_state, user_state]))Proactive Messaging
async def send_proactive_message(conversation_reference, message_text):
async def proactive_callback(turn_context):
await turn_context.send_activity(MessageFactory.text(message_text))
await adapter.continue_conversation(
conversation_reference,
proactive_callback
)Types
class ConversationReference:
"""Reference to a conversation for proactive messaging."""
activity_id: str
user: ChannelAccount
bot: ChannelAccount
conversation: ConversationAccount
channel_id: str
service_url: str
class ResourceResponse:
"""Response from Bot Framework operations."""
id: str
class TokenResponse:
"""OAuth token response."""
channel_id: str
connection_name: str
token: str
expiration: str
class ConnectorClient:
"""Client for Bot Framework Connector API."""
passInstall with Tessl CLI
npx tessl i tessl/pypi-botbuilder-core