itinai content

←back to Blog

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

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.

Target Audience Analysis

The target audience for this tutorial primarily consists of:

  • Data Scientists and AI Developers: Professionals looking to implement AI solutions in financial services.
  • Business Analysts: Individuals interested in automating financial calculations and improving decision-making processes.
  • Financial Institutions: Organizations seeking to enhance their services through AI-driven tools.

Pain Points: The audience may struggle with integrating various AI systems, managing complex data flows, and ensuring accurate financial calculations.

Goals: They aim to streamline financial processes, enhance user experience, and leverage AI for better insights.

Interests: Topics related to AI, financial technology, automation, and data analysis.

Communication Preferences: They prefer clear, concise technical documentation with practical examples and code snippets.

Getting Started with python-A2A

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-a2a

Creating the Agents

For this tutorial, we will be creating two agents:

  • EMI Agent: Calculates stock returns based on investment, rate, and time.
  • Inflation Agent: Adjusts 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"]
        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)

This EMI Calculator Agent is built using the python-a2a library and follows the decorator-based approach. The @agent decorator defines the agent’s name, description, and version. Inside the class, we define a skill using the @skill decorator, which performs the actual EMI calculation using the standard formula.

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:
            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)

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.”

Creating the Agent Network

Firstly, run both the agents in two separate terminals:

python emi_agent.py
python inflation_agent.py

Each 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.