{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": { "id": "copyright" }, "outputs": [], "source": [ "# Copyright 2026 Google LLC\n", "#\n", "# Licensed under the Apache License, Version 2.0 (the \"License\");\n", "# you may not use this file except in compliance with the License.\n", "# You may obtain a copy of the License at\n", "#\n", "# https://www.apache.org/licenses/LICENSE-2.0\n", "#\n", "# Unless required by applicable law or agreed to in writing, software\n", "# distributed under the License is distributed on an \"AS IS\" BASIS,\n", "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n", "# See the License for the specific language governing permissions and\n", "# limitations under the License." ] }, { "cell_type": "markdown", "metadata": { "id": "header" }, "source": [ "# Gemini Data Analytics: A2A HTTP API Sample\n", "\n", "This notebook demonstrates how to interact with the **DataA2AService** using standard HTTP requests. This is useful for environments where a high-level SDK is not available or when you want to minimize dependencies." ] }, { "cell_type": "markdown", "metadata": { "id": "view-in-github" }, "source": [ "\n", " \n", " \n", " \n", " \n", "
\n", " \n", " \"Google
Open in Colab\n", " \n", " 		\n", " \n", " \"Google
Open in Colab Enterprise\n", " \n", " 		\n", " \n", " \"Vertex
Open in Vertex AI Workbench\n", " \n", " 		\n", " \n", " \"GitHub
View on GitHub\n", " \n", "
\n", "\n", "
\n" ] }, { "cell_type": "markdown", "metadata": { "id": "background-and-overview" }, "source": [ "# Background and Overview\n", "The **Conversational Analytics API** (also known as Gemini Data Analytics) lets you chat with your BigQuery or Looker data anywhere. This notebook demonstrates how to use the **A2A (Agent-to-Agent)** interface via standard HTTP requests. This is useful for environments where a high-level SDK is not available or when you want to minimize dependencies.\n", "\n", "This is a **Pre-GA** product. See [documentation](https://cloud.google.com/gemini/docs/conversational-analytics-api/overview) for more details.\n", "\n", "Please provide feedback to conversational-analytics-api-feedback@google.com\n", "
\n", "### This notebook will help you\n", "1. Authenticate to Google Cloud\n", "2. Retrieve the Agent Card\n", "3. Send asynchronous messages and poll for results\n", "4. Process agent outputs (Artifacts)\n", "5. Cancel active tasks\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "setup", "executionInfo": { "status": "ok", "timestamp": 1776195682585, "user_tz": 420, "elapsed": 138, "user": { "displayName": "", "userId": "" } }, "outputId": "631853b2-2807-478e-d18c-761b72a77788" }, "outputs": [], "source": [ "# @title Setup and Authentication\n", "import json\n", "import os\n", "import time\n", "import uuid\n", "from google.auth import default\n", "from google.auth.transport.requests import Request\n", "from google.colab import auth\n", "import requests\n", "\n", "# Authenticate the user\n", "auth.authenticate_user()\n", "\n", "# Get credentials and project ID\n", "creds, _ = default()\n", "creds.refresh(Request())\n", "access_token = creds.token\n", "\n", "ENDPOINT = \"https://geminidataanalytics.googleapis.com\"\n", "LOCATION = \"global\" # @param {type:\"string\"}\n", "PROJECT_ID = \"[your-project-id]\" # @param {type:\"string\"}\n", "# AGENT_ID can be found from the Cloud URL, e.g.\n", "# https://console.cloud.google.com/bigquery/agents_hub/?project=\n", "AGENT_ID = \"your-agent-id\" # @param {type:\"string\"}\n", "\n", "if not PROJECT_ID or PROJECT_ID == \"[your-project-id]\"\n", " PROJECT_ID = str(os.environ.get(\"GOOGLE_CLOUD_PROJECT\"))\n", "if not LOCATION:\n", " LOCATION = os.environ.get(\"GOOGLE_CLOUD_REGION\")\n", " \n", "TENANT = f\"projects/{PROJECT_ID}/locations/{LOCATION}/agents/{AGENT_ID}\"\n", "BASE_URL = f\"{ENDPOINT}/v1beta/a2a/{TENANT}/v1\"\n", "HEADERS = {\n", " \"Authorization\": f\"Bearer {access_token}\",\n", " \"Content-Type\": \"application/json\",\n", "}\n", "print(f\"Target Tenant: {TENANT}\")\n", "print(f\"Access Token Length: {len(access_token) if access_token else 0}\")\n" ] }, { "cell_type": "markdown", "metadata": { "id": "get-agent-card-header" }, "source": [ "## 1. Get Agent Card\n", "\n", "First, let's retrieve the Agent Card to verify connectivity and see what the agent can do." ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "get-agent-card-code", "outputId": "437c7de7-c2ef-455d-e0b0-5f7bede70971" }, "outputs": [], "source": [ "url = f\"{BASE_URL}/card\"\n", "\n", "try:\n", " response = requests.get(url, headers=HEADERS, timeout=30)\n", " response.raise_for_status()\n", " print(\"Agent Card:\")\n", " print(json.dumps(response.json(), indent=2))\n", "except Exception as e:\n", " print(f\"Error fetching agent card: {e}\")" ] }, { "cell_type": "markdown", "metadata": { "id": "send-message-async-header" }, "source": [ "## 2. Send Message (Asynchronous + Polling)\n", "\n", "For long-running tasks, use `blocking=False`. This returns a `Task` object immediately, which you can poll for status." ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "send-message-async-code" }, "outputs": [], "source": [ "USER_QUERY = \"Hello\" # @param {type:\"string\"}\n", "\n", "def send_async_message(query):\n", " url = f\"{BASE_URL}/message:send\"\n", " payload = {\n", " \"tenant\": TENANT,\n", " \"message\": {\n", " \"message_id\": f\"msg-{uuid.uuid4()}\",\n", " \"role\": \"ROLE_USER\",\n", " \"content\": [{\"text\": query}],\n", " },\n", " \"configuration\": {\"blocking\": False},\n", " }\n", "\n", " response = requests.post(url, headers=HEADERS, json=payload, timeout=30)\n", " response.raise_for_status()\n", "\n", " res_json = response.json()\n", " \n", " task = res_json.get(\"task\")\n", " if task:\n", " print(f\"Task created: {task.get('id')}\")\n", " return task.get(\"id\")\n", " elif \"message\" in res_json:\n", " print(\"Received message directly instead of task.\")\n", " return None\n", " else:\n", " print(\"Response did not contain 'task' or 'message'\")\n", " return None\n", "\n", "\n", "def poll_task(task_id, max_retries=15):\n", " url = f\"{BASE_URL}/tasks/{task_id}\"\n", " retry_count = 0\n", " wait_time = 2\n", "\n", " while retry_count < max_retries:\n", " response = requests.get(url, headers=HEADERS, timeout=30)\n", " response.raise_for_status()\n", "\n", " task = response.json()\n", " state = task.get(\"status\", {}).get(\"state\")\n", " print(f\"Current State: {state}\")\n", "\n", " if state in [\n", " \"TASK_STATE_COMPLETED\",\n", " \"TASK_STATE_FAILED\",\n", " \"TASK_STATE_CANCELLED\",\n", " ]:\n", " return task\n", "\n", " time.sleep(wait_time)\n", " wait_time = min(wait_time * 1.5, 10)\n", " retry_count += 1\n", "\n", " print(\"Polling timed out.\")\n", " return None\n", "\n", "try:\n", " print(\"Starting Send Message...\")\n", " task_id = send_async_message(USER_QUERY)\n", " \n", " if task_id:\n", " print(f\"Polling task {task_id}...\")\n", " final_task = poll_task(task_id)\n", " if final_task:\n", " print(\"Final Task Result:\")\n", " print(json.dumps(final_task, indent=2))\n", " else:\n", " print(\"No task ID returned, skipping polling.\")\n", " \n", "except Exception as e:\n", " print(f\"Error during messaging/polling: {e}\")" ] }, { "cell_type": "markdown", "metadata": { "id": "artifact-parsing-header" }, "source": [ "## 3. Processing Agent Outputs (Artifacts)\n", "\n", "Agents often produce **Artifacts** (structured data, files, or references). Here is how to parse them from a completed task." ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "artifact-parsing-code" }, "outputs": [], "source": [ "USER_QUERY = \"Which item categories had the highest sales last year?\" # @param {type:\"string\"}\n", "\n", "\n", "def get_data_and_extract_artifacts(query):\n", " url = f\"{BASE_URL}/message:send\"\n", " payload = {\n", " \"tenant\": TENANT,\n", " \"message\": {\n", " \"message_id\": f\"msg-{uuid.uuid4()}\",\n", " \"role\": \"ROLE_USER\",\n", " \"content\": [{\"text\": query}],\n", " },\n", " # We set blocking=True to wait for the full response (including artifacts)\n", " \"configuration\": {\"blocking\": True},\n", " }\n", "\n", " print(f\"Sending analysis query: '{query}'\")\n", " print(\"This involves data processing, so it may take 30-60 seconds...\")\n", " \n", " # 120 second timeout to give it plenty of time to compute\n", " response = requests.post(url, headers=HEADERS, json=payload, timeout=120)\n", " response.raise_for_status()\n", "\n", " res_json = response.json()\n", " print(\"\\n--- Response Received ---\")\n", "\n", " # Case 1: The server returned a Task\n", " task = res_json.get(\"task\")\n", " if task:\n", " print(\"Server returned a Task. Processing artifacts...\")\n", " artifacts = task.get(\"artifacts\", [])\n", " display_artifacts(artifacts)\n", " return\n", "\n", " # Case 2: The server returned a Direct Message\n", " message = res_json.get(\"message\")\n", " if message:\n", " print(\"Server returned a direct Message.\")\n", " # Check if there are artifacts attached to the message or in the content\n", " content_parts = message.get(\"content\", [])\n", " \n", " print(f\"Message contains {len(content_parts)} content parts.\")\n", " for part in content_parts:\n", " # Print the text response\n", " if \"text\" in part:\n", " print(f\"\\nText Response:\\n{part['text']}\")\n", " \n", " # Check for metadata that might contain structured data (artifacts)\n", " metadata = part.get(\"metadata\", {})\n", " if metadata:\n", " print(f\"\\nMetadata found: {json.dumps(metadata, indent=2)}\")\n", " return\n", "\n", " print(f\"Unexpected response structure: {res_json}\")\n", "\n", "\n", "def display_artifacts(artifacts):\n", " if not artifacts:\n", " print(\"No artifacts found in the task.\")\n", " return\n", "\n", " print(f\"Found {len(artifacts)} artifacts:\")\n", " for art in artifacts:\n", " name = art.get(\"name\", \"Unnamed\")\n", " \n", " art_type = (\n", " art.get(\"metadata\", {})\n", " .get(\"fields\", {})\n", " .get(\"type\", {})\n", " .get(\"stringValue\", None)\n", " )\n", " \n", " if not art_type:\n", " art_type = art.get(\"metadata\", {}).get(\"type\", \"Unknown\")\n", " \n", " print(f\"\\n- [{art_type}] {name}: {art.get('description')}\")\n", " \n", " for part in art.get(\"parts\", []):\n", " if \"text\" in part:\n", " content = part[\"text\"]\n", " snippet = content[:500] + \"...\" if len(content) > 500 else content\n", " print(f\" Content:\\n{snippet}\")\n", "\n", "try:\n", " get_data_and_extract_artifacts(USER_QUERY)\n", "except Exception as e:\n", " print(f\"Error during artifact extraction: {e}\")" ] }, { "cell_type": "markdown", "metadata": { "id": "cancel-task-header" }, "source": [ "## 4. Cancel an Active Task\n", "\n", "If a task is taking too long or was sent in error, you can cancel it." ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "cancel-task-code" }, "outputs": [], "source": [ "USER_QUERY = \"Which item categories had the highest sales last year?\" # @param {type:\"string\"}\n", "\n", "def cancel_task(task_id):\n", " url = f\"{BASE_URL}/tasks/{task_id}:cancel\"\n", " response = requests.post(url, headers=HEADERS, timeout=30)\n", " response.raise_for_status()\n", "\n", " print(f\"Task {task_id} cancellation requested.\")\n", " return response.json()\n", "\n", "def send_async_message_with_timeout(query, timeout_secs=60):\n", " url = f\"{BASE_URL}/message:send\"\n", " payload = {\n", " \"tenant\": TENANT,\n", " \"message\": {\n", " \"message_id\": f\"msg-{uuid.uuid4()}\",\n", " \"role\": \"ROLE_USER\",\n", " \"content\": [{\"text\": query}],\n", " },\n", " \"configuration\": {\"blocking\": False},\n", " }\n", "\n", " response = requests.post(url, headers=HEADERS, json=payload, timeout=timeout_secs)\n", " response.raise_for_status()\n", " \n", " res_json = response.json()\n", " task = res_json.get(\"task\")\n", " if task:\n", " return task.get(\"id\")\n", " return None\n", "\n", "try:\n", " print(f\"Starting task with 60s timeout: '{USER_QUERY}'\")\n", " \n", " new_task_id = send_async_message_with_timeout(USER_QUERY, timeout_secs=60)\n", " \n", " if new_task_id:\n", " print(f\"Task created! ID: {new_task_id}. Cancelling now...\")\n", " cancel_response = cancel_task(new_task_id)\n", " print(f\"Cancel Response: {cancel_response}\")\n", " else:\n", " print(\"No task ID returned (the server might have answered immediately).\")\n", " \n", "except Exception as e:\n", " print(f\"Error during cancellation demo: {e}\")" ] }, { "cell_type": "markdown", "metadata": { "id": "cleanup-header" }, "source": [ "## 5. Cleanup\n", "\n", "It is good practice to clean up any temporary resources or state created during your session." ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "cleanup-code" }, "outputs": [], "source": [ "# @title Resource Cleanup\n", "print(\n", " \"No specific cloud resources were created in this demo that require manual\"\n", " \" deletion (e.g., storage buckets).\"\n", ")\n", "print(\n", " \"However, you can use this section to reset any local session state if\"\n", " \" needed.\"\n", ")" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.12" }, "colab": { "provenance": [] } }, "nbformat": 4, "nbformat_minor": 0 }