如何通过Python调用非HTTP触发的Google Cloud Function？

你遇到的401问题，根源在于非HTTP触发的Cloud Function不支持直接通过函数URL POST调用，哪怕你带上了正确的ID Token。gcloud functions call本质是调用了Cloud Functions API的call方法，而非直接请求函数端点。除了用subprocess调用gcloud，还有两种更优雅的实现方式：

方法一：使用Google Cloud Functions客户端库调用API #

这是最推荐的方式，官方库已封装认证和API调用细节，无需手动处理请求签名。

步骤： #

1. 安装依赖：

pip install google-cloud-functions

2. 编写调用代码：

from google.cloud import functions_v1

def invoke_storage_triggered_function(project_id, region, function_name, bucket, file_name):
    client = functions_v1.CloudFunctionsServiceClient()
    # 构造函数的资源名称
    function_path = client.function_path(project_id, region, function_name)
    # 构造触发事件的数据，和gcloud call的data格式一致
    request = functions_v1.CallFunctionRequest(
        name=function_path,
        data={
            "bucket": bucket,
            "name": file_name
        }
    )
    # 发起调用
    response = client.call_function(request=request)
    print(f"调用结果: {response.result}")

# 替换为你的实际参数
invoke_storage_triggered_function(
    project_id="<your-project>",
    region="europe-west3",
    function_name="<your-function>",
    bucket="<trigger-bucket>",
    file_name="<target-file>"
)

方法二：直接调用Cloud Functions REST API #

如果不想安装客户端库，也可以手动构造HTTP请求调用API，认证使用Google的OAuth2令牌（而非id_token）。

代码示例： #

import requests
from google.auth import default
from google.auth.transport.requests import Request

def invoke_via_rest_api(project_id, region, function_name, bucket, file_name):
    # 获取默认认证凭据（需要本地有gcloud认证或服务账号密钥）
    credentials, _ = default()
    credentials.refresh(Request())
    token = credentials.token

    api_url = f"https://cloudfunctions.googleapis.com/v1/projects/{project_id}/locations/{region}/functions/{function_name}:call"
    payload = {
        "data": {
            "bucket": bucket,
            "name": file_name
        }
    }
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    response = requests.post(api_url, json=payload, headers=headers)
    response.raise_for_status()
    print(f"调用结果: {response.json()}")

# 替换参数调用
invoke_via_rest_api(
    project_id="<your-project>",
    region="europe-west3",
    function_name="<your-function>",
    bucket="<trigger-bucket>",
    file_name="<target-file>"
)

关键说明： #

• 非HTTP触发的函数，必须通过Cloud Functions的call API触发，不能直接请求函数的https://<region>-<project>.cloudfunctions.net/<function> URL，这个URL只对HTTP触发的函数有效。
• 两种方法都需要调用者具备cloudfunctions.functions.call权限，确保你的认证账号（本地gcloud账号或服务账号）拥有这个权限。

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