Deploy: 8c89a8fbe906551039281c5c5a0089572c945b0b

Dockerfile

@@ -0,0 +1,24 @@
+# Start with Python
+FROM python:3.11-slim
+
+# Set the working directory inside the container
+WORKDIR /app
+
+# 1. Copy requirements (Relative to current folder, no 'backend/' prefix)
+COPY requirements.txt .
+
+# 2. Install dependencies
+# Added --no-cache-dir to keep image size small
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# This ARG changes whenever force rebuild is needed. 
+# We don't even need to pass a value; just the existence of a changed line forces a rebuild.
+ARG CACHEBUST=20251217 
+
+# 3. Copy the rest of the code (Current folder -> /app)
+COPY . .
+
+# 4. Run the application
+# Note: application:app works because application.py is now at /app/application.py
+CMD ["uvicorn", "application:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,9 @@
+---
+title: Midas Backend
+emoji: 🧠
+colorFrom: gray
+colorTo: blue
+sdk: docker
+app_port: 7860
+---
+# Midas Protocol Backend

agent_factory.py

@@ -0,0 +1,42 @@
+import ruamel.yaml as yaml  
+from typing import Dict
+from base_agent import SingleAgent
+from registry import ToolRegistry
+
+class AgentFactory:
+    def __init__(self, registry: ToolRegistry):
+        self.registry = registry
+        self.parser = yaml.YAML(typ='safe')
+        self.parser.pure = True 
+
+    def load_from_yaml(self, config_path: str) -> Dict[str, SingleAgent]:
+        """
+        Reads YAML and returns a dictionary of built Agents.
+        """
+        with open(config_path, "r") as f:
+            config = self.parser.load(f)
+        
+        agents_dict = {}
+
+        for agent_conf in config["agents"]:
+            name = agent_conf["name"]
+            subscriptions = agent_conf.get("subscriptions", [])
+            sys_prompt = agent_conf.get("system_prompt", "You are a helpful assistant.")
+            
+            unique_tools = set()
+            for category in subscriptions:
+                tools = self.registry.get_tools_by_category(category)
+                for t in tools:
+                    unique_tools.add(t)
+
+            agent_tools_list = list(unique_tools)
+
+            new_agent = SingleAgent(
+                name=name,
+                tools=agent_tools_list,
+                system_prompt=sys_prompt
+            )
+            agents_dict[name] = new_agent
+        
+        return agents_dict
+        

agents.yaml

@@ -0,0 +1,39 @@
+agents:
+  - name: "PriceWorker"
+    model: "llama-3.3-70b-versatile"  
+    description: "An expert in financial data and stock market valuations."
+    subscriptions:
+      - "finance"
+      - "utils"
+    system_prompt: |
+      You are a specialized financial analyst. 
+      Your goal is to retrieve accurate stock prices and perform precise calculations.
+      
+      CRITICAL TOOL INSTRUCTIONS:
+      1. When calling tools, use standard JSON format only.
+      2. NEVER use XML tags like <function> or <tool_code>.
+      3. Use 'get_stock_price' to find data.
+      4. NEVER do math in your head. ALWAYS use the 'calculator' tool.
+      5. **NO NESTED TOOLS**: You cannot pass a tool call as an argument to another tool.
+         - INCORRECT: calculator(x=get_stock_price("AAPL")...)
+         - CORRECT: 
+             Step 1: Call get_stock_price("AAPL")
+             Step 2: Wait for result.
+             Step 3: Call calculator(x=200, ...)
+      6. Output standard JSON for tool calls.
+
+  - name: "NewsWorker"
+    model: "llama-3.3-70b-versatile"
+    description: "An expert in market sentiment and company news."
+    subscriptions:
+      - "news"
+    system_prompt: |
+      You are a specialized news reporter.
+      Your goal is to summarize the latest stories affecting a company.
+      Focus on headlines that might impact stock value.
+      Be concise and objective.
+      
+      CRITICAL TOOL INSTRUCTIONS:
+      1. When calling tools, use standard JSON format only.
+      2. NEVER use XML tags like <function> or <tool_code>.
+      3. ALWAYS cite specific headlines in your final answer.

application.py

@@ -0,0 +1,231 @@
+from fastapi import FastAPI, HTTPException, Header
+from pydantic import BaseModel
+from dataclasses import dataclass
+from typing import Optional
+from enum import Enum
+from dotenv import load_dotenv
+
+from memory import TokenBufferMemory
+from agent_factory import AgentFactory
+from base_agent import ManagerAgent
+from tools import initialize_registry 
+from llm_provider import GroqProvider, GeminiProvider, QuotaExhaustedError, ProviderDownError, ProviderError
+
+import time
+import uvicorn
+import logging
+import os
+import sys
+
+load_dotenv()
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    handlers=[
+        logging.FileHandler("agent_debug.log"),
+        logging.StreamHandler(sys.stdout)
+    ]
+)
+logger = logging.getLogger("SystemBackend")
+
+class ChatRequest(BaseModel):
+    query: str
+    provider: Optional[str] = None  
+    api_key: Optional[str] = None  
+
+class ChatResponse(BaseModel):
+    success: bool
+    response: Optional[str] = None
+    provider_used: Optional[str] = None
+    agent_used: Optional[str] = None
+    error_type: Optional[str] = None
+    required_provider: Optional[str] = None 
+    message: Optional[str] = None
+
+app = FastAPI()
+registry = initialize_registry()
+factory = AgentFactory(registry)
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+YAML_PATH = os.path.join(BASE_DIR, "agents.yaml")
+try:
+    workers_map = factory.load_from_yaml(YAML_PATH)
+    logger.info(f"✅ Successfully loaded agents: {list(workers_map.keys())}")
+except FileNotFoundError:
+    logger.critical("❌ 'agents.yaml' not found! Please create this configuration file.")
+    sys.exit(1)
+
+agent_memory = TokenBufferMemory(max_tokens=4096)
+
+manager_agent = ManagerAgent(
+    name="Manager", 
+    sub_agents=workers_map,
+    memory=agent_memory
+    )
+
+
+class ProviderStatus(Enum):
+    ACTIVE = "Active"
+    DOWN = "Down"
+    QUOTA_EXHAUSTED = "QuotaExhausted"
+    ERROR = "Error"
+
+@dataclass
+class ProviderState:
+    name: str
+    status: ProviderStatus
+    reset_time: float
+
+class ProviderManager:
+    def __init__(self):
+        self.providers = {
+            "groq": ProviderState(name="groq", status=ProviderStatus.ACTIVE, reset_time=0.0),
+            "gemini": ProviderState(name="gemini", status=ProviderStatus.ACTIVE, reset_time=0.0),
+        }
+
+    def update_status(self, provider_name: str, status: ProviderStatus):
+        self.providers[provider_name].status = status
+        if status == ProviderStatus.QUOTA_EXHAUSTED:
+            self.providers[provider_name].reset_time = time.time() + (24 * 60 * 60) # 24 hours
+        elif status == ProviderStatus.DOWN:
+            self.providers[provider_name].reset_time = time.time() + 60 # 60 seconds
+
+    def get_provider(self):
+        for name, state in self.providers.items():
+            if state.status == ProviderStatus.ACTIVE:
+                return name
+            elif state.status in [ProviderStatus.DOWN, ProviderStatus.QUOTA_EXHAUSTED]:
+                if time.time() > state.reset_time:
+                    state.status = ProviderStatus.ACTIVE
+                    return name
+        return None
+
+    def is_provider_active(self, provider_name: str) -> bool:
+        if provider_name not in self.providers:
+            return False
+        state = self.providers[provider_name]
+        if state.status in [ProviderStatus.DOWN, ProviderStatus.QUOTA_EXHAUSTED]:
+            if time.time() > state.reset_time:
+                state.status = ProviderStatus.ACTIVE
+                return True
+            return False
+            
+        return state.status == ProviderStatus.ACTIVE
+
+provider_manager = ProviderManager()
+
+class RequestBody(BaseModel):
+    query: str
+    provider: Optional[str] = None 
+    api_key: Optional[str] = None  
+
+def has_server_key(name: str) -> bool:
+    if name == "groq" and os.getenv("GROQ_API_KEY"): return True
+    if name == "gemini" and os.getenv("GEMINI_API_KEY"): return True
+    return False
+
+def get_provider_instance(name: str, key: str):
+    if name == "groq": return GroqProvider(api_key=key)
+    elif name == "gemini": return GeminiProvider(api_key=key)
+    raise ValueError(f"Unknown provider: {name}")
+
+
+@app.post("/chat", response_model=ChatResponse)
+async def chat_endpoint(request: ChatRequest):
+    
+    # CASE 1: MANUAL OVERRIDE (User specifically asked for a provider)
+    if request.provider:
+        target = request.provider.lower()
+        
+        # A. Check if valid/active
+        if not provider_manager.is_provider_active(target):
+            return ChatResponse(
+                success=False, 
+                error_type="provider_down", 
+                required_provider=target,
+                message=f"Requested provider '{target}' is currently unavailable (Down/Quota)."
+            )
+
+        # B. Resolve Key
+        final_key = request.api_key if request.api_key else None
+        if not final_key and has_server_key(target):
+            final_key = os.getenv(f"{target.upper()}_API_KEY")
+            
+        logger.info(f"🔍 DEBUG CHECK: Target={target}, Key_Type={type(final_key)}, Has_Key={bool(final_key)}")
+
+        if not final_key:
+            return ChatResponse(
+                success=False, 
+                error_type="needs_key", 
+                required_provider=target, 
+                message=f"API Key missing for {target}."
+            )
+
+        # C. Execute (NO LOOP - Fail fast if user preference fails)
+        try:
+            logger.info(f"🔄 Executing User Preference: {target}")
+            llm = get_provider_instance(target, final_key)
+            result = manager_agent.process_query(request.query, llm)
+            return ChatResponse(
+                success=True, 
+                response=result.content, 
+                provider_used=target,
+                agent_used=manager_agent.name
+            )
+        except (QuotaExhaustedError, ProviderDownError) as e:
+            provider_manager.update_status(target, ProviderStatus.QUOTA_EXHAUSTED) # or check exception type
+            return ChatResponse(success=False, error_type="provider_down", message=str(e))
+        except Exception as e:
+            logger.error(f"Server Error: {e}")
+            return ChatResponse(success=False, error_type="server_error", message=str(e))
+
+    # CASE 2: AUTO-PILOT (Loop through available providers)
+    else:
+        while True:
+            current = provider_manager.get_provider()
+            
+            if not current:
+                return ChatResponse(
+                    success=False, 
+                    error_type="all_down", 
+                    message="All providers are currently down or exhausted."
+                )
+            
+            # Check Key for the Auto-Selected candidate
+            final_key = None
+            if has_server_key(current):
+                final_key = os.getenv(f"{current.upper()}_API_KEY")
+            
+            if not final_key:
+                # If Auto-Router picks a provider we have no key for, we must ask the user.
+                return ChatResponse(
+                    success=False, 
+                    error_type="needs_key", 
+                    required_provider=current, 
+                    message=f"Auto-switching to {current}, but API Key is missing."
+                )
+
+            try:
+                logger.info(f"🔄 Auto-Routing via: {current}")
+                llm = get_provider_instance(current, final_key)
+                result = manager_agent.process_query(request.query, llm)
+                return ChatResponse(
+                    success=True, 
+                    response=result.content, 
+                    provider_used=current,
+                    agent_used=manager_agent.name
+                )
+            except QuotaExhaustedError:
+                provider_manager.update_status(current, ProviderStatus.QUOTA_EXHAUSTED)
+                continue # Try next in loop
+            except ProviderDownError:
+                provider_manager.update_status(current, ProviderStatus.DOWN)
+                continue # Try next in loop
+            except Exception as e:
+                logger.error(f"Critical Error: {e}")
+                return ChatResponse(success=False, error_type="server_error", message=str(e))
+
+if __name__ == "__main__":
+    port = int(os.getenv("PORT", 7860)) 
+    uvicorn.run(app, host="0.0.0.0", port=port)

