Generative AI on Google Cloud

6 / 80·notebook

Chapter 06

Get started with A2A on Agent Engine

NotebookPython 3 (ipykernel)65 cells

In [ ]python · cell 1

python

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Get started with A2A on Agent Engine

Share to:

Authors
Joyce Liu
Rajesh Velicheti
Ivan Nardini
Max Gasztych

Overview

This notebook shows how to build, deploy, and interact with Agent2Agent (A2A) protocol agents hosted on the fully-managed, serverless Agent Platform.

A2A is an open standard, like HTTP for AI agents, enabling communication and collaboration between diverse AI agents by standardizing capability discovery (via Agent Cards) and interaction for complex tasks, thereby eliminating custom integrations.

Agent Platform is fully-managed, serverless platform for running A2A agents. It handles all the infrastructure, scaling, security, and monitoring so you can focus on your agent's logic.

In this tutorial, you will:

Build a simple, A2A-compliant agent using the Agent Platform SDK.
Test the agent locally to ensure it works as expected.
Deploy the agent to Agent Engine with a single command.
Query the managed agent endpoint using three different methods (Agent Platform SDK, A2A SDK, and direct HTTP requests).
Clean up the resources you've created.

Get started

Install required packages

First, we'll install the necessary packages.

a2a-sdk is the foundational open-source SDK for building A2A-compliant agents.
google-cloud-aiplatform is the Agent Platform SDK, containing the new Agent Engine template we'll use for deployment.

In [ ]python · cell 7

python

%pip install --upgrade --quiet "a2a-sdk>=1.0.0" --force-reinstall --quiet
%pip install --upgrade --quiet "google-cloud-aiplatform[agent_engines,adk]>=1.156.0" --force-reinstall --quiet

Authenticate your notebook environment (Colab only)

If you're running this notebook on Google Colab, run the cell below to authenticate your environment.

In [ ]python · cell 9

python

import sys

if "google.colab" in sys.modules:
    from google.colab import auth

    auth.authenticate_user()

Set Google Cloud project information

To get started using Agent Platform, you must have an existing Google Cloud project and enable the Agent Platform API.

Learn more about setting up a project and a development environment.

In [ ]python · cell 11

python

# Use the environment variable if the user doesn't provide Project ID.
import os

import vertexai
from google.genai import types

# fmt: off
PROJECT_ID = "[your-project-id]"  # @param {type: "string", placeholder: "[your-project-id]", isTemplate: true}
if not PROJECT_ID or PROJECT_ID == "[your-project-id]":
    PROJECT_ID = str(os.environ.get("GOOGLE_CLOUD_PROJECT"))

LOCATION = "[your-location]"  # @param {type: "string", placeholder: "[your-location]", isTemplate: true}
if not LOCATION or LOCATION == "[your-location]":
    LOCATION = str(os.environ.get("GOOGLE_CLOUD_REGION"))

BUCKET_NAME = "[your-bucket-name]"  # @param {type: "string", placeholder: "[your-bucket-name]", isTemplate: true}
if not BUCKET_NAME or BUCKET_NAME == "[your-bucket-name]":
    BUCKET_NAME = PROJECT_ID

BUCKET_URI = f"gs://{BUCKET_NAME}"

ENDPOINT = f"https://{LOCATION}-aiplatform.googleapis.com"


# !gsutil mb -l $LOCATION -p $PROJECT_ID $BUCKET_URI

# Initialize Agent Platform session
vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=BUCKET_URI,
    api_endpoint=ENDPOINT,  # This directs requests to the {$ENV} endpoint
)

# Initialize the Gen AI client using http_options
# The parameter customizes how the Agent Platform client communicates with Google Cloud's backend services.
# It's used here to access new, pre-release features.
client = vertexai.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options=types.HttpOptions(api_version="v1beta1", base_url=f"{ENDPOINT}/"),
)

Import libraries

Here, we're importing all the necessary Python classes and functions we'll use throughout the notebook.

In [ ]python · cell 13

python

import json
import logging
import time
from collections.abc import Awaitable, Callable
from datetime import datetime
from pprint import pprint
from typing import Any, NoReturn

import httpx
from IPython.display import Markdown, display
from google.auth import default
from google.auth.transport.requests import Request as req
from starlette.requests import Request

logging.getLogger().setLevel(logging.INFO)


# A2A
from a2a.client import ClientConfig, ClientFactory
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a import types as a2a_types
from a2a.utils import TransportProtocol

# ADK
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.sessions import VertexAiSessionService
from google.adk.tools import google_search_tool
from google.genai import types

# Agent Engine
from vertexai.agent_engines.templates.a2a import A2aAgent, create_agent_card

Helpers

These are simple utility functions to make our lives easier, especially for local testing. They help create mock HTTP requests (build_post_request, build_get_request) and fetch authentication tokens (get_bearer_token).

In [ ]python · cell 15

python

def receive_wrapper(data: dict) -> Callable[[], Awaitable[dict]]:
    """Creates a mock ASGI receive callable for testing."""

    async def receive():
        byte_data = json.dumps(data).encode("utf-8")
        return {"type": "http.request", "body": byte_data, "more_body": False}

    return receive


def build_post_request(
    data: dict[str, Any] | None = None, path_params: dict[str, str] | None = None
) -> Request:
    """Builds a mock Starlette Request object for a POST request with JSON data."""
    scope = {
        "type": "http",
        "http_version": "1.1",
        "headers": [(b"content-type", b"application/json")],
        "app": None,
    }
    if path_params:
        scope["path_params"] = path_params
    receiver = receive_wrapper(data)
    return Request(scope, receiver)


def build_get_request(path_params: dict[str, str]) -> Request:
    """Builds a mock Starlette Request object for a GET request."""
    scope = {
        "type": "http",
        "http_version": "1.1",
        "query_string": b"",
        "app": None,
    }
    if path_params:
        scope["path_params"] = path_params

    async def receive():
        return {"type": "http.disconnect"}

    return Request(scope, receive)


def get_bearer_token() -> str | None:
    """Fetches a Google Cloud bearer token using Application Default Credentials."""
    try:
        # Use an alias to avoid name collision with starlette.requests.Request
        credentials, project = default(
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )
        request = req()
        credentials.refresh(request)
        return credentials.token
    except Exception as e:
        print(f"Error getting credentials: {e}")
        print(
            "Please ensure you have authenticated with 'gcloud auth application-default login'."
        )
    return None

Build a simple ADK agent

Before we can build an A2A agent, we need an agent. We create an agent using the Agent Development Kit (ADK).

In [ ]python · cell 17

python

qna_agent = LlmAgent(
    # The LLM model to use
    model="gemini-2.5-flash",
    # Internal name for the agent (used in logging and sessions)
    name="qa_assistant",
    # Human-readable description
    description="I answer questions using web search.",
    # The system instruction that guides the agent's behavior
    # This is crucial for getting good results
    instruction="""You are a helpful Q&A assistant.
        When asked a question:
        1. Use Google Search to find current, accurate information
        2. Synthesize the search results into a clear answer
        3. Cite your sources when possible
        4. If you can't find a good answer, say so honestly

        Always aim for accuracy over speculation.""",
    # Tools available to the agent
    # The agent will automatically use these when needed
    tools=[google_search_tool.google_search],
)

Define the agent card

AgentCard is an important component of the A2A protocol. Think of it as a digital business card for your agent. It's a structured JSON document that tells other agents everything they need to know to interact with yours: its name, what it does, the skills it offers, and how to call its API endpoint.

We define an AgentSkill to describe our agent's Q&A capability. Then, we use the create_agent_card helper function to assemble the full card, including the agent's name, description, and the skill we just defined.

Note: The utility builds the card based on the limitations in the current integration: Streaming is turned off and supports Authenticated Extended Card is turned on. Also create_agent_card supports agent_card which allows you to supply an agent_card as dictionary. If an Agent Card is supplied as a dictionary, validation errors might show depending on whether the card meets the current integration limitations.

In [ ]python · cell 19

python

