feat:Bigquery ADK support for search catalog tool

contributing/samples/bigquery/README.md

@@ -55,6 +55,9 @@ distributed via the `google.adk.tools.bigquery` module. These tools include:
   `ARIMA_PLUS` model and then querying it with
   `ML.DETECT_ANOMALIES` to detect time series data anomalies.
 
+11. `search_catalog`
+  Searches for data entries across projects using the Dataplex Catalog. This allows discovery of datasets, tables, and other assets.
+
 ## How to use
 
 Set up environment variables in your `.env` file for using

pyproject.toml

@@ -37,6 +37,7 @@ dependencies = [
   "google-cloud-bigquery-storage>=2.0.0",
   "google-cloud-bigquery>=2.2.0",
   "google-cloud-bigtable>=2.32.0",                           # For Bigtable database
+  "google-cloud-dataplex>=1.7.0,<3.0.0",                    # For Dataplex Catalog Search tool
   "google-cloud-discoveryengine>=0.13.12, <0.14.0",          # For Discovery Engine Search Tool
   "google-cloud-pubsub>=2.0.0, <3.0.0",                      # For Pub/Sub Tool
   "google-cloud-secret-manager>=2.22.0, <3.0.0",             # Fetching secrets in RestAPI Tool

src/google/adk/tools/bigquery/bigquery_credentials.py

@@ -19,7 +19,10 @@
 from .._google_credentials import BaseGoogleCredentialsConfig
 
 BIGQUERY_TOKEN_CACHE_KEY = "bigquery_token_cache"
-BIGQUERY_DEFAULT_SCOPE = ["https://www.googleapis.com/auth/bigquery"]
+BIGQUERY_SCOPES = [
+    "https://www.googleapis.com/auth/bigquery",
+    "https://www.googleapis.com/auth/cloud-platform",
+]
 
 
 @experimental(FeatureName.GOOGLE_CREDENTIALS_CONFIG)
@@ -34,8 +37,7 @@ def __post_init__(self) -> BigQueryCredentialsConfig:
     super().__post_init__()
 
     if not self.scopes:
-      self.scopes = BIGQUERY_DEFAULT_SCOPE
-
+      self.scopes = BIGQUERY_SCOPES
     # Set the token cache key
     self._token_cache_key = BIGQUERY_TOKEN_CACHE_KEY
 

src/google/adk/tools/bigquery/bigquery_toolset.py

@@ -24,6 +24,7 @@
 from . import data_insights_tool
 from . import metadata_tool
 from . import query_tool
+from . import search_tool
 from ...features import experimental
 from ...features import FeatureName
 from ...tools.base_tool import BaseTool
@@ -87,6 +88,7 @@ async def get_tools(
             query_tool.analyze_contribution,
             query_tool.detect_anomalies,
             data_insights_tool.ask_data_insights,
+            search_tool.search_catalog,
         ]
     ]
 

src/google/adk/tools/bigquery/client.py

@@ -17,12 +17,16 @@
 from typing import Optional
 
 import google.api_core.client_info
+from google.api_core.gapic_v1 import client_info as gapic_client_info
 from google.auth.credentials import Credentials
 from google.cloud import bigquery
+from google.cloud import dataplex_v1
 
 from ... import version
 
-USER_AGENT = f"adk-bigquery-tool google-adk/{version.__version__}"
+USER_AGENT_BASE = f"google-adk/{version.__version__}"
+BQ_USER_AGENT = f"adk-bigquery-tool {USER_AGENT_BASE}"
+DP_USER_AGENT = f"adk-dataplex-tool {USER_AGENT_BASE}"
 
 
 from typing import List
