Send queries to a data agent

You can query a data agent in natural language to analyze your connected data and receive text, structured tables, and charts. By using natural language, you can analyze your connected data without needing to write complex code or SQL, making data insights more accessible to a wider range of users. To send queries, use the Python SDK or HTTP for stateful chat (managed by Google Cloud) or stateless chat (managed by your application).

Before you begin

Before you send queries, complete the following tasks:

1. Enable the Conversational Analytics API in your Google Cloud project.
2. Verify that you have the required Identity and Access Management (IAM) roles.
3. Verify that you have access to the data that you want to query (such as in BigQuery or Looker) or a configured data agent.

Required roles

To get the permissions that you need to send queries to a data agent, ask your administrator to grant you the following IAM roles:

• Chat with a persistent data agent: Gemini Data Analytics Data Agent User (roles/geminidataanalytics.dataAgentUser) on your project or the data agent
• Chat by using inline context: Gemini Data Analytics Stateless Chat User (roles/geminidataanalytics.dataAgentStatelessUser) on your project

For more information about granting roles, see Manage access to projects, folders, and organizations.

You might also be able to get the required permissions through custom roles or other predefined roles.

Configure settings and initialize your environment

Before you send requests to the Conversational Analytics API, configure your project settings, endpoint routing, and authentication by using either the Python SDK or HTTP:

Python SDK

The following sample code imports the google-cloud-geminidataanalytics client library, sets your project and location variables, and initializes the API clients with your regional endpoint settings:

from google.colab import auth
auth.authenticate_user()

from google.api_core import client_options
from google.cloud import geminidataanalytics

# Billing project
billing_project = "PROJECT_ID"
location = "LOCATION"

# System instructions
system_instruction = "SYSTEM_INSTRUCTIONS"

# Set client options based on location.
if not location or location == "global":
  endpoint = "geminidataanalytics.googleapis.com"
elif "-" in location:
  # Regional endpoints
  endpoint = f"geminidataanalytics-{location}.googleapis.com"
else:
  # Multi-regional endpoints
  endpoint = f"geminidataanalytics.{location}.rep.googleapis.com"
opts = client_options.ClientOptions(api_endpoint=endpoint)

data_agent_client = geminidataanalytics.DataAgentServiceClient(client_options=opts)
data_chat_client = geminidataanalytics.DataChatServiceClient(client_options=opts)

Replace the sample values as follows:

• PROJECT_ID: The ID of the Google Cloud project where you enabled the required APIs.
• LOCATION: The location where your resources are stored. To use the global default endpoint, specify global. To meet data residency requirements, specify a supported region (such as us-east4) or multi-region (such as eu or us). For more information, see Data residency.

Important: The location in the request parent or resource name must match the location of the endpoint that you call. For example, if you configure the client to use the geminidataanalytics.us.rep.googleapis.com endpoint, specify us in your request parent (such as projects/PROJECT_ID/locations/us).

• SYSTEM_INSTRUCTIONS: Guidance to customize the agent's behavior and formatting for your data needs, such as "Help the user analyze their data.". For more information, see Write effective system instructions.

HTTP

The following sample Python code uses the requests library to configure authentication headers, project variables, endpoint routing, and a streaming response helper function:

The HTTP code samples in this document use the Python requests library to demonstrate streaming. Standard HTTP clients cannot read HTTP/2 trailing error headers. If an error occurs while streaming, the API might return a generic status code, such as 503 Connection reset by peer or 500 Internal Server Error. For production applications, use the Python SDK, which uses gRPC to correctly parse trailing errors.

from google.colab import auth
auth.authenticate_user()

import json
import requests

access_token = !gcloud auth application-default print-access-token
headers = {
    "Authorization": f"Bearer {access_token[0]}",
    "Content-Type": "application/json",
    "x-server-timeout": "300",  # Custom timeout up to 600s
}