# Define a skill - a specific capability your agent offers
# Agents can have multiple skills for different tasks
qna_agent_skill = a2a_types.AgentSkill(
    # Unique identifier for this skill
    id="web_qa",
    # Human-friendly name
    name="Web Q&A",
    # Detailed description helps clients understand when to use this skill
    description="Answer questions using current web search results",
    # Tags for categorization and discovery
    # These help in agent marketplaces or registries
    tags=["question-answering", "search", "research"],
    # Examples show clients what kinds of requests work well
    # This is especially helpful for LLM-based clients
    examples=[
        "What is the current weather in Tokyo?",
        "Who won the latest Nobel Prize in Physics?",
        "What are the symptoms of the flu?",
        "How do I make sourdough bread?",
    ],
    # Optional: specify input/output modes
    # Default is text, but could include images, files, etc.
    input_modes=["text/plain"],
    output_modes=["text/plain"],
)

# Use the helper function to create a complete Agent Card
qna_agent_card = create_agent_card(
    agent_name="Q&A Agent",
    description="A helpful assistant agent that can answer questions.",
    skills=[qna_agent_skill],
)

Let's print the AgentCard we just created.

Take a look at the structure. You can see key fields like name, description, skills, and the url. For now, the URL points to localhost, which is perfect for local testing. When we deploy to Agent Engine, this URL will be automatically updated to point to the managed endpoint.

In [ ]python · cell 21

python

print(qna_agent_card)

Define the agent executor

The AgentExecutor is the bridge between the A2A protocol and our agent's internal logic. It's a class that you implement to handle incoming A2A requests. It has two main methods:

execute: This is the main entry point. When a message arrives, this method gets the user's query from the RequestContext, creates a TaskUpdater - a handy A2A SDK tool for managing the task's lifecycle (e.g., setting its state to working), calls the ADK Runner to process the query with the Gemini model and Google Search tool, asynchronously waits for the final response from the agent, packages the text response into an A2A Artifact—the official output of a task and finally, marks the task as completed.
cancel: Our simple agent doesn't support long-running, cancelable jobs, so we simply state that the operation is unsupported.

In [ ]python · cell 23

python

import os

import vertexai
from google.adk.runners import Runner


class QnAAgentExecutor(AgentExecutor):
    """Refactored Executor using VertexAiSessionService for persistence."""

    def __init__(self) -> None:
        self.agent = None
        self.runner = None

    def _init_agent(self) -> None:
        if self.agent is None:
            # 1. Initialize Agent Platform using environment-injected metadata
            project = os.environ.get("GOOGLE_CLOUD_PROJECT")
            location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
            # This ID is automatically provided by Agent Engine at runtime
            engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID")

            vertexai.init(project=project, location=location)

            self.agent = qna_agent

            # 2. Initialize the Session Service
            # If engine_id exists, we are deployed remotely -> use VertexAiSessionService.
            # If engine_id is None, we are local -> use InMemorySessionService.
            if engine_id:
                session_service = VertexAiSessionService(
                    project=project, location=location, agent_engine_id=engine_id
                )
            else:
                from google.adk.sessions.in_memory_session_service import (
                    InMemorySessionService,
                )

                session_service = InMemorySessionService()

            # 3. Setup Runner with the session service
            self.runner = Runner(
                app_name=self.agent.name,
                agent=self.agent,
                artifact_service=InMemoryArtifactService(),
                session_service=session_service,
                memory_service=InMemoryMemoryService(),
            )

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        if self.agent is None:
            self._init_agent()

        query = context.get_user_input()
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        user_id = (
            context.message.metadata["user_id"]
            if "user_id" in context.message.metadata
            else "vais-query-reasoning-engine"
        )

        task = a2a_types.Task(
            id=context.task_id,
            context_id=context.context_id,
            status=a2a_types.TaskStatus(
                state=a2a_types.TaskState.TASK_STATE_SUBMITTED
            ),
            history=[context.message] if context.message else [],
        )
        await event_queue.enqueue_event(task)

        await updater.start_work()

        try:
            # Using context_id (A2A) as session_id (Vertex) ensures continuity
            session = await self._get_or_create_session(context.context_id, user_id)

            content = types.Content(role="user", parts=[types.Part(text=query)])

            async for event in self.runner.run_async(
                session_id=session.id,
                user_id=user_id,
                new_message=content,
            ):
                if event.is_final_response():
                    answer = self._extract_answer(event)
                    await updater.add_artifact(
                        [a2a_types.Part(text=answer)],
                        name="answer",
                        last_chunk=True,
                    )
                    await updater.complete()
                    break

        except Exception as e:
            await updater.update_status(
                a2a_types.TaskState.TASK_STATE_FAILED,
                message=updater.new_agent_message(
                    [a2a_types.Part(text=f"An error occurred: {str(e)}")]
                ),
            )
            raise

    async def _get_or_create_session(self, context_id: str, user_id: str):
        engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID")
        app_name = engine_id if engine_id else self.agent.name

        session = await self.runner.session_service.get_session(
            app_name=app_name,
            session_id=context_id,
            user_id=user_id,
        )

        if not session:
            session = await self.runner.session_service.create_session(
                app_name=app_name,
                user_id=user_id,
            )
        return session

    def _extract_answer(self, event) -> str:
        parts = event.content.parts
        text_parts = [part.text for part in parts if part.text]
        return " ".join(text_parts) if text_parts else "No answer found."

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        """Handle task cancellation requests."""
        task_id = context.task_id
        updater = TaskUpdater(
            event_queue=event_queue,
            task_id=task_id or "",
            context_id=context.context_id or "",
        )
        await updater.cancel()

Test the agent locally

Before deploying anything to the cloud, a crucial step is to test locally. This allows for rapid iteration and debugging.

The A2aAgent class from the Agent Platform SDK is our deployable unit. It wraps our AgentCard and AgentExecutor together. Calling set_up() prepares an in-memory server, allowing us to simulate calls to the agent as if it were deployed.

In [ ]python · cell 25

python

a2a_agent = A2aAgent(agent_card=qna_agent_card, agent_executor_builder=QnAAgentExecutor,extended_agent_card=qna_agent_card)
a2a_agent.set_up()

Get the agent card

At this point, we can call the handle_authenticated_agent_card method on our local agent instance to simulate a client discovering our agent by requesting its "business card." It would return the agent's capabilities, skills, and its endpoint URL, confirming our local server is set up correctly.

In [ ]python · cell 27

python

from a2a.types import GetExtendedAgentCardRequest
from a2a.server.context import ServerCallContext

request = GetExtendedAgentCardRequest()
context = ServerCallContext()

response = await a2a_agent.on_get_extended_agent_card(
    request=request, context=context
)

print(response)

Send a query

Finally, let's call the on_message_send method, which is the A2A endpoint for starting a new task.

The agent immediately responds with a Task object in the TASK_STATE_SUBMITTED state. This is standard asynchronous behavior: the system acknowledges the request and gives us a task_id to track its progress.

In [ ]python · cell 29

python

from a2a.types import SendMessageRequest, Message, Part
from a2a.server.context import ServerCallContext

message = Message(
    message_id=f"msg-{os.urandom(8).hex()}",
    role="ROLE_USER",
    parts=[Part(text="What is the capital of France?")],
)
message.metadata["user_id"] = "custom-user-123"

request_params = SendMessageRequest(message=message)
context = ServerCallContext()

response = await a2a_agent.on_message_send(request=request_params, context=context)
print(response)

We simply extract the task_id from the previous response and print it. We'll use this ID in the next step to fetch our answer.

In [ ]python · cell 31

python

if hasattr(response, "id"):
    task_id = response.id
else:
    task_id = response["task"]["id"]

print(f"The Task ID is: {task_id}")

Get the response

With the task_id in hand, we can now poll for the result. We call the on_get_task method, which retrieves the current status of our task.

Since our ADK agent is fast, the task should have already moved to the TASK_STATE_COMPLETED state. Notice the artifacts field in the response. This contains the answer to our question, neatly packaged as a text part.

In [ ]python · cell 33

python

from a2a.types import GetTaskRequest
from a2a.server.context import ServerCallContext

