"""
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",
)