diff --git a/app/admin/metadata.py b/app/admin/metadata.py index b22269c6..aad83705 100644 --- a/app/admin/metadata.py +++ b/app/admin/metadata.py @@ -1,13 +1,187 @@ +import json import logging -from typing import List +from pathlib import Path +from typing import Any, Dict, List, Optional, Set from fastapi import APIRouter, Depends +from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +# Import functions from skill_processor +from app.admin.generator.skill_processor import ( + AVAILABLE_SKILL_CATEGORIES, + get_skill_default_api_key_provider, + get_skill_keyword_config, + get_skill_states, +) from intentkit.models.db import get_db from intentkit.models.llm import LLMModelInfo, LLMModelInfoTable, LLMProvider -from intentkit.models.skill import Skill, SkillTable + +# Import additional modules for our custom schema loading + +# Cache for skill schemas +_skill_schemas_cache: Dict[str, Dict[str, Any]] = {} + + +class SkillState(BaseModel): + """Detailed skill state information.""" + + name: str = Field(..., description="State name") + title: Optional[str] = Field(None, description="Display title for the state") + description: Optional[str] = Field( + None, description="Description of what this state does" + ) + default_value: str = Field("disabled", description="Default value for this state") + options: List[str] = Field( + default_factory=list, + description="Available options (disabled, public, private)", + ) + + +class BasicSkill(BaseModel): + """Basic skill information with comprehensive details.""" + + skill_name: str = Field(..., description="The skill category name") + title: Optional[str] = Field(None, description="Display title from schema") + description: Optional[str] = Field(None, description="Description from schema") + icon_url: Optional[str] = Field(None, description="URL to the skill icon") + x_tags: List[str] = Field( + default_factory=list, description="Tags/categories for the skill" + ) + states: List[SkillState] = Field( + default_factory=list, description="Available skill states with details" + ) + simple_states: List[str] = Field( + default_factory=list, description="Simple list of state names" + ) + api_key_provider_options: List[str] = Field( + default_factory=list, description="Available API key provider options" + ) + default_api_key_provider: str = Field( + "platform", description="Default API key provider" + ) + requires_user_api_key: bool = Field( + False, description="Whether user needs to provide their own API key" + ) + requires_agent_owner_api_key: bool = Field( + False, description="Whether agent owner must provide API key" + ) + has_schema: bool = Field(True, description="Whether the skill has a schema file") + keywords: List[str] = Field( + default_factory=list, description="Keywords for skill discovery" + ) + + +def load_skill_schema(skill_name: str) -> Optional[Dict[str, Any]]: + """Load schema.json for a specific skill with correct path.""" + if skill_name in _skill_schemas_cache: + return _skill_schemas_cache[skill_name] + + try: + # Get the correct path to skills directory + # From intentkit/app/admin/metadata.py, go to intentkit/intentkit/skills/ + current_file = Path(__file__) + skills_dir = current_file.parent.parent.parent / "intentkit" / "skills" + schema_path = skills_dir / skill_name / "schema.json" + + logging.debug(f"Looking for schema at: {schema_path}") + + if schema_path.exists(): + with open(schema_path, "r") as f: + schema = json.load(f) + _skill_schemas_cache[skill_name] = schema + logging.debug( + f"Successfully loaded schema for {skill_name}: title='{schema.get('title')}', description='{schema.get('description', '')[:50]}...'" + ) + return schema + else: + logging.warning( + f"Schema file not found for skill: {skill_name} at {schema_path}" + ) + return None + except Exception as e: + logging.error(f"Error loading schema for skill {skill_name}: {e}") + return None + + +def parse_skill_states_from_schema( + schema: Dict[str, Any], actual_states: Set[str] +) -> List[SkillState]: + """Parse detailed state information from schema.""" + states = [] + + if not schema or "properties" not in schema or "states" not in schema["properties"]: + # Fallback to actual states without details + return [ + SkillState(name=state, options=["disabled", "public", "private"]) + for state in sorted(actual_states) + ] + + states_schema = schema["properties"]["states"] + if "properties" not in states_schema: + return [ + SkillState(name=state, options=["disabled", "public", "private"]) + for state in sorted(actual_states) + ] + + for state_name, state_config in states_schema["properties"].items(): + if state_name in actual_states: + title = state_config.get("title", state_name) + description = state_config.get("description", "") + default_value = state_config.get("default", "disabled") + options = state_config.get("enum", ["disabled", "public", "private"]) + + states.append( + SkillState( + name=state_name, + title=title, + description=description, + default_value=default_value, + options=options, + ) + ) + + # Add any states that exist in actual_states but not in schema + schema_state_names = set(states_schema["properties"].keys()) + missing_states = actual_states - schema_state_names + for state_name in missing_states: + states.append( + SkillState(name=state_name, options=["disabled", "public", "private"]) + ) + + return sorted(states, key=lambda x: x.name) + + +def determine_api_key_requirements( + schema: Optional[Dict[str, Any]], default_provider: str +) -> tuple[bool, bool]: + """Determine API key requirements based on schema and default provider.""" + requires_user_api_key = False + requires_agent_owner_api_key = False + + if ( + not schema + or "properties" not in schema + or "api_key_provider" not in schema["properties"] + ): + return requires_user_api_key, requires_agent_owner_api_key + + api_key_provider = schema["properties"]["api_key_provider"] + provider_options = api_key_provider.get("enum", []) + + # If agent_owner is the only option, then agent owner MUST provide API key + if provider_options == ["agent_owner"]: + requires_agent_owner_api_key = True + # If default is agent_owner, then by default agent owner needs to provide + elif default_provider == "agent_owner": + requires_user_api_key = True # User (agent owner) needs to provide + # If both platform and agent_owner are available, user CAN provide their own + elif "agent_owner" in provider_options and "platform" in provider_options: + requires_user_api_key = False # Optional - user can choose + + return requires_user_api_key, requires_agent_owner_api_key + # Create a readonly router for metadata endpoints metadata_router_readonly = APIRouter(tags=["Metadata"]) @@ -21,25 +195,108 @@ class LLMModelInfoWithProviderName(LLMModelInfo): @metadata_router_readonly.get( "/metadata/skills", - response_model=List[Skill], + response_model=List[BasicSkill], summary="Get all skills", description="Returns a list of all available skills in the system", ) -async def get_skills(db: AsyncSession = Depends(get_db)): +async def get_skills(): """ Get all skills available in the system. **Returns:** - * `List[Skill]` - List of all skills + * `List[BasicSkill]` - List of all skills """ try: - # Query all skills from the database - stmt = select(SkillTable) - result = await db.execute(stmt) - skills = result.scalars().all() + skills = [] + + # Get keyword configuration for all skills + keyword_config = get_skill_keyword_config() + + # Process each available skill category + for skill_name in sorted(AVAILABLE_SKILL_CATEGORIES): + try: + # Load schema for this skill + schema = load_skill_schema(skill_name) + + # Get skill states + actual_states = get_skill_states(skill_name) + + # Get keywords for this skill + keywords = keyword_config.get(skill_name, [skill_name]) + + # Extract information from schema + title = schema.get("title") if schema else None + description = schema.get("description") if schema else None + icon_url = schema.get("x-icon") if schema else None + x_tags = schema.get("x-tags", []) if schema else [] + + # Get API key provider information + api_key_provider_options = [] + if ( + schema + and "properties" in schema + and "api_key_provider" in schema["properties"] + ): + api_key_provider = schema["properties"]["api_key_provider"] + if "enum" in api_key_provider: + api_key_provider_options = api_key_provider["enum"] + + # Get default API key provider + default_api_key_provider = get_skill_default_api_key_provider( + skill_name + ) + + # Determine API key requirements + requires_user_api_key, requires_agent_owner_api_key = ( + determine_api_key_requirements(schema, default_api_key_provider) + ) + + # Parse detailed state information + detailed_states = parse_skill_states_from_schema(schema, actual_states) + simple_states = sorted(list(actual_states)) if actual_states else [] + + # Create the basic skill object + skill = BasicSkill( + skill_name=skill_name, + title=title, + description=description, + icon_url=icon_url, + x_tags=x_tags, + states=detailed_states, + simple_states=simple_states, + api_key_provider_options=api_key_provider_options, + default_api_key_provider=default_api_key_provider, + requires_user_api_key=requires_user_api_key, + requires_agent_owner_api_key=requires_agent_owner_api_key, + has_schema=schema is not None, + keywords=keywords, + ) + + skills.append(skill) + + except Exception as e: + logging.warning(f"Error processing skill {skill_name}: {e}") + # Create a minimal entry for skills that failed to process + skill = BasicSkill( + skill_name=skill_name, + title=skill_name, + description=f"Skill {skill_name} (processing error)", + icon_url=None, + x_tags=[], + states=[], + simple_states=[], + api_key_provider_options=[], + default_api_key_provider="platform", + requires_user_api_key=False, + requires_agent_owner_api_key=False, + has_schema=False, + keywords=[skill_name], + ) + skills.append(skill) + + logging.info(f"Successfully processed {len(skills)} skills") + return skills - # Convert to Skill models - return [Skill.model_validate(skill) for skill in skills] except Exception as e: logging.error(f"Error getting skills: {e}") raise