request_params = GetTaskRequest(id=task_id)
context = ServerCallContext()

response = await a2a_agent.on_get_task(request=request_params, context=context)

print(response)

# Extract the artifacts from the Task Protobuf response
for artifact in response.artifacts:
    if artifact.parts:
        part = artifact.parts[0]
        if hasattr(part, "text") and part.text:
            display(Markdown(f"**Answer**:\n {part.text}"))
        else:
            print("Could not extract text from artifact parts.")

(Optional) Cancel a task

If your agent executes long run operations, you can always cancel the associated task as shown below:

python

from a2a.types import CancelTaskRequest
from a2a.server.context import ServerCallContext

# 1. Construct the cancel request with the task ID
request_params = CancelTaskRequest(id=task_id)
context = ServerCallContext()

# 2. Call the agent's handler to cancel the task
response = await a2a_agent.on_cancel_task(request=request_params, context=context)

print(response)

Deploy on Agent Engine

Now it is time to deploy the agent to a fully-managed, scalable platform, Agent Platform.

With a single agent_engines.create() call, the Agent Platform SDK performs a series of actions behind the scenes that allows you to scale your A2A agent. In order:

It takes our local a2a_agent object.
It serializes (pickles) the agent's code and its configuration.
It inspects the environment to determine the necessary Python package requirements.
It packages everything up and uploads it to the Cloud Storage bucket we configured earlier.
It provisions a secure, scalable, and fully-managed serverless endpoint on Agent Engine to host our agent.

In [ ]python · cell 36

python

remote_a2a_agent = client.agent_engines.create(
    # The actual agent to deploy
    agent=a2a_agent,
    config={
        # Display name shown in the console
        "display_name": a2a_agent.agent_card.name,
        # Description for documentation
        "description": a2a_agent.agent_card.description,
        # Python dependencies needed in Agent Engine
        "requirements": [
            "a2a-sdk>=1.0.0",
            "google-cloud-aiplatform[agent_engines,adk]>=1.156.0"
        ],
        # Http options
        "http_options": {
            "base_url": ENDPOINT,
            "api_version": "v1beta1",
        },
        # Staging bucket
        "staging_bucket": BUCKET_URI,
        "min_instances": 1,
        "max_instances": 1
    },
)

Get the remote agent card

The SDK handles the authentication and API call to our deployed endpoint, and we get back the AgentCard. The get method allows you to reconnect to an existing, already-deployed agent in a new session just by using its resource name.

Notice that the url field in the card now points to the public aiplatform.googleapis.com endpoint, not localhost.

In [ ]python · cell 38

python

from a2a.types import GetExtendedAgentCardRequest
from a2a.server.context import ServerCallContext

remote_a2a_agent_resource_name = remote_a2a_agent.api_resource.name
config = {"http_options": {"base_url": ENDPOINT}}

remote_a2a_agent = client.agent_engines.get(
    name=remote_a2a_agent_resource_name,
    config=config,
)

request = GetExtendedAgentCardRequest()
context = ServerCallContext()

remote_a2a_agent_card = await remote_a2a_agent.on_get_extended_agent_card(
    request=request, context=context
)

print(f"Agent: {remote_a2a_agent_card.name}")
print(f"Supported Interfaces: {remote_a2a_agent_card.supported_interfaces}")
print(f"Skills: {[s.description for s in remote_a2a_agent_card.skills]}")
print(f"Examples: {[s.examples for s in remote_a2a_agent_card.skills][0]}")

Query the remote A2A agent

Our agent is now live on Agent Platform! Let's interact with it.

Agent Engine and its A2A integration provides multiple ways to connect, catering to different developer needs and use cases. We'll explore three common methods:

Via Agent Platform SDK for Python
Via A2A Client
Via http request

Via Agent Platform SDK for Python

For Python developers, this is the simplest method. The remote_a2a_agent acts as a smart client or proxy that knows how to communicate with the deployed endpoint. This allows you to use the same methods you used for local testing to interact with the remote agent.

Send a message to start a task

Again, the code is nearly identical to our local test. We call on_message_send with our question. The SDK sends the request to the deployed agent, which kicks off the task on the agent engine. The response contains the task_id for our remote job.

In [ ]python · cell 42

python

from a2a.types import SendMessageRequest, Message, Part
from a2a.server.context import ServerCallContext

message = Message(
    message_id=f"msg-{os.urandom(8).hex()}",
    role="ROLE_USER",
    parts=[Part(text="What is the capital of Italy?")],
)

request_params = SendMessageRequest(message=message)
context = ServerCallContext()

# Invoke the remote agent
response = await remote_a2a_agent.on_message_send(request=request_params, context=context)

In [ ]python · cell 43

python

# The response contains a StreamResponse containing a Task object
task_object = None
for chunk in response:
    if hasattr(chunk, "task") and chunk.task.id:
        task_object = chunk.task
        break

if task_object:
    task_id = task_object.id
    print(f"Task started: {task_id}")
    print(f"Status: {task_object.status.state}")
else:
    print("Could not retrieve the task object from the response.")

Get the response

Using the task_id from the previous step, we call on_get_task.

The SDK polls the Agent Engine endpoint and retrieves the completed task, including the final answer in the artifacts field. We have successfully communicated with our deployed A2A agent.

Note: Running this cell might require few seconds depending on the use case.

In [ ]python · cell 45

python

from a2a.types import GetTaskRequest, TaskState
from a2a.server.context import ServerCallContext

request_params = GetTaskRequest(id=task_id, history_length=1)
context = ServerCallContext()

result = None
retries = 0
max_retries = 30

while True:
    try:
        # Get the task result
        result = await remote_a2a_agent.on_get_task(request=request_params, context=context)

        if result.status.state in [TaskState.TASK_STATE_COMPLETED, TaskState.TASK_STATE_FAILED]:
            break

        print(f"Task state: {result.status.state}. Waiting 1s...")
        time.sleep(1)

    except Exception as e:
        error_str = str(e)
        if "400 Bad Request" in error_str:
            retries += 1
            if retries <= max_retries:
                print(f"Received HTTP 400. Retrying in 1s ({retries}/{max_retries})...")
                time.sleep(1)
                continue
            else:
                print("Max retries reached.")
                raise
        else:
            raise

In [ ]python · cell 46

python

# Artifacts contain the actual results
for artifact in result.artifacts:
    if artifact.parts:
        part = artifact.parts[0]
        if hasattr(part, "text") and part.text:
            display(Markdown(f"**Answer**:\n {part.text}"))
        else:
            print("Could not extract text from artifact parts.")

(Optional) Cancel a task

If your remote agent executes long run operations, you can always cancel the associated task as shown below:

python

# Remote agent via Agent Platform SDK
task_data = {
    "id": task_id,
}
response = await remote_a2a_agent.on_cancel_task(**task_data)
print(response)

Via A2A Client

This method is for developers who want to use the standard, open-source a2a-sdk client directly. This is useful if you're building an application that needs to talk to various A2A agents, not just those hosted on Agent Engine, or if you prefer to work directly with the protocol's native objects.

Initialize A2A Client

Here, we set up the A2A SDK ClientFactory.

We start from the AgentCard we fetched from our deployed agent. This is crucial because the client needs the card (especially the url) to know where to send requests.

Next, we get standard Google Cloud authentication credentials. Then, we create a ClientConfig object, telling it to use standard HTTP transport and providing a httpx client pre-configured with our authentication headers.

Finally, the factory.create(remote_a2a_agent_card) call gives us a client instance ready to communicate with our specific agent endpoint.

In [ ]python · cell 50

python

# Get authentication token for Google Cloud
bearer_token = get_bearer_token()
headers = {
    "Authorization": f"Bearer {bearer_token}",
    "Content-Type": "application/json",
}