billing_project = "PROJECT_ID"
location = "LOCATION"
system_instruction = "SYSTEM_INSTRUCTIONS"

# Set the base URL based on location.
if not location or location == "global":
  base_url = "https://geminidataanalytics.googleapis.com"
elif "-" in location:
  # Regional endpoints
  base_url = f"https://geminidataanalytics-{location}.googleapis.com"
else:
  # Multi-regional endpoints
  base_url = f"https://geminidataanalytics.{location}.rep.googleapis.com"

# Helper function to parse streaming JSON chunks and print text responses
def process_stream(response, conversation_messages=None):
  acc = ""
  for line in response.iter_lines():
    if not line:
      continue
    decoded_line = line.decode("utf-8")
    if decoded_line == "[{":
      acc = "{"
    elif decoded_line == "}]":
      acc += "}"
    elif decoded_line == ",":
      continue
    else:
      acc += decoded_line

    try:
      data = json.loads(acc)
      acc = ""
      if conversation_messages is not None:
        conversation_messages.append(data)
      system_message = data.get("systemMessage", {})
      text_data = system_message.get("text", {})
      if "parts" in text_data:
        print("".join(text_data["parts"]))
    except json.JSONDecodeError:
      continue

Replace the sample values as follows:

• PROJECT_ID: The ID of the Google Cloud project where you enabled the required APIs.
• LOCATION: The location where your resources are stored. To use the global default endpoint, specify global. To meet data residency requirements, specify a supported region (such as us-east4) or multi-region (such as eu or us). For more information, see Data residency.

Important: The location in your request path must match the location of the endpoint that you call. For example, if you call the geminidataanalytics.us.rep.googleapis.com endpoint, specify us in your request path (such as /locations/us).

• SYSTEM_INSTRUCTIONS: Guidance to customize the agent's behavior and formatting for your data needs, such as "Help the user analyze their data.". For more information, see Write effective system instructions.

Agent responses and message types

When you send a query to the Conversational Analytics API, the API returns a stream of Message objects. Depending on the query, this stream can include text, structured data tables, and charts.

Text messages indicate the purpose of each response chunk by using the TextType field:

• THOUGHT: Shows the agent's internal reasoning as it plans and analyzes your query. Contains a summary (parts[0]) and the full thought text (parts[1]).
• PROGRESS: Reports execution progress, such as data retrieval or tool execution (returned for Looker data sources). Contains a summary (parts[0]) and the full progress text (parts[1]).
• FINAL_RESPONSE: Provides the final natural language answer to your query.

The code samples in this document print streaming text parts. To parse and render complete response types (such as data tables, SQL queries, and charts) by using sample helper functions, see Parse data agent responses.

Warning: The Python code that data agents return is designed to execute within a secure, Google-managed sandbox. Executing this code in a local or other unverified environment bypasses the sandbox's security protections and can expose your system to security risks, such as the execution of malicious code. For guidance on how to safely inspect and handle agent-generated Python code, see How can I safely inspect and handle agent-generated Python code?

For guidance on rendering responses in a user interface when using Looker data sources, see Render agent responses for Looker data sources.

Send stateful chat queries

Stateful chat lets you ask follow-up questions across multiple turns while Google Cloud manages the conversation history for you. You send only the current message with each request, and the API retains context from previous turns.

To send a stateful chat request, reference the Conversation resource and data agent that you created by using the Python SDK or HTTP:

Python SDK

The following sample code sends a stateful question to the data agent and prints the text responses from the stream:

# Create a request that contains a single user message (your question)
question = "QUESTION"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

data_agent_id = "DATA_AGENT_ID"
conversation_id = "CONVERSATION_ID"

# Create a conversation_reference
conversation_reference = geminidataanalytics.ConversationReference()
conversation_reference.conversation = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
conversation_reference.data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

# Form the request
request = geminidataanalytics.ChatRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    # credentials=credentials,  # Uncomment for Looker data sources
    messages=messages,
    conversation_reference=conversation_reference,
)

