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