How A2A Enables Agentic AI Collaboration at Scale
Now, To understand the concept better, let’s start with a simple example.
Consider various departments within a large travel organization that have developed Agentic AI using different multi-agent frameworks.
YESTERDAY: Phase 1 — Siloed Agentic Apps:
Flight Operations has developed an Agentic application using the Google Agent Development Kit (ADK) to handle flight-related queries.
#agent.pyimport os
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.tools import FunctionTool
from google.adk.tools.agent_tool import AgentTool
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext
from typing import AsyncGenerator
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "")
'''def external_approval_tool(content):
print("**** waiting for approval *****", content)
return "rejected"'''
def flight_advisor_agent():
#approval_tool = FunctionTool(func=external_approval_tool)
# Agent that prepares the request
"""validate_agent = LlmAgent(
name="validate_agent",
model=LiteLlm("openai/gpt-4o-mini"),
description="Help to validate whether user provided all information",
instruction='''Ask user if either origin or destination is missing''',
disallow_transfer_to_parent=True,
disallow_transfer_to_peers=True,
output_key="validate",
)"""
extract_agent = LlmAgent(
name="extract_agent",
model=LiteLlm("openai/gpt-4o-mini"),
description="Help to infer origin, destination and class from user query",
instruction='''Infer origin, destination and class details from user query.
Return the response as a JSON object formatted like this:
{
origin:
destination:
class:
}''',
disallow_transfer_to_parent=True,
disallow_transfer_to_peers=True,
output_key="request",
)
search_agent = LlmAgent(
name="search_agent",
model=LiteLlm("openai/gpt-4o-mini"),
description="Help users find best flight deals",
instruction='''Read state ['request'] and generate Best 3 search results for given resutls.
Return the response as a JSON object formatted like this:
{{
{{"flights": [
{
"flight_number":"Unique identifier for the flight, like BA123, AA31, etc."),
"departure": {{
"city_name": "Name of the departure city",
"timestamp": ("ISO 8601 departure date and time"),
}},
"arrival": {{
"city_name":"Name of the arrival city",
"timestamp": "ISO 8601 arrival date and time",
}},
"airlines": [
"Airline names, e.g., American Airlines, Emirates"
],
"price_in_usd": "Integer - Flight price in US dollars",
"number_of_stops": "Integer - indicating the number of stops during the flight",
}
]}}
}}''',
disallow_transfer_to_parent=True,
disallow_transfer_to_peers=True,
output_key="flight",
)
flight_agent = Agent(
model=LiteLlm("openai/gpt-4o-mini"),
description="""Helps users with flight details""",
name="flight_agent",
instruction='''Choose the one of the BEST flight option from state ['flight']
and "provide flights info" to user.
Note:Don't send all options to user.''',
tools=[
#AgentTool(agent=validate_agent),
AgentTool(agent=extract_agent),
AgentTool(agent=search_agent),
],
)
return flight_agent
This code defines a flight advisory agent that consists of three main components: a multi-agent system, a FastAPI backend, and a Streamlit-based chat user interface (UI). At the core is the `flight_advisor_agent`, which extracts flight-related details such as the origin, destination, and class from user queries before searching for and selecting the best flight options.
The agents operate collaboratively under a parent agent called `flight_agent`, utilizing GPT-4o-mini models, each assigned specific responsibilities and output formats. The `main.py` script sets up a FastAPI server that exposes the agent as a JSON-RPC endpoint. When a user submits a query, the server delegates the task to the agent and formats the results into a response object.
Finally, the `ui/app.py` file provides a chat interface using Streamlit, allowing users to interact with the application by capturing input, invoking the backend endpoint, and displaying the flight options returned by the agents.
For more information, please visit the GitHub page below.
Hotel Operations has created an Agentic application using the AutoGen framework for lodging recommendations.
# agents.py
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient# ensure your OPENAI_API_KEY is set in the environment
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "")
def stay_advisor_agent():
"""
Instantiate and return the list of AssistantAgents
in the order they should speak.
"""
client = OpenAIChatCompletionClient(model="gpt-4o-mini")
validation_agent = AssistantAgent(
"validation_agent",
model_client=client,
description="Validate incoming request",
system_message='''You are tasked to validate user request
to ensure whether user have provided desination details,
Otherwise you must say "TERMINATE" with reason of termination. '''
)
extract_agent = AssistantAgent(
"extract_agent",
model_client=client,
description="Help to infer destination from user query",
system_message='''Infer destination from user query.
Return the response as a JSON object formatted like this:
{
destination:
}
Don't make assumption'''
)
search_agent = AssistantAgent(
"search_agent",
model_client=client,
description="Help users find best hotel at destination",
system_message='''Generate TOP 3 hotel stay details only.
Return the response as a JSON object formatted like this:
{{
{{"hotels": [
{
"hotel": "hotel names, "
"price_in_usd": "US dollars",
"address": "location of the hotel",
}
]}}
}}'''
)
stay_summary_agent = AssistantAgent(
"stay_summary_agent",
model_client=client,
description="Compile final stay plan",
system_message=(
"You are an assistant that aggregates all advice into a final plan. "
"When done, TERMINATE it. Don't respond terminate wording at end"
)
)
return [extract_agent, search_agent, stay_summary_agent]
This codebase implements an AI-powered hotel accommodation agent system using a modular, multi-agent architecture. The `agents.py` file defines four distinct assistant agents (built with the AutoGen framework and OpenAI GPT-4o-mini): a validation agent to verify that the user input includes a destination, an extraction agent to infer the destination from the query, a search agent to find the top three hotel options in that destination, and a summary agent to compile a final stay plan. These agents collaborate in sequence to handle user queries related to travel accommodations.
The backend is implemented using FastAPI (`api.py`), which exposes a JSON-RPC endpoint to accept user queries and return structured hotel suggestions. The endpoint delegates logic to a service layer (the `provide_stay_plan` function, assumed to coordinate the agent flow).
A simple Streamlit frontend (`ui/app.py`) offers an interactive chat UI where users can enter accommodation-related queries. The UI sends requests to the FastAPI backend and renders the multi-agent responses as a conversational history, creating a cohesive end-to-end user experience for hotel planning.
For more information, please visit the GitHub page below.
So What’s the User experience:
Users should interact with two separate agentic applications to make a travel plan.
TODAY: Phase 2
Based on UX feedback, the organization explores options:
Option 1 — Rewrite Everything in ONE Framework
Migrate all agentic apps to a single technology stack.
- Pros: Unified architecture
- Cons: High cost, disruption, and loss of existing investments
Option 2 — Enable Cross-Agent Collaboration using A2A
Instead of a full rewrite, the organization adopts the Agent-to-Agent (A2A) protocol to enable seamless interoperability across frameworks.
The Travel team builds a LangGraph-based Travel Agentic AI App.
It collaborates via A2A with:
- Flight Agentic AI App (built on Google’s ADK) to fetch real-time flight data.
- Hotel Agentic AI App (built using Microsoft’s AutoGen) to retrieve hotel availability.
This enables modular, scalable collaboration — without re-architecting the entire stack.
For this sample, I have chosen Option 2.
# agent.pyfrom typing import List
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, AIMessage, HumanMessage, ToolMessage, BaseMessage
import httpx
import json
from langgraph.checkpoint.memory import MemorySaver
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END
from typing import Annotated
model_name = "gpt-4o-mini"
User_Prompt = """
You are a smart travel planning assistant.
**Step 1: Understand Intent**
Ask this clearly at the start if the intent is not obvious.
---
**Step 2: Collect Required Info Based on Intent**
**For Flights:**
**For Hotels:**
**For Full Travel Plan:**
---
**Smart Behavior:**
---
**Interaction Flow:**
1. Ask what kind of travel help the user needs (if not already clear)
2. Collect required inputs based on that intent
3. Confirm collected data with a summary
4. Proceed to show flight/hotel options or generate full travel plan
Be concise, friendly, and avoid unnecessary questions.
"""
def get_user_info(messages):
if not isinstance(messages, list):
messages = [messages]
return [SystemMessage(content=User_Prompt)] + messages
class TravelInfo(BaseModel):
"""Instructions on how to prompt the LLM."""
origin: str
destination: str
travel_date: str
travel_end: str
budget: str
#llm = ChatOpenAI(model=model_name, temperature=0, max_retries=1)
def travel_agent(state):
print("travel_agent")
messages = get_user_info(state["messages"])
print("travel_agent 1 \n", messages)
print("travel_agent -1 \n", messages[-1])
print("travel_agent 0 \n", messages[0])
llm = ChatOpenAI(model=model_name, temperature=0, max_retries=1)
try:
print("travel_agent 2")
llm_with_tool = llm.bind_tools([TravelInfo])
print("travel_agent 3")
response = llm_with_tool.invoke(messages)
print("travel_agent : ", response)
return {
"messages": [response]
}
except Exception as e:
print(f"{e}")
return e
def conclude_conversation(state):
response = {
"messages": [
ToolMessage(
content="Clarified and proceeding further",
tool_call_id=state["messages"][-1].tool_calls[0]["id"],
)
]
}
print("conclude_conversation : ", response)
return response
def is_clarified(state):
messages = state["messages"]
if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
return "yes" #"conclude_conversation"
else:
return "no" #"continue_conversation"
Intent_Prompt = "Generate user intent based on {reqs}"
def get_user_messages(messages: list):
tool_call = None
other_msgs = []
for m in messages:
if isinstance(m, AIMessage) and m.tool_calls:
tool_call = m.tool_calls[0]["args"]
if hasattr(m, 'tool_calls'):
delattr(m, 'tool_calls')
elif isinstance(m, ToolMessage):
continue
elif tool_call is not None:
other_msgs.append(m)
return[
SystemMessage(content=Intent_Prompt.format(reqs=tool_call)),
] + other_msgs
def generate_user_intent(state):
messages = get_user_messages(state["messages"])
llm = ChatOpenAI(model=model_name, temperature=0, max_retries=1)
response = llm.invoke(messages)
print("generate_user_intent : ", response)
return {
"user_intent": response
}
def agent_cards():
urls = [
"http://localhost:9001/.well-known/agent.json",
"http://localhost:9002/.well-known/agent.json"
]
responses = []
for url in urls:
try:
resp = httpx.get(url, timeout=5.0)
if resp.status_code == 200:
data = resp.json()
responses.append(data)
else:
print(f"Failed to fetch from {url}, status code: {resp.status_code}")
except Exception as e:
print(f"Error calling {url}: {e}")
#print("All responses:", responses)
return responses
Classify_Prompt = """Help to choose right agent from {agent_list} for given user query.
- Repond with Flights agent array from agent list if user is enquiring about flight relation information
- Repond with from agent list if user is enquiring about stay/hotels related information
- Repond with from agent list if user is enquiring about travel or flight and stay related
Respond json object only.
Don't make wild guess"""
def get_agent_info(messages):
agents = agent_cards()
return [SystemMessage(content=Classify_Prompt.format(agent_list=agents))] + messages
def discovery_agent(state):
print("discovery_agent")
messages = get_agent_info(state["user_intent"])
print("discovery_agent :", messages)
print("discovery_agent - user_intent:", state["user_intent"])
llm = ChatOpenAI(model=model_name, temperature=0, max_retries=1)
response = llm.invoke(messages)
print("discovery_agent : ", response)
return {"agent_card": response}
async def call_remote_agent(agent, user_query):
payload = {
"jsonrpc": "2.0",
"id": "101",
"method": "tasks/send",
"params": {
"id": "1011",
"message": {
"role": "user",
"parts": [
{
"type": "text",
"text": user_query
}
]
}
}
}
async with httpx.AsyncClient(timeout=120) as client:
try:
response = await client.post(agent, json=payload)
print("Agent response :", response.json())
return response.json()
except Exception as e:
print(e)
return e
async def collection_agent(state):
print("collection_agent")
agent_card = state['agent_card'][-1].content
user_query = state['user_intent'][-1].content
print("collection_agent - agent card : ", agent_card)
print("collection_agent - user_query : ", user_query)
if not agent_card.strip():
print("Error: agent_card is empty.")
return
else:
print(agent_card)
content = agent_card.strip()
# Remove triple backtick formatting if present
if content.startswith("```json"):
content = content[7:] # remove ```json
if content.endswith("```"):
content = content[:-3] # remove `
agents = json.loads(content)
print("Agents :", agents)
# Access elements
print("\n*** Discovered Agents ***\n")
agent_responses = []
for agent in agents:
print("URL:", agent["url"])
print("User Intent :", user_query)
agent_response = await call_remote_agent (agent["url"], user_query)
agent_responses.append ({
"name" : agent["name"],
"response" : agent_response
})
print("agent_responses: ", agent_responses)
return {
"messages": [
{
"role": "assistant",
"content": f"{json.dumps(agent_responses, indent=2)}"
}
]
}
Aggregate_Prompt = "Generate Travel Plan for provided {details}"
def get_aggregate_info(messages):
return [SystemMessage(content=Aggregate_Prompt.format(details=messages))]
def aggregate_agent(state):
agent_responses = state['messages'][-1].content
messages = get_aggregate_info(agent_responses)
llm = ChatOpenAI(model=model_name, temperature=0, max_retries=1)
response = llm.invoke(messages)
print("Final Plan :", response)
return {"messages": response}
class State(TypedDict):
messages: Annotated[list, add_messages]
initial_query: Annotated[list, add_messages]
user_intent: Annotated[list, add_messages]
agent_card: Annotated[list, add_messages]
max_iteration: int
iteration: int
def build_aggregator():
memory = MemorySaver()
workflow = StateGraph(State)
workflow.add_edge(START, "Planner")
workflow.add_node("Planner", travel_agent)
workflow.add_node("generate_user_intent", generate_user_intent)
workflow.add_node("Discoverer", discovery_agent)
workflow.add_node("Collector", collection_agent)
workflow.add_node("Aggregator", aggregate_agent)
workflow.add_node("conclude_conversation", conclude_conversation)
workflow.add_conditional_edges(
"Planner",
is_clarified,
{"yes": "conclude_conversation", "no": END})
workflow.add_edge("conclude_conversation", "generate_user_intent")
workflow.add_edge("generate_user_intent", "Discoverer")
workflow.add_edge("Discoverer", "Collector")
workflow.add_edge("Collector", "Aggregator")
workflow.add_edge("Aggregator", END)
graph = workflow.compile(checkpointer=memory)
return graph
This code implements a modular, agent-based travel planning application using LangGraph, OpenAI’s GPT-4o-mini, and FastAPI. The application is designed to interact with users, infer their travel-related intent (e.g., flights, hotels, or full travel plans), and coordinate with specialized external agents/agentic apps(via JSON-RPC over HTTP) to gather relevant travel information.
The conversation flow begins with a planning agent that prompts the user to clarify their request. If the user’s intent isn’t fully understood, the application elicits any missing details and proceeds only after receiving user confirmation.
Once the intent is confirmed, a user intent generation module interprets the query, and a discovery agent identifies whether it requires interaction with Hotel agentic app, Flight agentic app, or both to fulfil user queries. These agent cards are fetched from known endpoints. Then, a collection agent asynchronously queries the selected agents and gathers their responses. Finally, an aggregator agent synthesizes these results into a complete travel plan.
For more information, please visit the GitHub page below.
So What’s the User experience:
The user interacts with the Travel Agentic AI app, which in turn communicates with the Flight Agentic AI App, Hotel Agentic AI App, or both — depending on the user’s query.
Evolving to Agentic AI App Collaboration: What Improves in Phase 2
As Agentic AI apps gain traction, the architecture naturally shifts to support more dynamic, scalable collaboration. Phase 2 reflects this evolution, focusing on key enterprise priorities:
About The Author