base_agent.py

@@ -0,0 +1,238 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Dict, List, Any
+import json
+import logging
+
+from memory import BaseMemory
+from base_tool import BaseTool 
+from llm_provider import LLMProvider, LLMResponse
+
+logger = logging.getLogger("AgentFramework")
+
+@dataclass
+class AgentResponse:
+    content: str                  
+    metadata: Dict[str, Any] = field(default_factory=dict) 
+
+class BaseAgent(ABC):
+    """
+    The parent class for all agents. 
+    Now accepts a clean list of 'BaseTool' objects.
+    """
+    def __init__(self, name: str, tools: List[BaseTool], system_prompt: str = "You are a helpful assistant."):
+        self.name = name
+        self.system_prompt = system_prompt
+        
+        # 1. Build the Registry (Map Name -> Function) for execution
+        self.tool_registry = {tool.name: tool.run for tool in tools}
+        
+        # 2. Build the Definitions (List of Schemas) for the LLM
+        self.tool_definitions = [tool.get_schema() for tool in tools]
+
+    @abstractmethod
+    def process_query(self, user_query: str, provider: LLMProvider) -> AgentResponse:
+        pass
+
+
+class SingleAgent(BaseAgent):
+    """
+    A standard worker agent that uses the provided BaseTools to answer queries.
+    """
+    def __init__(self, name: str, tools: List[BaseTool], system_prompt: str = "You are a helpful assistant."):
+        # Pass the tool objects directly to the parent
+        super().__init__(name, tools, system_prompt)
+
+    def process_query(self, user_query: str, provider: LLMProvider) -> AgentResponse:
+        messages = [
+            {"role": "system", "content": self.system_prompt},
+            {"role": "user", "content": user_query}
+        ]
+
+        logger.info(f"\n🚀 [{self.name}] Starting Loop...")
+
+        for turn in range(5):
+            logger.info(f"--- Turn {turn + 1} ---")
+            
+            # 1. Ask the Provider (Using the internally built definitions)
+            response: LLMResponse = provider.get_response(messages, self.tool_definitions)
+
+            # 2. Handle Tool Calls
+            if response.tool_call:
+                tool_name = response.tool_call["name"]
+                tool_args = response.tool_call["args"]
+                tool_id = response.tool_call.get("id", "call_default")
+                
+                logger.info(f"🤖 Agent Intent: Call `{tool_name}` with {tool_args}")
+
+                if tool_name in self.tool_registry:
+                    messages.append({
+                        "role": "assistant",
+                        "content": None, 
+                        "tool_calls": [{"id": tool_id, "type": "function", "function": {"name": tool_name, "arguments": json.dumps(tool_args)}}]
+                    })
+
+                    try:
+                        # Execution uses the registry built in __init__
+                        tool_func = self.tool_registry[tool_name]
+                        raw_result = tool_func(**tool_args)
+                        result_str = json.dumps(raw_result) if not isinstance(raw_result, str) else raw_result
+                        
+                        logger.info(f"Tool Output: {result_str}")
+                        messages.append({"role": "tool", "tool_call_id": tool_id, "name": tool_name, "content": result_str})
+
+                    except Exception as e:
+                        error_msg = f"Tool Execution Failed: {str(e)}"
+                        logger.error(error_msg)
+                        messages.append({"role": "tool", "tool_call_id": tool_id, "name": tool_name, "content": error_msg})
+                    continue 
+                else:
+                    messages.append({"role": "tool", "tool_call_id": tool_id, "name": tool_name, "content": f"❌ Unknown tool '{tool_name}'"})
+                    continue 
+
+            # 3. Handle Final Answer
+            if response.content:
+                logger.info(f"[{self.name}] Final Answer: {response.content}")
+                return AgentResponse(content=response.content, metadata={"final_answer": response.content})
+            
+        return AgentResponse(content="Agent timed out.", metadata={"error": "Timeout"})
+
+class ManagerAgent(BaseAgent):
+    """
+    The Brain. 
+    It treats its sub-agents as 'Tools' and dynamically decides which one to call.
+    Now equipped with Short-Term Memory!
+    """
+    def __init__(self, name: str, sub_agents: Dict[str, SingleAgent], memory: BaseMemory, system_prompt: str = "You are a manager."):
+        super().__init__(name, tools=[], system_prompt=system_prompt)
+        self.sub_agents = sub_agents
+        self.memory = memory
+        self.delegation_definitions = self._build_delegation_definitions()
+
+    def _build_delegation_definitions(self) -> List[Dict]:
+        """
+        Dynamically creates OpenAI-compatible function schemas for each sub-agent.
+        """
+        definitions = []
+        for agent_name, agent in self.sub_agents.items():
+            agent_desc = getattr(agent, "description", "A helper agent.")   
+            schema = {
+                "type": "function",
+                "function": {
+                    "name": f"delegate_to_{agent_name}",
+                    "description": f"Delegate a query to the {agent_name}. Capability: {agent_desc}",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "query": {
+                                "type": "string", 
+                                "description": "The specific question or instruction for this worker."
+                            }
+                        },
+                        "required": ["query"]
+                    }
+                }
+            }
+            definitions.append(schema)
+        return definitions
+
+    def process_query(self, user_query: str, provider: LLMProvider) -> AgentResponse:
+        """
+        The Manager's Thinking Loop.
+        It decides: Do I answer myself? Or do I call a worker?
+        """
+
+        # 1. Save User Query to Memory
+        self.memory.add_message(role="user", content=user_query)
+
+        # 2. Construct the Context (System Prompt + History)
+        team_roster = ", ".join(self.sub_agents.keys())
+        enhanced_system_prompt = (
+            f"{self.system_prompt}\n"
+            f"You manage a team of agents: [{team_roster}].\n"
+            f"Delegate tasks to them using the available tools.\n"
+            f"Combine their outputs into a comprehensive final answer."
+            f"Use the conversation history to answer follow-up questions."
+        )
+
+        # Start with System Prompt
+        messages = [{"role": "system", "content": enhanced_system_prompt}]
+
+        # Add Conversation History
+        history = self.memory.get_history()
+        messages.extend(history)
+
+        logger.info(f"👑 [{self.name}] Starting Orchestration Loop...")
+
+        # 3. Start the Loop (Max 5 turns)
+        for turn in range(5):
+            logger.info(f"--- Manager Turn {turn + 1} ---")
+            
+            # A. Ask the Provider
+            response: LLMResponse = provider.get_response(messages, self.delegation_definitions)
+
+            # B. Handle "Virtual Tool" Calls (Delegation)
+            if response.tool_call:
+                tool_name = response.tool_call["name"]
+                tool_args = response.tool_call["args"]
+                tool_id = response.tool_call.get("id", "call_mgr")
+                
+                if tool_name.startswith("delegate_to_"):
+                    agent_name = tool_name.replace("delegate_to_", "")
+                    
+                    if agent_name in self.sub_agents:
+                        logger.info(f"👑 -> 👷 Delegating to {agent_name}: {tool_args.get('query')}")
+                        
+                        # Record the "Thought" (Tool Call)
+                        messages.append({
+                            "role": "assistant",
+                            "content": None,
+                            "tool_calls": [{
+                                "id": tool_id,
+                                "type": "function",
+                                "function": {"name": tool_name, "arguments": json.dumps(tool_args)}
+                            }]
+                        })
+
+                        # EXECUTE THE WORKER
+                        worker_agent = self.sub_agents[agent_name]
+                        worker_query = tool_args.get("query")
+                        
+                        try:
+                            # Worker runs its own loop (stateless for now)
+                            worker_response = worker_agent.process_query(worker_query, provider)
+                            worker_content = worker_response.content
+                            logger.info(f"👷 -> 👑 {agent_name} replied.")
+
+                        except Exception as e:
+                            worker_content = f"Error from {agent_name}: {str(e)}"
+                            logger.error(worker_content)
+
+                        # Record the "Observation" (Tool Output)
+                        messages.append({
+                            "role": "tool",
+                            "tool_call_id": tool_id,
+                            "name": tool_name,
+                            "content": f"Output from {agent_name}:\n{worker_content}"
+                        })
+                        continue
+                    else:
+                        logger.warning(f"❌ Manager tried to call unknown agent: {agent_name}")
+                        messages.append({
+                            "role": "tool",
+                            "tool_call_id": tool_id,
+                            "name": tool_name,
+                            "content": f"Error: Agent {agent_name} does not exist."
+                        })
+                        continue
+            
+            # C. Handle Final Answer (Synthesis)
+            if response.content:
+                logger.info(f"✅ [{self.name}] Final Synthesis: {response.content}")
+                
+                # 4. Save Assistant Answer to Memory
+                self.memory.add_message(role="assistant", content=response.content)
+                
+                return AgentResponse(content=response.content)
+
+        return AgentResponse(content="Manager timed out while coordinating agents.", metadata={"error": "timeout"})

