""" Transformation handler for LiteLLM database-backed skills. This module provides the SDK-level transformation layer that converts API requests to database operations via LiteLLMSkillsHandler. Pattern follows litellm/llms/litellm_proxy/responses/transformation.py """ from typing import TYPE_CHECKING, Any, Coroutine, Dict, List, Optional, Union from litellm.types.llms.anthropic_skills import ( DeleteSkillResponse, ListSkillsResponse, Skill, ) from litellm.types.utils import LlmProviders if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj class LiteLLMSkillsTransformationHandler: """ Transformation handler for skills API requests to LiteLLM database operations. This is used when custom_llm_provider="litellm_proxy" to store/retrieve skills from the LiteLLM proxy database instead of calling an external API. """ @property def custom_llm_provider(self) -> str: """Return the provider name for logging.""" return LlmProviders.LITELLM_PROXY.value def create_skill_handler( self, display_title: Optional[str] = None, description: Optional[str] = None, instructions: Optional[str] = None, files: Optional[List[Any]] = None, file_content: Optional[bytes] = None, file_name: Optional[str] = None, file_type: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, user_id: Optional[str] = None, _is_async: bool = False, logging_obj: Optional["LiteLLMLoggingObj"] = None, litellm_call_id: Optional[str] = None, **kwargs, ) -> Union[Skill, Coroutine[Any, Any, Skill]]: """ Create a skill in LiteLLM database. Args: display_title: Display title for the skill description: Description of the skill instructions: Instructions/prompt for the skill files: Files to upload - list of tuples (filename, content, content_type) file_content: Binary content of skill files (alternative to files) file_name: Original filename (alternative to files) file_type: MIME type (alternative to files) metadata: Additional metadata user_id: User ID for tracking _is_async: Whether to return a coroutine Returns: Skill object or coroutine that returns Skill """ # Pre-call logging if logging_obj: logging_obj.update_environment_variables( model=None, optional_params={"display_title": display_title}, litellm_params={"litellm_call_id": litellm_call_id}, custom_llm_provider=self.custom_llm_provider, ) # Extract file content from files parameter if provided # files is a list of tuples: [(filename, content, content_type), ...] if files and not file_content: if isinstance(files, list) and len(files) > 0: first_file = files[0] if isinstance(first_file, tuple) and len(first_file) >= 2: file_name = first_file[0] file_content = first_file[1] file_type = first_file[2] if len(first_file) > 2 else "application/zip" if _is_async: return self._async_create_skill( display_title=display_title, description=description, instructions=instructions, file_content=file_content, file_name=file_name, file_type=file_type, metadata=metadata, user_id=user_id, ) import asyncio return asyncio.get_event_loop().run_until_complete( self._async_create_skill( display_title=display_title, description=description, instructions=instructions, file_content=file_content, file_name=file_name, file_type=file_type, metadata=metadata, user_id=user_id, ) ) async def _async_create_skill( self, display_title: Optional[str] = None, description: Optional[str] = None, instructions: Optional[str] = None, file_content: Optional[bytes] = None, file_name: Optional[str] = None, file_type: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, user_id: Optional[str] = None, ) -> Skill: """Async implementation of create_skill.""" # Lazy import to avoid SDK dependency on proxy from litellm.llms.litellm_proxy.skills.handler import LiteLLMSkillsHandler from litellm.proxy._types import NewSkillRequest skill_request = NewSkillRequest( display_title=display_title, description=description, instructions=instructions, file_content=file_content, file_name=file_name, file_type=file_type, metadata=metadata, ) db_skill = await LiteLLMSkillsHandler.create_skill( data=skill_request, user_id=user_id, ) return self._db_skill_to_response(db_skill) def list_skills_handler( self, limit: int = 20, offset: int = 0, _is_async: bool = False, logging_obj: Optional["LiteLLMLoggingObj"] = None, litellm_call_id: Optional[str] = None, **kwargs, ) -> Union[ListSkillsResponse, Coroutine[Any, Any, ListSkillsResponse]]: """ List skills from LiteLLM database. Args: limit: Maximum number of skills to return offset: Number of skills to skip _is_async: Whether to return a coroutine logging_obj: LiteLLM logging object litellm_call_id: Call ID for logging Returns: ListSkillsResponse or coroutine that returns ListSkillsResponse """ # Pre-call logging if logging_obj: logging_obj.update_environment_variables( model=None, optional_params={"limit": limit, "offset": offset}, litellm_params={"litellm_call_id": litellm_call_id}, custom_llm_provider=self.custom_llm_provider, ) if _is_async: return self._async_list_skills(limit=limit, offset=offset) import asyncio return asyncio.get_event_loop().run_until_complete( self._async_list_skills(limit=limit, offset=offset) ) async def _async_list_skills( self, limit: int = 20, offset: int = 0, ) -> ListSkillsResponse: """Async implementation of list_skills.""" # Lazy import to avoid SDK dependency on proxy from litellm.llms.litellm_proxy.skills.handler import LiteLLMSkillsHandler db_skills = await LiteLLMSkillsHandler.list_skills( limit=limit, offset=offset, ) skills = [self._db_skill_to_response(s) for s in db_skills] return ListSkillsResponse( data=skills, has_more=len(skills) >= limit, next_page=None, ) def get_skill_handler( self, skill_id: str, _is_async: bool = False, logging_obj: Optional["LiteLLMLoggingObj"] = None, litellm_call_id: Optional[str] = None, **kwargs, ) -> Union[Skill, Coroutine[Any, Any, Skill]]: """ Get a skill from LiteLLM database. Args: skill_id: The skill ID to retrieve _is_async: Whether to return a coroutine logging_obj: LiteLLM logging object litellm_call_id: Call ID for logging Returns: Skill or coroutine that returns Skill """ # Pre-call logging if logging_obj: logging_obj.update_environment_variables( model=None, optional_params={"skill_id": skill_id}, litellm_params={"litellm_call_id": litellm_call_id}, custom_llm_provider=self.custom_llm_provider, ) if _is_async: return self._async_get_skill(skill_id=skill_id) import asyncio return asyncio.get_event_loop().run_until_complete( self._async_get_skill(skill_id=skill_id) ) async def _async_get_skill(self, skill_id: str) -> Skill: """Async implementation of get_skill.""" # Lazy import to avoid SDK dependency on proxy from litellm.llms.litellm_proxy.skills.handler import LiteLLMSkillsHandler db_skill = await LiteLLMSkillsHandler.get_skill(skill_id=skill_id) return self._db_skill_to_response(db_skill) def delete_skill_handler( self, skill_id: str, _is_async: bool = False, logging_obj: Optional["LiteLLMLoggingObj"] = None, litellm_call_id: Optional[str] = None, **kwargs, ) -> Union[DeleteSkillResponse, Coroutine[Any, Any, DeleteSkillResponse]]: """ Delete a skill from LiteLLM database. Args: skill_id: The skill ID to delete _is_async: Whether to return a coroutine logging_obj: LiteLLM logging object litellm_call_id: Call ID for logging Returns: DeleteSkillResponse or coroutine that returns DeleteSkillResponse """ # Pre-call logging if logging_obj: logging_obj.update_environment_variables( model=None, optional_params={"skill_id": skill_id}, litellm_params={"litellm_call_id": litellm_call_id}, custom_llm_provider=self.custom_llm_provider, ) if _is_async: return self._async_delete_skill(skill_id=skill_id) import asyncio return asyncio.get_event_loop().run_until_complete( self._async_delete_skill(skill_id=skill_id) ) async def _async_delete_skill(self, skill_id: str) -> DeleteSkillResponse: """Async implementation of delete_skill.""" # Lazy import to avoid SDK dependency on proxy from litellm.llms.litellm_proxy.skills.handler import LiteLLMSkillsHandler result = await LiteLLMSkillsHandler.delete_skill(skill_id=skill_id) return DeleteSkillResponse( id=result["id"], type=result.get("type", "skill_deleted"), ) def _db_skill_to_response(self, db_skill: Any) -> Skill: """ Convert a database skill record to Anthropic-compatible Skill response. Args: db_skill: LiteLLM_SkillsTable record Returns: Skill object """ created_at = "" updated_at = "" if hasattr(db_skill, "created_at") and db_skill.created_at: created_at = ( db_skill.created_at.isoformat() if hasattr(db_skill.created_at, "isoformat") else str(db_skill.created_at) ) if hasattr(db_skill, "updated_at") and db_skill.updated_at: updated_at = ( db_skill.updated_at.isoformat() if hasattr(db_skill.updated_at, "isoformat") else str(db_skill.updated_at) ) return Skill( id=db_skill.skill_id, created_at=created_at, updated_at=updated_at, display_title=db_skill.display_title, latest_version=db_skill.latest_version, source=db_skill.source or "custom", type="skill", )