Use an Agent2Agent agent
Stay organized with collections
Save and categorize content based on your preferences.
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
importvertexai
fromgoogle.genaiimport 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, andLOCATIONis 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:
pipinstalla2a-sdk>=0.3.4Then, get the agent's card to create a client instance. The A2AClient
handles the discovery and communication for you.
fromgoogle.authimport default
fromgoogle.auth.transport.requestsimport Request
froma2a.clientimport ClientConfig, ClientFactory
froma2a.typesimport TransportProtocol
importhttpx
# 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.
fromgoogle.authimport default
fromgoogle.auth.transport.requestsimport 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
froma2a.typesimport Message, Part, TextPart
importpprint
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
importhttpx
importjson
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
froma2a.typesimport 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
froma2a.typesimport 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))