Use an Agent Development Kit agent

Before you begin

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

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

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:

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"},}'

  • USER_ID: Choose your own user ID with a character limit of 128. For example, user-123.

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 the user ID you defined. For example, user-123.

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.

What's next