Build a data agent using HTTP and Python

This page guides you through using Python to make HTTP requests to the Conversational Analytics API (accessed through geminidataanalytics.googleapis.com).

The sample Python code on this page shows how to complete the following tasks:

Configure initial settings and authentication
Connect to a Looker, BigQuery, or Looker Studio data source
Create a data agent
Create a conversation
Use the API to ask questions
Create a stateless multi-turn conversation

You can also run the code samples on this page in the Conversational Analytics API HTTP Colaboratory notebook.

A complete version of the sample code is included at the end of the page, along with the helper functions that are used to stream the API response.

Configure initial settings and authentication

The following sample Python code performs these tasks:

Imports the required Python libraries
Defines variables for the billing project, system instructions, and a question for the data agent
Obtains an access token for HTTP authentication by using the Google Cloud CLI

from pygments import highlight, lexers, formatters
import pandas as pd
import json as json_lib
import requests
import json
import altair as alt
import IPython
from IPython.display import display, HTML
import google.auth
from google.auth.transport.requests import Request

from google.colab import auth
auth.authenticate_user()

billing_project = 'YOUR-BILLING-PROJECT'
system_description = 'YOUR-SYSTEM-INSTRUCTIONS'
question = 'YOUR-QUESTION-HERE'

access_token = !gcloud auth application-default print-access-token
headers = {
    "Authorization": f"Bearer {access_token[0]}",
    "Content-Type": "application/json",
}

Replace the sample values as follows:

YOUR-BILLING-PROJECT: The ID of the billing project where you've enabled the required APIs.
YOUR-SYSTEM-INSTRUCTIONS: System instructions to guide the agent's behavior and allow you to customize the agent for your needs. For example, you can use system instructions to define business terms (like what constitutes a "loyal customer"), control response length ("summarize in fewer than 20 words"), or set data formatting ("match company standards"). Replace the placeholder text with instructions relevant to your data and use case.
YOUR-QUESTION-HERE: A natural language question to send to the data agent.

Authenticate to Looker

If you plan to connect to a Looker data source, you will need to authenticate to the Looker instance.

Using API keys

The following Python code sample demonstrates how to authenticate your agent to a Looker instance using API keys.

looker_credentials = {
    "oauth": {
        "secret": {
            "client_id": "YOUR-LOOKER-CLIENT-ID",
            "client_secret": "YOUR-LOOKER-CLIENT-SECRET",
        }
    }
}

Replace the sample values as follows:

YOUR-LOOKER-CLIENT-ID: The client ID of your generated Looker API key.
YOUR-LOOKER-CLIENT-SECRET: The client secret of your generated Looker API key.

Using access tokens

The following Python code sample demonstrates how to authenticate your agent to a Looker instance using access tokens.

looker_credentials = {
    "oauth": {
        "token": {
            "access_token": "YOUR-TOKEN",
        }
    }
}

Replace the sample values as follows:

YOUR-TOKEN: The access_token value you generate to authenticate to Looker.

Connect to a data source

The following Python code samples demonstrate how to define the Looker, BigQuery, or Looker Studio data source for your agent to use.

Connect to Looker data

The following sample code defines a connection to a Looker Explore. To establish a connection with a Looker instance, verify that you've generated Looker API keys, as described in Authenticate and connect to a data source with the Conversational Analytics API.

looker_data_source = {
    "looker": {
        "explore_references": {
            "looker_instance_uri": "https://your_company.looker.com",
            "lookml_model": "your_model",
            "explore": "your_explore",
       },
    }
}

Replace the sample values as follows:

https://your_company.looker.com: The complete URL of your Looker instance.
your_model: The name of the LookML model that includes the Explore that you want to connect to.
your_explore: The name of the Looker Explore that you want the data agent to query.

Connect to BigQuery data

The following sample code defines a connection to a BigQuery table.

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "bigquery-public-data",
                "datasetId": "san_francisco",
                "tableId": "street_trees",
            }
        ]
    }
}

