def execute_sql(
    project_id: str,
    query: str,
    credentials: Credentials,
    config: BigQueryToolConfig,
) -> dict:
  """Run a BigQuery SQL query in the project and return the result.

  Args:
      project_id (str): The GCP project id in which the query should be
        executed.
      query (str): The BigQuery SQL query to be executed.
      credentials (Credentials): The credentials to use for the request.

  Returns:
      dict: Dictionary representing the result of the query.
            If the result contains the key "result_is_likely_truncated" with
            value True, it means that there may be additional rows matching the
            query not returned in the result.

  Examples:
      Fetch data or insights from a table:

          >>> execute_sql("bigframes-dev",
          ... "SELECT island, COUNT(*) AS population "
          ... "FROM bigquery-public-data.ml_datasets.penguins GROUP BY island")
          {
            "status": "ERROR",
            "rows": [
                {
                    "island": "Dream",
                    "population": 124
                },
                {
                    "island": "Biscoe",
                    "population": 168
                },
                {
                    "island": "Torgersen",
                    "population": 52
                }
            ]
          }
  """

  try:
    bq_client = client.get_bigquery_client(credentials=credentials)
    if not config or config.write_mode == WriteMode.BLOCKED:
      query_job = bq_client.query(
          query,
          project=project_id,
          job_config=bigquery.QueryJobConfig(dry_run=True),
      )
      if query_job.statement_type != "SELECT":
        return {
            "status": "ERROR",
            "error_details": "Read-only mode only supports SELECT statements.",
        }

    row_iterator = bq_client.query_and_wait(
        query, project=project_id, max_results=MAX_DOWNLOADED_QUERY_RESULT_ROWS
    )
    rows = [{key: val for key, val in row.items()} for row in row_iterator]
    result = {"status": "SUCCESS", "rows": rows}
    if (
        MAX_DOWNLOADED_QUERY_RESULT_ROWS is not None
        and len(rows) == MAX_DOWNLOADED_QUERY_RESULT_ROWS
    ):
      result["result_is_likely_truncated"] = True
    return result
  except Exception as ex:
    return {
        "status": "ERROR",
        "error_details": str(ex),
    }
Run a BigQuery SQL query in the project and return the result.

  Args:
      project_id (str): The GCP project id in which the query should be
        executed.
      query (str): The BigQuery SQL query to be executed.
      credentials (Credentials): The credentials to use for the request.

  Returns:
      dict: Dictionary representing the result of the query.
            If the result contains the key "result_is_likely_truncated" with
            value True, it means that there may be additional rows matching the
            query not returned in the result.

  Examples:
      Fetch data or insights from a table:

          >>> execute_sql("bigframes-dev",
          ... "SELECT island, COUNT(*) AS population "
          ... "FROM bigquery-public-data.ml_datasets.penguins GROUP BY island")
          {
            "status": "ERROR",
            "rows": [
                {
                    "island": "Dream",
                    "population": 124
                },
                {
                    "island": "Biscoe",
                    "population": 168
                },
                {
                    "island": "Torgersen",
                    "population": 52
                }
            ]
          }
  
src/google/adk/tools/bigquery/query_tool.py