Currently, I am developing a Teams Bot in Python (Flask). Due to certain circumstances,
I am proceeding with bot development without using BotFramework. I have been able to return an AdaptiveCard based on user messages.
However, when I press the button inside the AdaptiveCard, I encounter errors or end up in a state similar to the image below within the task module.
{"error":{"code":"BadArgument","message":"Unknown activity type"}}
The reference sites I used are as follows:
https://qiita.com/Derek/items/9e25da0efb8045c73b7a
https://zenn.dev/yusukeiwaki/articles/b6ae7bf77a81eb
I will also provide the Python code I am currently writing. I suspect that the issue lies in the JSON, but I'm not sure. I'm in a bit of a dilemma. I would appreciate any advice you can provide. Thank you in advance.
from flask import Flask, request, jsonify
import requests
import json
import os
from threading import Thread
MICROSOFT_TENANT_ID = "**********"
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)
# print(response.text)
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':
Thread(target=create_task_module_response, args=(json_body,)).start()
if json_body['type'] == 'message':
Thread(target=process_message, args=(json_body,)).start()
# HTTP 200 OKならなんでもよい
return jsonify(status=200)
def process_message(params):
service_url = params['serviceUrl']
conversation_id = params['conversation']['id']
access_token = fetch_access_token()
print(params)
try:
if params['text'] == 'aa':
reply = {
"type": "message",
"from": {
"id": "AI_Sakura@7AJmJNFYzCM",
"name": "AI_Sakura"
},
"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",
"body": {
"type": "task/fetch",
"title": "Open Task Module",
"value": {
"type": "AdaptiveCard",
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"body": [
{
"type": "TextBlock",
"text": "こんにちは",
"weight": "bolder",
"size": "large"
}
]
}
}
},
"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"
}
}
]
}
}
]
}
except KeyError:
print("module★")
reply = {
"type": "task/fetch",
"title": "Open Task Module",
"value": {
"type": "AdaptiveCard",
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"body": [
{
"type": "TextBlock",
"text": "こんにちは",
"weight": "bolder",
"size": "large"
}
]
}
}
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
# print("Response Body:", response.status_code)
# print("Response Headers:", response.text)
def create_task_module_response(params):
print('★★★★★★')
service_url = params['serviceUrl']
conversation_id = params['conversation']['id']
access_token = fetch_access_token()
reply = {
"type": "invoke",
"name": "task/fetch",
"value": {
"title": "タスク モジュールのタイトル",
"height": 400,
"width": 500,
}
}
headers = {
'Content-Type': 'application/json',
'Authorization': f"Bearer {access_token}"
}
resp =requests.post(f"{service_url}v3/conversations/{conversation_id}/activities", json=reply, headers=headers)
print(resp.text)
return resp
if __name__ == '__main__':
app.run(debug=True)