Replace the sample values as follows:

bigquery-public-data: The ID of the Google Cloud project that contains the BigQuery dataset and table that you want to connect to. To connect to a public dataset, specify bigquery-public-data.
san_francisco: The ID of the BigQuery dataset.
street_trees: The ID of the BigQuery table.

Connect to Looker Studio data

The following sample code defines a connection to a Looker Studio data source.

looker_studio_data_source = {
    "studio":{
        "studio_references": [
            {
              "studio_datasource_id": "studio_datasource_id"
            }
        ]
    }
}

Replace studio_datasource_id with the data source ID.

Create a data agent

The following sample code demonstrates how to create the data agent by sending an HTTP POST request to the data agent creation endpoint. The request payload includes the following details:

The full resource name for the agent. This value includes the project ID, location, and a unique identifier for the agent.
The data agent's description.
The data agent's context, including the system description (defined in Configure initial settings and authentication) and the data source that the agent uses (defined in Connect to a data source).

You can also optionally enable advanced analysis with Python by including the options parameter in the request payload.

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/dataAgents"

data_agent_id = "data_agent_1"

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of data_agent_1.", # Optional

      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              # Optional: To enable advanced analysis with Python, include the following options block:
              "options": {
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  }
              }
          }
      }
  }

params = {"data_agent_id": data_agent_id} # Optional

data_agent_response = requests.post(
    data_agent_url, params=params, json=data_agent_payload, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agent created successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error creating Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

Replace the sample values as follows:

data_agent_1: A unique identifier for the data agent. This value is used in the agent's resource name and as the data_agent_id URL query parameter.
This is the description of data_agent_1.: A description for the data agent.

Create a conversation

The following sample code demonstrates how to create a conversation with your data agent.

conversation_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/conversations"

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

conversation_payload = {
    "agents": [
        f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
    ],
    "name": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
}
params = {
    "conversation_id": conversation_id
}

conversation_response = requests.post(conversation_url, headers=headers, params=params, json=conversation_payload)

if conversation_response.status_code == 200:
    print("Conversation created successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error creating Conversation: {conversation_response.status_code}")
    print(conversation_response.text)

Replace the sample values as follows:

data_agent_1: The ID of data agent, as defined in the sample code block in Create a data agent.
conversation_1: A unique identifier for the conversation.

Use the API to ask questions

After you've created a data agent and a conversation, you can ask questions of your data.

The Conversational Analytics API supports multi-turn conversations, which let users ask follow-up questions that build on previous context. The API provides the following methods for managing conversation history:

Stateful chat: Google Cloud stores and manages the conversation history. Stateful chat is inherently multi-turn, as the API retains context from previous messages. You only need to send the current message for each turn.
Stateless chat: Your application manages the conversation history. You must include the relevant previous messages with each new message. For detailed examples on how to manage multi-turn conversations in stateless mode, see Create a stateless multi-turn conversation.

Stateful chat

Send a stateful chat request with a conversation reference

The following sample code demonstrates how to ask the API questions using the conversation that you defined in previous steps. This sample uses a get_stream helper function to stream the response.

chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "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}",
            # "credentials": looker_credentials
        }
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

Replace the sample values as follows:

data_agent_1: The ID of the data agent, as defined in the sample code block in Create a data agent.
conversation_1: A unique identifier for the conversation.
Make a bar graph for the top 5 states by the total number of airports was used as the sample prompt.

Stateless chat

Send a stateless chat request with a data agent reference

The following sample code demonstrates how to ask the API a stateless question by using the data agent that you defined in previous steps. This sample uses a get_stream helper function to stream the response.

chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "data_agent_1"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "data_agent_context": {
        "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
        # "credentials": looker_credentials
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

Replace the sample values as follows:

data_agent_1: The ID of the data agent, as defined in the sample code block in Create a data agent.
Make a bar graph for the top 5 states by the total number of airports was used as the sample prompt.

Send a stateless chat request with inline context

The following sample code demonstrates how to ask the API a stateless question by using inline context. This sample uses a get_stream helper function to stream the response and uses a BigQuery data source as an example.

You can also optionally enable advanced analysis with Python by including the options parameter in the request payload.

chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/global:chat"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "inline_context": {
        "datasource_references": bigquery_data_sources,
          # Optional: To enable advanced analysis with Python, include the following options block:
          "options": {
              "analysis": {
                  "python": {
                      "enabled": True
                  }
              }
          }
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

Create a stateless multi-turn conversation

To ask follow-up questions in a stateless conversation, your application must manage the conversation's context by sending the entire message history with each new request. The following sections show how to define and call helper functions to create a multi-turn conversation:

Send multi-turn requests
Process responses

Send multi-turn requests

The following multi_turn_Conversation helper function manages conversation context by storing messages in a list. This lets you send follow-up questions that build on previous turns. In the function's payload, you can reference a data agent or provide the data source directly by using inline context.

chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/global:chat"

# List that is used to track previous turns and is reused across requests
conversation_messages = []

data_agent_id = "data_agent_1"

# Helper function for calling the API
def multi_turn_Conversation(msg):

  userMessage = {
      "userMessage": {
          "text": msg
      }
  }

  # Send a multi-turn request by including previous turns and the new message
  conversation_messages.append(userMessage)

  # Construct the payload
  chat_payload = {
      "parent": f"projects/{billing_project}/locations/global",
      "messages": conversation_messages,
      # Use a data agent reference
      "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
          # "credentials": looker_credentials
      },
      # Use inline context
      # "inline_context": {
      #     "datasource_references": bigquery_data_sources,
      # }
  }

  # Call the get_stream_multi_turn helper function to stream the response
  get_stream_multi_turn(chat_url, chat_payload, conversation_messages)

In the previous example, replace data_agent_1 with the ID of the data agent, as defined in the sample code block in Create a data agent.

You can call the multi_turn_Conversation helper function for each turn of the conversation. The following sample code shows how to send an initial request and then a follow-up request that builds on the previous response.

# Send first-turn request
multi_turn_Conversation("Which species of tree is most prevalent?")

# Send follow-up-turn request
multi_turn_Conversation("Can you show me the results as a bar chart?")

In the previous example, replace the sample values as follows:

Which species of tree is most prevalent?: A natural language question to send to the data agent.
Can you show me the results as a bar chart?: A follow-up question that builds on or refines the previous question.

Process responses

The following get_stream_multi_turn function processes the streaming API response. This function is similar to the get_stream helper function, but it stores the response in the conversation_messages list to save the conversation context for the next turn.

def get_stream_multi_turn(url, json, conversation_messages):
    s = requests.Session()

    acc = ''

    with s.post(url, json=json, headers=headers, stream=True) as resp:
        for line in resp.iter_lines():
            if not line:
                continue

            decoded_line = str(line, encoding='utf-8')

            if decoded_line == '[{':
                acc = '{'
            elif decoded_line == '}]':
                acc += '}'
            elif decoded_line == ',':
                continue
            else:
                acc += decoded_line

            if not is_json(acc):
                continue

            data_json = json_lib.loads(acc)
            # Store the response that will be used in the next iteration
            conversation_messages.append(data_json)

            if not 'systemMessage' in data_json:
                if 'error' in data_json:
                    handle_error(data_json['error'])
                continue

            if 'text' in data_json['systemMessage']:
                handle_text_response(data_json['systemMessage']['text'])
            elif 'schema' in data_json['systemMessage']:
                handle_schema_response(data_json['systemMessage']['schema'])
            elif 'data' in data_json['systemMessage']:
                handle_data_response(data_json['systemMessage']['data'])
            elif 'chart' in data_json['systemMessage']:
                handle_chart_response(data_json['systemMessage']['chart'])
            else:
                colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
                print(colored_json)
            print('\n')
            acc = ''

End-to-end code sample

The following expandable code sample contains all the tasks that are covered in this guide.