Cloud Function调用Google Workflows回调接口时认证失败求助

开发者阿泽

2026-7-21

Google Workflows回调URL调用401未授权问题

问题概述

基于Google Workflows搭建数据管道，利用回调机制等待Pub/Sub触发的Cloud Function执行完毕
Workflow与Cloud Function使用同一服务账号部署，该账号已拥有Workflows Invoker角色，且二者处于同一GCP项目
调用回调URL时触发HTTP错误：requests.exceptions.HTTPError: 401 Client Error: Unauthorized for url: https://workflowexecutions.googleapis.com/v1/projects

Workflow定义

main:
  steps:
    - init:
        assign:
          - project: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
    - create-callback:
        call: events.create_callback_endpoint
        args:
            http_callback_method: "GET"
        result: callback_load
    - init-callback:
        assign:
          - message: {"name": "check_load_finished", "callback": "${callback_load.url}"}
          - base64Msg: ${base64.encode(json.encode(message))}
    - check-load-finished:
        call: googleapis.pubsub.v1.projects.topics.publish
        args:
          topic: ${"projects/" + project + "/topics/data-scripts"}
          body:
            messages:
              - data: ${base64Msg}
    - await-callback:
        call: events.await_callback
        args:
            callback: ${callback_load}
            timeout: 1200
        result: callback_request

Cloud Function回调代码

Cloud Function执行完成后调用以下函数发起回调请求：

import google.auth.transport.requests
import google.oauth2.id_token
import requests

def send_callback(callback_url):
    auth_req = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(auth_req, callback_url)
    print(f"ID Token: {id_token}")

    headers = {
        "accept": "application/json",
        "Authorization": f"Bearer {id_token}",
    }

    r = requests.get(
        url=callback_url,
        headers=headers,
    )
    r.raise_for_status()
    print(f"Callback sent to '{callback_url}'")