feat(sample): Simple bot integration for Microsoft Teams

app.py

@@ -0,0 +1,21 @@
+from botbuilder.core.integration import aiohttp_error_middleware
+from botbuilder.integration.aiohttp import CloudAdapter, ConfigurationBotFrameworkAuthentication
+
+from aiohttp import web
+from aiohttp.web import Application
+
+from bot import WebhookBot
+from handler import BotHandler
+from config import DefaultConfig
+
+CONFIG = DefaultConfig()
+
+ADAPTER = CloudAdapter(ConfigurationBotFrameworkAuthentication(CONFIG))
+
+BOT = WebhookBot(CONFIG)
+BOT_HANDLER = BotHandler(ADAPTER, BOT, CONFIG)
+
+APP = Application(middlewares=[aiohttp_error_middleware])
+APP.router.add_post("/api/messages", BOT_HANDLER.messages)
+
+web.run_app(APP, host="0.0.0.0", port=CONFIG.PORT)

bot.py

@@ -0,0 +1,42 @@
+from typing import Tuple
+
+from botbuilder.core import ActivityHandler, TurnContext
+from botbuilder.schema import ChannelAccount
+
+from config import DefaultConfig
+
+class WebhookBot(ActivityHandler):
+    """Webhook bot for Microsoft Teams."""
+    
+    def __init__(self, config: DefaultConfig) -> None:
+        super().__init__()
+        self.command_prefix = DefaultConfig.COMMAND_PREFIX
+        
+    def is_command(self, text: str) -> tuple[bool, str]:    
+        """Check if the text is a command."""
+        if text.startswith(self.command_prefix) and len(text) > len(self.command_prefix):
+            return (True, text[len(self.command_prefix):])
+        return (False, text)    
+    
+    async def on_command(self, command: str, turn_context: TurnContext) -> None:
+        """Handle the command."""
+        await turn_context.send_activity(f"Command {command} received.")
+        return
+        
+    async def on_members_added_activity(
+        self, members_added: list[ChannelAccount], turn_context: TurnContext
+    ) -> None:
+        """Handle the members added activity."""
+        if turn_context.activity.recipient is None:
+            return
+        for member in members_added:
+            if member.id != turn_context.activity.recipient.id:
+                await turn_context.send_activity("Hello and welcome!")
+        return
+    
+    async def on_message_activity(self, turn_context: TurnContext) -> None:
+        """Handle the message activity."""
+        is_command, command = self.is_command(turn_context.activity.text)
+        if is_command:
+            await self.on_command(command, turn_context)
+        return

config.py

@@ -0,0 +1,12 @@
+import os
+
+class DefaultConfig:
+    """Default configuration for the bot."""
+
+    COMMAND_PREFIX = "/"
+    PORT = 3978
+    APP_ID = os.getenv("APP_ID")
+    APP_PASSWORD = os.getenv("APP_PASSWORD")
+    APP_TYPE = os.environ.get("APP_TYPE", "MultiTenant")
+    APP_TENANTID = os.getenv("APP_TENANTID")
+    

handler.py

@@ -0,0 +1,31 @@
+from botbuilder.core import TurnContext
+from botbuilder.integration.aiohttp import CloudAdapter
+
+from aiohttp.web import Request, Response, json_response
+
+from bot import WebhookBot
+from config import DefaultConfig
+
+import sys
+import traceback
+
+class BotHandler:
+    def __init__(self, adapter: CloudAdapter, bot: WebhookBot, config: DefaultConfig) -> None:
+        self.adapter = adapter
+        self.bot = bot
+
+    async def messages(self, req: Request) -> Response:
+        response = await self.adapter.process(req, self.bot)
+        if response is not None:
+            return response
+        return json_response(status=204)
+
+    async def on_error(self, context: TurnContext, error: Exception) -> None:
+        print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
+        traceback.print_exc()
+        
+        await context.send_activity("The bot encountered an error or bug.")
+        await context.send_activity(
+            "To continue to run this bot, please fix the bot source code."
+        )
+        return

requirements.txt

@@ -0,0 +1,5 @@
+botbuilder-integration-aiohttp>=4.15.0
+botbuilder-schema>=4.15.0
+botbuilder-core>=4.15.0
+python-dotenv>=1.0.0
+aiohttp>=3.13.2