# Configure the A2A client factory
# This handles the protocol implementation details
factory = ClientFactory(
    ClientConfig(
        supported_protocol_bindings=[TransportProtocol.HTTP_JSON],
        # Use client preferences for protocol negotiation
        use_client_preference=True,
        # Configure HTTP client with authentication
        httpx_client=httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {bearer_token}",
                "Content-Type": "application/json",
            },
            timeout=60.0
        ),
    )
)
# Temporary workaround for faulty URL returned by the server
correct_engine_id = remote_a2a_agent.api_resource.name.split("/")[-1]
for interface in remote_a2a_agent_card.supported_interfaces:
    if interface.url and "test-agent-engine" in interface.url:
        interface.url = interface.url.replace("test-agent-engine", correct_engine_id)
# end of the workaround
# Create a client for our specific agent
# The client uses the Agent Card to understand capabilities
a2a_client = factory.create(remote_a2a_agent_card)

Get the agent card

This is a simple call to the A2A client get_extended_agent_card() method to verify that our connection and authentication are configured correctly. It should return the same agent card we've seen before.

In [ ]python · cell 52

python

from a2a.types import GetExtendedAgentCardRequest
from a2a.server.context import ServerCallContext

response = await a2a_client.get_extended_agent_card(GetExtendedAgentCardRequest())

print(f"Agent: {response.name}")
print(f"Supported Interfaces: {response.supported_interfaces}")
print(f"Skills: {[s.description for s in response.skills]}")
print(f"Examples: {[s.examples for s in response.skills][0] if response.skills else 'No skills'}")

Send a message to start a task

Once again, we manually construct a Message object with a role, parts, and a unique message_id. We then call a2a_client.send_message(). This method returns an async generator, so we loop through it to get the resulting chunks, which contain the submitted Task object. From there, we extract the task_id.

In [ ]python · cell 54

python

from a2a.types import SendMessageRequest

# Send a message using A2A protocol objects
message = Message(
    message_id=f"message-{os.urandom(8).hex()}",
    role="ROLE_USER",
    parts=[Part(text="What is the weather in Paris today?")],
)

# In A2A 1.0, send_message expects SendMessageRequest
request_params = SendMessageRequest(message=message)

# Get response (async iterator yielding StreamResponse)
response = a2a_client.send_message(request_params)

task_id = None
task_object = None

# The response is an async generator
async for response_chunk in response:
    # Check if the chunk has the task field populated
    if response_chunk.HasField("task"):
        task_object = response_chunk.task
        task_id = task_object.id
        break

if task_id:
    print(f"Task started: {task_id}")
    print(f"Status: {task_object.status.state}")
else:
    print("Could not retrieve task_id from stream.")

Get the response

Using the task_id, we create a TaskQueryParams object and pass it to the a2a_client.get_task() method. This fetches the final result from the Agent Engine endpoint, demonstrating a successful interaction using the generic A2A client SDK.

Note: Running this cell might require few seconds depending on the use case.

In [ ]python · cell 56

python

from a2a.types import GetTaskRequest

# Poll for completion
task_data = {
    "id": task_id,
    "history_length": 1,
}
response = None
retries = 0
max_retries = 30  # Set a reasonable maximum number of retries

while True:
    try:
        print(f"Attempting to get task {task_id} (Retry {retries}/{max_retries})...")
        response = await a2a_client.get_task(GetTaskRequest(id=task_id, history_length=1))

        # If we get a response, check the task state
        if response.status.state == TaskState.TASK_STATE_COMPLETED:
            print(f"Task {task_id} completed successfully.")
            break  # Exit loop if task is completed
        elif response.status.state == TaskState.TASK_STATE_FAILED:
            print(f"Task {task_id} failed with state: {response.status.state}.")
            break  # Exit loop if task is failed
        else:
            # If still in progress, wait and retry
            print(
                f"Task {task_id} is still in state: {response.status.state}. Waiting 1 second..."
            )
            # Wait for a second before checking again to avoid spamming the API.
            time.sleep(1)
            continue

    except Exception as e:
        error_str = str(e)
        if "400 Bad Request" in error_str:
            retries += 1
            if retries < max_retries:
                time.sleep(1)
                continue  # Retry
            else:
                print(
                    f"Max retries ({max_retries}) reached for HTTP 400 Bad Request for task {task_id}."
                )
                raise  # Re-raise if max retries reached
        else:
            print(f"An error occurred for task {task_id}: {e}")
            raise

# Check if it has artifacts only if the task completed successfully
if (
    response
    and response.status.state == TaskState.TASK_STATE_COMPLETED
    and hasattr(response, "artifacts")
    and response.artifacts
):
    for artifact in response.artifacts:
        if artifact.parts:
            part = artifact.parts[0]
            if hasattr(part, "text") and part.text:
                display(Markdown(f"**Result**\n: {part.text}"))
            else:
                print("Could not extract text from artifact parts.")
elif response and response.status.state == TaskState.TASK_STATE_FAILED:
    print(f"Task {task_id} failed. No artifacts to display.")
else:
    print("No artifacts found or task not completed successfully.")

Via http request

This is the most fundamental way to interact with our agent: making direct HTTP requests.

This approach is perfect for debugging with tools like curl or for integrating from languages that don't have a dedicated A2A or Agent Platform SDK.

Get the agent card

To start, we get the endpoint URL from the agent card we fetched earlier. Next, we obtain a bearer token for authentication. Then we set the necessary Authorization and Content-Type headers. And finally, we use the httpx library to make the GET request and print the resulting JSON response.

In [ ]python · cell 59

python

# Prepare authentication headers
headers = {
    "Authorization": f"Bearer {get_bearer_token()}",
    "Content-Type": "application/json",
    "A2A-Version": "1.0"
}

# Get the agent card endpoint from supported_interfaces in A2A 1.0
remote_agent_card_url = None
for interface in remote_a2a_agent_card.supported_interfaces:
    if "HTTP" in interface.protocol_binding:
        remote_agent_card_url = interface.url
        # Remove trailing slash if present to avoid double slashes in endpoints
        if remote_agent_card_url.endswith("/"):
             remote_agent_card_url = remote_agent_card_url[:-1]
        break

if not remote_agent_card_url:
    raise ValueError("Could not find HTTP interface in remote agent card.")

remote_agent_card_endpoint = f"{remote_agent_card_url}/extendedAgentCard"
print(f"Fetching card from: {remote_agent_card_endpoint}")

try:
    # Send the HTTP request
    response = httpx.get(remote_agent_card_endpoint, headers=headers)
    response.raise_for_status()
    # Parse the response
    result = response.json()
    print(json.dumps(result, indent=2))
except httpx.HTTPStatusError as e:
    print(f"HTTP error occurred: {e}")
    print(f"Response body: {e.response.text}")
except httpx.RequestError as e:
    print(f"An error occurred while trying to send the request: {e}")

Send a message to start a task

Now you can make a POST request to the /message:send endpoint. We construct the JSON payload manually, following the structure defined by the A2A protocol. We then send the request using httpx.post with the same headers as before. The response is the JSON object for the submitted task, from which we extract the task_id.

Note: Running this cell might require few seconds depending on the use case.

In [ ]python · cell 61

python

# Construct the A2A message payload for A2A 1.0
payload = {
    "message": {
        "messageId": f"msg-{os.urandom(8).hex()}",
        "role": "ROLE_USER",
        "parts": [{"text": "Who is the current UN Secretary-General?"}],
    },
    "metadata": {
        # Optional metadata for tracking/debugging
        "source": "tutorial",
        "timestamp": datetime.now().isoformat(),
        "user_agent": "test_script",
    },
}

try:
    # Send the HTTP request
    response = httpx.post(
        f"{remote_agent_card_url}/message:send", json=payload, headers=headers
    )
    response.raise_for_status()
    # Parse the response
    result = response.json()
    print(json.dumps(result, indent=2))
except httpx.HTTPStatusError as e:
    print(f"HTTP error occurred: {e}")
    print(f"Response body: {e.response.text}")
except httpx.RequestError as e:
    print(f"An error occurred while trying to send the request: {e}")

# The response contains a task
# Protobuf JSON might return 'id' directly or nested depending on how SendMessageResponse is structured
# Let's try to extract it safely
task_id = result.get("id") or result.get("task", {}).get("id")
task_status = result.get("status", {}).get("state") or result.get("task", {}).get("status", {}).get("state")

