Bot Framework SDK Python 应用程序因 UserAssignedMSI 而失败，但可在 Azure 上使用多租户

Question

我有一个使用 Python 中的 Bot Framework SDK 构建的机器人应用程序，它与 MicrosoftAppType: MultiTenant 完美配合。但是，当我尝试将其配置为使用 MicrosoftAppType: UserAssignedMSI 时，通过 Azure Bot 服务的“在 Web 聊天中测试”功能进行测试时，机器人无法连接。

设置详情： 我的代码部署在 Azure Web 应用程序上。 我正在使用 Azure 机器人服务连接到机器人。 该设置适用于 MicrosoftAppType：MultiTenant。 由于特定要求，我无法使用 MultiTenant 或 SingleTenant 作为 MicrosoftAppType 的选项。 我遇到了与 MicrosoftAppType: UserAssignedMSI 相关的问题。 在配置 UserAssignedMSI 时，我已经在 Azure Web App 的标识部分中添加了用户标识。 我在下面包含了 app.py 和 config.py 的代码片段。我正在寻求帮助来了解为什么 UserAssignedMSI 会出现此问题以及如何修复它。也欢迎任何替代解决方案的建议。

应用程序.py

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

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 (
    TurnContext,
)
from botbuilder.core.integration import aiohttp_error_middleware
from botbuilder.integration.aiohttp import CloudAdapter, ConfigurationBotFrameworkAuthentication
from botbuilder.schema import Activity, ActivityTypes

from bots import EchoBot
from config import DefaultConfig

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):
    # 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
    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."
    )
    # Send a trace activity if we're talking to the Bot Framework Emulator
    if context.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
        await context.send_activity(trace_activity)


ADAPTER.on_turn_error = on_error

# Create the Bot
BOT = EchoBot()


# Listen for incoming requests on /api/messages
async def messages(req: Request) -> Response:
    return await ADAPTER.process(req, BOT)


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 error:
        raise error

配置.py

import os

""" Bot Configuration """


class DefaultConfig:
    """ Bot Configuration """

    PORT = 3978
    APP_ID = os.environ.get("MicrosoftAppId", "MyMicrosoftAppId")
    APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "")
    APP_TYPE = os.environ.get("MicrosoftAppType", "UserAssignedMSI")
    APP_TENANTID = os.environ.get("MicrosoftAppTenantId", "MyMicrosoftAppTenantId")

配置的多租户设置：

我最初使用 MicrosoftAppType: MultiTenant 配置机器人，在 Azure 机器人服务的“在 Web 聊天中测试”功能中进行测试时，它运行良好。机器人按预期响应，表明连接和身份验证成功。 切换到 UserAssignedMSI：

我更改了配置以使用 MicrosoftAppType: UserAssignedMSI 以满足特定要求。我将用户身份添加到 Azure Web 应用程序的身份部分，并验证了 UserAssignedMSI 是否在 Azure 中正确设置。 预期结果：

我希望机器人能够使用 UserAssignedMSI 正确连接和进行身份验证，类似于它与 MultiTenant 的工作方式，并在“网络聊天测试”功能中进行测试时正确响应。 实际结果：

使用 UserAssignedMSI 时，机器人无法连接或进行身份验证，并且我看不到任何有助于诊断问题的详细错误消息。 我不确定我是否错过了一些特定于 UserAssignedMSI 的配置步骤，或者是否需要其他权限。任何解决此问题的帮助或使用 UserAssignedMSI 的替代建议将不胜感激。

Answer 1

将以下代码用

init_func()

括在

app.py

中：

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 error:
        raise error

修改

app.py

：

CONFIG = DefaultConfig()

SETTINGS = BotFrameworkAdapterSettings(CONFIG.APP_ID, CONFIG.APP_PASSWORD)
ADAPTER = BotFrameworkAdapter(SETTINGS)

async def on_error(context: TurnContext, error: Exception):
    print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
    traceback.print_exc()

    # Send a message to the user
    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."
    )
    # Send a trace activity if we're talking to the Bot Framework Emulator
    if context.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",
        )
        await context.send_activity(trace_activity)


ADAPTER.on_turn_error = on_error

BOT = MyBot()

async def messages(req: Request) -> Response:
    # Main bot message handler.
    if "application/json" in req.headers["Content-Type"]:
        body = await req.json()
    else:
        return Response(status=415)

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

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


def init_func(argv):
    APP = web.Application(middlewares=[aiohttp_error_middleware])
    APP.router.add_post("/api/messages", messages)
    return APP
if __name__ == "__main__":
    APP = init_func(None)

    try:
        web.run_app(APP, host="0.0.0.0", port=CONFIG.PORT)
    except Exception as error:
        raise error

参考 MSDOC 使用用户分配的托管标识将 Python SDK echo bot 部署到 Azure Bot：

创建启用托管身份和应用服务的 Python Azure 机器人。

通过在

Azure Bot=>Configuration=>Message Endpoint

(

https://<app_name>.azurewebsites.net/api/messages

) 中添加应用服务端点来配置 Azure Bot 中的消息传递端点：

将 Azure 机器人与应用服务链接：

转到

App Service=>Identity=>User Assigned=>Add=>Select your Bot User assigned managed Identity

：

• 点击添加的

  Bot Managed Identity=>Properties

  ，复制租户ID、客户端ID。

config.py：

class DefaultConfig:
    """ Bot Configuration """

    PORT = 3978
    APP_ID = os.environ.get("MicrosoftAppId", "MicrosoftApp_Id")
    APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "")
    APP_TYPE = os.environ.get("MicrosoftAppType", "MicrosoftApp_Type")
    APP_TENANTID = os.environ.get("MicrosoftAppTenantId", "MicrosoftApp_TenantId")

将机器人部署到 Azure 应用服务并在

Azure App Service=>Configuration=>General settings

中添加启动命令（如下所示）：

python3 -m aiohttp.web -H 0.0.0.0 -P 8000 app:init_func

能够在Azure Bot中测试Web聊天：

检查

App Service=>Monitoring=>Log Stream

以识别阻止机器人正确响应的错误日志。