Generative AI Google Brain

agents-gemini-data-analytics-a2a-sdk-sample

3 min read

Gemini Data Analytics: A2A SDK API Sample

Source notebook

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

Uses the A2A Python SDK to send synchronous-style requests to Gemini Data Analytics.

Summary

The notebook teaches how to authenticate to Google Cloud, install a2a-sdk, and build a high-level client for the DataA2Aservice endpoint. It constructs an A2A AgentCard for a Gemini Data Analytics tenant, sends a user query through the SDK, and prints streamed status, artifact, or message responses. The workflow assumes an existing project, location, and agent ID, and creates no cloud resources to clean up.

Key code patterns

Authenticate and build tenant

auth.authenticate_user()
creds, _ = default()
creds.refresh(Request())
access_token = creds.token
TENANT = f"projects/{PROJECT_ID}/locations/{LOCATION}/agents/{AGENT_ID}"

Gets a bearer token and formats the agent resource used by the A2A endpoint.

Configure A2A client

httpx_client = httpx.AsyncClient(
    headers={"Authorization": f"Bearer {self.token}"},
    timeout=60.0
)
client_config = A2AClientConfig(
    streaming=True,
    polling=True,
    httpx_client=httpx_client,
    supported_transports=[A2ATransport.http_json]
)

Uses authenticated HTTP+JSON transport with streaming and polling enabled.

Create agent card

card = AgentCard(
    url=f"{self.endpoint}/v1beta/a2a/{TENANT}/",
    name="TargetAgent",
    description="Test Agent",
    version="1.0",
    preferred_transport="HTTP+JSON",
    capabilities=AgentCapabilities()
)

Points the SDK factory at the Gemini Data Analytics A2A tenant endpoint.

Send and handle responses

message = A2AMessage(
    message_id=str(uuid.uuid4()),
    role="user",
    parts=[{"text": text}]
)
responses = self.client.send_message(message)
async for response in responses:
    print(response)

Wraps a natural-language query as an A2A message and iterates streamed responses.

Models & APIs used

  • APIs / services: Conversational Analytics API, Gemini Data Analytics, BigQuery, Looker
  • SDKs / libraries: a2a-sdk, google-auth, httpx, nest_asyncio, requests

When to use this

Use this pattern when you need a Python client to chat with an existing Gemini Data Analytics agent over the high-level A2A SDK.

Gotchas & caveats

  • The notebook labels Gemini Data Analytics as a Pre-GA product.
  • PROJECT_ID and AGENT_ID must be set for an existing agent tenant.
  • The example uses LOCATION=“global”.
  • Authentication depends on google.colab.auth.authenticate_user().
  • The notebook says to restart the runtime if the a2a-sdk import fails after installation.
  • The code uses nest_asyncio.apply() before running async SDK calls from a notebook.

Best practices

  • Use Google Cloud user authentication and refresh credentials before creating the client.
  • Pass the bearer token through an httpx.AsyncClient Authorization header.
  • Use the high-level a2a-sdk instead of manual stubs.
  • Enable streaming and polling in ClientConfig.
  • Use a UUID for each A2A message_id.
  • Include a cleanup section and note that no demo cloud resources were created.

Graph View

Backlinks