print(f"Task started: {task_id}")
print(f"Status: {task_status}")

Get the response

Finally, we construct the URL for the specific task using the task_id and make a GET request to the /tasks/{task_id} endpoint. The response is the full JSON object for the completed task, containing the final answer. This confirms we can interact with our agent using nothing but standard HTTP calls.

In [ ]python · cell 63

python

# Poll for completion
task_url = f"{remote_agent_card_url}/tasks/{task_id}"
print(f"Polling for results at: {task_url}")

task_data = {}
state = None
retries = 0
max_retries = 30

while True:
    try:
        # Poll the task endpoint until it reaches a terminal state.
        response = httpx.get(task_url, headers=headers, params={"historyLength": 1})
        response.raise_for_status()
        task_data = response.json()
        state = task_data["status"]["state"]

        if state in ["TASK_STATE_COMPLETED", "TASK_STATE_FAILED"]:
            print(f"Task finished with state: {state}")
            break

        # Wait for a second before checking again to avoid spamming the API.
        time.sleep(1)

    except httpx.HTTPStatusError as e:
        if e.response.status_code == 400:
            retries += 1
            if retries <= max_retries:
                print(
                    f"Received HTTP 400 Bad Request. Retrying in 1s ({retries}/{max_retries})..."
                )
                time.sleep(1)
                continue
            else:
                print("Max retries reached.")
                raise
        else:
            raise

# Extract the result
if state == "TASK_STATE_COMPLETED":
    artifacts = task_data.get("artifacts", [])
    for artifact in artifacts:
        for part in artifact["parts"]:
            if "text" in part:
                display(Markdown(f"**Result**\n: {part['text']}"))
                break

Cleaning up

Time to clean up. Run the cell below prevents you from incurring ongoing costs for the services you've provisioned during this tutorial.

In [ ]python · cell 65

python

delete_agent_engine = True

if delete_agent_engine:
    remote_a2a_agent.delete(force=True)

← 返回课程

Generative AI on Google Cloud

6 / 80·notebook

Chapter 06

Get started with A2A on Agent Engine

NotebookPython 3 (ipykernel)65 cells

In [ ]python · cell 1

python

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Get started with A2A on Agent Engine

Share to:

Authors
Joyce Liu
Rajesh Velicheti
Ivan Nardini
Max Gasztych

Overview

This notebook shows how to build, deploy, and interact with Agent2Agent (A2A) protocol agents hosted on the fully-managed, serverless Agent Platform.

Agent Platform is fully-managed, serverless platform for running A2A agents. It handles all the infrastructure, scaling, security, and monitoring so you can focus on your agent's logic.

In this tutorial, you will:

Build a simple, A2A-compliant agent using the Agent Platform SDK.
Test the agent locally to ensure it works as expected.
Deploy the agent to Agent Engine with a single command.
Query the managed agent endpoint using three different methods (Agent Platform SDK, A2A SDK, and direct HTTP requests).
Clean up the resources you've created.

Get started

Install required packages

First, we'll install the necessary packages.

a2a-sdk is the foundational open-source SDK for building A2A-compliant agents.
google-cloud-aiplatform is the Agent Platform SDK, containing the new Agent Engine template we'll use for deployment.

In [ ]python · cell 7

python

%pip install --upgrade --quiet "a2a-sdk>=1.0.0" --force-reinstall --quiet
%pip install --upgrade --quiet "google-cloud-aiplatform[agent_engines,adk]>=1.156.0" --force-reinstall --quiet

Authenticate your notebook environment (Colab only)

If you're running this notebook on Google Colab, run the cell below to authenticate your environment.

In [ ]python · cell 9

python

import sys

if "google.colab" in sys.modules:
    from google.colab import auth

    auth.authenticate_user()

Set Google Cloud project information

To get started using Agent Platform, you must have an existing Google Cloud project and enable the Agent Platform API.

Learn more about setting up a project and a development environment.

In [ ]python · cell 11

python

# Use the environment variable if the user doesn't provide Project ID.
import os

import vertexai
from google.genai import types

# fmt: off
PROJECT_ID = "[your-project-id]"  # @param {type: "string", placeholder: "[your-project-id]", isTemplate: true}
if not PROJECT_ID or PROJECT_ID == "[your-project-id]":
    PROJECT_ID = str(os.environ.get("GOOGLE_CLOUD_PROJECT"))

LOCATION = "[your-location]"  # @param {type: "string", placeholder: "[your-location]", isTemplate: true}
if not LOCATION or LOCATION == "[your-location]":
    LOCATION = str(os.environ.get("GOOGLE_CLOUD_REGION"))

BUCKET_NAME = "[your-bucket-name]"  # @param {type: "string", placeholder: "[your-bucket-name]", isTemplate: true}
if not BUCKET_NAME or BUCKET_NAME == "[your-bucket-name]":
    BUCKET_NAME = PROJECT_ID

BUCKET_URI = f"gs://{BUCKET_NAME}"

ENDPOINT = f"https://{LOCATION}-aiplatform.googleapis.com"


# !gsutil mb -l $LOCATION -p $PROJECT_ID $BUCKET_URI

# Initialize Agent Platform session
vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=BUCKET_URI,
    api_endpoint=ENDPOINT,  # This directs requests to the {$ENV} endpoint
)

# Initialize the Gen AI client using http_options
# The parameter customizes how the Agent Platform client communicates with Google Cloud's backend services.
# It's used here to access new, pre-release features.
client = vertexai.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options=types.HttpOptions(api_version="v1beta1", base_url=f"{ENDPOINT}/"),
)

Import libraries

Here, we're importing all the necessary Python classes and functions we'll use throughout the notebook.

In [ ]python · cell 13

python

import json
import logging
import time
from collections.abc import Awaitable, Callable
from datetime import datetime
from pprint import pprint
from typing import Any, NoReturn

import httpx
from IPython.display import Markdown, display
from google.auth import default
from google.auth.transport.requests import Request as req
from starlette.requests import Request

logging.getLogger().setLevel(logging.INFO)


# A2A
from a2a.client import ClientConfig, ClientFactory
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a import types as a2a_types
from a2a.utils import TransportProtocol

# ADK
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.sessions import VertexAiSessionService
from google.adk.tools import google_search_tool
from google.genai import types

# Agent Engine
from vertexai.agent_engines.templates.a2a import A2aAgent, create_agent_card

Helpers

In [ ]python · cell 15

python

def receive_wrapper(data: dict) -> Callable[[], Awaitable[dict]]:
    """Creates a mock ASGI receive callable for testing."""

    async def receive():
        byte_data = json.dumps(data).encode("utf-8")
        return {"type": "http.request", "body": byte_data, "more_body": False}

    return receive


def build_post_request(
    data: dict[str, Any] | None = None, path_params: dict[str, str] | None = None
) -> Request:
    """Builds a mock Starlette Request object for a POST request with JSON data."""
    scope = {
        "type": "http",
        "http_version": "1.1",
        "headers": [(b"content-type", b"application/json")],
        "app": None,
    }
    if path_params:
        scope["path_params"] = path_params
    receiver = receive_wrapper(data)
    return Request(scope, receiver)


def build_get_request(path_params: dict[str, str]) -> Request:
    """Builds a mock Starlette Request object for a GET request."""
    scope = {
        "type": "http",
        "http_version": "1.1",
        "query_string": b"",
        "app": None,
    }
    if path_params:
        scope["path_params"] = path_params

    async def receive():
        return {"type": "http.disconnect"}

    return Request(scope, receive)


def get_bearer_token() -> str | None:
    """Fetches a Google Cloud bearer token using Application Default Credentials."""
    try:
        # Use an alias to avoid name collision with starlette.requests.Request
        credentials, project = default(
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )
        request = req()
        credentials.refresh(request)
        return credentials.token
    except Exception as e:
        print(f"Error getting credentials: {e}")
        print(
            "Please ensure you have authenticated with 'gcloud auth application-default login'."
        )
    return None

Build a simple ADK agent

