如何通过Microsoft 365 Workflow发送Airflow失败通知至Microsoft Teams

微软已逐步弃用旧版MessageCard格式的webhook支持，改用Adaptive Cards作为Microsoft 365 Workflow环境下的标准通知格式。以下是适配后的代码，保持原代码核心逻辑的同时兼容新规范：

from airflow.models import Variable
import logging
import requests

def failed_task_notify_teams(context):
    logging.info("Send notification to the Teams Group")
    task_key = context['task_instance_key_str']
    dag_id = context['dag'].dag_id
    log_url = context['task_instance'].log_url
    ds = context['ds']

    # 使用Adaptive Cards格式（微软365环境推荐标准）
    payload = {
        "type": "AdaptiveCard",
        "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
        "version": "1.4",
        "body": [
            {
                "type": "TextBlock",
                "size": "Large",
                "weight": "Bolder",
                "text": "Airflow Task Error",
                "color": "Attention"
            },
            {
                "type": "TextBlock",
                "text": f"**Task**: {task_key} : Failed",
                "spacing": "Small"
            },
            {
                "type": "TextBlock",
                "text": f"**DAG**: {dag_id}",
                "spacing": "Small"
            },
            {
                "type": "FactSet",
                "facts": [
                    {
                        "title": "Date:",
                        "value": ds
                    },
                    {
                        "title": "Log URL:",
                        "value": f"[{log_url}]({log_url})"
                    }
                ]
            }
        ],
        "actions": [
            {
                "type": "Action.OpenUrl",
                "title": "See Logs",
                "url": log_url
            }
        ]
    }
    
    headers = {"content-type": "application/json"}
    try:
        response = requests.post(Variable.get('teams_webhook_secret'), json=payload, headers=headers)
        response.raise_for_status()
        logging.info("Teams notification sent to the group!")
    except requests.exceptions.RequestException as e:
        logging.error(f"Failed to send Teams notification: {str(e)}")

关键修改说明 #

• 替换通知格式：从旧版MessageCard切换为Adaptive Cards，这是Microsoft 365 Workflow支持的标准格式
• 保留核心信息：原代码中的任务标识、DAG ID、日期、日志链接等核心通知内容完全保留
• 增加错误处理：添加try-except块捕获请求异常，便于排查通知失败问题
• 优化样式展示：通过Adaptive Cards的原生样式属性（如color、weight）提升通知可读性

注意事项 #

• 确保你的Teams webhook是在Microsoft 365环境中创建的新入站webhook，旧版webhook将不再被支持
• 验证Airflow变量teams_webhook_secret中存储的是正确的新webhook URL
• 可根据需求调整Adaptive Cards的样式（如修改color值、调整布局结构）

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