Building Financial AI Agents with python-A2A and Google’s Agent-to-Agent Protocol

Introduction to python-A2A and Agent Communication

Python A2A is a Python implementation of Google’s Agent-to-Agent (A2A) protocol. This protocol allows AI agents to communicate seamlessly using a shared, standardized format, eliminating the complexities of custom integrations between different services.

Using Decorators to Define Agents and Skills

The python-a2a library uses a decorator-based approach to simplify agent creation. By applying @agent and @skill decorators, developers can easily define an agent's identity and functionalities without managing low-level communication details.

Installing python-a2a

To begin, install the python-a2a library using pip:

pip install python-a2a

Creating the EMI Calculator Agent

We create an EMI Calculator Agent that computes monthly EMI given principal, interest rate, and loan duration. The agent uses @agent to define metadata and @skill to implement the EMI calculation.

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"]
 
        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
 
            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
 
if __name__ == "__main__":
    agent = EMIAgent()
    run_server(agent, port=4737)

Creating the Inflation Adjustment Agent

This agent calculates the future value of an amount adjusted for inflation over a number of years.

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:
            amount_match = re.search(r"₹?(\d{3,10})", text)
            amount = float(amount_match.group(1)) if amount_match else None
 
            rate_match = re.search(r"(\d+(\.\d+)?)\s*(%|percent)", text, re.IGNORECASE)
            rate = float(rate_match.group(1)) if rate_match else None
 
            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)

Creating an Agent Network

Run each agent in separate terminals:

python emi_agent.py
python inflation_agent.py

Each agent exposes a REST API at ports 4737 and 4747 respectively.

Add these agents to a network:

from python_a2a import AgentNetwork, A2AClient, AIAgentRouter
 
network = AgentNetwork(name="Economics Calculator")
network.add("EMI", "http://localhost:4737")
network.add("Inflation", "http://localhost:4747")

Using a Router to Route Queries

Create a router that uses a Large Language Model client to route queries to the appropriate agent.

router = AIAgentRouter(
    llm_client=A2AClient("http://localhost:5000/openai"),
    agent_network=network
)
 
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")
 
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")
 
agent = network.get_agent(agent_name)
response = agent.ask(query)
print(f"Response: {response}")

This setup demonstrates how multiple financial agents can be connected and queried uniformly using python-a2a and Google’s A2A protocol.