Before we can build an A2A agent, we need an agent. We create an agent using the Agent Development Kit (ADK).

In [ ]python · cell 17

python

qna_agent = LlmAgent(
    # The LLM model to use
    model="gemini-2.5-flash",
    # Internal name for the agent (used in logging and sessions)
    name="qa_assistant",
    # Human-readable description
    description="I answer questions using web search.",
    # The system instruction that guides the agent's behavior
    # This is crucial for getting good results
    instruction="""You are a helpful Q&A assistant.
        When asked a question:
        1. Use Google Search to find current, accurate information
        2. Synthesize the search results into a clear answer
        3. Cite your sources when possible
        4. If you can't find a good answer, say so honestly

        Always aim for accuracy over speculation.""",
    # Tools available to the agent
    # The agent will automatically use these when needed
    tools=[google_search_tool.google_search],
)

Define the agent card

Note: The utility builds the card based on the limitations in the current integration: Streaming is turned off and supports Authenticated Extended Card is turned on. Also create_agent_card supports agent_card which allows you to supply an agent_card as dictionary. If an Agent Card is supplied as a dictionary, validation errors might show depending on whether the card meets the current integration limitations.

In [ ]python · cell 19

python

# Define a skill - a specific capability your agent offers
# Agents can have multiple skills for different tasks
qna_agent_skill = a2a_types.AgentSkill(
    # Unique identifier for this skill
    id="web_qa",
    # Human-friendly name
    name="Web Q&A",
    # Detailed description helps clients understand when to use this skill
    description="Answer questions using current web search results",
    # Tags for categorization and discovery
    # These help in agent marketplaces or registries
    tags=["question-answering", "search", "research"],
    # Examples show clients what kinds of requests work well
    # This is especially helpful for LLM-based clients
    examples=[
        "What is the current weather in Tokyo?",
        "Who won the latest Nobel Prize in Physics?",
        "What are the symptoms of the flu?",
        "How do I make sourdough bread?",
    ],
    # Optional: specify input/output modes
    # Default is text, but could include images, files, etc.
    input_modes=["text/plain"],
    output_modes=["text/plain"],
)

# Use the helper function to create a complete Agent Card
qna_agent_card = create_agent_card(
    agent_name="Q&A Agent",
    description="A helpful assistant agent that can answer questions.",
    skills=[qna_agent_skill],
)

Let's print the AgentCard we just created.

In [ ]python · cell 21

python

print(qna_agent_card)

Define the agent executor

The AgentExecutor is the bridge between the A2A protocol and our agent's internal logic. It's a class that you implement to handle incoming A2A requests. It has two main methods:

execute: This is the main entry point. When a message arrives, this method gets the user's query from the RequestContext, creates a TaskUpdater - a handy A2A SDK tool for managing the task's lifecycle (e.g., setting its state to working), calls the ADK Runner to process the query with the Gemini model and Google Search tool, asynchronously waits for the final response from the agent, packages the text response into an A2A Artifact—the official output of a task and finally, marks the task as completed.
cancel: Our simple agent doesn't support long-running, cancelable jobs, so we simply state that the operation is unsupported.

In [ ]python · cell 23

python

import os

import vertexai
from google.adk.runners import Runner


class QnAAgentExecutor(AgentExecutor):
    """Refactored Executor using VertexAiSessionService for persistence."""

    def __init__(self) -> None:
        self.agent = None
        self.runner = None

    def _init_agent(self) -> None:
        if self.agent is None:
            # 1. Initialize Agent Platform using environment-injected metadata
            project = os.environ.get("GOOGLE_CLOUD_PROJECT")
            location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
            # This ID is automatically provided by Agent Engine at runtime
            engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID")

            vertexai.init(project=project, location=location)

            self.agent = qna_agent

            # 2. Initialize the Session Service
            # If engine_id exists, we are deployed remotely -> use VertexAiSessionService.
            # If engine_id is None, we are local -> use InMemorySessionService.
            if engine_id:
                session_service = VertexAiSessionService(
                    project=project, location=location, agent_engine_id=engine_id
                )
            else:
                from google.adk.sessions.in_memory_session_service import (
                    InMemorySessionService,
                )

                session_service = InMemorySessionService()

            # 3. Setup Runner with the session service
            self.runner = Runner(
                app_name=self.agent.name,
                agent=self.agent,
                artifact_service=InMemoryArtifactService(),
                session_service=session_service,
                memory_service=InMemoryMemoryService(),
            )

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        if self.agent is None:
            self._init_agent()

        query = context.get_user_input()
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        user_id = (
            context.message.metadata["user_id"]
            if "user_id" in context.message.metadata
            else "vais-query-reasoning-engine"
        )

        task = a2a_types.Task(
            id=context.task_id,
            context_id=context.context_id,
            status=a2a_types.TaskStatus(
                state=a2a_types.TaskState.TASK_STATE_SUBMITTED
            ),
            history=[context.message] if context.message else [],
        )
        await event_queue.enqueue_event(task)

        await updater.start_work()

        try:
            # Using context_id (A2A) as session_id (Vertex) ensures continuity
            session = await self._get_or_create_session(context.context_id, user_id)

            content = types.Content(role="user", parts=[types.Part(text=query)])

            async for event in self.runner.run_async(
                session_id=session.id,
                user_id=user_id,
                new_message=content,
            ):
                if event.is_final_response():
                    answer = self._extract_answer(event)
                    await updater.add_artifact(
                        [a2a_types.Part(text=answer)],
                        name="answer",
                        last_chunk=True,
                    )
                    await updater.complete()
                    break

        except Exception as e:
            await updater.update_status(
                a2a_types.TaskState.TASK_STATE_FAILED,
                message=updater.new_agent_message(
                    [a2a_types.Part(text=f"An error occurred: {str(e)}")]
                ),
            )
            raise

    async def _get_or_create_session(self, context_id: str, user_id: str):
        engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID")
        app_name = engine_id if engine_id else self.agent.name

        session = await self.runner.session_service.get_session(
            app_name=app_name,
            session_id=context_id,
            user_id=user_id,
        )

        if not session:
            session = await self.runner.session_service.create_session(
                app_name=app_name,
                user_id=user_id,
            )
        return session

    def _extract_answer(self, event) -> str:
        parts = event.content.parts
        text_parts = [part.text for part in parts if part.text]
        return " ".join(text_parts) if text_parts else "No answer found."

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        """Handle task cancellation requests."""
        task_id = context.task_id
        updater = TaskUpdater(
            event_queue=event_queue,
            task_id=task_id or "",
            context_id=context.context_id or "",
        )
        await updater.cancel()

Test the agent locally

Before deploying anything to the cloud, a crucial step is to test locally. This allows for rapid iteration and debugging.

In [ ]python · cell 25

python

a2a_agent = A2aAgent(agent_card=qna_agent_card, agent_executor_builder=QnAAgentExecutor,extended_agent_card=qna_agent_card)
a2a_agent.set_up()

Get the agent card

In [ ]python · cell 27

python

from a2a.types import GetExtendedAgentCardRequest
from a2a.server.context import ServerCallContext

request = GetExtendedAgentCardRequest()
context = ServerCallContext()

response = await a2a_agent.on_get_extended_agent_card(
    request=request, context=context
)

print(response)

Send a query

Finally, let's call the on_message_send method, which is the A2A endpoint for starting a new task.

In [ ]python · cell 29

python

from a2a.types import SendMessageRequest, Message, Part
from a2a.server.context import ServerCallContext

message = Message(
    message_id=f"msg-{os.urandom(8).hex()}",
    role="ROLE_USER",
    parts=[Part(text="What is the capital of France?")],
)
message.metadata["user_id"] = "custom-user-123"

request_params = SendMessageRequest(message=message)
context = ServerCallContext()

response = await a2a_agent.on_message_send(request=request_params, context=context)
print(response)

We simply extract the task_id from the previous response and print it. We'll use this ID in the next step to fetch our answer.

In [ ]python · cell 31

python

if hasattr(response, "id"):
    task_id = response.id
else:
    task_id = response["task"]["id"]