@@ -48,7 +52,7 @@ def get_bigquery_client(
     A BigQuery client.
   """
 
-  user_agents = [USER_AGENT]
+  user_agents = [BQ_USER_AGENT]
   if user_agent:
     if isinstance(user_agent, str):
       user_agents.append(user_agent)
@@ -67,3 +71,35 @@ def get_bigquery_client(
   )
 
   return bigquery_client
+
+
+def get_dataplex_catalog_client(
+    *,
+    credentials: Credentials,
+    user_agent: Optional[Union[str, List[str]]] = None,
+) -> dataplex_v1.CatalogServiceClient:
+  """Get a Dataplex CatalogServiceClient with minimal necessary arguments.
+
+  Args:
+    credentials: The credentials to use for the request.
+    user_agent: Additional user agent string(s) to append.
+
+  Returns:
+    A Dataplex Client.
+  """
+
+  user_agents = [DP_USER_AGENT]
+  if user_agent:
+    if isinstance(user_agent, str):
+      user_agents.append(user_agent)
+    else:
+      user_agents.extend([ua for ua in user_agent if ua])
+
+  client_info = gapic_client_info.ClientInfo(user_agent=" ".join(user_agents))
+
+  dataplex_client = dataplex_v1.CatalogServiceClient(
+      credentials=credentials,
+      client_info=client_info,
+  )
+
+  return dataplex_client

src/google/adk/tools/bigquery/search_tool.py

@@ -0,0 +1,148 @@
+# Copyright 2026 Google LLC
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Optional
+
+from google.api_core import exceptions as api_exceptions
+from google.auth.credentials import Credentials
+from google.cloud import dataplex_v1
+
+from . import client
+from .config import BigQueryToolConfig
+
+
+def _construct_search_query_helper(
+    predicate: str, operator: str, items: List[str]
+) -> str:
+  if not items:
+    return ""
+  if len(items) == 1:
+    return f'{predicate}{operator}"{items[0]}"'
+
+  clauses = [f'{predicate}{operator}"{item}"' for item in items]
+  return "(" + " OR ".join(clauses) + ")"
+
+
+def search_catalog(
+    prompt: str,
+    project_id: str,
+    credentials: Credentials,
+    settings: BigQueryToolConfig,
+    location: Optional[str] = None,
+    page_size: int = 10,
+    project_ids_filter: Optional[List[str]] = None,
+    dataset_ids_filter: Optional[List[str]] = None,
+    types_filter: Optional[List[str]] = None,
+) -> Dict[str, Any]:
+  """Search for BigQuery assets within Dataplex.
+
+  Args:
+      prompt (str): The base search query (natural language or keywords).
+      project_id (str): The Google Cloud project ID to scope the search.
+      credentials (Credentials): Credentials for the request.
+      settings (BigQueryToolConfig): BigQuery tool settings.
+      location (str): The Dataplex location to use.
+      page_size (int): Maximum number of results.
+      project_ids_filter (Optional[List[str]]): Specific project IDs to include in the search results.
+                                              If None, defaults to the scoping project_id.
+      dataset_ids_filter (Optional[List[str]]): BigQuery dataset IDs to filter by.
+      types_filter (Optional[List[str]]): Entry types to filter by (e.g., "TABLE", "DATASET").
+
+  Returns:
+      dict: Search results or error.
+  """
+  try:
+    if not project_id:
+      return {
+          "status": "ERROR",
+          "error_details": "project_id must be provided.",
+      }
+
+    dataplex_client = client.get_dataplex_catalog_client(
+        credentials=credentials,
+        user_agent=[settings.application_name, "search_catalog"],
+    )
+
+    query_parts = []
+    if prompt:
+      query_parts.append(f"({prompt})")
+
+    # Filter by project IDs
+    projects_to_filter = (
+        project_ids_filter if project_ids_filter else [project_id]
+    )
+    if projects_to_filter:
+      query_parts.append(
+          _construct_search_query_helper("projectid", "=", projects_to_filter)
+      )
+
+    # Filter by dataset IDs
+    if dataset_ids_filter:
+      dataset_resource_filters = [
+          f'linked_resource:"//bigquery.googleapis.com/projects/{pid}/datasets/{did}/*"'
+          for pid in projects_to_filter
+          for did in dataset_ids_filter
+      ]
+      if dataset_resource_filters:
+        query_parts.append(f"({' OR '.join(dataset_resource_filters)})")
+    # Filter by entry types
+    if types_filter:
+      query_parts.append(
+          _construct_search_query_helper("type", "=", types_filter)
+      )
+
+    # Always scope to BigQuery system
+    query_parts.append("system=BIGQUERY")
+
+    full_query = " AND ".join(filter(None, query_parts))
+
+    search_location = location or settings.location or "global"
+    search_scope = f"projects/{project_id}/locations/{search_location}"
+
+    request = dataplex_v1.SearchEntriesRequest(
+        name=search_scope,
+        query=full_query,
+        page_size=page_size,
+        semantic_search=True,
+    )
+
+    response = dataplex_client.search_entries(request=request)
+
+    results = []
+    for result in response.results:
+      entry = result.dataplex_entry
+      source = entry.entry_source
+      results.append({
+          "name": entry.name,
+          "display_name": source.display_name or "",
+          "entry_type": entry.entry_type,
+          "update_time": str(entry.update_time),
+          "linked_resource": source.resource or "",
+          "description": source.description or "",
+          "location": source.location or "",
+      })
+    return {"status": "SUCCESS", "results": results}
+
+  except api_exceptions.GoogleAPICallError as e:
+    logging.exception("search_catalog tool: API call failed")
+    return {"status": "ERROR", "error_details": f"Dataplex API Error: {str(e)}"}
+  except Exception as ex:
+    logging.exception("search_catalog tool: Unexpected error")
+    return {"status": "ERROR", "error_details": str(ex)}

tests/unittests/tools/bigquery/test_bigquery_credentials.py

@@ -46,7 +46,10 @@ def test_valid_credentials_object_auth_credentials(self):
     assert config.credentials == auth_creds
     assert config.client_id is None
     assert config.client_secret is None
-    assert config.scopes == ["https://www.googleapis.com/auth/bigquery"]
+    assert config.scopes == [
+        "https://www.googleapis.com/auth/bigquery",
+        "https://www.googleapis.com/auth/cloud-platform",
+    ]
 
   def test_valid_credentials_object_oauth2_credentials(self):
     """Test that providing valid Credentials object works correctly with
@@ -86,7 +89,10 @@ def test_valid_client_id_secret_pair_default_scope(self):
     assert config.credentials is None
     assert config.client_id == "test_client_id"
     assert config.client_secret == "test_client_secret"
-    assert config.scopes == ["https://www.googleapis.com/auth/bigquery"]
+    assert config.scopes == [
+        "https://www.googleapis.com/auth/bigquery",
+        "https://www.googleapis.com/auth/cloud-platform",
+    ]
 
   def test_valid_client_id_secret_pair_w_scope(self):
     """Test that providing client ID and secret with explicit scopes works.
@@ -128,7 +134,10 @@ def test_valid_client_id_secret_pair_w_empty_scope(self):
     assert config.credentials is None
     assert config.client_id == "test_client_id"
     assert config.client_secret == "test_client_secret"
-    assert config.scopes == ["https://www.googleapis.com/auth/bigquery"]
+    assert config.scopes == [
+        "https://www.googleapis.com/auth/bigquery",
+        "https://www.googleapis.com/auth/cloud-platform",
+    ]
 
   def test_missing_client_secret_raises_error(self):
     """Test that missing client secret raises appropriate validation error.