Cómo usar Python-A2A para crear y conectar agentes financieros con el protocolo de agente a agente (A2A) de Google

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)

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)

python inflation_agent.py

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

router = AIAgentRouter(
    llm_client=A2AClient("http://localhost:5000/openai"),  # LLM for making routing decisions
    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")

# 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}")