In addition to the general instructions for using an agent
,
this page describes features that are specific to AdkApp
.
Before you begin
This tutorial assumes that you have read and followed the instructions in:
- Develop an Agent Development Kit agent
: to develop
agentas an instance ofAdkApp. - User authentication to authenticate as a user for querying the agent.
To query an ADK application, you need to first create a new ADK application instance or get an existing instance .
To get the ADK application corresponding to a specific resource ID:
Vertex AI SDK for Python
Run the following code:
from
vertexai
import
agent_engines
adk_app
=
agent_engines
.
get
(
RESOURCE_ID
)
Alternatively, you can provide the full resource name of the agent:
adk_app
=
agent_engines
.
get
(
"projects/ PROJECT_ID
/locations/ LOCATION
/reasoningEngines/ RESOURCE_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
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
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.
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
"
)
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.
List sessions
To list the sessions for a user:
Vertex AI SDK for Python
await
adk_app
.
async_list_sessions
(
user_id
=
" USER_ID
"
)
requests
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
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.
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
"
)
requests
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
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
"},}'
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
"
)
requests
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
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
)
requests
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
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?",
}
}'
