Python A2A is an implementation of the Agent-to-Agent (A2A) protocol from Google, permitting AI brokers to speak with one another utilizing shared, standardized codecs.
This tutorial makes use of a decorator-based method offered by the Python-A2A library. Easy @agent and @talent Whereas decorators can help you outline the id and conduct of brokers, libraries deal with protocol processing and message circulation.
This methodology is ideal for shortly constructing task-focused brokers with out worrying about low-level communication logic.
Putting in Dependencies
To get began, it’s worthwhile to set up the Python-A2A library. The Python-A2A library gives a clear abstraction for constructing and working brokers following the A2A protocol.
Open a terminal and run it.
Creating an Agent
On this tutorial, you’ll create two brokers. One is to calculate inventory returns based mostly on funding, charges and hours, and the opposite is to regulate quantities based mostly on inflation over the 12 months.
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", "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, process):
input_text = process.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}"
process.artifacts = [{
"parts": [{"type": "text", "text": emi_text}]
}]
process.standing = TaskStatus(state=TaskState.COMPLETED)
return process
# 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 a decorator-based method. On the high, use @agent A decorator that defines the identify, description, and model of the agent. This registers the agent to speak utilizing the A2A protocol.
Throughout the class, @skillDecorator. This talent known as Calculate_emiperforms the precise EMI calculation utilizing commonplace formulation. The method contains three parameters: mortgage principal, annual rate of interest, and mortgage time period. Convert your annual charge to month-to-month charges and use it to calculate your month-to-month EMI.
handle_task The strategy is the core of the agent. It receives the consumer’s enter message, extracts the related quantity utilizing a easy common expression, and passes it to the Calculate_EMI methodology.
Lastly, on the backside of the file, run_server() Options on ports 4737able to obtain A2A protocol messages. This design makes the agent easy and modular, with extra expertise attributable to future expertise.
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 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|12 months)", textual content, re.IGNORECASE)
years = int(years_match.group(1)) if years_match else None
if quantity shouldn't be None and charge shouldn't be None and years shouldn't be 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 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, process):
textual content = process.message["content"]["text"]
consequence = self.handle_input(textual content)
process.artifacts = [{
"parts": [{"type": "text", "text": result}]
}]
process.standing = TaskStatus(state=TaskState.COMPLETED)
return process
if __name__ == "__main__":
agent = InflationAgent()
run_server(agent, port=4747)This agent will enable you calculate how worthwhile a selected quantity might be sooner or later after adjusting for inflation. It makes use of the identical decorator-based construction offered by the Python-A2A library. @agent The decorator defines the metadata for this agent @talent The decorator registers the principle logic below the identify “inflation adjustment.”
handle_input The strategy is the place the principle processing takes place. Use a easy common expression to extract portions, inflation charges, and years from consumer enter. If all three values exist, use the usual future worth method to calculate the inflation adjustment.
Adjusted worth = Quantity x (1 + Reate/100) ^ 12 months.
If there is no such thing as a worth, the agent returns a helpful immediate, together with an instance, telling the consumer what to supply. handle_task The perform connects all the things by getting the consumer’s message, passing it to the talent perform, and returning the formatted consequence again to the consumer.
Lastly, the agent is used and began run_server() Within the port 4747able to course of the A2A question.
Creating an Agent Community
First, run each brokers on two separate terminals
python inflation_agent.pyEvery of those brokers exposes REST API endpoints (http://localhost:4737 for EMI, http://localhost:4747) utilizing the A2A protocol. They take heed to incoming duties (resembling “Calculate EMI at £2,00,000”) and reply with textual solutions.
Subsequent, add these two brokers to the 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, create a router to intelligently direct queries to the most effective brokers. That is the core utility for the A2A protocol. Outline commonplace process codecs to permit brokers to be queried uniformly, permitting routers to make use of LLM to make clever routing selections.
router = AIAgentRouter(
llm_client=A2AClient("http://localhost:5000/openai"), # LLM for making routing selections
agent_network=community
)Lastly, we’ll question the agent
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}")
Please examine Pocket book – inflation_agent.py, network.ipynb and emi_agent.py. All credit for this examine might be despatched to researchers on this mission. Additionally, please be at liberty to comply with us Twitter And do not forget to affix us 100k+ ml subreddit And subscribe Our Newsletter.
I’m a civil engineering graduate (2022) from Jamia Milia Islamia, New Delhi, and have a powerful curiosity in information science, notably neural networks and purposes in quite a lot of fields.
