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

OAuth & Authentication

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

OAuth token management and user authentication functionality. Includes token providers, authentication utilities, and integration with Bot Framework authentication services.

Capabilities

ExtendedUserTokenProvider

Extended user token provider interface that provides comprehensive OAuth token management capabilities including token retrieval, sign-out, and token exchange functionality.

class ExtendedUserTokenProvider:
    async def get_user_token(self, turn_context: TurnContext, connection_name: str, magic_code: str = None, oauth_app_credentials=None):
        """
        Get OAuth token for user.
        
        Args:
            turn_context (TurnContext): Current turn context
            connection_name (str): OAuth connection name
            magic_code (str, optional): Magic code from OAuth flow
            oauth_app_credentials (optional): OAuth app credentials
            
        Returns:
            TokenResponse: Token response or None
        """
    
    async def sign_out_user(self, turn_context: TurnContext, connection_name: str, user_id: str = None, oauth_app_credentials=None):
        """
        Sign out user from OAuth provider.
        
        Args:
            turn_context (TurnContext): Current 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, turn_context: TurnContext, connection_name: str, oauth_app_credentials=None, final_redirect: str = None):
        """
        Get OAuth sign-in link.
        
        Args:
            turn_context (TurnContext): Current turn context
            connection_name (str): OAuth connection name
            oauth_app_credentials (optional): OAuth app credentials
            final_redirect (str, optional): Final redirect URL
            
        Returns:
            str: OAuth sign-in URL
        """
    
    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): Current turn context
            connection_name (str): OAuth connection name
            user_id (str): User ID
            exchange_request: Token exchange request
            
        Returns:
            TokenResponse: Exchange response
        """

UserTokenProvider

Base user token provider interface that defines the core OAuth functionality for token management in bot applications.

class UserTokenProvider:
    async def get_user_token(self, turn_context: TurnContext, connection_name: str, magic_code: str = None):
        """Get user token from OAuth provider."""
    
    async def sign_out_user(self, turn_context: TurnContext, connection_name: str, user_id: str = None):
        """Sign out user from OAuth provider."""
    
    async def get_oauth_sign_in_link(self, turn_context: TurnContext, connection_name: str):
        """Get OAuth sign-in link."""

ConnectorClientBuilder

Abstract base class for building connector clients with authentication support for Bot Framework communication.

class ConnectorClientBuilder:
    async def create_connector_client(self, service_url: str, identity=None, audience: str = None):
        """
        Create connector client with authentication.
        
        Args:
            service_url (str): Service URL
            identity (optional): Claims identity
            audience (str, optional): Target audience
            
        Returns:
            ConnectorClient: Authenticated connector client
        """

Usage Examples

Basic OAuth Flow

from botbuilder.core import ActivityHandler, TurnContext, MessageFactory, CardFactory

class OAuthBot(ActivityHandler):
    def __init__(self, connection_name: str):
        self.connection_name = connection_name
    
    async def on_message_activity(self, turn_context: TurnContext):
        text = turn_context.activity.text.lower()
        
        if text in ["login", "signin"]:
            await self.send_oauth_card(turn_context)
        elif text in ["logout", "signout"]:
            await self.sign_out_user(turn_context)
        elif text == "profile":
            await self.show_user_profile(turn_context)
        else:
            await turn_context.send_activity(MessageFactory.text("Say 'login' to authenticate"))
    
    async def on_token_response_event(self, turn_context: TurnContext):
        # Handle successful OAuth token response
        token_response = turn_context.activity.value
        if token_response and token_response.get("token"):
            await turn_context.send_activity(MessageFactory.text("Authentication successful!"))
            await self.show_user_profile(turn_context)
        else:
            await turn_context.send_activity(MessageFactory.text("Authentication failed"))
    
    async def send_oauth_card(self, turn_context: TurnContext):
        # Create OAuth card
        card = CardFactory.oauth_card(
            connection_name=self.connection_name,
            title="Please sign in",
            text="Click below to sign in with your account"
        )
        
        reply = MessageFactory.attachment(card)
        await turn_context.send_activity(reply)
    
    async def sign_out_user(self, turn_context: TurnContext):
        # Get the adapter (assumes it implements ExtendedUserTokenProvider)
        adapter = turn_context.adapter
        
        await adapter.sign_out_user(turn_context, self.connection_name)
        await turn_context.send_activity(MessageFactory.text("You have been signed out"))
    
    async def show_user_profile(self, turn_context: TurnContext):
        # Get user token
        adapter = turn_context.adapter
        token_response = await adapter.get_user_token(turn_context, self.connection_name)
        
        if token_response:
            # Use token to call external API (example with Microsoft Graph)
            user_info = await self.get_user_info_from_graph(token_response.token)
            await turn_context.send_activity(
                MessageFactory.text(f"Hello {user_info.get('displayName', 'User')}!")
            )
        else:
            await self.send_oauth_card(turn_context)
    
    async def get_user_info_from_graph(self, token: str):
        # Example: Call Microsoft Graph API
        import aiohttp
        
        headers = {"Authorization": f"Bearer {token}"}
        async with aiohttp.ClientSession() as session:
            async with session.get("https://graph.microsoft.com/v1.0/me", headers=headers) as response:
                if response.status == 200:
                    return await response.json()
        return {}

Magic Code Flow

async def on_message_activity(self, turn_context: TurnContext):
    text = turn_context.activity.text
    
    # Check if message contains magic code (6-digit number)
    if text.isdigit() and len(text) == 6:
        await self.process_magic_code(turn_context, text)
    else:
        await self.send_oauth_card(turn_context)

async def process_magic_code(self, turn_context: TurnContext, magic_code: str):
    adapter = turn_context.adapter
    
    # Try to get token using magic code
    token_response = await adapter.get_user_token(
        turn_context, 
        self.connection_name, 
        magic_code=magic_code
    )
    
    if token_response:
        await turn_context.send_activity(MessageFactory.text("Authentication successful with magic code!"))
        await self.show_user_profile(turn_context)
    else:
        await turn_context.send_activity(MessageFactory.text("Invalid magic code. Please try again."))
        await self.send_oauth_card(turn_context)

Token Exchange

async def handle_token_exchange(self, turn_context: TurnContext, sso_token: str):
    adapter = turn_context.adapter
    
    # Create token exchange request
    exchange_request = {
        "token": sso_token,
        "uri": "https://graph.microsoft.com/.default"
    }
    
    # Exchange SSO token for access token
    token_response = await adapter.exchange_token(
        turn_context,
        self.connection_name,
        turn_context.activity.from_property.id,
        exchange_request
    )
    
    if token_response:
        await turn_context.send_activity(MessageFactory.text("Token exchange successful!"))
        # Use the exchanged token
        await self.use_access_token(token_response.token)
    else:
        await turn_context.send_activity(MessageFactory.text("Token exchange failed"))
        await self.send_oauth_card(turn_context)

Types

class TokenResponse:
    """OAuth token response."""
    def __init__(self):
        self.channel_id: str = None
        self.connection_name: str = None
        self.token: str = None
        self.expiration: str = None

class TokenExchangeRequest:
    """Token exchange request."""
    def __init__(self):
        self.uri: str = None
        self.token: str = None

class OAuthCard:
    """OAuth card for authentication."""
    def __init__(self):
        self.text: str = None
        self.connection_name: str = None
        self.buttons: list = None

Install with Tessl CLI