print(f"The Task ID is: {task_id}")

Get the response

With the task_id in hand, we can now poll for the result. We call the on_get_task method, which retrieves the current status of our task.

In [ ]python · cell 33

python

from a2a.types import GetTaskRequest
from a2a.server.context import ServerCallContext

request_params = GetTaskRequest(id=task_id)
context = ServerCallContext()

response = await a2a_agent.on_get_task(request=request_params, context=context)

print(response)

# Extract the artifacts from the Task Protobuf response
for artifact in response.artifacts:
    if artifact.parts:
        part = artifact.parts[0]
        if hasattr(part, "text") and part.text:
            display(Markdown(f"**Answer**:\n {part.text}"))
        else:
            print("Could not extract text from artifact parts.")

(Optional) Cancel a task

If your agent executes long run operations, you can always cancel the associated task as shown below:

python

from a2a.types import CancelTaskRequest
from a2a.server.context import ServerCallContext

# 1. Construct the cancel request with the task ID
request_params = CancelTaskRequest(id=task_id)
context = ServerCallContext()

# 2. Call the agent's handler to cancel the task
response = await a2a_agent.on_cancel_task(request=request_params, context=context)

print(response)

Deploy on Agent Engine

Now it is time to deploy the agent to a fully-managed, scalable platform, Agent Platform.

With a single agent_engines.create() call, the Agent Platform SDK performs a series of actions behind the scenes that allows you to scale your A2A agent. In order:

It takes our local a2a_agent object.
It serializes (pickles) the agent's code and its configuration.
It inspects the environment to determine the necessary Python package requirements.
It packages everything up and uploads it to the Cloud Storage bucket we configured earlier.
It provisions a secure, scalable, and fully-managed serverless endpoint on Agent Engine to host our agent.

In [ ]python · cell 36

python

remote_a2a_agent = client.agent_engines.create(
    # The actual agent to deploy
    agent=a2a_agent,
    config={
        # Display name shown in the console
        "display_name": a2a_agent.agent_card.name,
        # Description for documentation
        "description": a2a_agent.agent_card.description,
        # Python dependencies needed in Agent Engine
        "requirements": [
            "a2a-sdk>=1.0.0",
            "google-cloud-aiplatform[agent_engines,adk]>=1.156.0"
        ],
        # Http options
        "http_options": {
            "base_url": ENDPOINT,
            "api_version": "v1beta1",
        },
        # Staging bucket
        "staging_bucket": BUCKET_URI,
        "min_instances": 1,
        "max_instances": 1
    },
)

Get the remote agent card

Notice that the url field in the card now points to the public aiplatform.googleapis.com endpoint, not localhost.

In [ ]python · cell 38

python

from a2a.types import GetExtendedAgentCardRequest
from a2a.server.context import ServerCallContext

remote_a2a_agent_resource_name = remote_a2a_agent.api_resource.name
config = {"http_options": {"base_url": ENDPOINT}}

remote_a2a_agent = client.agent_engines.get(
    name=remote_a2a_agent_resource_name,
    config=config,
)

request = GetExtendedAgentCardRequest()
context = ServerCallContext()

remote_a2a_agent_card = await remote_a2a_agent.on_get_extended_agent_card(
    request=request, context=context
)

print(f"Agent: {remote_a2a_agent_card.name}")
print(f"Supported Interfaces: {remote_a2a_agent_card.supported_interfaces}")
print(f"Skills: {[s.description for s in remote_a2a_agent_card.skills]}")
print(f"Examples: {[s.examples for s in remote_a2a_agent_card.skills][0]}")

Query the remote A2A agent

Our agent is now live on Agent Platform! Let's interact with it.

Agent Engine and its A2A integration provides multiple ways to connect, catering to different developer needs and use cases. We'll explore three common methods:

Via Agent Platform SDK for Python
Via A2A Client
Via http request

Via Agent Platform SDK for Python

Send a message to start a task

In [ ]python · cell 42

python

from a2a.types import SendMessageRequest, Message, Part
from a2a.server.context import ServerCallContext

message = Message(
    message_id=f"msg-{os.urandom(8).hex()}",
    role="ROLE_USER",
    parts=[Part(text="What is the capital of Italy?")],
)

request_params = SendMessageRequest(message=message)
context = ServerCallContext()

# Invoke the remote agent
response = await remote_a2a_agent.on_message_send(request=request_params, context=context)

In [ ]python · cell 43

python

# The response contains a StreamResponse containing a Task object
task_object = None
for chunk in response:
    if hasattr(chunk, "task") and chunk.task.id:
        task_object = chunk.task
        break

if task_object:
    task_id = task_object.id
    print(f"Task started: {task_id}")
    print(f"Status: {task_object.status.state}")
else:
    print("Could not retrieve the task object from the response.")

Get the response

Using the task_id from the previous step, we call on_get_task.

The SDK polls the Agent Engine endpoint and retrieves the completed task, including the final answer in the artifacts field. We have successfully communicated with our deployed A2A agent.

Note: Running this cell might require few seconds depending on the use case.

In [ ]python · cell 45

python

from a2a.types import GetTaskRequest, TaskState
from a2a.server.context import ServerCallContext

request_params = GetTaskRequest(id=task_id, history_length=1)
context = ServerCallContext()

result = None
retries = 0
max_retries = 30

while True:
    try:
        # Get the task result
        result = await remote_a2a_agent.on_get_task(request=request_params, context=context)

        if result.status.state in [TaskState.TASK_STATE_COMPLETED, TaskState.TASK_STATE_FAILED]:
            break

        print(f"Task state: {result.status.state}. Waiting 1s...")
        time.sleep(1)

    except Exception as e:
        error_str = str(e)
        if "400 Bad Request" in error_str:
            retries += 1
            if retries <= max_retries:
                print(f"Received HTTP 400. Retrying in 1s ({retries}/{max_retries})...")
                time.sleep(1)
                continue
            else:
                print("Max retries reached.")
                raise
        else:
            raise

In [ ]python · cell 46

python

# Artifacts contain the actual results
for artifact in result.artifacts:
    if artifact.parts:
        part = artifact.parts[0]
        if hasattr(part, "text") and part.text:
            display(Markdown(f"**Answer**:\n {part.text}"))
        else:
            print("Could not extract text from artifact parts.")

(Optional) Cancel a task

If your remote agent executes long run operations, you can always cancel the associated task as shown below:

python

# Remote agent via Agent Platform SDK
task_data = {
    "id": task_id,
}
response = await remote_a2a_agent.on_cancel_task(**task_data)
print(response)

Via A2A Client

Initialize A2A Client

Here, we set up the A2A SDK ClientFactory.

We start from the AgentCard we fetched from our deployed agent. This is crucial because the client needs the card (especially the url) to know where to send requests.

Finally, the factory.create(remote_a2a_agent_card) call gives us a client instance ready to communicate with our specific agent endpoint.

In [ ]python · cell 50

python

# Get authentication token for Google Cloud
bearer_token = get_bearer_token()
headers = {
    "Authorization": f"Bearer {bearer_token}",
    "Content-Type": "application/json",
}

# Configure the A2A client factory
# This handles the protocol implementation details
factory = ClientFactory(
    ClientConfig(
        supported_protocol_bindings=[TransportProtocol.HTTP_JSON],
        # Use client preferences for protocol negotiation
        use_client_preference=True,
        # Configure HTTP client with authentication
        httpx_client=httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {bearer_token}",
                "Content-Type": "application/json",
            },
            timeout=60.0
        ),
    )
)
# Temporary workaround for faulty URL returned by the server
correct_engine_id = remote_a2a_agent.api_resource.name.split("/")[-1]
for interface in remote_a2a_agent_card.supported_interfaces:
    if interface.url and "test-agent-engine" in interface.url:
        interface.url = interface.url.replace("test-agent-engine", correct_engine_id)
# end of the workaround
# Create a client for our specific agent
# The client uses the Agent Card to understand capabilities
a2a_client = factory.create(remote_a2a_agent_card)

Get the agent card

In [ ]python · cell 52

python

