Add Google API toolset for agents to access 1000+ APIs

providers/common/ai/docs/index.rst

@@ -254,6 +254,7 @@ Extra           Dependencies
 ``parquet``     ``pyarrow>=18.0.0; python_version < '3.14'``, ``pyarrow>=22.0.0; python_version >= '3.14'``
 ``sql``         ``apache-airflow-providers-common-sql>=1.33.0``, ``sqlglot>=30.0.0``
 ``aws``         ``apache-airflow-providers-amazon>=9.0.0``
+``gcp``         ``apache-airflow-providers-google>=22.0.0``
 ``common.sql``  ``apache-airflow-providers-common-sql>=1.33.0``
 ``langchain``   ``langchain>=1.0.0``
 ``llamaindex``  ``dataclasses-json>=0.6.7``, ``llama-index-core>=0.13.0``, ``llama-index-embeddings-openai>=0.6.0``, ``llama-index-llms-openai>=0.6.0``

providers/common/ai/docs/toolsets.rst

@@ -24,20 +24,23 @@ Airflow's 350+ provider hooks already have typed methods, rich docstrings,
 and managed credentials. Toolsets expose them as pydantic-ai tools so that
 LLM agents can call them during multi-turn reasoning.
 
-Four toolsets are included:
+Five core toolsets are included:
 
 - :class:`~airflow.providers.common.ai.toolsets.aws.AWSToolset` — configured
   AWS services toolset for agent access to AWS APIs through Airflow-managed
   AWS connections.
 - :class:`~airflow.providers.common.ai.toolsets.hook.HookToolset` — generic
   adapter for any Airflow Hook.
+- :class:`~airflow.providers.common.ai.toolsets.google.GoogleCloudToolset` —
+  allow-listed access to Google APIs published through Google's Discovery
+  service.
 - :class:`~airflow.providers.common.ai.toolsets.mcp.MCPToolset` — connect to
   `MCP servers <https://modelcontextprotocol.io/>`__ configured via Airflow
   connections.
 - :class:`~airflow.providers.common.ai.toolsets.sql.SQLToolset` — curated
   4-tool database toolset.
 
-All four implement pydantic-ai's
+All five implement pydantic-ai's
 `AbstractToolset <https://ai.pydantic.dev/toolsets/>`__ interface and can be
 passed to any pydantic-ai ``Agent``, including via
 :class:`~airflow.providers.common.ai.operators.agent.AgentOperator`.
@@ -51,6 +54,44 @@ passed to any pydantic-ai ``Agent``, including via
     integration, and the connection UI, but you are not locked in.
 
 
+Choosing between HookToolset and provider API toolsets
+------------------------------------------------------
+
+For provider-backed actions, Airflow supports two complementary approaches.
+
+Use ``HookToolset`` when the task maps to methods that already exist on an
+Airflow Hook. Its allow-list contains Python method names on that hook, such as
+``list`` or ``get_records``. This is a good fit when the provider already wraps
+the operation in a stable, workflow-oriented hook method.
+
+Use provider API toolsets, such as ``AWSToolset`` and ``GoogleCloudToolset``,
+when the task needs selected operations from the cloud provider API surface
+itself. Their allow-lists use the provider's operation language, such as AWS
+actions or Google Discovery REST methods. This is useful when the provider API
+has operations that are not wrapped by Airflow Hooks, or when one agent task
+needs to inspect several low-level services during troubleshooting.
+
+For example, use ``HookToolset`` when an agent should call an existing storage
+hook to list objects, an existing database hook to run a query, or an existing
+HTTP hook to call an internal API.
+
+Use a provider API toolset when an agent should compare signals across services,
+such as storage object arrivals, query job status, and monitoring time series,
+or inspect cloud resources and configuration where Airflow does not provide a
+dedicated hook method.
+
+The distinction is the allow-list contract:
+
+- ``HookToolset`` allow-lists Airflow Hook method names.
+- ``AWSToolset`` allow-lists AWS service/API actions.
+- ``GoogleCloudToolset`` allow-lists Google Discovery REST methods.
+
+Toolset allow-lists are application-level guardrails: they limit what the agent
+can ask the toolset to call. They are not a replacement for least-privilege
+cloud credentials. The connection used by the toolset should still be scoped to
+the minimum permissions needed for the agent's task.
+
+
 Using Toolsets Directly with PydanticAI
 ---------------------------------------
 
@@ -207,6 +248,78 @@ Parameters
   (``DESCRIBE``/``SHOW``) statements are permitted.
 - ``max_rows``: Maximum rows returned from the ``query`` tool. Default ``50``.
 
