Before you begin
This tutorial assumes that you have read and followed the instructions in:
- Develop an Agent2Agent agent
to develop an agent as an instance of
A2aAgent. - 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 A2aAgent
, you need to first create a new instance
or get an existing instance
.
To get the A2aAgent
corresponding to a specific resource ID:
Vertex AI SDK for Python
import
vertexai
from
google.genai
import
types
PROJECT_ID
=
" PROJECT_ID
"
LOCATION
=
" LOCATION
"
RESOURCE_ID
=
" RESOURCE_ID
"
RESOURCE_NAME
=
f
"projects/
{
PROJECT_ID
}
/locations/
{
LOCATION
}
/reasoningEngines/
{
RESOURCE_ID
}
"
client
=
vertexai
.
Client
(
project
=
PROJECT_ID
,
location
=
LOCATION
,
http_options
=
types
.
HttpOptions
(
api_version
=
"v1beta1"
)
)
remote_agent
=
client
.
agent_engines
.
get
(
name
=
RESOURCE_NAME
)
print
(
remote_agent
)
where
-
PROJECT_IDis the Google Cloud project ID under which you develop and deploy agents, and -
LOCATIONis one of the supported regions . -
RESOURCE_IDis the ID of the deployed agent as areasoningEngineresource .
A2A Python SDK
This method uses the official A2A Python SDK, which provides a client library for interacting with A2A-compliant agents. For more information, see the A2A Python SDK documentation .
First, install the SDK:
pip
install
a2a-sdk> =
0
.3.4
Then, get the agent's card to create a client instance. The A2AClient
handles the discovery and communication for you.
from
google.auth
import
default
from
google.auth.transport.requests
import
Request
from
a2a.client
import
ClientConfig
,
ClientFactory
from
a2a.types
import
TransportProtocol
import
httpx
# We assume 'agent_card' is an existing AgentCard object.
# Fetch credentials for authentication for demo purpose. Use your own auth
credentials
,
_
=
default
(
scopes
=
[
'https://www.googleapis.com/auth/cloud-platform'
])
credentials
.
refresh
(
Request
())
# Create the client by chaining the factory and config initialization.
factory
=
ClientFactory
(
ClientConfig
(
supported_transports
=
[
TransportProtocol
.
http_json
],
# only support http_json
use_client_preference
=
True
,
httpx_client
=
httpx
.
AsyncClient
(
headers
=
{
"Authorization"
:
f
"Bearer
{
credentials
.
token
}
"
,
"Content-Type"
:
"application/json"
,
}
),
)
)
a2a_client
=
factory
.
create
(
agent_card
)
Python requests library
The A2A protocol is built on standard HTTP endpoints. You can interact with these endpoints using any HTTP client.
Retrieve the A2A URL from the agent card and define the request headers.
from
google.auth
import
default
from
google.auth.transport.requests
import
Request
# We assume 'agent_card' is an existing object
a2a_url
=
agent_card
.
url
# Get an authentication token for demonstration purposes. Use your own authentication mechanism.
credentials
,
_
=
default
(
scopes
=
[
'https://www.googleapis.com/auth/cloud-platform'
])
credentials
.
refresh
(
Request
())
headers
=
{
"Authorization"
:
f
"Bearer
{
credentials
.
token
}
"
,
"Content-Type"
:
"application/json"
,
}
When using the Vertex AI SDK for Python, the remote_agent
object corresponds to an AgentEngine
class that contains the following:
- an
agent.api_resourcewith information about the deployed agent. You can also callagent.operation_schemas()to return the list of operations that the agent supports. See Supported operations for details. - an
agent.api_clientthat allows for synchronous service interactions - an
agent.async_api_clientthat allows for asynchronous service interactions
The rest of this section assumes that you have an AgentEngine
instance, named as remote_agent
.
Supported operations
An A2A agent hosted on Agent Engine exposes a set of operations that correspond directly to the A2A protocol's API endpoints.
-
on_message_send: Sends a new message to the agent to start a task. -
on_get_task: Retrieves the status and artifacts of an existing task. -
on_cancel_task: Cancels a running task. -
handle_authenticated_agent_card: Retrieves the agent's full capabilities and skills.
Retrieve the agent card
Note that Agent Engine does not serve the public agent card. To retrieve the authenticated agent card:
Vertex AI SDK for Python
response
=
await
remote_agent
.
handle_authenticated_agent_card
()
A2A Python SDK
response
=
await
a2a_client
.
get_card
()
Python requests library
card_endpoint
=
f
"
{
a2a_url
}
/v1/card"
response
=
httpx
.
get
(
card_endpoint
,
headers
=
headers
)
print
(
json
.
dumps
(
response
.
json
(),
indent
=
4
))
Send a message
To send a message:
Vertex AI SDK for Python
message_data
=
{
"messageId"
:
"remote-agent-message-id"
,
"role"
:
"user"
,
"parts"
:
[{
"kind"
:
"text"
,
"text"
:
"What is the exchange rate from USD to EUR today?"
}],
}
response
=
await
remote_agent
.
on_message_send
(
**
message_data
)
A2A Python SDK
from
a2a.types
import
Message
,
Part
,
TextPart
import
pprint
message
=
Message
(
message_id
=
"remote-agent-message-id"
,
role
=
"user"
,
parts
=
[
Part
(
root
=
TextPart
(
text
=
"What's the currency rate of USD and EUR"
))],
)
response_iterator
=
a2a_client
.
send_message
(
message
)
async
for
chunk
in
response_iterator
:
pprint
.
pp
(
chunk
)
Python requests library
import
httpx
import
json
endpoint
=
f
"
{
a2a_url
}
/v1/message:send"
payload
=
{
"message"
:
{
"messageId"
:
"remote-agent-message-id"
,
"role"
:
"1"
,
"content"
:
[{
"text"
:
"What is the exchange rate from USD to EUR today?"
}],
},
"metadata"
:
{
"source"
:
"python_script"
},
}
response
=
httpx
.
post
(
endpoint
,
json
=
payload
,
headers
=
headers
)
print
(
json
.
dumps
(
response
.
json
(),
indent
=
4
))
Get a task
To get a task and its status
Vertex AI SDK for Python
task_data
=
{
"id"
:
task_id
,
}
response
=
await
remote_agent
.
on_get_task
(
**
task_data
)
A2A Python SDK
from
a2a.types
import
TaskQueryParams
task_data
=
{
"id"
:
task_id
,
}
response
=
await
a2a_client
.
get_task
(
TaskQueryParams
(
**
task_data
))
Python requests library
task_end_point
=
f
"
{
a2a_url
}
/v1/tasks/
{
task_id
}
"
response
=
httpx
.
get
(
task_end_point
,
headers
=
headers
)
print
(
json
.
dumps
(
response
.
json
(),
indent
=
4
))
Cancel a task
To cancel a task:
Vertex AI SDK for Python
task_data
=
{
"id"
:
task_id
,
}
response
=
await
remote_agent
.
on_cancel_task
(
**
task_data
)
A2A Python SDK
from
a2a.types
import
TaskQueryParams
task_data
=
{
"id"
:
task_id
,
}
response
=
await
a2a_client
.
cancel_task
(
TaskQueryParams
(
**
task_data
))
Python requests library
task_end_point
=
f
"
{
a2a_url
}
/v1/tasks/
{
task_id
}
:cancel"
response
=
httpx
.
post
(
task_end_point
,
headers
=
headers
)
print
(
json
.
dumps
(
response
.
json
(),
indent
=
4
))
