How to create and connect financial agents using Python-A2A and Google’s Agents and Agents (A2A) protocol
Python A2A is an implementation of Google’s proxy-to-proxy (A2A) protocol that allows AI agents to communicate with each other in a shared, standardized format, thus determining the need for custom integration between services.
In this tutorial, we will use the decorative-based approach provided by the Python-A2A library. Simple @agent and @Skill Decorators, you can define the identity and behavior of the agent, while the library is responsible for protocol processing and message flow.
This approach is ideal for quickly building useful, task-centric agents without having to worry about low-level communication logic.
Install dependencies
First, you need to install the Python-A2A library, which provides a clean abstraction to build and run a proxy that follows the A2A protocol.
Open the terminal and run:
Create a proxy
In this tutorial, we will create two agents – one for calculating stock returns based on investment, interest rates and time, and the other for adjusting the amount for years based on inflation.
EMI agent (emi_agent.py)
from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
import re
@agent(
name="EMI Calculator Agent",
description="Calculates EMI for a given principal, interest rate, and loan duration",
version="1.0.0"
)
class EMIAgent(A2AServer):
@skill(
name="Calculate EMI",
description="Calculates EMI given principal, annual interest rate, and duration 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 loan of ₹{principal:.0f} at {annual_rate:.2f}% interest for {months} months is ₹{emi:.2f}"
def handle_task(self, task):
input_text = task.message["content"]["text"]
# Extract values from natural 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)
try:
principal = float(principal_match.group(1)) if principal_match else 100000
rate = 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}, Rate: {rate}, Months: {months}")
emi_text = self.calculate_emi(principal, rate, months)
except Exception as e:
emi_text = f"Sorry, I couldn't parse your input. Error: {e}"
task.artifacts = [{
"parts": [{"type": "text", "text": emi_text}]
}]
task.status = TaskStatus(state=TaskState.COMPLETED)
return task
# Run the server
if __name__ == "__main__":
agent = EMIAgent()
run_server(agent, port=4737)This EMI calculator proxy is built using the Python-A2A library and follows a decor-based approach. On the top, we use @agent Decorator that defines the name, description and version of the agent. This registers the proxy so that it can communicate using the A2A protocol.
In class, we use @Skill Decorator. This skill is called calculate_emiperform actual EMI calculations using standard formulas. The formula takes three parameters: the loan principal, the annual interest rate and the loan duration within a few months. We convert the annual rate to a monthly rate and use it to calculate the monthly EMI.
this handle_task Methods are at the heart of the agent. It receives input messages from the user, extracts the relevant numbers using a simple regular expression, and then passes them to the calculation method.
Finally, at the bottom of the file, we use run_server() Functions on the port 4737prepare to receive A2A protocol messages. This design keeps agents simple, modular and easy to scale to have more skills in the future.
Inflation agent (inflation_agent.py)
from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
import re
@agent(
name="Inflation Adjusted Amount Agent",
description="Calculates the future value adjusted for inflation",
version="1.0.0"
)
class InflationAgent(A2AServer):
@skill(
name="Inflation Adjustment",
description="Adjusts an amount for inflation over time",
tags=["inflation", "adjustment", "future value"]
)
def handle_input(self, text: str) -> str:
try:
# Extract amount
amount_match = re.search(r"₹?(d{3,10})", text)
amount = float(amount_match.group(1)) if amount_match else None
# Extract rate (e.g. 6%, 7.5 percent)
rate_match = re.search(r"(d+(.d+)?)s*(%|percent)", text, re.IGNORECASE)
rate = float(rate_match.group(1)) if rate_match else None
# Extract years (e.g. 5 years)
years_match = re.search(r"(d+)s*(years|year)", text, re.IGNORECASE)
years = int(years_match.group(1)) if years_match else None
if amount is not None and rate is not None and years is not None:
adjusted = amount * ((1 + rate / 100) ** years)
return f"₹{amount:.2f} adjusted for {rate:.2f}% inflation over {years} years is ₹{adjusted:.2f}"
return (
"Please provide amount, inflation rate (e.g. 6%) and duration (e.g. 5 years).n"
"Example: 'What is ₹10000 worth after 5 years at 6% inflation?'"
)
except Exception as e:
return f"Sorry, I couldn't compute that. Error: {e}"
def handle_task(self, task):
text = task.message["content"]["text"]
result = self.handle_input(text)
task.artifacts = [{
"parts": [{"type": "text", "text": result}]
}]
task.status = TaskStatus(state=TaskState.COMPLETED)
return task
if __name__ == "__main__":
agent = InflationAgent()
run_server(agent, port=4747)This agent helps calculate the value of a given quantity in the future after adjusting for inflation. It uses the same decor-based structure provided by the Python-A2A library. this @agent The decorator defines metadata for the proxy, @Skill The decorator registers the main logic under the name “inflation adjustment”.
this handle_input Methods are the main way to deal with where they occur. It uses a simple regular expression to extract the number, inflation rate and quantity of user input from the user’s input. If all three values exist, it uses the standard future value formula to calculate the inflation-adjusted amount:
Adjustment value = quantity × (1 + rate/100) ^ years.
If any value is missing, the proxy returns a useful prompt telling the user what can be provided, including examples. this handle_task The function connects everything by getting the user’s message, passing it to the skill function, and then returning the formatted result to the user.
Finally, use run_server() In the port 4747ready to process A2A queries.
Create a proxy network
First run two agents in two separate terminals
python inflation_agent.pyEach of these reagents reveals REST API endpoints (e.g. EMI, inflation) using the A2A protocol. They listen to incoming tasks (such as “calculated selling price is ₹2,00,000…”) and respond with a text answer.
Now, we will add these two agents to our network
from python_a2a import AgentNetwork, A2AClient, AIAgentRouter
# Create an agent network
network = AgentNetwork(name="Economics Calculator")
# Add agents to the network
network.add("EMI", "")
network.add("Inflation", "")Next, we will create a router to intelligently turn queries to the best proxy. This is the core utility of the A2A protocol – it defines the standard task format so that the query proxy can be unified, and the router can make intelligent routing decisions using LLMS.
router = AIAgentRouter(
llm_client=A2AClient(" # LLM for making routing decisions
agent_network=network
)Finally, we will query the agent
query = "Calculate EMI for ₹200000 at 5% interest over 18 months."
agent_name, confidence = router.route_query(query)
print(f"Routing to {agent_name} with {confidence:.2f} confidence")
# Get the selected agent and ask the question
agent = network.get_agent(agent_name)
response = agent.ask(query)
print(f"Response: {response}")query = "What is ₹1500000 worth if inflation is 9% for 10 years?"
agent_name, confidence = router.route_query(query)
print(f"Routing to {agent_name} with {confidence:.2f} confidence")
# Get the selected agent and ask the question
agent = network.get_agent(agent_name)
response = agent.ask(query)
print(f"Response: {response}")
Check Notebook – Inflation_agent.py,,,,, network.ipynb and emi_agent.py. All credits for this study are to the researchers on the project. Also, please stay tuned for us twitter And don’t forget to join us 100K+ ml reddit And subscribe Our newsletter.
I am a civil engineering graduate in Islamic Islam in Jamia Milia New Delhi (2022) and I am very interested in data science, especially neural networks and their applications in various fields.