from a2a.types import GetExtendedAgentCardRequest
from a2a.server.context import ServerCallContext

response = await a2a_client.get_extended_agent_card(GetExtendedAgentCardRequest())

print(f"Agent: {response.name}")
print(f"Supported Interfaces: {response.supported_interfaces}")
print(f"Skills: {[s.description for s in response.skills]}")
print(f"Examples: {[s.examples for s in response.skills][0] if response.skills else 'No skills'}")

Send a message to start a task

In [ ]python · cell 54

python

from a2a.types import SendMessageRequest

# Send a message using A2A protocol objects
message = Message(
    message_id=f"message-{os.urandom(8).hex()}",
    role="ROLE_USER",
    parts=[Part(text="What is the weather in Paris today?")],
)

# In A2A 1.0, send_message expects SendMessageRequest
request_params = SendMessageRequest(message=message)

# Get response (async iterator yielding StreamResponse)
response = a2a_client.send_message(request_params)

task_id = None
task_object = None

# The response is an async generator
async for response_chunk in response:
    # Check if the chunk has the task field populated
    if response_chunk.HasField("task"):
        task_object = response_chunk.task
        task_id = task_object.id
        break

if task_id:
    print(f"Task started: {task_id}")
    print(f"Status: {task_object.status.state}")
else:
    print("Could not retrieve task_id from stream.")

Get the response

Note: Running this cell might require few seconds depending on the use case.

In [ ]python · cell 56

python

from a2a.types import GetTaskRequest

# Poll for completion
task_data = {
    "id": task_id,
    "history_length": 1,
}
response = None
retries = 0
max_retries = 30  # Set a reasonable maximum number of retries

while True:
    try:
        print(f"Attempting to get task {task_id} (Retry {retries}/{max_retries})...")
        response = await a2a_client.get_task(GetTaskRequest(id=task_id, history_length=1))

        # If we get a response, check the task state
        if response.status.state == TaskState.TASK_STATE_COMPLETED:
            print(f"Task {task_id} completed successfully.")
            break  # Exit loop if task is completed
        elif response.status.state == TaskState.TASK_STATE_FAILED:
            print(f"Task {task_id} failed with state: {response.status.state}.")
            break  # Exit loop if task is failed
        else:
            # If still in progress, wait and retry
            print(
                f"Task {task_id} is still in state: {response.status.state}. Waiting 1 second..."
            )
            # Wait for a second before checking again to avoid spamming the API.
            time.sleep(1)
            continue

    except Exception as e:
        error_str = str(e)
        if "400 Bad Request" in error_str:
            retries += 1
            if retries < max_retries:
                time.sleep(1)
                continue  # Retry
            else:
                print(
                    f"Max retries ({max_retries}) reached for HTTP 400 Bad Request for task {task_id}."
                )
                raise  # Re-raise if max retries reached
        else:
            print(f"An error occurred for task {task_id}: {e}")
            raise

# Check if it has artifacts only if the task completed successfully
if (
    response
    and response.status.state == TaskState.TASK_STATE_COMPLETED
    and hasattr(response, "artifacts")
    and response.artifacts
):
    for artifact in response.artifacts:
        if artifact.parts:
            part = artifact.parts[0]
            if hasattr(part, "text") and part.text:
                display(Markdown(f"**Result**\n: {part.text}"))
            else:
                print("Could not extract text from artifact parts.")
elif response and response.status.state == TaskState.TASK_STATE_FAILED:
    print(f"Task {task_id} failed. No artifacts to display.")
else:
    print("No artifacts found or task not completed successfully.")

Via http request

This is the most fundamental way to interact with our agent: making direct HTTP requests.

This approach is perfect for debugging with tools like curl or for integrating from languages that don't have a dedicated A2A or Agent Platform SDK.

Get the agent card

In [ ]python · cell 59

python

# Prepare authentication headers
headers = {
    "Authorization": f"Bearer {get_bearer_token()}",
    "Content-Type": "application/json",
    "A2A-Version": "1.0"
}

# Get the agent card endpoint from supported_interfaces in A2A 1.0
remote_agent_card_url = None
for interface in remote_a2a_agent_card.supported_interfaces:
    if "HTTP" in interface.protocol_binding:
        remote_agent_card_url = interface.url
        # Remove trailing slash if present to avoid double slashes in endpoints
        if remote_agent_card_url.endswith("/"):
             remote_agent_card_url = remote_agent_card_url[:-1]
        break

if not remote_agent_card_url:
    raise ValueError("Could not find HTTP interface in remote agent card.")

remote_agent_card_endpoint = f"{remote_agent_card_url}/extendedAgentCard"
print(f"Fetching card from: {remote_agent_card_endpoint}")

try:
    # Send the HTTP request
    response = httpx.get(remote_agent_card_endpoint, headers=headers)
    response.raise_for_status()
    # Parse the response
    result = response.json()
    print(json.dumps(result, indent=2))
except httpx.HTTPStatusError as e:
    print(f"HTTP error occurred: {e}")
    print(f"Response body: {e.response.text}")
except httpx.RequestError as e:
    print(f"An error occurred while trying to send the request: {e}")

Send a message to start a task

Note: Running this cell might require few seconds depending on the use case.

In [ ]python · cell 61

python

# Construct the A2A message payload for A2A 1.0
payload = {
    "message": {
        "messageId": f"msg-{os.urandom(8).hex()}",
        "role": "ROLE_USER",
        "parts": [{"text": "Who is the current UN Secretary-General?"}],
    },
    "metadata": {
        # Optional metadata for tracking/debugging
        "source": "tutorial",
        "timestamp": datetime.now().isoformat(),
        "user_agent": "test_script",
    },
}

try:
    # Send the HTTP request
    response = httpx.post(
        f"{remote_agent_card_url}/message:send", json=payload, headers=headers
    )
    response.raise_for_status()
    # Parse the response
    result = response.json()
    print(json.dumps(result, indent=2))
except httpx.HTTPStatusError as e:
    print(f"HTTP error occurred: {e}")
    print(f"Response body: {e.response.text}")
except httpx.RequestError as e:
    print(f"An error occurred while trying to send the request: {e}")

# The response contains a task
# Protobuf JSON might return 'id' directly or nested depending on how SendMessageResponse is structured
# Let's try to extract it safely
task_id = result.get("id") or result.get("task", {}).get("id")
task_status = result.get("status", {}).get("state") or result.get("task", {}).get("status", {}).get("state")

print(f"Task started: {task_id}")
print(f"Status: {task_status}")

Get the response

In [ ]python · cell 63

python

# Poll for completion
task_url = f"{remote_agent_card_url}/tasks/{task_id}"
print(f"Polling for results at: {task_url}")

task_data = {}
state = None
retries = 0
max_retries = 30

while True:
    try:
        # Poll the task endpoint until it reaches a terminal state.
        response = httpx.get(task_url, headers=headers, params={"historyLength": 1})
        response.raise_for_status()
        task_data = response.json()
        state = task_data["status"]["state"]

        if state in ["TASK_STATE_COMPLETED", "TASK_STATE_FAILED"]:
            print(f"Task finished with state: {state}")
            break

        # Wait for a second before checking again to avoid spamming the API.
        time.sleep(1)

    except httpx.HTTPStatusError as e:
        if e.response.status_code == 400:
            retries += 1
            if retries <= max_retries:
                print(
                    f"Received HTTP 400 Bad Request. Retrying in 1s ({retries}/{max_retries})..."
                )
                time.sleep(1)
                continue
            else:
                print("Max retries reached.")
                raise
        else:
            raise

# Extract the result
if state == "TASK_STATE_COMPLETED":
    artifacts = task_data.get("artifacts", [])
    for artifact in artifacts:
        for part in artifact["parts"]:
            if "text" in part:
                display(Markdown(f"**Result**\n: {part['text']}"))
                break

Cleaning up

Time to clean up. Run the cell below prevents you from incurring ongoing costs for the services you've provisioned during this tutorial.

In [ ]python · cell 65

python

delete_agent_engine = True

if delete_agent_engine:
    remote_a2a_agent.delete(force=True)