googlecloudplatform--generative-ai/agents/gemini_data_analytics/a2a_sdk_sample.ipynb

{
  "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 SDK API Sample\n",
        "\n",
        "This notebook demonstrates how to interact with the **DataA2Aservice** using the high-level A2A Python SDK."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github"
      },
      "source": [
        "<table align=\"left\">\n",
        "  <td style=\"text-align: center\">\n",
        "    <a href=\"https://colab.research.google.com/github/GoogleCloudPlatform/generative-ai/blob/main/agents/gemini_data_analytics/a2a_sdk_sample.ipynb\">\n",
        "      <img width=\"32px\" src=\"https://www.gstatic.com/pantheon/images/bigquery/welcome_page/colab-logo.svg\" alt=\"Google Colaboratory logo\"><br> Open in Colab\n",
        "    </a>\n",
        "  </td>\n",
        "  <td style=\"text-align: center\">\n",
        "    <a href=\"https://console.cloud.google.com/vertex-ai/colab/import/https:%2F%2Fraw.githubusercontent.com%2FGoogleCloudPlatform%2Fgenerative-ai%2Fmain%2Fagents%2Fgemini_data_analytics%2Fa2a_sdk_sample.ipynb\">\n",
        "      <img width=\"32px\" src=\"https://lh3.googleusercontent.com/JmcxdQi-qOpctIvWKgPtrzZdJJK-J3sWE1RsfjZNwshCFgE_9fULcNpuXYTilIR2hjwN\" alt=\"Google Cloud Colab Enterprise logo\"><br> Open in Colab Enterprise\n",
        "    </a>\n",
        "  </td>\n",
        "  <td style=\"text-align: center\">\n",
        "    <a href=\"https://console.cloud.google.com/vertex-ai/workbench/deploy-notebook?download_url=https://raw.githubusercontent.com/GoogleCloudPlatform/generative-ai/main/agents/gemini_data_analytics/a2a_sdk_sample.ipynb\">\n",
        "      <img src=\"https://www.gstatic.com/images/branding/gcpiconscolors/vertexai/v1/32px.svg\" alt=\"Vertex AI logo\"><br> Open in Vertex AI Workbench\n",
        "    </a>\n",
        "  </td>\n",
        "  <td style=\"text-align: center\">\n",
        "    <a href=\"https://github.com/GoogleCloudPlatform/generative-ai/blob/main/agents/gemini_data_analytics/a2a_sdk_sample.ipynb\">\n",
        "      <img width=\"32px\" src=\"https://storage.googleapis.com/github-repo/generative-ai/logos/GitHub_Invertocat_Dark.svg\" alt=\"GitHub logo\"><br> View on GitHub\n",
        "    </a>\n",
        "  </td>\n",
        "</table>\n",
        "\n",
        "<div style=\"clear: both;\"></div>\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 high-level usage patterns off standard developer operations.\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",
        "<br>\n",
        "### This notebook will help you\n",
        "1. Authenticate to Google Cloud and Setup Environment\n",
        "2. Install a2a-sdk Library\n",
        "3. Instantiate a High-level A2A Client\n",
        "4. Fulfill requests synchronously without needing async loops\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "setup"
      },
      "outputs": [],
      "source": [
        "# @title 1. Environment Setup\n",
        "# Install core dependencies\n",
        "%pip install google-auth requests httpx nest_asyncio\n",
        "\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",
        "\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",
        "PROJECT_ID = \"[your-project-id]\"  # @param {type:\"string\"}\n",
        "LOCATION = \"global\"  # @param {type:\"string\"}\n",
        "# AGENT_ID can be found from the Cloud URL, e.g.\n",
        "# https://console.cloud.google.com/bigquery/agents_hub/<your-agent-id>?project=<your-project-id>\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",
        "\n",
        "print(f\"Target Tenant: {TENANT}\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "sdk-install-header"
      },
      "source": [
        "## 2. Install A2A SDK\n",
        "\n",
        "Pulling library mounts drawn from high level ADK frameworks instead of manual stubs!"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "sdk-install-code"
      },
      "outputs": [],
      "source": [
        "# @title Fetch and reference SDK\n",
        "# Running external library installs natively on Sandbox environment\n",
        "%pip install a2a-sdk\n",
        "\n",
        "from a2a.client import Client as A2AClient\n",
        "from a2a.types import AgentCard\n",
        "\n",
        "print(\"High-level SDK imported successfully. Please restart runtime if import fails directly after execution!\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "grpc-client-header"
      },
      "source": [
        "## 3. Implementation: SDK Native Client\n",
        "\n",
        "Create constructor wrappers using authentic SDK client hooks operations of pure synchronous execution!"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "grpc-client-code"
      },
      "outputs": [],
      "source": [
        "import uuid\n",
        "import asyncio\n",
        "import httpx\n",
        "import nest_asyncio\n",
        "from a2a.client import Client as A2AClient\n",
        "from a2a.client.client_factory import ClientFactory as A2AClientFactory\n",
        "from a2a.client.client import ClientConfig as A2AClientConfig\n",
        "from a2a.types import TransportProtocol as A2ATransport, AgentCapabilities\n",
        "from a2a.types import AgentCard\n",
        "\n",
        "nest_asyncio.apply()\n",
        "\n",
        "class DataA2AClient:\n",
        "\n",
        "  def __init__(self, endpoint, token):\n",
        "    self.endpoint = f\"https://{endpoint}\" if not endpoint.startswith(\"http\") else endpoint\n",
        "    self.token = token\n",
        "    httpx_client = httpx.AsyncClient(\n",
        "        headers={\"Authorization\": f\"Bearer {self.token}\"},\n",
        "        timeout=60.0\n",
        "    )\n",
        "    client_config = A2AClientConfig(\n",
        "        streaming=True,\n",
        "        polling=True,\n",
        "        httpx_client=httpx_client,\n",
        "        supported_transports=[A2ATransport.http_json]\n",
        "    )\n",
        "    factory = A2AClientFactory(config=client_config)\n",
        "    card = AgentCard(\n",
        "        url=f\"{self.endpoint}/v1beta/a2a/{TENANT}/\", \n",
        "        name=\"TargetAgent\",\n",
        "        description=\"Test Agent\",\n",
        "        version=\"1.0\",\n",
        "        preferred_transport=\"HTTP+JSON\",\n",
        "        capabilities=AgentCapabilities(),\n",
        "        default_input_modes=[],\n",
        "        default_output_modes=[],\n",
        "        skills=[]\n",
        "    )\n",
        "    self.client = factory.create(card)\n",
        "\n",
        "  async def send_message(self, tenant, text):\n",
        "    from a2a.types import Message as A2AMessage\n",
        "    \n",
        "    message = A2AMessage(\n",
        "        message_id=str(uuid.uuid4()),\n",
        "        role=\"user\",\n",
        "        parts=[{\"text\": text}]\n",
        "    )\n",
        "    \n",
        "    responses = self.client.send_message(message)\n",
        "    async for response in responses:\n",
        "      if hasattr(response, \"status\") and response.status:\n",
        "        print(f\"[Status] {response.status.state}\")\n",
        "      elif hasattr(response, \"artifact\") and response.artifact:\n",
        "        print(f\"[Artifact] {response.artifact.name}: {response.artifact.description}\")\n",
        "      elif hasattr(response, \"parts\") and response.parts:\n",
        "        part = response.parts[0]\n",
        "        if hasattr(part, \"root\") and part.root:\n",
        "          print(f\"[Message] {part.root.text}\")\n",
        "        elif hasattr(part, \"text\") and part.text:\n",
        "          print(f\"[Message] {part.text}\")\n",
        "\n",
        "client = DataA2AClient(ENDPOINT, access_token)\n",
        "print(\"Client created successfully using authentic A2A SDK!\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "streaming-demo-header"
      },
      "source": [
        "## 4. Example: Standard Resolution\n",
        "\n",
        "We can now receive targeted execution states naturally mapped on list loops!"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "streaming-demo-code"
      },
      "outputs": [],
      "source": [
        "query = \"Show me sales trends for 2025\"\n",
        "print(f\"Sending request: {query}\\n\")\n",
        "\n",
        "try:\n",
        "  asyncio.run(client.send_message(TENANT, query))\n",
        "except Exception as e:\n",
        "  print(f\"Error: {e}\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cleanup-header"
      },
      "source": [
        "## 5. Cleanup\n",
        "\n",
        "It is good practice to clean up any temporary resources or local session state."
      ]
    },
    {
      "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.\"\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-pointer",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.10.12"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}