botbuilder-integration-aiohttp

import os import sys import traceback from datetime import datetime from http import HTTPStatus from aiohttp import web from aiohttp.web import Request, Response, json_response from botbuilder.core import ( ActivityHandler, TurnContext, ) from botbuilder.core.integration import aiohttp_error_middleware from botbuilder.integration.aiohttp import CloudAdapter, ConfigurationBotFrameworkAuthentication from botbuilder.schema import Activity, ActivityTypes # Configuration class DefaultConfig: APP_ID = os.environ.get('MicrosoftAppId', '') APP_PASSWORD = os.environ.get('MicrosoftAppPassword', '') PORT = 3978 CONFIG = DefaultConfig() # Create adapter. See https://aka.ms/about-bot-adapter to learn more about how bots work. ADAPTER = CloudAdapter(ConfigurationBotFrameworkAuthentication(CONFIG)) # Catch-all for errors. async def on_error(context: TurnContext, error: Exception): 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.") if context.activity.channel_id == "emulator": trace_activity = Activity( label="TurnError", name="on_turn_error Trace", timestamp=datetime.utcnow(), type=ActivityTypes.trace, value=f"{error}", value_type="https://www.botframework.com/schemas/error", ) await context.send_activity(trace_activity) ADAPTER.on_turn_error = on_error # Define your bot logic class EchoBot(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: [ChannelAccount], turn_context: TurnContext): for member in members_added: if member.id != turn_context.activity.recipient.id: await turn_context.send_activity("Hello and welcome!") BOT = EchoBot() # Listen for incoming requests on /api/messages async def messages(req: Request) -> Response: if "application/json" in req.headers["Content-Type"]: body = await req.json() else: return Response(status=HTTPStatus.UNSUPPORTED_MEDIA_TYPE) activity = Activity().deserialize(body) auth_header = req.headers["Authorization"] if "Authorization" in req.headers else "" try: response = await ADAPTER.process_activity(activity, auth_header, BOT.on_turn) if response: return json_response(data=response.body, status=response.status) return Response(status=HTTPStatus.OK) except Exception as e: raise e app = web.Application(middlewares=[aiohttp_error_middleware]) app.router.add_post("/api/messages", messages) if __name__ == "__main__": try: web.run_app(app, host="localhost", port=CONFIG.PORT) except Exception as e: print(f"Error starting server: {e}")