This is expected. You're getting HTTP 403
because your requests are not being authenticated.
Setting the GOOGLE_APPLICATION_CREDENTIALS
variable to the Service Account will not automagically set the authentication headers.
In addition the role you need is Cloud Functions Invoker
and NOT Cloud Function Viewer
. Cloud Function Viewer
is used to view the functions, not to trigger them.
You can try this as seen in this answer:
from google.oauth2 import service_account
from google.auth.transport.requests import AuthorizedSession
url = 'https://test-123456.cloudfunctions.net/my-cloud-function'
creds = service_account.IDTokenCredentials.from_service_account_file(
'/path/to/service-account-credentials.json', target_audience=url)
authed_session = AuthorizedSession(creds)
# make authenticated request and print the response, status_code
resp = authed_session.get(url)
print(resp.status_code)
print(resp.text)
Or this code presented by Jonh Hanley here:
import json
import base64
import requests
import google.auth.transport.requests
from google.oauth2.service_account import IDTokenCredentials
# The service account JSON key file to use to create the Identity Token
sa_filename = 'service-account.json'
# Endpoint to call
endpoint = 'https://us-east1-replace_with_project_id.cloudfunctions.net/main'
# The audience that this ID token is intended for (example Google Cloud Functions service URL)
aud = 'https://us-east1-replace_with_project_id.cloudfunctions.net/main'
def invoke_endpoint(url, id_token):
headers = {'Authorization': 'Bearer ' + id_token}
r = requests.get(url, headers=headers)
if r.status_code != 200:
print('Calling endpoint failed')
print('HTTP Status Code:', r.status_code)
print(r.content)
return None
return r.content.decode('utf-8')
if __name__ == '__main__':
credentials = IDTokenCredentials.from_service_account_file(
sa_filename,
target_audience=aud)
request = google.auth.transport.requests.Request()
credentials.refresh(request)
# This is debug code to show how to decode Identity Token
# print('Decoded Identity Token:')
# print_jwt(credentials.token.encode())
response = invoke_endpoint(endpoint, credentials.token)
if response is not None:
print(response)