MS Teams Helper to get user email address and inject messages to the conversation with Rasa bot. The inputs are dependent on Rasa Custom Botframework Connector: https://gist.github.com/nirajch/788a73c65a00bc575666671f83ef9827

MSTeamsHelper.py

import os
import logging
from typing import Dict, Text, Any, List, Union, Optional
from rasa_sdk import Action, Tracker]
import ruamel.yaml

import requests
import json
import datetime

from requests.auth import HTTPBasicAuth

logger = logging.getLogger(__name__)
vers = "vers: 0.1.0, date: Apr 2, 2020"
logger.debug(vers)

"""Get teams details from the yml file."""
msteams_config = ruamel.yaml.safe_load(open("msteams_credentials.yml", "r")) or {}
localmode = msteams_config.get("localmode", True)
logger.debug(f"Local mode: {localmode}")

MICROSOFT_OAUTH2_URL = "https://login.microsoftonline.com"
MICROSOFT_OAUTH2_PATH = "botframework.com/oauth2/v2.0/token"


class MSTeams:
    token_expiration_date = datetime.datetime.now()

    headers = None

    def __init__(self, tracker: Tracker) -> None:
        events = tracker.current_state()["events"]
        for e in events:
            if e["event"] == "user":
                self.conversation_id = e['metadata']['conversation_id']
                service_url = e['metadata']['service_url']
                break
        service_url = (
            f"{service_url}/" if not service_url.endswith("/") else service_url
        )
        self.client_id = msteams_config.get("client_id")
        self.client_secret = msteams_config.get("client_secret")
        self.service_url = f"{service_url}v3/"

    async def _get_headers(self) -> Optional[Dict[Text, Any]]:
        if MSTeams.token_expiration_date < datetime.datetime.now():
            uri = f"{MICROSOFT_OAUTH2_URL}/{MICROSOFT_OAUTH2_PATH}"
            grant_type = "client_credentials"
            scope = "https://api.botframework.com/.default"
            payload = {
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "grant_type": grant_type,
                "scope": scope,
            }

            token_response = requests.post(uri, data=payload)

            if token_response.ok:
                token_data = token_response.json()
                access_token = token_data["access_token"]
                token_expiration = token_data["expires_in"]

                delta = datetime.timedelta(seconds=int(token_expiration))
                MSTeams.token_expiration_date = datetime.datetime.now() + delta

                MSTeams.headers = {
                    "content-type": "application/json",
                    "Authorization": "Bearer %s" % access_token,
                }
                return MSTeams.headers
            else:
                logger.error("Could not get BotFramework token")
        else:
            return MSTeams.headers

    async def get_email(self) -> Optional[Text]:
        get_user_uri = f'{self.service_url}conversations/{self.conversation_id}/members'
        headers = await self._get_headers()
        get_user = requests.get(
            get_user_uri, headers=headers
        )

        if get_user.ok:
            user_response = json.loads(get_user.text)[0]
            email = user_response['email']
        else:
            logger.error(
                "Error trying to get email - BotFramework. Response: %s",
                get_user.text,
            )
        return email

    async def send(self, text: Text) -> None:
        data = {
            "type": "message",
            "conversation": {"id": self.conversation_id},
            "channelData": {"notification": {"alert": "true"}},
            "text": text,
        }

        send_message_uri = f"{self.service_url}conversations/{self.conversation_id}/activities"
        headers = await self._get_headers()
        post_send_message = requests.post(
            send_message_uri, headers=headers, data=json.dumps(data)
        )

        if not post_send_message.ok:
            logger.error(
                "Error trying to send BotFramework message. Response: %s",
                post_send_message.text,
            )