Python A2A is an implementation of Google’s Agent-to-Agent (A2A) protocol, which enables AI agents to communicate with each other using a shared, standardized format—eliminating the need for custom integration between services.
In this tutorial, we’ll use the decorator-based approach provided by the python-a2a library. With simple @agent and @skill decorators, you can define your agent’s identity and behavior, while the library takes care of protocol handling and message flow.
This method is perfect for quickly building useful, task-focused agents without worrying about low-level communication logic.
Installing the dependencies
To get started, you’ll need to install the python-a2a library, which provides a clean abstraction to build and run agents that follow the A2A protocol.
Open your terminal and run:
pip install python-a2aCreating the Agents
For this tutorial, we will be creating two agents – one for calculating stock returns based on investment, rate, and time, and another for adjusting an amount based on inflation over a period of years.
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 Agent is built using the python-a2a library and follows the decorator-based approach. At the top, we use the @agent decorator to define the agent’s name, description, and version. This registers the agent so that it can communicate using the A2A protocol.
Inside the class, we define a single skill using the @skill decorator. This skill, called calculate_emi, performs the actual EMI calculation using the standard formula. The formula takes in three parameters: the loan principal, the annual interest rate, and the loan duration in months. We convert the annual rate into a monthly rate and use it to compute the monthly EMI.
The handle_task method is the core of the agent. It receives the user’s input message, extracts relevant numbers using simple regular expressions, and passes them to the calculate_emi method.
Finally, at the bottom of the file, we launch the agent using the run_server() function on port 4737, making it ready to receive A2A protocol messages. This design keeps the agent simple, modular, and easy to extend with 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 how much a given amount would be worth in the future after adjusting for inflation. It uses the same decorator-based structure provided by the python-a2a library. The @agent decorator defines the metadata for this agent, and the @skill decorator registers the main logic under the name “Inflation Adjustment.”
The handle_input method is where the main processing happens. It extracts the amount, inflation rate, and number of years from the user’s input using simple regular expressions. If all three values are present, it uses the standard future value formula to calculate the inflation-adjusted amount:
Adjusted Value = amount × (1 + rate/100) ^ years.
If any value is missing, the agent returns a helpful prompt telling the user what to provide, including an example. The handle_task function connects everything by taking the user’s message, passing it to the skill function, and returning the formatted result back to the user.
Finally, the agent is launched using run_server() on port 4747, making it ready to handle A2A queries.
Creating the Agent Network
Firstly run both the agents in two separate terminals
python emi_agent.pypython inflation_agent.pyEach of these agents exposes a REST API endpoint (e.g. http://localhost:4737 for EMI, http://localhost:4747 for Inflation) using the A2A protocol. They listen for incoming tasks (like “calculate EMI for ₹2,00,000…”) and respond with text answers.
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", "http://localhost:4737")
network.add("Inflation", "http://localhost:4747")Next we will create a router to intelligently direct queries to the best agent. This is a core utility of the A2A protocol—it defines a standard task format so agents can be queried uniformly, and routers can make intelligent routing decisions using LLMs.
router = AIAgentRouter(
llm_client=A2AClient("http://localhost:5000/openai"), # LLM for making routing decisions
agent_network=network
)Lastly, we will query the agents
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 out the Notebooks- inflation_agent.py, network.ipynb and emi_agent.py. All credit for this research goes to the researchers of this project. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter.
The post How to Use python-A2A to Create and Connect Financial Agents with Google’s Agent-to-Agent (A2A) Protocol appeared first on MarkTechPost.
Source: Read MoreÂ
