Question: For a requirement I want to call/invoke a cloud function from inside a cloud composer pipeline but I cant find much info on it, I tried using SimpleHTTP airflow operator but I get this error: [2021-09-10 10:35:46,649] {taskinstance.py:1503} ERROR - Task failed with exception Traceback (most recent call last): File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1333, in _prepare_and_execute_task_with_callbacks result = self._execute_task(context, task_copy) File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1363, in _execute_task result = task_copy.execute(context=context) File "/home/airflow/gcs/dags/to_gcf.py", line 51, in execute if not self.response_check(response): File "/home/airflow/gcs/dags/to_gcf.py", line 83, in <lambda> response_check=lambda response: False if len(response.json()) == 0 else True, File "/opt/python3.8/lib/python3.8/site-packages/requests/models.py", line 900, in json return complexjson.loads(self.text, **kwargs) File "/opt/python3.8/lib/python3.8/json/__init__.py", line 357, in loads return _default_decoder.decode(s) File "/opt/python3.8/lib/python3.8/json/decoder.py", line 337, in decode obj, end = self.raw_decode(s, idx=_w(s, 0).end()) File "/opt/python3.8/lib/python3.8/json/decoder.py", line 355, in raw_decode raise JSONDecodeError("Expecting value", s, err.value) from None  Thanks in advance!!

Answer: I faced the same issue as you, but I managed to figure it out by studying the Airflow 2.0 provider packages for Google and using a PythonOperator instead. from airflow.providers.google.common.utils import id_token_credentials as id_token_credential_utils import google.auth.transport.requests from google.auth.transport.requests import AuthorizedSession def invoke_cloud_function(): url = "<your_cloud_function_url>" #the url is also the target audience. request = google.auth.transport.requests.Request() #this is a request for obtaining the the credentials id_token_credentials = id_token_credential_utils.get_default_id_token_credentials(url, request=request) # If your cloud function url has query parameters, remove them before passing to the audience resp = AuthorizedSession(id_token_credentials).request("GET", url=url) # the authorized session object is used to access the Cloud Function print(resp.status_code) # should return 200 print(resp.content) # the body of the HTTP response  Thus, invoke the function as below:  task = PythonOperator(task_id="invoke_cf", python_callable=invoke_cloud_function)  From my understanding, accessing an authenticated HTTP Cloud Function strictly requires a credential based on ID Tokens. Thus to obtain the required type of credentials, get_default_id_token_credentials() executes the Application Default Credentials(ADC) authorization flow, which is a process that obtains credentials from environment variables, known locations. or the Compute Engine metadata server. Composer should have the associated service account keyfile made avaliable via environment variables (probably GOOGLE_APPLICATION_CREDENTIALS). Once you have the right type of credentials, you can use the AuthorizedSessions object to authenticate your requests to the cloud function.