# Send the request
stream = data_chat_client.chat(request=request, timeout=300)  # Custom timeout up to 600s

# Handle the response stream
for response in stream:
  if response.system_message.text:
    print("".join(response.system_message.text.parts))

Replace the sample values as follows:

• QUESTION: A natural language question about your connected data.
• DATA_AGENT_ID: The unique identifier for the data agent, as defined in Create a data agent.
• CONVERSATION_ID: The unique identifier for the conversation, as defined in Create a conversation.

HTTP

The following sample code sends a stateful question to the data agent by using the requests library and prints the streaming text responses:

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "DATA_AGENT_ID"
conversation_id = "CONVERSATION_ID"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,  # Uncomment for Looker data sources
    "messages": [
        {
            "userMessage": {
                "text": "QUESTION"
            }
        }
    ],
    "conversation_reference": {
        "conversation": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
        "data_agent_context": {
            "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
        },
    },
}

# Send the request and process the response stream
response = requests.post(
    chat_url, json=chat_payload, headers=headers, stream=True
)
process_stream(response)

Replace the sample values as follows:

• QUESTION: A natural language question about your connected data.
• DATA_AGENT_ID: The unique identifier for the data agent, as defined in Create a data agent.
• CONVERSATION_ID: The unique identifier for the conversation, as defined in Create a conversation.

Send stateless chat queries

In stateless chat, your application manages conversation history and context. Because the API doesn't store session history between requests, your application provides any previous messages or data context with each request. You can send queries by referencing an existing DataAgent resource or by providing context inline within the request payload.

Send a stateless chat request with a data agent reference

To send a stateless query that uses the context, instructions, and data sources configured in an existing data agent, pass a DataAgentContext object:

Python SDK

The following sample code sends a stateless question to an existing data agent and prints the text responses from the stream:

# Create a request that contains a single user message (your question)
question = "QUESTION"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

data_agent_id = "DATA_AGENT_ID"

# Reference the existing data agent
data_agent_context = geminidataanalytics.DataAgentContext()
data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

# Form the request
request = geminidataanalytics.ChatRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    # credentials=credentials,  # Uncomment for Looker data sources
    messages=messages,
    data_agent_context=data_agent_context,
)

# Send the request
stream = data_chat_client.chat(request=request, timeout=300)  # Custom timeout up to 600s

# Handle the response stream
for response in stream:
  if response.system_message.text:
    print("".join(response.system_message.text.parts))

Replace the sample values as follows:

• QUESTION: A natural language question about your connected data.
• DATA_AGENT_ID: The unique identifier for the data agent, as defined in Create a data agent.

HTTP

The following sample code sends a stateless question to an existing data agent by using the requests library and prints the streaming text responses:

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "DATA_AGENT_ID"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,  # Uncomment for Looker data sources
    "messages": [
        {
            "userMessage": {
                "text": "QUESTION"
            }
        }
    ],
    "data_agent_context": {
        "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
    },
}

# Send the request and process the response stream
response = requests.post(
    chat_url, json=chat_payload, headers=headers, stream=True
)
process_stream(response)

Replace the sample values as follows:

• QUESTION: A natural language question about your connected data.
• DATA_AGENT_ID: The unique identifier for the data agent, as defined in Create a data agent.

Send a stateless chat request with inline context

The following sample code demonstrates how to use the inline_context parameter to provide data context directly within your stateless chat request.

The following code samples use the inline_context object that you configured in Set up context for stateless chat.

Python SDK

# Create a request that contains a single user message (your question)
question = "QUESTION"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

# Form the request
request = geminidataanalytics.ChatRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    # credentials=credentials,  # Uncomment for Looker data sources
    messages=messages,
    inline_context=inline_context,
)

# Send the request
stream = data_chat_client.chat(request=request, timeout=300)  # Custom timeout up to 600s

