Python A2A is an implementation of Google’s Agent-to-Agent (A2A) protocol, which allows AI brokers to speak with one another utilizing a shared, standardized format—eliminating the necessity for customized integration between providers.
On this tutorial, we’ll use the decorator-based strategy offered by the python-a2a library. With easy @agent and @talent decorators, you’ll be able to outline your agent’s id and conduct, whereas the library takes care of protocol dealing with and message stream.
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 supplies a clear abstraction to construct and run brokers that comply with the A2A protocol.
Open your terminal and run:
Creating the Brokers
For this tutorial, we will likely be creating two brokers – one for calculating inventory returns based mostly on funding, fee, and time, and one other for adjusting an quantity based mostly 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(
identify="EMI Calculator Agent",
description="Calculates EMI for a given principal, rate of interest, and mortgage length",
model="1.0.0"
)
class EMIAgent(A2AServer):
@talent(
identify="Calculate EMI",
description="Calculates EMI given principal, annual rate of interest, and length in months",
tags=("emi", "mortgage", "curiosity")
)
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 material")("textual content")
# 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)
attempt:
principal = float(principal_match.group(1)) if principal_match else 100000
fee = 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: {fee}, Months: {months}")
emi_text = self.calculate_emi(principal, fee, months)
besides Exception as e:
emi_text = f"Sorry, I could not parse your enter. Error: {e}"
activity.artifacts = ({
"elements": ({"kind": "textual content", "textual content": 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 strategy. On the high, we use the @agent decorator to outline the agent’s identify, description, and model. This registers the agent in order that it might talk utilizing the A2A protocol.
Inside the category, we outline a single talent utilizing the .kill decorator. This talent, referred to as calculate_emiperforms the precise EMI calculation utilizing the usual formulation. The formulation takes in three parameters: the mortgage principal, the annual rate of interest, and the mortgage length in months. We convert the annual fee right into a month-to-month fee 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() perform on port 4737making 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(
identify="Inflation Adjusted Quantity Agent",
description="Calculates the longer term worth adjusted for inflation",
model="1.0.0"
)
class InflationAgent(A2AServer):
@talent(
identify="Inflation Adjustment",
description="Adjusts an quantity for inflation over time",
tags=("inflation", "adjustment", "future worth")
)
def handle_input(self, textual content: str) -> str:
attempt:
# Extract quantity
amount_match = re.search(r"₹?(d{3,10})", textual content)
quantity = float(amount_match.group(1)) if amount_match else None
# Extract fee (e.g. 6%, 7.5 p.c)
rate_match = re.search(r"(d+(.d+)?)s*(%|p.c)", textual content, re.IGNORECASE)
fee = float(rate_match.group(1)) if rate_match else None
# Extract years (e.g. 5 years)
years_match = re.search(r"(d+)s*(years|12 months)", textual content, re.IGNORECASE)
years = int(years_match.group(1)) if years_match else None
if quantity is just not None and fee is just not None and years is just not None:
adjusted = quantity * ((1 + fee / 100) ** years)
return f"₹{quantity:.2f} adjusted for {fee:.2f}% inflation over {years} years is ₹{adjusted:.2f}"
return (
"Please present quantity, inflation fee (e.g. 6%) and length (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 material")("textual content")
end result = self.handle_input(textual content)
activity.artifacts = ({
"elements": ({"kind": "textual content", "textual content": end 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 could 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 identify “Inflation Adjustment.”
The handle_input technique is the place the principle processing occurs. It extracts the quantity, inflation fee, 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 formulation to calculate the inflation-adjusted quantity:
Adjusted Worth = quantity × (1 + fee/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 perform connects every part by taking the person’s message, passing it to the talent perform, and returning the formatted end result again to the person.
Lastly, the agent is launched utilizing run_server() on port 4747making 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(identify="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 one of the best agent. This can be a core utility of the A2A protocol—it defines a typical activity format so brokers may 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 venture. Additionally, be at liberty to comply with us on Twitter and don’t neglect to affix 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 Knowledge Science, particularly Neural Networks and their utility in numerous areas.