+``GoogleCloudToolset``
+----------------------
+
+Curated toolset that gives an agent allow-listed access to Google APIs
+published through Google's Discovery REST surface. Requires the ``gcp`` extra:
+``pip install "apache-airflow-providers-common-ai[gcp]"``.
+
+.. exampleinclude:: /../../ai/src/airflow/providers/common/ai/example_dags/example_gcp_toolset.py
+    :language: python
+    :start-after: [START howto_operator_agent_gcp]
+    :end-before: [END howto_operator_agent_gcp]
+
+For a multi-service troubleshooting example, use a small read-only allow-list
+across Cloud Storage, BigQuery, and Cloud Monitoring:
+
+.. exampleinclude:: /../../ai/src/airflow/providers/common/ai/example_dags/example_gcp_troubleshooting_agent.py
+    :language: python
+    :start-after: [START howto_operator_agent_gcp_troubleshooting]
+    :end-before: [END howto_operator_agent_gcp_troubleshooting]
+
+``GoogleCloudToolset`` exposes ``list_gcp_methods``,
+``describe_gcp_method``, and ``call_gcp``. The allow-list is required and
+deny-by-default. Entries use ``"<api>/<version>:<resource.method>"`` form:
+
+.. code-block:: python
+
+    GoogleCloudToolset(
+        gcp_conn_id="google_cloud_default",
+        allowed_methods=[
+            "storage/v1:buckets.list",
+            "storage/v1:objects.list",
+            "pubsub/v1:projects.topics.list",
+            "bigquery/v2:jobs.query",
+        ],
+    )
+
+The API and version must be explicit, for example ``pubsub/v1`` rather than
+``pubsub``. The method part accepts ``*``/``?`` wildcards, but methods that
+return credentials or decrypted secrets are never matched by a wildcard; each
+must be listed verbatim.
+
+The toolset covers APIs present in the bundled Discovery documents shipped
+with ``google-api-python-client``. This includes many Google Cloud control and
+metadata APIs such as Cloud Storage, Pub/Sub, BigQuery REST methods, Compute
+Engine, Cloud SQL Admin, and Workspace APIs that publish Discovery documents.
+It does not cover APIs outside that Discovery REST surface, and credentials
+still need IAM permissions for the requested call.
+
+Credentials, impersonation, and the project come from ``gcp_conn_id`` via the
+Google provider. With ``enforce_project=True`` (the default), missing
+``project``/``projectId`` parameters are filled from the connection, and
+model-supplied values for another project are rejected.
+
+Parameters
+^^^^^^^^^^
+
+- ``gcp_conn_id``: Airflow Google connection ID. Default
+  ``"google_cloud_default"``.
+- ``allowed_methods``: Required list of allowed methods in
+  ``"<api>/<version>:<resource.method>"`` form. The method part accepts
+  wildcards; the API and version do not.
+- ``impersonation_chain``: Optional service account or chain passed through to
+  ``GoogleBaseHook``.
+- ``enforce_project``: Fill or reject project parameters based on the
+  connection's project. Default ``True``.
+- ``allow_remote_discovery``: Allow APIs missing from the bundled Discovery
+  documents and fetch their documents lazily. Default ``False``.
+- ``max_pages``: Maximum number of pages followed for paginated methods.
+  Default ``5``.
+- ``max_output_bytes``: Maximum serialized response size returned to the
+  agent. Larger responses are truncated and marked as such. Default ``65536``.
+
 ``DataFusionToolset``
 ---------------------
 

providers/common/ai/provider.yaml

@@ -455,6 +455,7 @@ toolsets:
   - integration-name: Common AI
     python-modules:
       - airflow.providers.common.ai.toolsets.aws
+      - airflow.providers.common.ai.toolsets.google
       - airflow.providers.common.ai.toolsets.hook
       - airflow.providers.common.ai.toolsets.sql
       - airflow.providers.common.ai.toolsets.datafusion

providers/common/ai/pyproject.toml

