Use an Agent Development Kit agent

Before you begin

This tutorial assumes that you have read and followed the instructions in:

Develop an Agent Development Kit agent: to develop agent as an instance of AdkApp.
User authentication to authenticate as a user for querying the agent.
Import and initialize the SDK to initialize the client for getting a deployed instance (if needed).

Get an instance of an agent

To query an AdkApp, you need to first create a new instance or get an existing instance.

To get the AdkApp corresponding to a specific resource ID:

Vertex AI SDK for Python

Run the following code:

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    project="PROJECT_ID",
    location="LOCATION",
)

adk_app = client.agent_engines.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(adk_app)

where

PROJECT_ID is the Google Cloud project ID under which you develop and deploy agents,
LOCATION is one of the supported regions, and
RESOURCE_ID is the ID of the deployed agent as a reasoningEngine resource.

Python requests library

Run the following code:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

response = requests.get(
f"https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID

When using the Vertex AI SDK for Python, the adk_app object corresponds to an AgentEngine class that contains the following:

an adk_app.api_resource with information about the deployed agent. You can also call adk_app.operation_schemas() to return the list of operations that the adk_app supports. See Supported operations for details.
an adk_app.api_client that allows for synchronous service interactions
an adk_app.async_api_client that allows for asynchronous service interactions

The rest of this section assumes that you have an AgentEngine instance, named as adk_app.

Supported operations

The following operations are supported for AdkApp:

async_stream_query: for streaming a response to a query.
async_create_session: for creating a new session.
async_list_sessions: for listing the sessions available.
async_get_session: for retrieving a specific session.
async_delete_session: for deleting a specific session.
async_add_session_to_memory: for generating memories of a session.
async_search_memory: for retrieving memories.

To list all supported operations:

Vertex AI SDK for Python

Run the following code:

adk_app.operation_schemas()

Python requests library

Run the following code:

import json

json.loads(response.content).get("spec").get("classMethods")

REST API

Represented in spec.class_methods from the response to the curl request.

Manage sessions

AdkApp uses cloud-based managed sessions after you deploy the agent to Vertex AI Agent Engine. This section describes how to use managed sessions.

Create a session

To create a session for a user:

Vertex AI SDK for Python

session = await adk_app.async_create_session(user_id="USER_ID")

print(session)

Python requests library

Run the following code:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_create_session",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_create_session", "input": {"user_id": "USER_ID"},}'

where USER_ID is a user-defined ID with a character limit of 128.

The session is created as the dictionary representation of an ADK session object.

List sessions

To list the sessions for a user:

Vertex AI SDK for Python

response = await adk_app.async_list_sessions(user_id="USER_ID"):
for session in response.sessions:
    print(session)

Python requests library

Run the following code:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_list_sessions",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_list_sessions", "input": {"user_id": "USER_ID"},}'

where USER_ID is a user-defined ID with a character limit of 128.

If any sessions are returned, they use the dictionary form of an ADK session object.

Get a session

To get a specific session, you need both the user ID and the session ID:

Vertex AI SDK for Python

session = await adk_app.async_get_session(user_id="USER_ID", session_id="SESSION_ID")

print(session)

Python requests library

Run the following code:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_get_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_get_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

The session is the dictionary representation of an ADK session object.

Delete a session

To delete a session, you need both the user ID and the session ID:

Vertex AI SDK for Python

await adk_app.async_delete_session(user_id="USER_ID", session_id="SESSION_ID")

Python requests library

Run the following code:

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_delete_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_delete_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

Stream a response to a query

To stream responses from an agent in a session:

Vertex AI SDK for Python

async for event in adk_app.async_stream_query(
    user_id="USER_ID",
    session_id="SESSION_ID",  # Optional
    message="What is the exchange rate from US dollars to SEK today?",
):
  print(event)

Python requests library

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

requests.post(
    f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:streamQuery",
    headers={
        "Content-Type": "application/json",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "async_stream_query",
        "input": {
            "user_id": "USER_ID",
            "session_id": "SESSION_ID",
            "message": "What is the exchange rate from US dollars to SEK today?",
        },
    }),
    stream=True,
)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:streamQuery?alt=sse -d '{
  "class_method": "async_stream_query",
  "input": {
    "user_id": "USER_ID",
    "session_id": "SESSION_ID",
    "message": "What is the exchange rate from US dollars to SEK today?",
  }
}'

If you are using the Vertex AI SDK for Python, you should receive a continuation of the conversation like the following sequence of dictionaries:

{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
                                                   'currency_from': 'USD',
                                                   'currency_to': 'SEK'},
                                          'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                          'name': 'get_exchange_rate'}}],
             'role': 'model'},
 'id': 'bOPHtzji',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_response': {'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                              'name': 'get_exchange_rate',
                                              'response': {'amount': 1.0,
                                                           'base': 'USD',
                                                           'date': '2025-04-03',
                                                           'rates': {'SEK': 9.6607}}}}],
             'role': 'user'},
 'id': '9AoDFmiL',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
                                '2025-04-03 is 1 USD to 9.6607 SEK.'}],
             'role': 'model'},
 'id': 'hmle7trT',
 # ...
}

Manage memories

AdkApp uses Vertex AI Agent Engine Memory Bank if you include a PreloadMemoryTool in the agent definition and deploy the agent to Vertex AI Agent Engine. This section describes how to use generate and retrieve memories from the agent through the default implementation of the ADK memory service.

Add session to memory

To retain memory of meaningful information in a session (that can be used in future sessions), use the async_add_session_to_memory method:

Vertex AI SDK for Python

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

where SESSION_DICT is the dictionary form of an ADK session object.

Search for memories

To search through the memories of the agent, you can use the async_search_memory method:

Vertex AI SDK for Python

response = await adk_app.async_search_memory(
    user_id="USER_ID",
    query="QUERY",
)
print(response)

where

USER_ID is the scope for relevant memories.
QUERY is the query for which to perform similarity search.

Use an Agent Development Kit agent Stay organized with collections Save and categorize content based on your preferences.

Before you begin

Get an instance of an agent

Vertex AI SDK for Python

Python requests library

REST API

Supported operations

Vertex AI SDK for Python

Python requests library

REST API

Manage sessions

Create a session

Vertex AI SDK for Python

Python requests library

REST API

List sessions

Vertex AI SDK for Python

Python requests library

REST API

Get a session

Vertex AI SDK for Python

Python requests library

REST API

Delete a session

Vertex AI SDK for Python

Python requests library

REST API

Stream a response to a query

Vertex AI SDK for Python

Python requests library

REST API

Manage memories

Add session to memory

Vertex AI SDK for Python

Search for memories

Vertex AI SDK for Python

What's next

Use an Agent Development Kit agent