As an agent developer, one of the key distinctions to make when implementing the Agent-to-Agent (A2A) protocol is when to use a Message versus a Task object. While both are fundamental to the A2A protocol, they serve distinct purposes. This guide will walk you through the nuances of each and provide a clear mental thought process for when to use them in your agent’s design.
At its core, the A2A protocol empowers agents to communicate and collaborate. Messages are the lifeblood of this communication. They are perfect for:
-
Discovery: When a user interacts with your agent, they might want to know its capabilities. This includes chit-chat where the user is trying to figure out what they want. A simple message exchange is the most natural way to handle this.
-
Quick Questions and Clarifications: If the client needs a simple piece of information to proceed which can be fetched quickly.
On the other hand, a Task object brings structure and clarity to more complex interactions. Tasks are preferable for:
-
Goal-Oriented Actions: When a user has a clear intent to accomplish something, a
Taskprovides a dedicated space to track & collaborate over that goal. -
Long-Running Operations: For processes that might take time, a
Taskallows for asynchronous tracking, so the user can check on the status without being blocked. -
Disambiguation: Tasks prevent confusion when multiple goals are being pursued in parallel within the same conversation.
Message-only approach
It is possible for an agent to just reply with Message objects. Here’s how a conversation might unfold using only messages:
User: What can you do?
Booking Agent: I can book flights, hotels, and car rentals.
User: I want to book a flight from NY to SF.
Booking Agent: Sure, I can help with that.
User: I also need to book a hotel room in Lake Tahoe.
Booking Agent: I am on it.
User: Book it for the 19th of this month.
Herein lies the ambiguity. Is the user referring to the flight or the hotel booking? The agent now has to infer the user’s intent. While the agent could ask for clarification, this creates a clunky user experience, especially if it happens repeatedly.
This ambiguity isn’t limited to the user’s messages. Consider the agent’s perspective:
Booking Agent: How many travelers to book for?
Is the agent asking about the flight or the hotel? The user is now the one who has to do the guesswork.
This problem is magnified as we increase the number of goals being parallely collaborated upon. For a parent agent orchestrating multiple decomposed-goals, this ambiguity can quickly become an issue.
The Task object to the rescue
This is where the Task object shines. By creating a dedicated Task for each goal, we introduce a clear and unambiguous way to collaborate. Each Task has a unique ID, so both the user and the agent can refer to a specific goal without confusion. All interaction pertaining to that user goal is within the context of that Task.
What if we only used Tasks?
One might be tempted to use Task objects for all interactions. For example, every response to a user could be a COMPLETED task (the full JSON structure is omitted for brevity).
User: Message(text: "What can you do?")
Booking Agent: Task(id: 456, status: COMPLETED, message: "I can book flights, hotels, etc.")
While this is technically possible, it introduces its own set of problems:
-
Task Proliferation: Your system will be flooded with COMPLETED task objects for even the most trivial exchanges.
-
Difficult History Review: When a user wants to review their past interactions, it becomes difficult to distinguish between simple chit-chat and meaningful goal-oriented tasks.
The best of both worlds: A hybrid approach
The most effective approach is to use both Messages and Tasks in a way that plays to their respective strengths.
-
Use Messages for:
-
Initial greetings and capability discovery.
-
To return responses for quick internal actions.
-
General conversation that doesn’t express a clear intent to perform an action.
-
-
Use Tasks when:
-
A user expresses a clear intent to achieve a goal (e.g., “Book a flight,” “Find a hotel”).
-
The goal completion can require the agent to collaborate over multiple turns.
-
The operation is long-running and requires asynchronous tracking.
-
Benefits of the hybrid approach
This approach offers several advantages:
-
Clarity and Disambiguation: Task IDs eliminate confusion when multiple goals are being pursued in parallel.
-
Simplified Client Design: Clients have a dedicated getTask API for fetching the state of a Task, simplifying the process of tracking progress.
-
Simpler Client UIs: Users can easily view all their active and completed tasks, providing a clear overview of their interactions with the agent. When an agent needs more information, that request can be displayed within the context of the relevant task and all associated previous messages, making it easy for the user to understand what’s being asked.
Let’s revisit our booking agent scenario, this time with Tasks (the full JSON structure is omitted for brevity):
User: Message(text: "I want to book a flight from NY to SF.")
Booking Agent: Task(id: "flight-123", status: "working", message: "Sure, I can help with that.")
User: Message(text: "I also need to book a hotel room in Lake Tahoe.")
Booking Agent: Task(id: "hotel-456", status: "working", message: "I am on it.")
User: Message(task_id: "hotel-456", text: "Book it for the 19th of this month.")
Booking Agent: Task(id: "flight-123", status: "input-required", message: "How many travellers?")
Notice how the task_id field in the Message object clearly indicates which task the user is referring to. Similarly, when the booking agent asks for more input, it specifies the task_id. The conversation is now unambiguous and easy to follow for both the user and the agent.
Advanced: Dynamically upgrade Message to a Task
As previously stated, messages can be used to return responses from fast internal tools or APIs. Like in the case of a Booking Agent, assuming it is determined that check-flight-status API is fast enough, user queries for flight status can be responded to as a Message.
Though it is not always the case that the internal API returns within the expected timeout of the user request. In such cases, the agent can internally keep a timer for a pre-defined timeout. If the internal API comes back with the response within that timer, then the response is piped back to the user as a Message.
Otherwise, a new Task is created in WORKING state. Once the internal API response is available, then the task is populated with the results and marked as COMPLETED.
Code example walkthrough
Link to the Colab notebook.
Setup :
Install dependencies
%pip install google-cloud-aiplatform httpx "a2a-sdk" --quiet
%pip install --upgrade --quiet "google-adk"
Imports & authenticate
import json
import uuid
import pprint
from collections.abc import AsyncIterable
from typing import Any, Optional
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.apps import A2AStarletteApplication
from a2a.server.events import EventQueue
from pydantic import BaseModel
from enum import Enum
from a2a.types import (
Part,
Task,
TaskState,
TextPart,
MessageSendParams,
Role,
Message,
)
from a2a.utils import (
new_agent_parts_message,
new_agent_text_message,
new_task,
)
from a2a.server.request_handlers.default_request_handler import (
DefaultRequestHandler,
)
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.utils.errors import MethodNotImplementedError
# Build agent with adk
from google.adk.events import Event
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.tool_context import ToolContext
from google.adk.agents.llm_agent import LlmAgent
# Evaluate agent
from google.cloud import aiplatform
from google.genai import types
pp = pprint.PrettyPrinter(indent=2, width=120)
from google.colab import auth
try:
auth.authenticate_user()
print('Colab user authenticated.')
except Exception as e:
print(
f'Not in a Colab environment or auth failed: {e}. Assuming local gcloud auth.'
)
import os
if not PROJECT_ID:
raise ValueError('Please set your PROJECT_ID.')
os.environ['GOOGLE_CLOUD_PROJECT'] = PROJECT_ID
os.environ['GOOGLE_CLOUD_LOCATION'] = LOCATION
os.environ['GOOGLE_GENAI_USE_VERTEXAI'] = 'True'
aiplatform.init(project=PROJECT_ID, location=LOCATION)
Router
The idea is to define a router agent, whose sole job is to look at user queries and figure out if it’s chit-chat vs. it maps to set of pre-defined actions. Let’s take the booking agent example.
Below is the definition of a router agent. It determines the next step and associated user message. Since this agent has access to user conversation, it can summarise that as input to the next step as well.
class RouterActionType(str, Enum):
NONE = "NONE"
BOOK_FLIGHT = "BOOK_FLIGHT"
BOOK_HOTEL = "BOOK_HOTEL"
class RouterOutput(BaseModel):
message: str
next_step: RouterActionType
next_step_input: Optional[str] = None
We defined the instructions for the ADK agent, and used ADK output schema configuration to restrict the output.
class RouterAgent:
"""An agent that determines whether to call internal API vs chit-chat"""
def __init__(self) -> None:
self._agent = self._build_agent()
self._user_id = 'remote_agent'
self._runner = Runner(
app_name=self._agent.name,
agent=self._agent,
session_service=InMemorySessionService(),
)
def _build_agent(self) -> LlmAgent:
"""Builds the LLM agent for the router agent."""
return LlmAgent(
model='gemini-2.5-flash',
name='router_agent',
output_schema=RouterOutput,
instruction="""
You are an agent who reponds to user queries on behalf of a booking company. The booking company can book flights, hotels & cars rentals.
Based on user query, you need to suggest next step. Follow below guidelines to choose below next step:
- BOOK_FLIGHT: If the user shows intent to book a flight.
- BOOK_HOTEL: If the user shows intent to book a hotel.
- Otherwise the next step is NONE.
Your reponses should be in JSON in below schema:
{{
"next_step": "NONE | BOOK_FLIGHT | BOOK_HOTEL",
"next_step_input": "Optional. Not needed in case of next step is NONE. Relevant info from user converstaion required for the selceted next step.",
"message": "A user visible message, based on the suggested next step. Assume the suggested next step would be auto executed."
}}
""",
)
The agent can be called through run() method and it returns the RouterOutput:
class RouterAgent:
...
async def run(self, query, session_id) -> RouterOutput:
session = await self._runner.session_service.get_session(
app_name=self._agent.name,
user_id=self._user_id,
session_id=session_id,
)
content = types.Content(
role='user', parts=[types.Part.from_text(text=query)]
)
if session is None:
session = await self._runner.session_service.create_session(
app_name=self._agent.name,
user_id=self._user_id,
state={},
session_id=session_id,
)
async for event in self._runner.run_async(
user_id=self._user_id, session_id=session.id, new_message=content
):
if event.is_final_response():
response = ''
if (
event.content
and event.content.parts
and event.content.parts[0].text
):
response = '\n'.join(
[p.text for p in event.content.parts if p.text]
)
return RouterOutput.model_validate_json(response)
raise Exception("Router failed")
A2A agent executor
Now we define an A2A executor which consumes this router to first detect the next step and run dummy book_flight & book_hotel actions. It creates Task objects only if the router suggests the next step to be BOOK_FLIGHT or BOOK_HOTEL.
class BookingAgentExecutor(AgentExecutor):
"""Booking AgentExecutor Example."""
def __init__(self) -> None:
self.router_agent = RouterAgent()
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
query = context.get_user_input()
task = context.current_task
router_output = await self.router_agent.run(query, str(uuid.uuid4()))
if router_output.next_step == RouterActionType.NONE:
await event_queue.enqueue_event(new_agent_text_message(router_output.message, context_id=context.context_id))
return
# Time to create a task.
if not task:
task = new_task(context.message)
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, task.context_id)
await updater.update_status(
TaskState.working,
new_agent_text_message(router_output.message, context_id=context.context_id)
)
booking_response = ''
if router_output.next_step == RouterActionType.BOOK_FLIGHT:
booking_response = await self.book_flight()
elif router_output.next_step == RouterActionType.BOOK_HOTEL:
booking_response = await self.book_hotel()
await updater.add_artifact(
[Part(root=TextPart(text=booking_response))], name='Booking ID'
)
await updater.complete()
async def book_flight(self) -> str:
return "PNR: FY1234"
async def book_hotel(self) -> str:
return "Hotel Reference No: H789"
Demo
For a quick demo, we are not setting up the entire A2A API server, just the request handler from A2A SDK and directly calling.
request_handler = DefaultRequestHandler(
agent_executor=BookingAgentExecutor(),
task_store=InMemoryTaskStore(),
)
import pprint
pp = pprint.PrettyPrinter(indent=4, width=120)
async def send_message(query: str = "hi"):
task_id = None
context_id = None
user_message = Message(
role=Role.user,
parts=[Part(root=TextPart(text=query))],
message_id=str(uuid.uuid4()),
task_id=task_id,
context_id=context_id,
)
params=MessageSendParams(
message=user_message
)
response_stream = request_handler.on_message_send_stream(params=params)
async for ev in response_stream:
pp.pprint(ev.model_dump(exclude_none=True))
For simple chit-chat, it returns a Message:
await send_message("hey, could you help book my trip")
{ 'contextId': 'c615c445-8369-4a9a-a1dd-3b2644794fca',
'kind': 'message',
'messageId': '7f217339-e889-451e-9ad5-448b0f67a628',
'parts': [{'kind': 'text', 'text': 'I can help with that. Are you looking to book a flight or a hotel?'}],
'role': <Role.agent: 'agent'>}
For booking a flight, it returns a Task with all streams of TaskStatusUpdateEvent objects.
await send_message("book a flight from NY to SF")
# Task Created
{ 'contextId': '06bc71c6-f694-444e-845f-51bba6959c3d',
'history': [ { 'contextId': '06bc71c6-f694-444e-845f-51bba6959c3d',
'kind': 'message',
'messageId': '6486aae5-4795-4654-9325-0f2c8085b159',
'parts': [{'kind': 'text', 'text': 'book a flight from NY to SF'}],
'role': <Role.user: 'user'>,
'taskId': '4683b10f-30fd-4c16-8def-a67fd7a37a5d'}],
'id': '4683b10f-30fd-4c16-8def-a67fd7a37a5d',
'kind': 'task',
'status': {'state': <TaskState.submitted: 'submitted'>}}
# Working
{ 'contextId': '06bc71c6-f694-444e-845f-51bba6959c3d',
'final': False,
'kind': 'status-update',
'status': { 'message': { 'contextId': '06bc71c6-f694-444e-845f-51bba6959c3d',
'kind': 'message',
'messageId': 'f2058f4c-2cb7-418c-b930-d9b2a73a8829',
'parts': [{'kind': 'text', 'text': 'OK. I am booking a flight from NY to SF.'}],
'role': <Role.agent: 'agent'>},
'state': <TaskState.working: 'working'>,
'timestamp': '2025-08-15T21:50:47.343116+00:00'},
'taskId': '4683b10f-30fd-4c16-8def-a67fd7a37a5d'}
# Artifact Created
{ 'artifact': { 'artifactId': 'acd7a390-95c4-44bd-a6fa-92f3e346306e',
'name': 'Booking ID',
'parts': [{'kind': 'text', 'text': 'PNR: FY1234'}]},
'contextId': '06bc71c6-f694-444e-845f-51bba6959c3d',
'kind': 'artifact-update',
'taskId': '4683b10f-30fd-4c16-8def-a67fd7a37a5d'}
# Task Finished
{ 'contextId': '06bc71c6-f694-444e-845f-51bba6959c3d',
'final': True,
'kind': 'status-update',
'status': {'state': <TaskState.completed: 'completed'>, 'timestamp': '2025-08-15T21:50:47.343253+00:00'},
'taskId': '4683b10f-30fd-4c16-8def-a67fd7a37a5d'}
