Как отправлять уведомления об ошибках воздушного потока в Microsoft Teams с помощью Microsoft 365 Worfklow

Посмотрев немного дальше, я смог найти решение. Код выглядит практически так же, мне нужно только изменить «параметры» полезной нагрузки. Вот код, работающий с модификациями:

from airflow.models import Variable
import logging
import requests

def failed_task_notify_teams(context):
    logging.info("Send notification to the Teams Group")
    payload = {
        "type": "message",
        "attachments" : [{
            "contentType": "application/vdn.microsoft.card.adaptive",
            "contentUrl": None,
            "content": {
                "$schema": "http://adaptivecard.io/schemas/adaptive-card.json",
                "type": "AdaptiveCard",
                "version": "1.2",
                "body": [
                    {
                        "type": "TextBlock",
                        "text": f"Task {context['task_instance_key_str']} : Failed",
                        "weight": "bolder",
                        "size": "medium",
                        "color": "attention",
                    },
                    {
                        "type": "TextBlock",
                        "text": f"DAG : {context['dag'].dag_id}",
                        "spacing": "none"
                    },
                    {
                        "type": "FactSet",
                        "facts": [
                            {
                                "title": "Date :",
                                "value": context['ds']
                            },
                            {
                                "title": "Log URL :",
                                "value": f"[Click here]({context['task_instance'].log_url})"
                            }
                        ]
                    }
                ],
                "actions": [
                    {
                        "type": "Action.OpenUrl",
                        "title": "See logs",
                        "url": context['task_instance'].log_url
                    }
                ]
            }
        }]
    }
    
    headers = {"content-type": "application/json"}
    response = requests.post(Variable.get('teams_azure_webook_secret'), json=payload, headers=headers)

    if response.status_code == 200:
        logging.info("Notification sent to Teams Group")
    else:
        logging.error(f"Failed to sent notification to Teams Group: {response.status_code} - {response.text}")