How to use Python-A2A to create and connect financial agents with Google's agent-to-agent (A2A) protocol

How to use Python-A2A to create and connect financial agents with Google’s agent-to-agent (A2A) protocol

by

Python A2A is an implementation of Google’s agent-to-agent (A2A) protocol, which allows AIA agents to communicate with each other using a shared, standardized format-in-what-felt need for custom integration between services.

In this tutorial we 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.

Installation of the addictions

To get started, install the Python-A2A library, which provides a pure abstraction to build and run agents that follow the A2A protocol.

Open your terminal and run:

Creating the agents

For this tutorial, we create two agents – one for calculating stock returns based on investment, rate and time and another to adjust an amount based on inflation over a period of year.

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 is built using the Python-A2A library and follows the decorator-based approach. At the top we use @agent Decorator to define the agent’s name, description and version. This detects the agent so it can communicate using the A2A protocol.

Inside the class we define a single skill using @skill Decorator. This skill, called Calculate_emiPerforms the actual EMI calculation using the standard formula. The formula occupies three parameters: the loan principle, the annual interest rate and the loan value for months. We convert the annual rate to a monthly rate and use it to calculate the monthly EMI.

The Handler_task Method is at the heart 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 Run_server () Function on port 4737Get ready to receive A2A protocol messages. This design keeps the agent simple, modular and easy to expand 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 remedy helps to 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 @skill Decorator detects the most important logic under the name “Inflation adjustment.”

The Handler_input Method is where the main treatment happens. It extracts the amount, inflation rate and the 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 useful prompt that tells the user what to give, including an example. The Handler_task Function connects everything by taking the user’s message, passing it on to the skill function and returning the formatted result back to the user.

Finally, the agent is launched using Run_server () At port 4747Get ready to handle A2A queries.

Creating the Agent Network

First run both agents in two separate terminals

python inflation_agent.py

Each of these agents exposes a residual API end point (eg http: // localhost: 4737 for EMI, http: // localhost: 4747 to inflation) using the A2A protocol. They listen to incoming tasks (which “calculate EMI for £ 2.00,000 …”) and respond with text answers.

Now we 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 create a router for intelligent direct queries to the best agent. This is a core tool of the A2A protocol – it defines a standard task format so that agents can be spured uniform and routers can take intelligent routing decisions using LLMs.

router = AIAgentRouter(
    llm_client=A2AClient("http://localhost:5000/openai"),  # LLM for making routing decisions
    agent_network=network
)

Eventually we will ask 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 Notebooks- inflation_Agent.pyAt Network.ipynb and EMI_AGENT.PY. All credit for this research goes to the researchers in this project. You are also welcome to follow us on Twitter And don’t forget to join our 100k+ ml subbreddit and subscribe to Our newsletter.


I am a candidate for a civil engineer (2022) from Jamia Millia Islamia, New Delhi, and I have a great interest in data sciences, especially neural networks and their use in different areas.

Leave a Comment