In addition to the general instructions for using an agent , this page describes features that are specific to A2A agents.
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.
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.
Setup
Vertex AI SDK for Python
You can interact with A2A agents deployed to Agent Engine using the Vertex AI SDK for Python, using the same syntax.
To set up A2A with Agent Engine, get an instance of the deployed agent. This instance wraps the underlying A2A endpoints, allowing you to call the endpoints as Python methods.
import
vertexai
from
google.genai
import
types
# Replace with your actual values
PROJECT_ID
=
"your-project-id"
LOCATION
=
"your-location"
REASONING_ENGINE_ID
=
"your-reasoning-engine-id"
AGENT_ENGINE_RESOURCE
=
f
"projects/
{
PROJECT_ID
}
/locations/
{
LOCATION
}
/reasoningEngines/
{
REASONING_ENGINE_ID
}
"
client
=
vertexai
.
Client
(
project
=
PROJECT_ID
,
location
=
LOCATION
,
http_options
=
types
.
HttpOptions
(
api_version
=
"v1beta1"
)
)
remote_agent
=
client
.
agent_engines
.
get
(
name
=
AGENT_ENGINE_RESOURCE
,
)
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"
,
}
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
))
