botbuilder-core

import os import asyncio from aiohttp import web from botbuilder.core import ActivityHandler, TurnContext from botbuilder.schema import Activity from botbuilder.integration.aiohttp import BotFrameworkHttpAdapter class MyEchoBot(ActivityHandler): async def on_message_activity(self, turn_context: TurnContext): await turn_context.send_activity(f"You said: {turn_context.activity.text}") async def on_members_added_activity(self, members_added: [object], turn_context: TurnContext): for member in members_added: if member.id != turn_context.activity.recipient.id: await turn_context.send_activity("Hello and welcome!") # Configuration (replace with your actual values or env vars) # For local testing without authentication, these can be empty strings APP_ID = os.environ.get("MicrosoftAppId", "") APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "") # Create the BotFrameworkHttpAdapter adapter = BotFrameworkHttpAdapter(APP_ID, APP_PASSWORD) # Create the bot instance bot = MyEchoBot() # Listen for incoming requests on /api/messages async def messages(request): if "application/json" in request.headers["Content-Type"]: body = await request.json() else: return web.Response(status=400) activity = Activity().deserialize(body) auth_header = request.headers["Authorization"] if "Authorization" in request.headers else "" try: response = await adapter.process_activity(activity, auth_header, bot.on_turn) if response: return web.json_response(data=response.body, status=response.status) return web.Response(status=200) except Exception as e: print(f"Error processing activity: {e}") return web.Response(status=500, text=str(e)) # Setup AIOHTTP web application app = web.Application() app.router.post("/api/messages", messages) if __name__ == "__main__": try: web.run_app(app, host="localhost", port=3978) except Exception as e: raise e