app.py

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.

importsys
importtraceback
fromdatetimeimportdatetime
fromhttpimportHTTPStatus

fromaiohttpimportweb
fromaiohttp.webimportRequest, Response, json_response
frombotbuilder.coreimport (
BotFrameworkAdapterSettings,
TurnContext,
BotFrameworkAdapter,
)
frombotbuilder.core.integrationimportaiohttp_error_middleware
frombotbuilder.schemaimportActivity, ActivityTypes
frombotsimportTeamsTaskModuleBot
fromconfigimportDefaultConfig

# Load configuration
CONFIG=DefaultConfig()

# Create adapter.
# See https://aka.ms/about-bot-adapter to learn more about how bots work.
SETTINGS=BotFrameworkAdapterSettings(CONFIG.APP_ID, CONFIG.APP_PASSWORD)
ADAPTER=BotFrameworkAdapter(SETTINGS)

# Catch-all for errors.
asyncdefon_error(context: TurnContext, error: Exception):
"""
 This function is called when an error occurs during a turn.

 :param context: The context object for the turn.
 :param error: The exception that was thrown.
 """
# This check writes out errors to console log vs. app insights.
# NOTE: In production environment, you should consider logging this to Azure
# application insights.
print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
traceback.print_exc()

# Send a message to the user
awaitcontext.send_activity("The bot encountered an error or bug.")
awaitcontext.send_activity(
"To continue to run this bot, please fix the bot source code."
 )
# Send a trace activity if we're talking to the Bot Framework Emulator
ifcontext.activity.channel_id=="emulator":
# Create a trace activity that contains the error object
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",
 )
# Send a trace activity, which will be displayed in Bot Framework Emulator
awaitcontext.send_activity(trace_activity)

# Set the error handler on the adapter
ADAPTER.on_turn_error=on_error

# Create the Bot
BOT=TeamsTaskModuleBot(CONFIG)

# Listen for incoming requests on /api/messages
asyncdefmessages(req: Request) ->Response:
"""
 Main bot message handler.

 :param req: The incoming request.
 :return: The response to the request.
 """
if"application/json"inreq.headers["Content-Type"]:
body=awaitreq.json()
else:
returnResponse(status=HTTPStatus.UNSUPPORTED_MEDIA_TYPE)

activity=Activity().deserialize(body)
auth_header=req.headers["Authorization"] if"Authorization"inreq.headerselse""

invoke_response=awaitADAPTER.process_activity(activity, auth_header, BOT.on_turn)
ifinvoke_response:
returnjson_response(data=invoke_response.body, status=invoke_response.status)
returnResponse(status=HTTPStatus.OK)

# Create the web application
APP=web.Application(middlewares=[aiohttp_error_middleware])
APP.router.add_post("/api/messages", messages)
APP.router.add_static("/", path="./pages/", name="pages")

if__name__=="__main__":
try:
web.run_app(APP, host="localhost", port=CONFIG.PORT)
exceptExceptionaserror:
raiseerror