Python A2A is an implementation of Google’s Agent-to-Agent (A2A) protocol, which permits AI brokers to speak with one another utilizing a shared, standardized format—eliminating the necessity for customized integration between companies.
On this tutorial, we’ll use the decorator-based method offered by the python-a2a library. With easy @agent and @talent decorators, you may outline your agent’s id and conduct, whereas the library takes care of protocol dealing with and message movement.
This technique is ideal for rapidly constructing helpful, task-focused brokers with out worrying about low-level communication logic.
Putting in the dependencies
To get began, you’ll want to put in the python-a2a library, which gives a clear abstraction to construct and run brokers that observe the A2A protocol.
Open your terminal and run:
Creating the Brokers
For this tutorial, we can be creating two brokers – one for calculating inventory returns primarily based on funding, charge, and time, and one other for adjusting an quantity primarily based on inflation over a interval of years.
EMI Agent (emi_agent.py)
from python_a2a import A2AServer, talent, agent, run_server, TaskStatus, TaskState
import re
@agent(
title="EMI Calculator Agent",
description="Calculates EMI for a given principal, rate of interest, and mortgage period",
model="1.0.0"
)
class EMIAgent(A2AServer):
@talent(
title="Calculate EMI",
description="Calculates EMI given principal, annual rate of interest, and period in months",
tags=["emi", "loan", "interest"]
)
def calculate_emi(self, principal: float, annual_rate: float, months: int) -> str:
monthly_rate = annual_rate / (12 * 100)
emi = (principal * monthly_rate * ((1 + monthly_rate) ** months)) / (((1 + monthly_rate) ** months) - 1)
return f"The EMI for a mortgage of ₹{principal:.0f} at {annual_rate:.2f}% curiosity for {months} months is ₹{emi:.2f}"
def handle_task(self, activity):
input_text = activity.message["content"]["text"]
# Extract values from pure language
principal_match = re.search(r"₹?(d{4,10})", input_text)
rate_match = re.search(r"(d+(.d+)?)s*%", input_text)
months_match = re.search(r"(d+)s*(months|month)", input_text, re.IGNORECASE)
strive:
principal = float(principal_match.group(1)) if principal_match else 100000
charge = float(rate_match.group(1)) if rate_match else 10.0
months = int(months_match.group(1)) if months_match else 12
print(f"Inputs → Principal: {principal}, Price: {charge}, Months: {months}")
emi_text = self.calculate_emi(principal, charge, months)
besides Exception as e:
emi_text = f"Sorry, I could not parse your enter. Error: {e}"
activity.artifacts = [{
"parts": [{"type": "text", "text": emi_text}]
}]
activity.standing = TaskStatus(state=TaskState.COMPLETED)
return activity
# Run the server
if __name__ == "__main__":
agent = EMIAgent()
run_server(agent, port=4737)This EMI Calculator Agent is constructed utilizing the python-a2a library and follows the decorator-based method. On the high, we use the @agent decorator to outline the agent’s title, description, and model. This registers the agent in order that it might probably talk utilizing the A2A protocol.
Inside the category, we outline a single talent utilizing the @skill decorator. This talent, known as calculate_emi, performs the precise EMI calculation utilizing the usual components. The components takes in three parameters: the mortgage principal, the annual rate of interest, and the mortgage period in months. We convert the annual charge right into a month-to-month charge and use it to compute the month-to-month EMI.
The handle_task technique is the core of the agent. It receives the person’s enter message, extracts related numbers utilizing easy common expressions, and passes them to the calculate_emi technique.
Lastly, on the backside of the file, we launch the agent utilizing the run_server() operate on port 4737, making it able to obtain A2A protocol messages. This design retains the agent easy, modular, and straightforward to increase with extra expertise sooner or later.
Inflation Agent (inflation_agent.py)
from python_a2a import A2AServer, talent, agent, run_server, TaskStatus, TaskState
import re
@agent(
title="Inflation Adjusted Quantity Agent",
description="Calculates the longer term worth adjusted for inflation",
model="1.0.0"
)
class InflationAgent(A2AServer):
@talent(
title="Inflation Adjustment",
description="Adjusts an quantity for inflation over time",
tags=["inflation", "adjustment", "future value"]
)
def handle_input(self, textual content: str) -> str:
strive:
# Extract quantity
amount_match = re.search(r"₹?(d{3,10})", textual content)
quantity = float(amount_match.group(1)) if amount_match else None
# Extract charge (e.g. 6%, 7.5 %)
rate_match = re.search(r"(d+(.d+)?)s*(%|%)", textual content, re.IGNORECASE)
charge = float(rate_match.group(1)) if rate_match else None
# Extract years (e.g. 5 years)
years_match = re.search(r"(d+)s*(years|yr)", textual content, re.IGNORECASE)
years = int(years_match.group(1)) if years_match else None
if quantity isn't None and charge isn't None and years isn't None:
adjusted = quantity * ((1 + charge / 100) ** years)
return f"₹{quantity:.2f} adjusted for {charge:.2f}% inflation over {years} years is ₹{adjusted:.2f}"
return (
"Please present quantity, inflation charge (e.g. 6%) and period (e.g. 5 years).n"
"Instance: 'What's ₹10000 value after 5 years at 6% inflation?'"
)
besides Exception as e:
return f"Sorry, I could not compute that. Error: {e}"
def handle_task(self, activity):
textual content = activity.message["content"]["text"]
outcome = self.handle_input(textual content)
activity.artifacts = [{
"parts": [{"type": "text", "text": result}]
}]
activity.standing = TaskStatus(state=TaskState.COMPLETED)
return activity
if __name__ == "__main__":
agent = InflationAgent()
run_server(agent, port=4747)This agent helps calculate how a lot a given quantity can be value sooner or later after adjusting for inflation. It makes use of the identical decorator-based construction offered by the python-a2a library. The @agent decorator defines the metadata for this agent, and the @talent decorator registers the principle logic underneath the title “Inflation Adjustment.”
The handle_input technique is the place the principle processing occurs. It extracts the quantity, inflation charge, and variety of years from the person’s enter utilizing easy common expressions. If all three values are current, it makes use of the usual future worth components to calculate the inflation-adjusted quantity:
Adjusted Worth = quantity × (1 + charge/100) ^ years.
If any worth is lacking, the agent returns a useful immediate telling the person what to supply, together with an instance. The handle_task operate connects every little thing by taking the person’s message, passing it to the talent operate, and returning the formatted outcome again to the person.
Lastly, the agent is launched utilizing run_server() on port 4747, making it able to deal with A2A queries.
Creating the Agent Community
Firstly run each the brokers in two separate terminals
python inflation_agent.pyEvery of those brokers exposes a REST API endpoint (e.g. http://localhost:4737 for EMI, http://localhost:4747 for Inflation) utilizing the A2A protocol. They hear for incoming duties (like “calculate EMI for ₹2,00,000…”) and reply with textual content solutions.
Now, we’ll add these two brokers to our community
from python_a2a import AgentNetwork, A2AClient, AIAgentRouter
# Create an agent community
community = AgentNetwork(title="Economics Calculator")
# Add brokers to the community
community.add("EMI", "http://localhost:4737")
community.add("Inflation", "http://localhost:4747")Subsequent we’ll create a router to intelligently direct queries to the perfect agent. This can be a core utility of the A2A protocol—it defines an ordinary activity format so brokers could be queried uniformly, and routers could make clever routing choices utilizing LLMs.
router = AIAgentRouter(
llm_client=A2AClient("http://localhost:5000/openai"), # LLM for making routing choices
agent_network=community
)Lastly, we’ll question the brokers
question = "Calculate EMI for ₹200000 at 5% curiosity over 18 months."
agent_name, confidence = router.route_query(question)
print(f"Routing to {agent_name} with {confidence:.2f} confidence")
# Get the chosen agent and ask the query
agent = community.get_agent(agent_name)
response = agent.ask(question)
print(f"Response: {response}")
question = "What's ₹1500000 value if inflation is 9% for 10 years?"
agent_name, confidence = router.route_query(question)
print(f"Routing to {agent_name} with {confidence:.2f} confidence")
# Get the chosen agent and ask the query
agent = community.get_agent(agent_name)
response = agent.ask(question)
print(f"Response: {response}")
Try the Notebooks- inflation_agent.py, community.ipynb and emi_agent.py. All credit score for this analysis goes to the researchers of this mission. Additionally, be at liberty to observe us on Twitter and don’t neglect to hitch our 100k+ ML SubReddit and Subscribe to our Publication.
I’m a Civil Engineering Graduate (2022) from Jamia Millia Islamia, New Delhi, and I’ve a eager curiosity in Information Science, particularly Neural Networks and their software in varied areas.