@@ -113,6 +113,12 @@ dependencies = [
 "aws" = [
     "apache-airflow-providers-amazon>=9.0.0",
 ]
+# GoogleCloudToolset: allow-listed Google API access for agents. The google
+# provider supplies credential resolution (GoogleBaseHook) and brings
+# google-api-python-client with the bundled discovery documents.
+"gcp" = [
+    "apache-airflow-providers-google>=22.0.0",
+]
 "common.sql" = [
     "apache-airflow-providers-common-sql>=1.33.0"
 ]

providers/common/ai/src/airflow/providers/common/ai/example_dags/example_gcp_toolset.py

@@ -0,0 +1,53 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Example Dag: agent with allow-listed Google API access via GoogleCloudToolset."""
+
+from __future__ import annotations
+
+from airflow.providers.common.ai.operators.agent import AgentOperator
+from airflow.providers.common.ai.toolsets.google import GoogleCloudToolset
+from airflow.providers.common.compat.sdk import dag
+
+
+# [START howto_operator_agent_gcp]
+@dag(tags=["example"])
+def example_agent_gcp_toolset():
+    AgentOperator(
+        task_id="gcs_auditor",
+        prompt="Which buckets exist, and roughly how many objects are in 'data-lake-raw'?",
+        llm_conn_id="pydanticai_default",
+        system_prompt=(
+            "You are a Google Cloud operations assistant. Discover what you "
+            "are allowed to call, check parameter shapes before calling, and "
+            "answer with concrete numbers."
+        ),
+        toolsets=[
+            GoogleCloudToolset(
+                gcp_conn_id="google_cloud_default",
+                allowed_methods=[
+                    "storage/v1:buckets.list",
+                    "storage/v1:buckets.get",
+                    "storage/v1:objects.list",
+                ],
+            )
+        ],
+    )
+
+
+# [END howto_operator_agent_gcp]
+
+example_agent_gcp_toolset()

providers/common/ai/src/airflow/providers/common/ai/example_dags/example_gcp_troubleshooting_agent.py

@@ -0,0 +1,87 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Dag: troubleshoot a Google Cloud data pipeline with GoogleCloudToolset.
+
+Required Airflow Variables:
+- ``common_ai_gcp_project_id``
+- ``common_ai_gcp_bucket``
+
+Optional Airflow Variables:
+- ``common_ai_gcp_dataset``; default: ``pipeline_observability``
+- ``common_ai_gcp_pipeline_table``; default: ``pipeline_runs``
+- ``common_ai_gcp_raw_prefix``; default: ``raw/``
+
+For a useful demo, point the variables at a project with:
+- GCS objects under ``gs://<bucket>/<raw_prefix>``
+- a BigQuery table with recent pipeline status rows
+- Cloud Monitoring metrics from recent BigQuery and GCS activity
+"""
+
+from __future__ import annotations
+
+from airflow.providers.common.ai.operators.agent import AgentOperator
+from airflow.providers.common.ai.toolsets.google import GoogleCloudToolset
+from airflow.providers.common.compat.sdk import Variable, dag, task
+
+
+# [START howto_operator_agent_gcp_troubleshooting]
+@dag(tags=["example"])
+def example_gcp_troubleshooting_agent():
+    @task
+    def build_triage_prompt() -> str:
+        project_id = Variable.get("common_ai_gcp_project_id")
+        bucket_name = Variable.get("common_ai_gcp_bucket")
+        dataset_id = Variable.get("common_ai_gcp_dataset", default="pipeline_observability")
+        pipeline_table = Variable.get("common_ai_gcp_pipeline_table", default="pipeline_runs")
+        raw_prefix = Variable.get("common_ai_gcp_raw_prefix", default="raw/")
+
+        return (
+            f"Pipeline is slow today in project {project_id}. "
+            f"Check whether raw files arrived in gs://{bucket_name}/{raw_prefix}, "
+            "inspect recent BigQuery jobs and rows from "
+            f"{dataset_id}.{pipeline_table}, check active Cloud Monitoring alert policies, "
+            "and compare BigQuery query count and GCS request count metrics for today versus yesterday. "
+            "Summarize the likely cause and the next action."
+        )
+
+    AgentOperator(
+        task_id="triage_pipeline",
+        prompt=build_triage_prompt(),
+        llm_conn_id="pydanticai_default",
+        system_prompt="Use the available Google tools to investigate the pipeline. Do not invent missing values.",
+        toolsets=[
+            GoogleCloudToolset(
+                gcp_conn_id="google_cloud_default",
+                allowed_methods=[
+                    "storage/v1:objects.list",
+                    "bigquery/v2:jobs.list",
+                    "bigquery/v2:jobs.get",
+                    "bigquery/v2:jobs.query",
+                    "monitoring/v3:projects.alertPolicies.list",
+                    "monitoring/v3:projects.timeSeries.list",
+                ],
+                max_pages=2,
+                max_output_bytes=20000,
+            )
+        ],
+    )
+
+
+# [END howto_operator_agent_gcp_troubleshooting]
+
+example_gcp_troubleshooting_agent()

providers/common/ai/src/airflow/providers/common/ai/toolsets/__init__.py

@@ -20,7 +20,14 @@
 
 from airflow.providers.common.ai.toolsets.hook import HookToolset
 
-__all__ = ["AWSToolset", "HookToolset", "MCPToolset", "SQLToolset", "airflow_toolset_to_langchain_tools"]
+__all__ = [
+    "AWSToolset",
+    "GoogleCloudToolset",
+    "HookToolset",
+    "MCPToolset",
+    "SQLToolset",
+    "airflow_toolset_to_langchain_tools",
+]
 
 
 def __getattr__(name: str):
@@ -54,4 +61,12 @@ def __getattr__(name: str):
 
             raise AirflowOptionalProviderFeatureException() from e
         return AWSToolset
+    if name == "GoogleCloudToolset":
+        try:
+            from airflow.providers.common.ai.toolsets.google import GoogleCloudToolset
+        except ImportError as e:
+            from airflow.providers.common.compat.sdk import AirflowOptionalProviderFeatureException
+
+            raise AirflowOptionalProviderFeatureException() from e
+        return GoogleCloudToolset
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")