不使用BotFramework的Python Teams Bot如何打开Task Module？

问题背景 #

使用Python Flask开发Teams Bot（未使用BotFramework），已实现基于用户消息返回AdaptiveCard的功能，但点击卡片内按钮时，Task Module触发错误：

{"error":{"code":"BadArgument","message":"Unknown activity type"}}

核心问题分析 #

1. Invoke请求处理逻辑错误：点击AdaptiveCard的Action.Submit触发task/fetch类型的invoke请求时，Teams要求直接从webhook返回响应，而非通过serviceUrl发送新活动。
2. AdaptiveCard触发结构不规范：卡片中msteams字段的task/fetch格式嵌套了多余层级，不符合Teams规范。
3. 响应结构错误：返回的Task Module响应类型不正确，不应使用invoke类型，需遵循task/fetch的响应格式要求。

具体修正步骤 #

1. 修正Webhook的Invoke请求处理逻辑 #

将webhook函数中处理invoke请求的逻辑改为直接返回响应，无需异步线程或调用serviceUrl API：

@app.route('/message', methods=['POST'])
def webhook():
    json_body = request.get_json()
    print(f"Parameter: {json_body}")
    
    if json_body['type'] == 'invoke':
        # 直接返回task/fetch响应，同步处理
        return create_task_module_response(json_body)

    if json_body['type'] == 'message':
        Thread(target=process_message, args=(json_body,)).start()
    
    return jsonify(status=200)

2. 修正AdaptiveCard的Task Module触发结构 #

移除msteams字段内多余的body嵌套，使用标准触发格式：

# 修正后的Action.Submit示例
{
    "type": "Action.Submit",
    "title": "Adaptive Card",
    "data": {
        "msteams": {
            "type": "task/fetch",
            "value": {
                "cardType": "AdaptiveCard"
            }
        },
        "data": "AdaptiveCard"
    }
}

3. 重构Task Module响应函数 #

create_task_module_response需直接返回符合规范的task/fetch响应，无需调用外部API：

def create_task_module_response(params):
    print('★★★★★★')
    trigger_data = params['value']['data']
    
    # 根据触发按钮的data返回对应内容
    if trigger_data == "AdaptiveCard":
        task_card = {
            "type": "AdaptiveCard",
            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
            "version": "1.0",
            "body": [
                {
                    "type": "TextBlock",
                    "text": "こんにちは",
                    "weight": "bolder",
                    "size": "large"
                }
            ]
        }
        return jsonify({
            "task": {
                "type": "continue",
                "value": {
                    "title": "Adaptive Card Task Module",
                    "height": 400,
                    "width": 500,
                    "card": task_card
                }
            }
        })
    elif trigger_data == "CustomForm":
        # 自定义表单示例
        task_card = {
            "type": "AdaptiveCard",
            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
            "version": "1.0",
            "body": [
                {
                    "type": "Input.Text",
                    "id": "username",
                    "placeholder": "请输入用户名"
                }
            ]
        }
        return jsonify({
            "task": {
                "type": "continue",
                "value": {
                    "title": "自定义表单",
                    "height": 400,
                    "width": 500,
                    "card": task_card
                }
            }
        })
    elif trigger_data == "YouTube":
        # 嵌入外部页面示例
        return jsonify({
            "task": {
                "type": "continue",
                "value": {
                    "title": "YouTube 视频",
                    "height": 400,
                    "width": 500,
                    "url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
                }
            }
        })
    
    # 默认返回关闭任务模块
    return jsonify({"task": {"type": "complete"}})

4. 清理错误的异常处理分支 #

删除process_message中except KeyError分支里错误的task/fetch响应结构，普通message请求仅返回message类型活动。

完整修正代码 #

from flask import Flask, request, jsonify
import requests
import json
from threading import Thread

MICROSOFT_CLIENT_ID = "********************************"
MICROSOFT_CLIENT_SECRET = "********************************"

app = Flask(__name__)

def fetch_access_token():
    token_url = f"https://login.microsoftonline.com/botframework.com/oauth2/v2.0/token"
    payload = {
        'grant_type': 'client_credentials',
        'client_id': MICROSOFT_CLIENT_ID,
        'client_secret': MICROSOFT_CLIENT_SECRET,
        'scope': 'https://api.botframework.com/.default',
    }
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
    }
    
    response = requests.post(token_url, headers=headers, data=payload)
    access_token = json.loads(response.text)['access_token']
    return access_token