base_tool.py

@@ -0,0 +1,36 @@
+from abc import ABC, abstractmethod
+from typing import List, Dict, Any
+
+class BaseTool(ABC):
+    """
+    The universal contract for any capability.
+    Enforces structure so the Agent Loader can handle any tool automatically.
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """Unique identifier (e.g., 'get_stock_price')."""
+        pass
+
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """Natural language instruction for the LLM."""
+        pass
+
+    @property
+    @abstractmethod
+    def categories(self) -> List[str]:
+        """Tags for the Registry (e.g., ['finance', 'public_api'])."""
+        pass
+
+    @abstractmethod
+    def get_schema(self) -> Dict[str, Any]:
+        """Returns the JSON schema for LLM function calling."""
+        pass
+
+    @abstractmethod
+    def run(self, **kwargs) -> Any:
+        """Executes the tool logic."""
+        pass

llm_provider.py

@@ -0,0 +1,171 @@
+import json
+from abc import ABC, abstractmethod
+from typing import List, Dict, Any, Optional
+from pydantic import BaseModel
+from groq import Groq
+from google import genai
+from google.genai import types
+
+
+class ProviderError(Exception):
+    """Base class for all provider issues."""
+    pass
+
+class QuotaExhaustedError(ProviderError):
+    """Raised when we run out of credits/limit."""
+    pass
+
+class ProviderDownError(ProviderError):
+    """Raised when the provider is temporarily broken (500, 429)."""
+    pass
+
+
+class LLMResponse(BaseModel):
+    content: Optional[str] = None
+    tool_call: Optional[Dict[str, Any]] = None
+
+
+class LLMProvider(ABC):
+    @abstractmethod
+    def get_response(self, messages: List[Dict[str, str]], tools: List[Dict]) -> LLMResponse:
+        """
+        Args:
+            messages: Full conversation history [{"role": "user", "content": "..."}, ...]
+            tools: JSON Schema definitions for tools.
+        """
+        pass
+
+
+class GroqProvider(LLMProvider):
+    def __init__(self, api_key: str, model_name: str = 'llama-3.1-8b-instant'):
+        self.client = Groq(api_key=api_key)
+        self.model_name = model_name
+
+    def get_response(self, messages: List[Dict[str, str]], tools: List[Dict]) -> LLMResponse:
+        try:
+            # Groq/OpenAI native format
+            response = self.client.chat.completions.create(
+                model=self.model_name,
+                messages=messages,
+                tools=tools,
+                tool_choice="auto",
+                temperature=0.1
+            )
+
+            candidate = response.choices[0]
+            
+            # Check for tool calls
+            if candidate.message.tool_calls:
+                # We take the first tool call
+                tool_call_data = candidate.message.tool_calls[0]
+                return LLMResponse(
+                    tool_call={
+                        "name": tool_call_data.function.name,
+                        "args": json.loads(tool_call_data.function.arguments),
+                        "id": tool_call_data.id # Store ID for history tracking
+                    }
+                )
+            
+            # Return text content
+            return LLMResponse(content=candidate.message.content)
+
+        except Exception as e:
+            error_msg = str(e).lower()
+            if "resource_exhausted" in error_msg or "quota" in error_msg:
+                raise QuotaExhaustedError("Groq Quota Exhausted")
+            else:
+                raise ProviderDownError(f"Groq Error: {e}")
+
+
+class GeminiProvider(LLMProvider):
+    def __init__(self, api_key: str, model_name: str = 'gemini-2.0-flash'):
+        self.client = genai.Client(api_key=api_key)
+        self.model_name = model_name
+
+    def _map_tools(self, tools: List[Dict]) -> List[types.Tool]:
+        """
+        Converts OpenAI/Groq-style tool definitions into Gemini types.Tool objects.
+        """
+        gemini_tools = []
+        for t in tools:
+            # Check if it matches the OpenAI schema {"type": "function", "function": {...}}
+            if t.get("type") == "function":
+                func_def = t["function"]
+                
+                # Create the Gemini-specific FunctionDeclaration
+                fn_decl = types.FunctionDeclaration(
+                    name=func_def["name"],
+                    description=func_def.get("description"),
+                    parameters=func_def.get("parameters")
+                )
+                
+                # Wrap it in a Tool object
+                gemini_tools.append(types.Tool(function_declarations=[fn_decl]))
+        return gemini_tools
+
+    def _default_history_format(self, messages: List[Dict]) -> str:
+         formatted_prompt = ""
+         for msg in messages:
+             role = msg["role"]
+             content = msg.get("content", "") or ""
+             if role == "system":
+                 formatted_prompt += f"System Instruction: {content}\n\n"
+             elif role == "user":
+                 formatted_prompt += f"User: {content}\n"
+             elif role == "assistant":
+                 if "tool_calls" in msg:
+                     tc = msg["tool_calls"][0]
+                     formatted_prompt += f"Assistant (Thought): I will call tool '{tc['function']['name']}' with args {tc['function']['arguments']}.\n"
+                 else:
+                     formatted_prompt += f"Assistant: {content}\n"
+             elif role == "tool":
+                 formatted_prompt += f"Tool Output ({msg.get('name')}): {content}\n"
+         formatted_prompt += "\nBased on the history above, provide the next response or tool call."
+         return formatted_prompt
+
+    def get_response(self, messages: List[Dict[str, str]], tools: List[Dict]) -> LLMResponse:
+        try:
+            # 1. Translate History
+            full_prompt = self._default_history_format(messages)
+            
+            gemini_messages = [
+                types.Content(role="user", parts=[types.Part(text=full_prompt)])
+            ]
+
+            # 2. Translate Tools 
+            mapped_tools = self._map_tools(tools)
+
+            config = types.GenerateContentConfig(
+                tools=mapped_tools, 
+                temperature=0.0
+            )
+
+            response = self.client.models.generate_content(
+                model=self.model_name,
+                contents=gemini_messages,
+                config=config,
+            )
+            
+            candidate = response.candidates[0]
+            function_call_part = None
+            for part in candidate.content.parts:
+                if part.function_call:
+                    function_call_part = part
+                    break
+            
+            if function_call_part:
+                return LLMResponse(
+                    tool_call={
+                        "name": function_call_part.function_call.name,
+                        "args": function_call_part.function_call.args
+                    }
+                )
+            
+            return LLMResponse(content=candidate.content.parts[0].text)
+
+        except Exception as e:
+            error_msg = str(e).lower()
+            if "resource_exhausted" in error_msg or "quota" in error_msg:
+                raise QuotaExhaustedError("Gemini Quota Exhausted")
+            else:
+                raise ProviderDownError(f"Gemini Error: {e}")

memory.py

@@ -0,0 +1,68 @@
+import tiktoken
+from abc import ABC, abstractmethod
+from typing import List, Dict
+import logging
+
+logger = logging.getLogger("FinancialAgent")
+
+class BaseMemory(ABC):
+    """
+    Abstract Base Class for memory management.
+    """
+    @abstractmethod
+    def add_message(self, role: str, content: str):
+        pass
+
+    @abstractmethod
+    def get_history(self) -> List[Dict[str, str]]:
+        pass
+
+    @abstractmethod
+    def clear(self):
+        pass
+
+class TokenBufferMemory(BaseMemory):
+    """
+    Memory that keeps conversation within a strict token limit.
+    Uses a FIFO (First-In-First-Out) eviction strategy when full.
+    """
+    def __init__(self, max_tokens: int = 4096, encoding_name: str = "cl100k_base"):
+        self.max_tokens = max_tokens
+        self.messages = []
+        # cl100k_base is the encoding for GPT-4 and acts as a good standard proxy
+        self.tokenizer = tiktoken.get_encoding(encoding_name) 
+
+    def _count_tokens(self, text: str) -> int:
+        """Helper to count tokens in a string."""
+        try:
+            return len(self.tokenizer.encode(text))
+        except Exception:
+            # Fallback for empty strings or weird encoding errors
+            return 0
+
+    def _evict_if_needed(self):
+        """
+        Removes oldest messages until we are under the token limit.
+        Safety: Never deletes the most recent message (index -1), 
+        so we always have at least the latest context.
+        """
+        while len(self.messages) > 1:  
+            current_buffer_tokens = sum(self._count_tokens(m["content"]) for m in self.messages)
+            
+            if current_buffer_tokens <= self.max_tokens:
+                break
+            
+            # Remove the oldest message
+            removed = self.messages.pop(0)
+            logger.info(f"🧹 Memory Full. Evicted message: {removed['role']} ({len(removed['content'])} chars)")
+
+    def add_message(self, role: str, content: str):
+        """Adds a message and triggers eviction check."""
+        self.messages.append({"role": role, "content": content})
+        self._evict_if_needed()
+
+    def get_history(self) -> List[Dict[str, str]]:
+        return self.messages
+
+    def clear(self):
+        self.messages = []

registry.py

@@ -0,0 +1,29 @@
+from typing import Dict, List
+from base_tool import BaseTool
+
+class ToolRegistry:
+    def __init__(self):
+        # Index 1: Look up by Name (for execution)
+        self._tools_by_name: Dict[str, BaseTool] = {}
+        
+        # Index 2: Look up by Category (for subscription)
+        self._tools_by_category: Dict[str, List[BaseTool]] = {}
+
+    def register(self, tool: BaseTool):
+        if tool.name in self._tools_by_name:
+            raise ValueError(f"Tool '{tool.name}' is already registered.")
+        
+        # 1. Add to Main Index
+        self._tools_by_name[tool.name] = tool
+
+        # 2. Add to Category Index
+        for category in tool.categories:
+            if category not in self._tools_by_category:
+                self._tools_by_category[category] = []
+            self._tools_by_category[category].append(tool)
+
+    def get_tool(self, name: str) -> BaseTool:
+        return self._tools_by_name.get(name)
+
+    def get_tools_by_category(self, category: str) -> List[BaseTool]:
+        return self._tools_by_category.get(category, [])

requirements.txt

@@ -0,0 +1,93 @@
+altair==6.0.0
+annotated-doc==0.0.4
+annotated-types==0.7.0
+anyio==4.12.0
+attrs==25.4.0
+beautifulsoup4==4.14.3
+blinker==1.9.0
+cachetools==6.2.2
+certifi==2025.11.12
+cffi==2.0.0
+charset-normalizer==3.4.4
+click==8.3.1
+curl_cffi==0.13.0
+distro==1.9.0
+docopt==0.6.2
+dotenv==0.9.9
+fastapi==0.123.9
+frozendict==2.4.7
+gitdb==4.0.12
+GitPython==3.1.45
+google-ai-generativelanguage==0.6.15
+google-api-core==2.28.1
+google-api-python-client==2.187.0
+google-auth==2.43.0
+google-auth-httplib2==0.2.1
+google-genai==1.53.0
+googleapis-common-protos==1.72.0
+groq==0.37.1
+grpcio==1.76.0
+grpcio-status==1.71.2
+h11==0.16.0
+httpcore==1.0.9
+httplib2==0.31.0
+httpx==0.28.1
+idna==3.11
+iniconfig==2.3.0
+Jinja2==3.1.6
+jsonschema==4.25.1
+jsonschema-specifications==2025.9.1
+MarkupSafe==3.0.3
+multitasking==0.0.12
+narwhals==2.13.0
+numpy==2.3.5
+packaging==25.0
+pandas==2.3.3
+peewee==3.18.3
+pillow==12.0.0
+pipreqs==0.4.13
+platformdirs==4.5.1
+pluggy==1.6.0
+proto-plus==1.26.1
+protobuf==5.29.5
+pyarrow==22.0.0
+pyasn1==0.6.1
+pyasn1_modules==0.4.2
+pycparser==2.23
+pydantic==2.12.5
+pydantic_core==2.41.5
+pydeck==0.9.1
+Pygments==2.19.2
+pyparsing==3.2.5
+pytest==9.0.2
+python-dateutil==2.9.0.post0
+python-dotenv==1.2.1
+pytz==2025.2
+referencing==0.37.0
+regex==2025.11.3
+requests==2.32.5
+rpds-py==0.30.0
+rsa==4.9.1
+ruamel.yaml==0.18.16
+ruamel.yaml.clib==0.2.15
+singleton==0.1.0
+six==1.17.0
+smmap==5.0.2
+sniffio==1.3.1
+soupsieve==2.8
+starlette==0.50.0
+streamlit==1.52.0
+tenacity==9.1.2
+tiktoken==0.12.0
+toml==0.10.2
+tornado==6.5.2
+tqdm==4.67.1
+typing-inspection==0.4.2
+typing_extensions==4.15.0
+tzdata==2025.2
+uritemplate==4.2.0
+urllib3==2.5.0
+uvicorn==0.38.0
+websockets==15.0.1
+yarg==0.1.10
+yfinance==0.2.66

tools.py

@@ -0,0 +1,147 @@
+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