使用Python开发Teams机器人（Bot Framework）发送消息时遇access_token错误

背景 #

使用Python版Bot Framework SDK结合FastAPI开发Teams机器人，需求是用户发送初始消息后，机器人每60秒向其发送定期更新。当前在on_message_activity方法内使用turn_context能正常响应初始消息，但在该方法外发送定期更新时出现access_token错误，不确定如何正确创建新的TurnContext实现定期消息发送。

代码实现 #

import asyncio
from botbuilder.core import ActivityHandler, TurnContext, MessageFactory, BotFrameworkAdapterSettings, BotFrameworkAdapter
from botbuilder.schema import Activity, ConversationReference
from botbuilder.core.teams import TeamsActivityHandler
from typing import Dict
from botframework.connector.auth import ClaimsIdentity, MicrosoftAppCredentials
from fastapi import APIRouter
import json

# 日志配置（假设已正确配置日志器）
import logging
logger = logging.getLogger(__name__)

# Bot配置
APP_ID = 'your-app-id-here'
APP_PASSWORD = 'your-app-password-here'

# 适配器配置
CONFIGS = BotFrameworkAdapterSettings(APP_ID, APP_PASSWORD)
ADAPTER = BotFrameworkAdapter(CONFIGS)

# FastAPI路由配置
router = APIRouter()

class NotifyBot(TeamsActivityHandler):
    def __init__(self, adapter):
        super().__init__()
        self.adapter = adapter
        self.conversation_references: Dict[str, ConversationReference] = {}
        self.scheduled_tasks: Dict[str, asyncio.Task] = {}
        logger.info("NotifyBot已初始化")

    async def on_message_activity(self, turn_context: TurnContext):
        # 用户发送消息时存储会话引用
        self.add_conversation_reference(turn_context.activity)

        # 从传入活动中提取用户ID
        user_id = turn_context.activity.from_property.id

        # 发送响应确认收到用户消息
        await turn_context.send_activity("感谢您的消息，我将开始为您发送定期更新。")

        # 若未已调度，则开始发送定期消息
        if user_id not in self.scheduled_tasks:
            logger.info(f"为用户{user_id}调度定期消息")
            self.scheduled_tasks[user_id] = asyncio.create_task(self.send_periodic_messages(user_id))

    def add_conversation_reference(self, activity: Activity):
        # 添加或更新用户的会话引用
        conversation_reference = TurnContext.get_conversation_reference(activity)
        self.conversation_references[conversation_reference.user.id] = conversation_reference
        logger.info(f"已为用户{conversation_reference.user.id}添加会话引用")

    async def send_periodic_messages(self, user_id: str):
        while True:
            try:
                # 每60秒发送一次消息
                await asyncio.sleep(60)
                await self.send_personal_message(user_id)
            except asyncio.CancelledError:
                logger.info(f"已取消用户{user_id}的定期消息调度")
                break
            except Exception as e:
                logger.error(f"向用户{user_id}发送定期消息时出错：{str(e)}")

    async def send_personal_message(self, user_id: str):
        conversation_reference = self.conversation_references.get(user_id)

        if conversation_reference:
            try:
                logger.info(f"正在向用户{user_id}发送消息")

                # 使用continue_conversation发送消息
                await self.adapter.continue_conversation(
                    bot_app_id=APP_ID,
                    reference=conversation_reference,
                    logic=lambda turn_context: asyncio.create_task(self.send_message_callback(turn_context, "定期更新：您好！"))
                )

            except Exception as e:
                logger.error(f"向用户{user_id}发送消息时出错：{str(e)}")
        else:
            logger.error(f"未找到用户{user_id}的会话引用")

    async def send_message_callback(self, turn_context: TurnContext, message: str):
        try:
            await turn_context.send_activity(MessageFactory.text(message))
            logger.info(f"已向{turn_context.activity.from_property.id}发送消息")
        except Exception as e:
            logger.error(f"发送消息时出错：{str(e)}")

遇到的问题 #

• 机器人能在on_message_activity中正常响应初始消息，但无法按预期每60秒发送定期消息；
• 使用continue_conversation发送消息时，收到access_token错误；
• 不确定如何处理令牌过期问题，以及如何正确创建新的TurnContext。

已尝试操作 #

• 确认APP_ID和APP_PASSWORD配置正确；
• 尝试使用MicrosoftAppCredentials获取新访问令牌，但实现方式存疑；
• 查阅continue_conversation和ClaimsIdentity的官方文档及示例，未解决问题。

疑问 #

1. 如何正确实现每60秒向用户发送定期消息？
2. 使用continue_conversation时，如何正确处理access_token和ClaimsIdentity？
3. 是否需要为每条消息创建新的TurnContext？若需要，该如何实现？

解决方案 #

1. 修复continue_conversation的身份验证问题 #

主动发起对话时，需要传递合法的ClaimsIdentity供适配器完成身份验证，修改send_personal_message方法：

async def send_personal_message(self, user_id: str):
    conversation_reference = self.conversation_references.get(user_id)

    if conversation_reference:
        try:
            logger.info(f"正在向用户{user_id}发送消息")
            
            # 创建机器人主动对话的身份标识
            claims_identity = ClaimsIdentity(
                claims={"aud": APP_ID},
                auth_type="BotFramework"
            )

            await self.adapter.continue_conversation(
                bot_app_id=APP_ID,
                reference=conversation_reference,
                logic=lambda turn_context: asyncio.create_task(self.send_message_callback(turn_context, "定期更新：您好！")),
                claims_identity=claims_identity
            )

        except Exception as e:
            logger.error(f"向用户{user_id}发送消息时出错：{str(e)}")
    else:
        logger.error(f"未找到用户{user_id}的会话引用")

2. 令牌自动刷新处理 #

Python Bot Framework SDK的BotFrameworkAdapter会通过MicrosoftAppCredentials自动获取、刷新令牌，只需确保：

• APP_ID和APP_PASSWORD与Azure Bot注册信息完全一致；
• 机器人应用在Azure AD中拥有Teams.SendMessage等必要权限。

3. 定期消息的优化实现 #

当前的定时任务逻辑可行，但需补充会话清理机制：

• 监听会话结束事件，移除对应的conversation_reference并取消定时任务；
• 给定时任务添加重试逻辑，避免单次异常导致任务终止。

4. TurnContext的创建说明 #

调用adapter.continue_conversation时，适配器会自动创建新的TurnContext，无需手动实例化。只需提供会话引用和消息处理逻辑即可。

内容的提问来源于stack exchange，提问作者NoNam4