# Handle the response stream
for response in stream:
  if response.system_message.text:
    for part in response.system_message.text.parts:
      print(part, end="")

Replace QUESTION with a natural language question about your connected data.

HTTP

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/{location}",
    # "credentials": looker_credentials,  # Uncomment for Looker data sources
    "messages": [
        {
            "userMessage": {
                "text": "QUESTION"
            }
        }
    ],
    "inline_context": inline_context,
}

# Send the request and process the response stream
response = requests.post(
    chat_url, json=chat_payload, headers=headers, stream=True
)
process_stream(response)

Replace QUESTION with a natural language question about your connected data.

Maintain multi-turn context in stateless chat

In stateless chat, your application maintains multi-turn context by accumulating user messages and agent responses in a list and sending the entire conversation history with each new request. You can maintain multi-turn context by referencing a data agent or by providing context inline.

The following code samples use a data agent reference. To use inline context instead, uncomment the inline_context line and comment out the data_agent_context parameter.

Python SDK

The following sample code defines a reusable helper function to send questions, stream responses, and append messages to a local conversation history list:

# List used to track message history across requests
conversation_messages = []

data_agent_id = "DATA_AGENT_ID"

# Configure data agent context
data_agent_context = geminidataanalytics.DataAgentContext()
data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

# Helper function to send multi-turn messages
def send_chat_message(prompt_text):
  user_message = geminidataanalytics.Message()
  user_message.user_message.text = prompt_text
  conversation_messages.append(user_message)

  request = geminidataanalytics.ChatRequest(
      parent=f"projects/{billing_project}/locations/{location}",
      # credentials=credentials,  # Uncomment for Looker data sources
      messages=conversation_messages,
      data_agent_context=data_agent_context,
      # inline_context=inline_context,  # To use inline context instead
  )

  # Send the request
  stream = data_chat_client.chat(request=request, timeout=300)  # Custom timeout up to 600s

  # Handle the response stream and append messages to conversation history
  for response in stream:
    if response.system_message.text:
      for part in response.system_message.text.parts:
        print(part, end="")
    conversation_messages.append(response)
  print("\n")

# Send the initial question
send_chat_message("QUESTION")

# Send a follow-up question
send_chat_message("FOLLOW_UP_QUESTION")

Replace the sample values as follows:

• DATA_AGENT_ID: The unique identifier for the data agent, as defined in Create a data agent.
• QUESTION: A natural language question about your connected data.
• FOLLOW_UP_QUESTION: A follow-up question that builds on or refines the previous question, such as "Can you show me the results as a bar chart?".

HTTP

The following sample code defines a reusable helper function to send questions by using the requests library, stream responses, and append messages to a local conversation history list:

chat_url = f"{base_url}/v1/projects/{billing_project}/locations/{location}:chat"
data_agent_id = "DATA_AGENT_ID"

# List used to track message history across requests
conversation_messages = []

# Helper function to send multi-turn messages
def send_chat_message(prompt_text):
  user_message = {
      "userMessage": {
          "text": prompt_text
      }
  }
  conversation_messages.append(user_message)

  chat_payload = {
      "parent": f"projects/{billing_project}/locations/{location}",
      # "credentials": looker_credentials,  # Uncomment for Looker data sources
      "messages": conversation_messages,
      "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
      },
      # "inline_context": inline_context,  # To use inline context instead
  }

  response = requests.post(
      chat_url, json=chat_payload, headers=headers, stream=True
  )
  process_stream(response, conversation_messages=conversation_messages)
  print("\n")

# Send the initial question
send_chat_message("QUESTION")

# Send a follow-up question
send_chat_message("FOLLOW_UP_QUESTION")

Replace the sample values as follows:

• DATA_AGENT_ID: The unique identifier for the data agent, as defined in Create a data agent.
• QUESTION: A natural language question about your connected data.
• FOLLOW_UP_QUESTION: A follow-up question that builds on or refines the previous question, such as "Can you show me the results as a bar chart?".