@app.route('/')
def home():
    return 'It works!'

@app.route('/message', methods=['POST'])
def webhook():
    json_body = request.get_json()
    print(f"Parameter: {json_body}")
    
    if json_body['type'] == 'invoke':
        return create_task_module_response(json_body)

    if json_body['type'] == 'message':
        Thread(target=process_message, args=(json_body,)).start()
    
    return jsonify(status=200)

def process_message(params):
    service_url = params['serviceUrl']
    conversation_id = params['conversation']['id']
    access_token = fetch_access_token()
    print(params)
    
    if params['text'] == 'aa':
        reply = { 
            "type": "message",
            "text": f"Reply for {params['text']}"
        }
    elif params['text'] == 'test':
        reply = {
            "type": "message",
            "attachments": [
                {
                    "contentType": "application/vnd.microsoft.card.adaptive",
                    "content": {
                        "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                        "version": "1.5",
                        "type": "AdaptiveCard",
                        "body": [
                            {
                                "type": "TextBlock",
                                "text": "Task Module Invocation from Adaptive Card",
                                "weight": "Bolder",
                                "size": 3
                            }
                        ],
                        "actions": [
                            {
                                "type": "Action.Submit",
                                "title": "Adaptive Card",
                                "data": {
                                    "msteams": {
                                        "type": "task/fetch",
                                        "value": {
                                            "cardType": "AdaptiveCard"
                                        }
                                    },
                                    "data": "AdaptiveCard"
                                }
                            },
                            {
                                "type": "Action.Submit",
                                "title": "Custom Form",
                                "data": {
                                    "msteams": {
                                        "type": "task/fetch"
                                    },
                                    "data": "CustomForm"
                                }
                            },
                            {
                                "type": "Action.Submit",
                                "title": "YouTube",
                                "data": {
                                    "msteams": {
                                        "type": "task/fetch"
                                    },
                                    "data": "YouTube"
                                }
                            }
                        ]
                    }
                }
            ]
        }
    else:
        reply = {
            "type": "message",
            "text": "Unknown command"
        }

    headers = {
        'Content-Type': 'application/json',
        'Authorization': f"Bearer {access_token}"
    }
    res = requests.post(f"{service_url}v3/conversations/{conversation_id}/activities", json=reply, headers=headers)
    print(res.text)
    return res

def create_task_module_response(params):
    print('★★★★★★')
    trigger_data = params['value']['data']
    
    if trigger_data == "AdaptiveCard":
        task_card = {
            "type": "AdaptiveCard",
            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
            "version": "1.0",
            "body": [
                {
                    "type": "TextBlock",
                    "text": "こんにちは",
                    "weight": "bolder",
                    "size": "large"
                }
            ]
        }
        return jsonify({
            "task": {
                "type": "continue",
                "value": {
                    "title": "Adaptive Card Task Module",
                    "height": 400,
                    "width": 500,
                    "card": task_card
                }
            }
        })
    elif trigger_data == "CustomForm":
        task_card = {
            "type": "AdaptiveCard",
            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
            "version": "1.0",
            "body": [
                {
                    "type": "Input.Text",
                    "id": "username",
                    "placeholder": "请输入用户名"
                },
                {
                    "type": "Input.Password",
                    "id": "password",
                    "placeholder": "请输入密码"
                }
            ],
            "actions": [
                {
                    "type": "Action.Submit",
                    "title": "提交"
                }
            ]
        }
        return jsonify({
            "task": {
                "type": "continue",
                "value": {
                    "title": "自定义表单",
                    "height": 400,
                    "width": 500,
                    "card": task_card
                }
            }
        })
    elif trigger_data == "YouTube":
        return jsonify({
            "task": {
                "type": "continue",
                "value": {
                    "title": "YouTube 视频",
                    "height": 400,
                    "width": 500,
                    "url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
                }
            }
        })
    
    return jsonify({"task": {"type": "complete"}})

if __name__ == '__main__':
    app.run(debug=True)

关键说明 #

• Teams的invoke请求（如task/fetch）要求同步返回响应，异步处理或通过serviceUrl发送活动会触发类型错误。
• task/fetch响应必须包含task字段，type为continue时需指定value内的card（自适应卡片）或url（外部页面）。
• AdaptiveCard中Action.Submit的msteams.type设为task/fetch后，Teams会自动向webhook发送invoke请求，无需额外嵌套结构。

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