Django: build a Microsoft Teams bot

import json
import os
import sys
from http import HTTPStatus

from asgiref.sync import async_to_sync
from azure.core.exceptions import DeserializationError
from botbuilder.core import (
    BotFrameworkAdapter,
    BotFrameworkAdapterSettings,
    TurnContext,
)
from botbuilder.core.teams import TeamsActivityHandler
from botbuilder.schema import Activity, ActivityTypes, InvokeResponse
from django.conf import settings
from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.core.wsgi import get_wsgi_application
from django.urls import path


class BotHandler(TeamsActivityHandler):
    """
    Determines what to do for incoming events with per-category methods.

    https://learn.microsoft.com/en-us/microsoftteams/platform/bots/bot-basics?tabs=python
    """

    async def on_message_activity(self, turn_context: TurnContext) -> None:
        """
        Handle “message activity” events, which correspond to the bot being
        directly messaged.
        """
        if not turn_context.activity.conversation.is_group:
            # Respond to direct messages only.
            return await turn_context.send_activity(
                Activity(
                    type=ActivityTypes.message,
                    text_format="markdown",
                    text="Beep boop 🤖",
                )
            )


bot = BotHandler()

bot_adapter = BotFrameworkAdapter(
    BotFrameworkAdapterSettings(
        # Replace these with settings from environment variables in a real app.
        # None values allow requests from the Bot Framework Emulator.
        app_id=None,
        app_password=None,
    )
)


@async_to_sync
async def call_bot(activity: Activity, auth_header: str) -> InvokeResponse | None:
    """Call the bot to respond to an incoming activity."""
    return await bot_adapter.process_activity(
        activity,
        auth_header,
        bot.on_turn,
    )


# Single-file Django project per:
# https://adamj.eu/tech/2019/04/03/django-versus-flask-with-single-file-applications/

settings.configure(
    DEBUG=(os.environ.get("DEBUG", "") == "1"),
    # Disable host header validation
    ALLOWED_HOSTS=["*"],
    # Make this module the urlconf
    ROOT_URLCONF=__name__,
    # We aren't using any security features but Django requires a secret key
    SECRET_KEY="django-insecure-whatever",
)


@csrf_exempt
@require_POST
def webhook(request):
    """
    Respond to an event from Microsoft Teams.
    """
    if request.content_type != "application/json":
        return HttpResponse(status=HTTPStatus.UNSUPPORTED_MEDIA_TYPE)

    payload = json.loads(request.body)

    # React to the activity
    try:
        activity = Activity.deserialize(payload)
    except DeserializationError:
        return HttpResponse(status=HTTPStatus.BAD_REQUEST)

    auth_header = request.headers.get("authorization", "")
    try:
        invoke_response = call_bot(activity, auth_header)
    # Note: more more except blocks may be needed, per:
    # https://github.com/microsoft/botbuilder-python/blob/main/libraries/botbuilder-core/botbuilder/core/integration/aiohttp_channel_service_exception_middleware.py#L19
    except TypeError:
        response = HttpResponse(status=HTTPStatus.BAD_REQUEST)
    else:
        if invoke_response:
            response = JsonResponse(
                data=invoke_response.body, status=invoke_response.status
            )
        else:
            response = JsonResponse({})

    return response


urlpatterns = [
    path("bot/", webhook),
]

app = get_wsgi_application()

if __name__ == "__main__":
    from django.core.management import execute_from_command_line

    execute_from_command_line(sys.argv)

$ python example.py test
Found 4 test(s).
System check identified no issues (0 silenced).
....
----------------------------------------------------------------------
Ran 4 tests in 0.024s

OK

from http import HTTPStatus

import requests_mock
from django.test import SimpleTestCase


class BotTests(SimpleTestCase):
    def setUp(self):
        self.mock_requests = self.enterContext(requests_mock.Mocker())

    def test_incorrect_content_type(self):
        response = self.client.post(
            "/bot/",
            content_type="text/plain",
        )
        assert response.status_code == HTTPStatus.UNSUPPORTED_MEDIA_TYPE

    def test_post_non_dict(self):
        response = self.client.post(
            "/bot/",
            content_type="application/json",
            data=[],
        )
        assert response.status_code == HTTPStatus.BAD_REQUEST

    def test_post_empty_dict(self):
        response = self.client.post(
            "/bot/",
            content_type="application/json",
            data={},
        )
        assert response.status_code == HTTPStatus.BAD_REQUEST

    def test_post_message(self):
        self.mock_requests.post(
            "http://localhost:50096/v3/conversations/19%3Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa%40thread.tacv2%3Bmessageid%3D1111111111111/activities/bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
        )
        # Data based on a payload captured from Bot Framework Emulator
        payload = {
            "text": "Hi!",
            "textFormat": "plain",
            "type": "message",
            "channelId": "msteams",
            "from": {
                "id": "82d12900-783f-496d-9449-43dcd216666a",
                "name": "User",
                "role": "user",
            },
            "localTimestamp": "2024-09-04T12:05:50+01:00",
            "localTimezone": "Europe/London",
            "timestamp": "2024-09-04T12:05:50.281Z",
            "channelData": {
                "channel": {
                    "id": "19:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa@thread.tacv2",
                },
            },
            "conversation": {
                "id": "19:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa@thread.tacv2;messageid=1111111111111",
            },
            "id": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
            "recipient": {
                "id": "cccccccc-cccc-cccc-cccc-cccccccccccc",
                "name": "Bot",
                "role": "bot",
            },
            "serviceUrl": "http://localhost:50096",
        }

        response = self.client.post(
            "/bot/",
            content_type="application/json",
            data=payload,
        )

        assert response.status_code == HTTPStatus.OK
        assert len(self.mock_requests.request_history) == 1
        data = self.mock_requests.request_history[0].json()
        assert data["type"] == "message"
        assert data["text"] == "Beep boop 🤖"