GCP桶触发Azure DevOps Pipeline后，如何在Teams频道打印触发用户ID

当GCP存储桶中的文件更新触发Azure CI流水线后，需要在Teams频道中输出CI触发用户ID，但现有代码中SYSTEM_ACCESSTOKEN相关的认证逻辑无法正常获取到触发用户ID，流水线日志里没有生成预期的ID。

尝试的代码：

def get_triggered_user_id():
    organization_url = "https://dev.azure.com/{organisation_name}/"
    project_name = os.environ.get('ComponentName')
    build_id = os.environ.get('BUILD_BUILDID')  # Assuming this environment variable is available

    api_url = f"{organization_url}{project_name}/_apis/build/builds/{build_id}?api-version=6.0"
    headers = {
        'Authorization': f'Basic {os.environ["SYSTEM_ACCESSTOKEN"]}'  ##Here the id is not generating in my logs in devops pipeline
    }

    response = requests.get(api_url, headers=headers)
    response.raise_for_status()

    triggering_user_id = response.json().get('requestedFor').get('id')
    return triggering_user_id

1. 开启流水线的OAuth令牌访问权限 #

Azure DevOps默认不会把SYSTEM_ACCESSTOKEN暴露给脚本，必须手动开启：

• 编辑流水线，找到运行Python脚本的目标任务
• 在任务的「附加选项」里勾选允许脚本访问OAuth令牌
• 保存流水线后重新运行

2. 修正Basic认证的编码逻辑 #

Azure DevOps的Basic认证要求将:{SYSTEM_ACCESSTOKEN}的组合进行Base64编码，而非直接使用原始token。修改代码中的认证头部分：

import base64

# ... 其他代码 ...

token = os.environ.get("SYSTEM_ACCESSTOKEN")
if not token:
    raise ValueError("SYSTEM_ACCESSTOKEN environment variable not found")

# 构造认证字符串：空用户名 + 冒号 + token，再做Base64编码
auth_str = f":{token}"
auth_bytes = auth_str.encode('utf-8')
base64_auth = base64.b64encode(auth_bytes).decode('utf-8')

headers = {
    'Authorization': f'Basic {base64_auth}'
}

3. 验证环境变量的有效性 #

在调用API前先打印关键环境变量，确认是否存在或取值正确：

print(f"BUILD_BUILDID: {os.environ.get('BUILD_BUILDID')}")
print(f"ComponentName: {os.environ.get('ComponentName')}")

如果ComponentName为空，可尝试替换为Azure DevOps默认的项目名环境变量SYSTEM_TEAMPROJECT。

4. 处理API响应的空值情况 #

若流水线由服务账号或系统自动触发，requestedFor字段可能为空，需添加判断避免报错：

build_data = response.json()
requested_for = build_data.get('requestedFor')
if requested_for:
    triggering_user_id = requested_for.get('id')
else:
    triggering_user_id = "System/Service Account"  # 自定义默认值

import os
import requests
import base64

def get_triggered_user_id():
    organization_url = "https://dev.azure.com/{organisation_name}/"
    # 优先用ComponentName，不存在则用默认项目变量SYSTEM_TEAMPROJECT
    project_name = os.environ.get('ComponentName') or os.environ.get('SYSTEM_TEAMPROJECT')
    build_id = os.environ.get('BUILD_BUILDID')

    if not project_name or not build_id:
        raise ValueError("Missing required environment variables: project_name or build_id")

    api_url = f"{organization_url}{project_name}/_apis/build/builds/{build_id}?api-version=6.0"
    
    token = os.environ.get("SYSTEM_ACCESSTOKEN")
    if not token:
        raise ValueError("SYSTEM_ACCESSTOKEN is not available. Enable script access to OAuth token in pipeline settings.")

    # 构造正确的Basic认证头
    auth_str = f":{token}"
    base64_auth = base64.b64encode(auth_str.encode('utf-8')).decode('utf-8')
    headers = {
        'Authorization': f'Basic {base64_auth}'
    }

    response = requests.get(api_url, headers=headers)
    response.raise_for_status()

    build_data = response.json()
    requested_for = build_data.get('requestedFor')
    triggering_user_id = requested_for.get('id') if requested_for else "System/Service Account"
    
    return triggering_user_id

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