Generative AI Google Brain

agents-gemini-data-analytics-a2a-http-sample

3 min read

Gemini Data Analytics: A2A HTTP API Sample

Source notebook

Repo path: agents/gemini_data_analytics/a2a_http_sample.ipynb · Open on GitHub · intermediate

Calls Gemini Data Analytics DataA2AService over HTTP for agent cards, messages, artifacts, and cancellation.

Summary

This notebook teaches how to authenticate to Google Cloud and call the Gemini Data Analytics A2A interface with standard HTTP requests. The workflow builds the A2A tenant URL, retrieves the Agent Card, sends blocking and nonblocking messages, polls task status, extracts artifacts or direct message content, and cancels active tasks.

Key code patterns

Bearer token authentication

auth.authenticate_user()
creds, _ = default()
creds.refresh(Request())
access_token = creds.token
HEADERS = {
  "Authorization": f"Bearer {access_token}",
  "Content-Type": "application/json",
}

Uses Colab user auth and a refreshed Google Cloud token for direct HTTP calls.

A2A tenant URL

ENDPOINT = "https://geminidataanalytics.googleapis.com"
TENANT = f"projects/{PROJECT_ID}/locations/{LOCATION}/agents/{AGENT_ID}"
BASE_URL = f"{ENDPOINT}/v1beta/a2a/{TENANT}/v1"
url = f"{BASE_URL}/card"
response = requests.get(url, headers=HEADERS, timeout=30)

Constructs the service path for a specific Gemini Data Analytics agent and verifies connectivity with the Agent Card.

Async message and polling

payload = {
  "tenant": TENANT,
  "message": {"message_id": f"msg-{uuid.uuid4()}", "role": "ROLE_USER", "content": [{"text": query}]},
  "configuration": {"blocking": False},
}
task_id = requests.post(url, headers=HEADERS, json=payload, timeout=30).json().get("task", {}).get("id")
task = requests.get(f"{BASE_URL}/tasks/{task_id}", headers=HEADERS, timeout=30).json()

Starts long-running analysis work without blocking and reads task status through the tasks endpoint.

Artifact extraction

payload["configuration"] = {"blocking": True}
res_json = requests.post(url, headers=HEADERS, json=payload, timeout=120).json()
artifacts = res_json.get("task", {}).get("artifacts", [])
for art in artifacts:
  name = art.get("name", "Unnamed")
  for part in art.get("parts", []):
    if "text" in part:
      print(part["text"][:500])

Shows how completed agent tasks can return structured outputs, files, references, or text parts.

Task cancellation

def cancel_task(task_id):
  url = f"{BASE_URL}/tasks/{task_id}:cancel"
  response = requests.post(url, headers=HEADERS, timeout=30)
  response.raise_for_status()
  return response.json()

Cancels an active task when it is slow or was submitted in error.

Models & APIs used

  • APIs / services: Conversational Analytics API, Gemini Data Analytics, DataA2AService, A2A HTTP API, BigQuery, Looker
  • SDKs / libraries: requests, google.auth, google.colab

When to use this

Use this pattern when you need to call Gemini Data Analytics A2A from an environment without a high-level SDK or with minimal dependencies.

Gotchas & caveats

  • The notebook states Gemini Data Analytics is Pre-GA.
  • PROJECT_ID and AGENT_ID must be set before calls succeed.
  • AGENT_ID is taken from the Cloud URL under bigquery/agents_hub.
  • Long-running data processing may take 30-60 seconds, so the blocking artifact call uses a 120 second timeout.
  • message:send can return either a task or a direct message, so callers must handle both shapes.

Best practices

  • Retrieve the Agent Card first to verify connectivity and agent capabilities.
  • Use uuid.uuid4() to create unique message_id values.
  • Use blocking=False for long-running tasks and poll terminal task states.
  • Use bounded backoff while polling task status.
  • Inspect both task artifacts and direct message content or metadata.
  • Cancel active tasks that take too long or were sent in error.
  • Keep a cleanup section even when no cloud resources were created.

Graph View

Backlinks