Teams Incoming Webhook退役后，Airflow Teams通知迁移方案咨询

1. 退役后实现Airflow Teams通知的方案 #

微软官方推荐两种替代方案，适配Airflow场景的选择如下：

• Teams Outgoing Webhook：在Teams团队内创建专用的Outgoing Webhook，获取验证密钥和请求URL，Airflow通过带密钥的HTTP请求发送消息，适配简单通知场景。
• Microsoft Graph API：通过Azure AD注册应用获取权限，调用Graph API的频道消息发送接口，适合需要复杂权限控制、自定义消息格式的场景。

2. 迁移复杂度等级 #

整体属于低到中等：

• 选择Outgoing Webhook的话，迁移仅需替换URL和添加简单的验证头，代码改动极小，复杂度低。
• 选择Graph API的话，需要额外配置Azure AD应用权限和令牌获取逻辑，代码适配量稍大，但官方文档清晰，落地难度不高，复杂度中等。

3. 是否需要修改原有代码 #

需要修改，但改动量取决于所选方案：

• 若用Outgoing Webhook：仅需替换原Incoming Webhook的URL，同时在请求头中添加Authorization: Bearer {密钥}，原有JSON消息格式可直接复用。
• 若用Graph API：需要重构消息发送逻辑，包括令牌获取、API调用适配，消息格式也要符合Graph API的要求，改动量相对较大。

迁移分步操作流程（Outgoing Webhook 方案） #

步骤1：在Teams中创建Outgoing Webhook #

1. 打开目标Teams团队的目标频道，点击右上角...选择Connectors。
2. 在连接器列表找到Outgoing Webhook，点击Configure。
3. 填写Webhook名称、可选上传图标，关联目标频道后点击Create。
4. 保存生成的Webhook URL和Security Token，这两个值是后续Airflow请求的核心参数。

步骤2：修改Airflow通知代码 #

替换原有Incoming Webhook的调用逻辑，添加验证头即可，示例Python代码：

import requests

# 替换为你的Outgoing Webhook参数
TEAMS_WEBHOOK_URL = "你的Outgoing Webhook URL"
SECURITY_TOKEN = "你的Security Token"

def send_teams_alert(message):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {SECURITY_TOKEN}"
    }
    payload = {"text": message}
    response = requests.post(TEAMS_WEBHOOK_URL, json=payload, headers=headers)
    response.raise_for_status()

# 调用示例
send_teams_alert("Airflow任务 dag_demo 执行成功")

步骤3：测试验证 #

触发Airflow任务，检查Teams频道是否收到通知；若报错，优先核对URL和令牌是否正确，请求头是否配置完整。

迁移分步操作流程（Graph API 方案） #

步骤1：Azure AD应用注册与权限配置 #

1. 登录Azure门户，进入Azure Active Directory -> 应用注册 -> 新注册。
2. 填写应用名称，选择账户类型（如“仅限本组织目录中的账户”），完成注册。
3. 记录应用的客户端ID、租户ID，并在证书和密码中创建客户端密码，保存生成的密码值（仅显示一次）。
4. 进入API权限 -> 添加权限 -> Microsoft Graph -> 应用权限，添加ChannelMessage.Send权限，点击授予管理员同意。
5. 获取目标团队ID和频道ID：可通过Graph Explorer查询对应团队和频道的ID。

步骤2：修改Airflow通知代码 #

实现令牌获取和API调用逻辑，示例Python代码：

import requests

# 替换为你的Azure AD和Teams参数
TENANT_ID = "你的租户ID"
CLIENT_ID = "你的客户端ID"
CLIENT_SECRET = "你的客户端密码"
TEAM_ID = "目标团队ID"
CHANNEL_ID = "目标频道ID"

def get_graph_access_token():
    token_url = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "https://graph.microsoft.com/.default"
    }
    response = requests.post(token_url, data=payload)
    response.raise_for_status()
    return response.json()["access_token"]

def send_teams_alert(message):
    token = get_graph_access_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "body": {"content": f"<p>{message}</p>"}
    }
    api_url = f"https://graph.microsoft.com/v1.0/teams/{TEAM_ID}/channels/{CHANNEL_ID}/messages"
    response = requests.post(api_url, json=payload, headers=headers)
    response.raise_for_status()

# 调用示例
send_teams_alert("Airflow任务 dag_demo 执行失败，请及时排查")

步骤3：测试验证 #

运行测试代码，检查Teams频道是否收到通知；若出现权限报错，确认已完成ChannelMessage.Send权限的管理员同意操作。

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