import yfinance as yf from typing import Dict, Any, List from base_tool import BaseTool from registry import ToolRegistry import logging logger = logging.getLogger("FinancialAgent") # backend/tools.py # backend/tools.py from typing import Dict, Any, List # import BaseTool if it's in a separate file, or assume it's available class CalculatorTool(BaseTool): name = "calculator" description = "Perform basic arithmetic operations. Use this for calculating differences, ratios, or percentages." categories = ["utils"] def get_schema(self) -> Dict[str, Any]: return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": { "type": "object", "properties": { "operation": { "type": "string", "enum": ["add", "subtract", "multiply", "divide"], "description": "The math operation to perform." }, "x": { "type": "number", "description": "The first number." }, "y": { "type": "number", "description": "The second number." } }, "required": ["operation", "x", "y"] } } } def run(self, operation: str, x: float, y: float) -> Any: try: # Ensure numbers are floats (in case string is passed) x, y = float(x), float(y) if operation == "add": return {"result": x + y} elif operation == "subtract": return {"result": x - y} elif operation == "multiply": return {"result": x * y} elif operation == "divide": if y == 0: return {"error": "Error: Division by zero"} return {"result": x / y} else: return {"error": f"Unknown operation: {operation}"} except Exception as e: return {"error": f"Math execution failed: {str(e)}"} class StockPriceTool(BaseTool): name = "get_stock_price" description = "Get the current price of a stock using its Ticker Symbol." categories = ["finance"] def get_schema(self) -> Dict[str, Any]: return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": { "type": "object", "properties": { "ticker_symbol": {"type": "string", "description": "The stock ticker (e.g., AAPL)"} }, "required": ["ticker_symbol"] } } } def run(self, ticker_symbol: str) -> Dict: try: ticker = yf.Ticker(ticker_symbol) info = ticker.fast_info return {"ticker": ticker_symbol, "price": info.last_price, "currency": info.currency} except Exception as e: return {"error": str(e)} class CompanyNewsTool(BaseTool): name = "get_company_news" description = "Get the latest news summaries about a company." categories = ["news"] def get_schema(self) -> Dict[str, Any]: return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": { "type": "object", "properties": { "ticker_symbol": {"type": "string", "description": "The stock ticker"}, "num_stories": {"type": "integer", "description": "Number of stories"} }, "required": ["ticker_symbol", "num_stories"] } } } def run(self, ticker_symbol: str, num_stories: int = 5) -> Dict: try: ticker = yf.Ticker(ticker_symbol) news = ticker.news[:num_stories] if ticker.news else [] if len(news) == 0: return {"error": f'HTTP Error 404: ${ticker_symbol}: possibly delisted; Quote not found for symbol'} return {"ticker": ticker_symbol, "news": [n['content']['title'] for n in news], "storiesFetched": len(news)} except Exception as e: return {"error": str(e)} def initialize_registry() -> ToolRegistry: registry = ToolRegistry() tools_list = [ StockPriceTool(), CompanyNewsTool(), CalculatorTool() ] for tool in tools_list: registry.register(tool) return registry if __name__ == "__main__": pass