app.py

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

importsys
importtraceback
importuuid
fromdatetimeimportdatetime
fromhttpimportHTTPStatus

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

frombotsimportTeamsConversationBot
fromconfigimportDefaultConfig

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 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)


ADAPTER.on_turn_error=on_error

# If the channel is the Emulator, and authentication is not in use, the AppId will be null.
# We generate a random AppId for this case only. This is not required for production, since
# the AppId will have a value.
APP_ID=SETTINGS.app_idifSETTINGS.app_idelseuuid.uuid4()

# Create the Bot
BOT=TeamsConversationBot(CONFIG.APP_ID, CONFIG.APP_PASSWORD)


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

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


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)
exceptExceptionaserror:
raiseerror