Newer
Older
cortex-hub / ai-hub / app / core / services / tool.py
from typing import List, Dict, Any, Optional
from sqlalchemy.orm import Session
from app.db import models
from app.core.skills.base import BaseSkill
import logging

logger = logging.getLogger(__name__)

class ToolService:
    """
    Orchestrates AI tools (Skills) available to users.
    Handles discovery, permission checks, and execution routing.
    """
    
    def __init__(self, local_skills: List[BaseSkill] = []):
        self._local_skills = {s.name: s for s in local_skills}

    def get_available_tools(self, db: Session, user_id: str) -> List[Dict[str, Any]]:
        """
        Retrieves all tools the user is authorized to use.
        """
        # 1. Start with system/local skills
        tools = [s.to_tool_definition() for s in self._local_skills.values()]
        
        # 2. Add DB-defined skills with permission checks
        db_skills = db.query(models.Skill).filter(
            (models.Skill.is_system == True) | 
            (models.Skill.owner_id == user_id)
        ).all()
        
        # TODO: Implement more complex group-based permission logic
        for ds in db_skills:
            # Prevent duplicates if name overlaps with local
            if any(t["function"]["name"] == ds.name for t in tools):
                continue
            
            tools.append({
                "type": "function",
                "function": {
                    "name": ds.name,
                    "description": ds.description,
                    "parameters": ds.config.get("parameters", {})
                }
            })
            
        return tools

    async def call_tool(self, tool_name: str, arguments: Dict[str, Any], **context) -> Any:
        """
        Executes a registered skill.
        """
        if tool_name in self._local_skills:
            skill = self._local_skills[tool_name]
            result = await skill.execute(**arguments)
            return result.dict()
        
        # TODO: Handle remote/gRPC skills or MCP skills here
        logger.error(f"Tool '{tool_name}' not found or handled yet.")
        return {"success": False, "error": "Tool not found"}