10.3 A2A Protocol in Practice
A2A (Agent-to-Agent) is a protocol that supports direct communication and collaboration between agents.
10.3.1 Protocol Design Motivation
The MCP protocol solved the interaction between agents and tools, while the A2A protocol solves the collaboration problem between agents. In a task requiring multi-agent (such as researcher, writer, editor) collaboration, they need to communicate, delegate tasks, negotiate capabilities, and synchronize states.
Traditional central coordinator (star topology) solutions have three main problems:
- Single Point of Failure: Coordinator failure leads to overall system paralysis.
- Performance Bottleneck: All communication goes through the central node, limiting concurrency.
- Difficult to Scale: Adding or modifying agents requires changing central logic.
The A2A protocol adopts a peer-to-peer (P2P) architecture (mesh topology), allowing agents to communicate directly, fundamentally solving the above problems. Its core is the two abstract concepts of Task and Artifact, which is its biggest difference from MCP, as shown in Table 10.7.
Table 10.7 A2A Core Concepts
To implement management of the collaboration process, A2A defines a standardized lifecycle for tasks, including states such as creation, negotiation, delegation, in-progress, completion, and failure, as shown in Figure 10.7.
Figure 10.7 A2A Task Lifecycle
This mechanism enables agents to perform task negotiation, progress tracking, and exception handling.
The A2A request lifecycle is a sequence that details the four main steps a request follows: agent discovery, authentication, send message API, and send message stream API. Figure 10.8 below, borrowed from the official website's flowchart, shows the operational flow, illustrating the interaction between client, A2A server, and authentication server.
Figure 10.8 A2A Request Lifecycle
10.3.2 A2A Protocol in Practice
Most existing A2A implementations are Sample Code, and even Python implementations are quite cumbersome. Therefore, here we only adopt a method that simulates the protocol's ideas, implementing partial functionality through the A2A-SDK.
(2) Creating a Simple A2A Agent
Let's create an A2A agent, again using the calculator case as a demonstration:
python
from hello_agents.protocols.a2a.implementation import A2AServer, A2A_AVAILABLE
def create_calculator_agent():
"""Create a calculator agent"""
if not A2A_AVAILABLE:
print("❌ A2A SDK not installed, please run: pip install a2a-sdk")
return None
print("🧮 Creating calculator agent")
# Create A2A server
calculator = A2AServer(
name="calculator-agent",
description="Professional mathematical calculation agent",
version="1.0.0",
capabilities={
"math": ["addition", "subtraction", "multiplication", "division"],
"advanced": ["power", "sqrt", "factorial"]
}
)
# Add basic calculation skills
@calculator.skill("add")
def add_numbers(query: str) -> str:
"""Addition calculation"""
try:
# Simple parsing of "calculate 5 + 3" format
parts = query.replace("calculate", "").replace("plus", "+").replace("add", "+")
if "+" in parts:
numbers = [float(x.strip()) for x in parts.split("+")]
result = sum(numbers)
return f"Calculation result: {' + '.join(map(str, numbers))} = {result}"
else:
return "Please use format: calculate 5 + 3"
except Exception as e:
return f"Calculation error: {e}"
@calculator.skill("multiply")
def multiply_numbers(query: str) -> str:
"""Multiplication calculation"""
try:
parts = query.replace("calculate", "").replace("times", "*").replace("×", "*")
if "*" in parts:
numbers = [float(x.strip()) for x in parts.split("*")]
result = 1
for num in numbers:
result *= num
return f"Calculation result: {' × '.join(map(str, numbers))} = {result}"
else:
return "Please use format: calculate 5 * 3"
except Exception as e:
return f"Calculation error: {e}"
@calculator.skill("info")
def get_info(query: str) -> str:
"""Get agent information"""
return f"I am {calculator.name}, can perform basic mathematical calculations. Supported skills: {list(calculator.skills.keys())}"
print(f"✅ Calculator agent created successfully, supported skills: {list(calculator.skills.keys())}")
return calculator
# Create agent
calc_agent = create_calculator_agent()
if calc_agent:
# Test skills
print("\n🧪 Testing agent skills:")
test_queries = [
"Get information",
"Calculate 10 + 5",
"Calculate 6 * 7"
]
for query in test_queries:
if "information" in query.lower():
result = calc_agent.skills["info"](query)
elif "+" in query:
result = calc_agent.skills["add"](query)
elif "*" in query or "×" in query:
result = calc_agent.skills["multiply"](query)
else:
result = "Unknown query type"
print(f" 📝 Query: {query}")
print(f" 🤖 Reply: {result}")
print()(2) Custom A2A Agent
You can also create your own A2A agent, here's a simple demonstration:
python
from hello_agents.protocols.a2a.implementation import A2AServer, A2A_AVAILABLE
def create_custom_agent():
"""Create custom agent"""
if not A2A_AVAILABLE:
print("Please install A2A SDK first: pip install a2a-sdk")
return None
# Create agent
agent = A2AServer(
name="my-custom-agent",
description="My custom agent",
capabilities={"custom": ["skill1", "skill2"]}
)
# Add skills
@agent.skill("greet")
def greet_user(name: str) -> str:
"""Greet user"""
return f"Hello, {name}! I am a custom agent."
@agent.skill("calculate")
def simple_calculate(expression: str) -> str:
"""Simple calculation"""
try:
# Safe calculation (only supports basic operations)
allowed_chars = set('0123456789+-*/(). ')
if all(c in allowed_chars for c in expression):
result = eval(expression)
return f"Calculation result: {expression} = {result}"
else:
return "Error: Only basic mathematical operations supported"
except Exception as e:
return f"Calculation error: {e}"
return agent
# Create and test custom agent
custom_agent = create_custom_agent()
if custom_agent:
# Test skills
print("Testing greeting skill:")
result1 = custom_agent.skills["greet"]("Zhang San")
print(result1)
print("\nTesting calculation skill:")
result2 = custom_agent.skills["calculate"]("10 + 5 * 2")
print(result2)10.3.3 Using HelloAgents A2A Tools
HelloAgents provides a unified A2A tool interface.
(1) Creating A2A Agent Server
First, let's create an Agent server:
python
from hello_agents.protocols import A2AServer
import threading
import time
# Create researcher Agent service
researcher = A2AServer(
name="researcher",
description="Agent responsible for searching and analyzing materials",
version="1.0.0"
)
# Define skills
@researcher.skill("research")
def handle_research(text: str) -> str:
"""Handle research requests"""
import re
match = re.search(r'research\s+(.+)', text, re.IGNORECASE)
topic = match.group(1).strip() if match else text
# Actual research logic (simplified here)
result = {
"topic": topic,
"findings": f"Research results about {topic}...",
"sources": ["Source 1", "Source 2", "Source 3"]
}
return str(result)
# Start service in background
def start_server():
researcher.run(host="localhost", port=5000)
if __name__ == "__main__":
server_thread = threading.Thread(target=start_server, daemon=True)
server_thread.start()
print("✅ Researcher Agent service started at http://localhost:5000")
# Keep program running
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\nService stopped")(2) Creating A2A Agent Client
Now, let's create a client to communicate with the server:
python
from hello_agents.protocols import A2AClient
# Create client to connect to researcher Agent
client = A2AClient("http://localhost:5000")
# Send research request
response = client.execute_skill("research", "research AI applications in healthcare")
print(f"Received response: {response.get('result')}")
# Output:
# Received response: {'topic': 'AI applications in healthcare', 'findings': 'Research results about AI applications in healthcare...', 'sources': ['Source 1', 'Source 2', 'Source 3']}(3) Creating Agent Network
For collaboration among multiple Agents, we can connect multiple Agents to each other:
python
from hello_agents.protocols import A2AServer, A2AClient
import threading
import time
# 1. Create multiple Agent services
researcher = A2AServer(
name="researcher",
description="Researcher"
)
@researcher.skill("research")
def do_research(text: str) -> str:
import re
match = re.search(r'research\s+(.+)', text, re.IGNORECASE)
topic = match.group(1).strip() if match else text
return str({"topic": topic, "findings": f"Research results for {topic}"})
writer = A2AServer(
name="writer",
description="Writer"
)
@writer.skill("write")
def write_article(text: str) -> str:
import re
match = re.search(r'write\s+(.+)', text, re.IGNORECASE)
content = match.group(1).strip() if match else text
# Try to parse research data
try:
data = eval(content)
topic = data.get("topic", "Unknown topic")
findings = data.get("findings", "No research results")
except:
topic = "Unknown topic"
findings = content
return f"# {topic}\n\nBased on research: {findings}\n\nArticle content..."
editor = A2AServer(
name="editor",
description="Editor"
)
@editor.skill("edit")
def edit_article(text: str) -> str:
import re
match = re.search(r'edit\s+(.+)', text, re.IGNORECASE)
article = match.group(1).strip() if match else text
result = {
"article": article + "\n\n[Edited and optimized]",
"feedback": "Article quality is good",
"approved": True
}
return str(result)
# 2. Start all services
threading.Thread(target=lambda: researcher.run(port=5000), daemon=True).start()
threading.Thread(target=lambda: writer.run(port=5001), daemon=True).start()
threading.Thread(target=lambda: editor.run(port=5002), daemon=True).start()
time.sleep(2) # Wait for services to start
# 3. Create clients to connect to each Agent
researcher_client = A2AClient("http://localhost:5000")
writer_client = A2AClient("http://localhost:5001")
editor_client = A2AClient("http://localhost:5002")
# 4. Collaboration workflow
def create_content(topic):
# Step 1: Research
research = researcher_client.execute_skill("research", f"research {topic}")
research_data = research.get('result', '')
# Step 2: Write
article = writer_client.execute_skill("write", f"write {research_data}")
article_content = article.get('result', '')
# Step 3: Edit
final = editor_client.execute_skill("edit", f"edit {article_content}")
return final.get('result', '')
# Usage
result = create_content("AI applications in healthcare")
print(f"\nFinal result:\n{result}")10.3.4 Using A2A Tools in Agents
Now let's see how to integrate A2A into HelloAgents agents.
(1) Using A2ATool Wrapper
python
from hello_agents import SimpleAgent, HelloAgentsLLM
from hello_agents.tools import A2ATool
from dotenv import load_dotenv
load_dotenv()
llm = HelloAgentsLLM()
# Assume a researcher Agent service is already running at http://localhost:5000
# Create coordinator Agent
coordinator = SimpleAgent(name="Coordinator", llm=llm)
# Add A2A tool, connect to researcher Agent
researcher_tool = A2ATool(
name="researcher",
description="Researcher Agent, can search and analyze materials",
agent_url="http://localhost:5000"
)
coordinator.add_tool(researcher_tool)
# Coordinator can call researcher Agent
response = coordinator.run("Please have the researcher help me research AI applications in education")
print(response)(2) Practical Case: Intelligent Customer Service System
Let's build a complete intelligent customer service system with three Agents:
- Receptionist: Analyzes customer question types
- Technical Expert: Answers technical questions
- Sales Consultant: Answers sales questions
python
from hello_agents import SimpleAgent, HelloAgentsLLM
from hello_agents.tools import A2ATool
from hello_agents.protocols import A2AServer
import threading
import time
from dotenv import load_dotenv
load_dotenv()
llm = HelloAgentsLLM()
# 1. Create technical expert Agent service
tech_expert = A2AServer(
name="tech_expert",
description="Technical expert, answers technical questions"
)
@tech_expert.skill("answer")
def answer_tech_question(text: str) -> str:
import re
match = re.search(r'answer\s+(.+)', text, re.IGNORECASE)
question = match.group(1).strip() if match else text
# In actual applications, this would call LLM or knowledge base
return f"Technical answer: Regarding '{question}', I suggest you check our technical documentation..."
# 2. Create sales consultant Agent service
sales_advisor = A2AServer(
name="sales_advisor",
description="Sales consultant, answers sales questions"
)
@sales_advisor.skill("answer")
def answer_sales_question(text: str) -> str:
import re
match = re.search(r'answer\s+(.+)', text, re.IGNORECASE)
question = match.group(1).strip() if match else text
return f"Sales answer: Regarding '{question}', we have special offers..."
# 3. Start services
threading.Thread(target=lambda: tech_expert.run(port=6000), daemon=True).start()
threading.Thread(target=lambda: sales_advisor.run(port=6001), daemon=True).start()
time.sleep(2)
# 4. Create receptionist Agent (using HelloAgents' SimpleAgent)
receptionist = SimpleAgent(
name="Receptionist",
llm=llm,
system_prompt="""You are a customer service receptionist, responsible for:
1. Analyzing customer question types (technical questions or sales questions)
2. Forwarding questions to appropriate experts
3. Organizing expert answers and returning them to customers
Please remain polite and professional."""
)
# Add technical expert tool
tech_tool = A2ATool(
agent_url="http://localhost:6000",
name="tech_expert",
description="Technical expert, answers technical-related questions"
)
receptionist.add_tool(tech_tool)
# Add sales consultant tool
sales_tool = A2ATool(
agent_url="http://localhost:6001",
name="sales_advisor",
description="Sales consultant, answers price and purchase-related questions"
)
receptionist.add_tool(sales_tool)
# 5. Handle customer inquiries
def handle_customer_query(query):
print(f"\nCustomer inquiry: {query}")
print("=" * 50)
response = receptionist.run(query)
print(f"\nCustomer service reply: {response}")
print("=" * 50)
# Test different types of questions
if __name__ == "__main__":
handle_customer_query("How do I call your API?")
handle_customer_query("What is the price of the enterprise version?")
handle_customer_query("How do I integrate it into my Python project?")(3) Advanced Usage: Agent Negotiation
The A2A protocol also supports negotiation mechanisms between Agents:
python
from hello_agents.protocols import A2AServer, A2AClient
import threading
import time
# Create two Agents that need to negotiate
agent1 = A2AServer(
name="agent1",
description="Agent 1"
)
@agent1.skill("propose")
def handle_proposal(text: str) -> str:
"""Handle negotiation proposals"""
import re
# Parse proposal
match = re.search(r'propose\s+(.+)', text, re.IGNORECASE)
proposal_str = match.group(1).strip() if match else text
try:
proposal = eval(proposal_str)
task = proposal.get("task")
deadline = proposal.get("deadline")
# Evaluate proposal
if deadline >= 7: # Need at least 7 days
result = {"accepted": True, "message": "Proposal accepted"}
else:
result = {
"accepted": False,
"message": "Timeline too tight",
"counter_proposal": {"deadline": 7}
}
return str(result)
except:
return str({"accepted": False, "message": "Invalid proposal format"})
agent2 = A2AServer(
name="agent2",
description="Agent 2"
)
@agent2.skill("negotiate")
def negotiate_task(text: str) -> str:
"""Initiate negotiation"""
import re
# Parse task and deadline
match = re.search(r'negotiate\s+task:(.+?)\s+deadline:(\d+)', text, re.IGNORECASE)
if match:
task = match.group(1).strip()
deadline = int(match.group(2))
# Send proposal to agent1
proposal = {"task": task, "deadline": deadline}
return str({"status": "negotiating", "proposal": proposal})
else:
return str({"status": "error", "message": "Invalid negotiation request"})
# Start services
threading.Thread(target=lambda: agent1.run(port=7000), daemon=True).start()
threading.Thread(target=lambda: agent2.run(port=7001), daemon=True).start()