tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-08-05 08:23:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Feat] New API - Claude Skills API (Anthropic) (#17042)

Browse Source 
* init readme

* init BaseSkillsAPIConfig

* init types for Skills APIs

* add feat: add create, list, retrieve skills

* add base skills config

* add BaseSkillsAPIConfig

* add get_provider_skills_api_config

* init skills

* add ANTHROPIC_SKILLS_API_BETA_VERSION

* init skills APIs

* working list, get skills

* working e2e skills API anthropic API

* add _prepare_skill_multipart_request

* add skills routes to llm api routes

* router _initialize_skills_endpoints

* add fix skills endpoints

* add convert_upload_files_to_file_data

* fix routing skills endpoints

* fix route llm request

* Potential fix for code scanning alert no. 3806: Clear-text logging of sensitive information

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* Potential fix for code scanning alert no. 3809: Clear-text logging of sensitive information

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* fix ruff checks

* test_initialize_skills_endpoints

* fix claude skills mypy linting errors

---------

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
This commit is contained in:
Ishaan Jaff
2025-11-24 15:01:40 -08:00
committed by GitHub
co-authored by Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
parent a807fe4450
commit 4e195d639e
25 changed files with 2821 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/__init__.py
+24
View File
@@ -1271,6 +1271,8 @@ from .llms.openai.chat.o_series_transformation import (
OpenAIOSeriesConfig as OpenAIO1Config, # maintain backwards compatibility
OpenAIOSeriesConfig,
)
from .llms.anthropic.skills.transformation import AnthropicSkillsConfig
from .llms.base_llm.skills.transformation import BaseSkillsAPIConfig
from .llms.gradient_ai.chat.transformation import GradientAIConfig
@@ -1367,6 +1369,18 @@ from .llms.cometapi.embed.transformation import CometAPIEmbeddingConfig
from .llms.lemonade.chat.transformation import LemonadeChatConfig
from .llms.snowflake.embedding.transformation import SnowflakeEmbeddingConfig
from .main import * # type: ignore
# Skills API
from .skills.main import (
create_skill,
acreate_skill,
list_skills,
alist_skills,
get_skill,
aget_skill,
delete_skill,
adelete_skill,
)
from .integrations import *
from .llms.custom_httpx.async_client_cleanup import close_litellm_async_clients
from .exceptions import (
@@ -1404,6 +1418,16 @@ from .batch_completion.main import * # type: ignore
from .rerank_api.main import *
from .llms.anthropic.experimental_pass_through.messages.handler import *
from .responses.main import *
from .skills.main import (
create_skill,
acreate_skill,
list_skills,
alist_skills,
get_skill,
aget_skill,
delete_skill,
adelete_skill,
)
from .containers.main import *
from .ocr.main import *
from .search.main import *
litellm/constants.py
+1
View File
@@ -291,6 +291,7 @@ DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE = os.getenv(
############### LLM Provider Constants ###############
### ANTHROPIC CONSTANTS ###
ANTHROPIC_SKILLS_API_BETA_VERSION = "skills-2025-10-02"
ANTHROPIC_WEB_SEARCH_TOOL_MAX_USES = {
"low": 1,
"medium": 5,
litellm/llms/anthropic/skills/__init__.py
+6
View File
@@ -0,0 +1,6 @@
"""Anthropic Skills API integration"""
from .transformation import AnthropicSkillsConfig
__all__ = ["AnthropicSkillsConfig"]
litellm/llms/anthropic/skills/readme.md
+17
View File
@@ -0,0 +1,17 @@
# Anthropic Skills API
This folder maintains the integration for the Anthropic Skills API.
You can do the following with the Anthropic Skills API:
1. Create a new skill
2. List all skills
3. Get a skill
4. Delete a skill
Versions:
- Create Skill Version
- List Skill Versions
- Get Skill Version
- Delete Skill Version
litellm/llms/anthropic/skills/transformation.py
+211
View File
@@ -0,0 +1,211 @@
"""
Anthropic Skills API configuration and transformations
"""
from typing import Any, Dict, Optional, Tuple
import httpx
from litellm._logging import verbose_logger
from litellm.llms.base_llm.skills.transformation import (
BaseSkillsAPIConfig,
LiteLLMLoggingObj,
)
from litellm.types.llms.anthropic_skills import (
CreateSkillRequest,
DeleteSkillResponse,
ListSkillsParams,
ListSkillsResponse,
Skill,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
class AnthropicSkillsConfig(BaseSkillsAPIConfig):
"""Anthropic-specific Skills API configuration"""
@property
def custom_llm_provider(self) -> LlmProviders:
return LlmProviders.ANTHROPIC
def validate_environment(
self, headers: dict, litellm_params: Optional[GenericLiteLLMParams]
) -> dict:
"""Add Anthropic-specific headers"""
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
# Get API key
api_key = None
if litellm_params:
api_key = litellm_params.api_key
api_key = AnthropicModelInfo.get_api_key(api_key)
if not api_key:
raise ValueError("ANTHROPIC_API_KEY is required for Skills API")
# Add required headers
headers["x-api-key"] = api_key
headers["anthropic-version"] = "2023-06-01"
# Add beta header for skills API
from litellm.constants import ANTHROPIC_SKILLS_API_BETA_VERSION
if "anthropic-beta" not in headers:
headers["anthropic-beta"] = ANTHROPIC_SKILLS_API_BETA_VERSION
elif isinstance(headers["anthropic-beta"], list):
if ANTHROPIC_SKILLS_API_BETA_VERSION not in headers["anthropic-beta"]:
headers["anthropic-beta"].append(ANTHROPIC_SKILLS_API_BETA_VERSION)
elif isinstance(headers["anthropic-beta"], str):
if ANTHROPIC_SKILLS_API_BETA_VERSION not in headers["anthropic-beta"]:
headers["anthropic-beta"] = [headers["anthropic-beta"], ANTHROPIC_SKILLS_API_BETA_VERSION]
headers["content-type"] = "application/json"
return headers
def get_complete_url(
self,
api_base: Optional[str],
endpoint: str,
skill_id: Optional[str] = None,
) -> str:
"""Get complete URL for Anthropic Skills API"""
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
if api_base is None:
api_base = AnthropicModelInfo.get_api_base()
if skill_id:
return f"{api_base}/v1/skills/{skill_id}?beta=true"
return f"{api_base}/v1/{endpoint}?beta=true"
def transform_create_skill_request(
self,
create_request: CreateSkillRequest,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Dict:
"""Transform create skill request for Anthropic"""
verbose_logger.debug(
"Transforming create skill request: %s", create_request
)
# Anthropic expects the request body directly
request_body = {k: v for k, v in create_request.items() if v is not None}
return request_body
def transform_create_skill_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> Skill:
"""Transform Anthropic response to Skill object"""
response_json = raw_response.json()
verbose_logger.debug(
"Transforming create skill response: %s", response_json
)
return Skill(**response_json)
def transform_list_skills_request(
self,
list_params: ListSkillsParams,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
"""Transform list skills request for Anthropic"""
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
api_base = AnthropicModelInfo.get_api_base(
litellm_params.api_base if litellm_params else None
)
url = self.get_complete_url(api_base=api_base, endpoint="skills")
# Build query parameters
query_params: Dict[str, Any] = {}
if "limit" in list_params and list_params["limit"]:
query_params["limit"] = list_params["limit"]
if "page" in list_params and list_params["page"]:
query_params["page"] = list_params["page"]
if "source" in list_params and list_params["source"]:
query_params["source"] = list_params["source"]
verbose_logger.debug(
"List skills request made to Anthropic Skills endpoint with params: %s", query_params
)
return url, query_params
def transform_list_skills_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ListSkillsResponse:
"""Transform Anthropic response to ListSkillsResponse"""
response_json = raw_response.json()
verbose_logger.debug(
"Transforming list skills response: %s", response_json
)
return ListSkillsResponse(**response_json)
def transform_get_skill_request(
self,
skill_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
"""Transform get skill request for Anthropic"""
url = self.get_complete_url(
api_base=api_base, endpoint="skills", skill_id=skill_id
)
verbose_logger.debug("Get skill request - URL: %s", url)
return url, headers
def transform_get_skill_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> Skill:
"""Transform Anthropic response to Skill object"""
response_json = raw_response.json()
verbose_logger.debug(
"Transforming get skill response: %s", response_json
)
return Skill(**response_json)
def transform_delete_skill_request(
self,
skill_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
"""Transform delete skill request for Anthropic"""
url = self.get_complete_url(
api_base=api_base, endpoint="skills", skill_id=skill_id
)
verbose_logger.debug("Delete skill request - URL: %s", url)
return url, headers
def transform_delete_skill_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> DeleteSkillResponse:
"""Transform Anthropic response to DeleteSkillResponse"""
response_json = raw_response.json()
verbose_logger.debug(
"Transforming delete skill response: %s", response_json
)
return DeleteSkillResponse(**response_json)
litellm/llms/base_llm/skills/__init__.py
+6
View File
@@ -0,0 +1,6 @@
"""Base Skills API configuration"""
from .transformation import BaseSkillsAPIConfig
__all__ = ["BaseSkillsAPIConfig"]
litellm/llms/base_llm/skills/transformation.py
+246
View File
@@ -0,0 +1,246 @@
"""
Base configuration class for Skills API
"""
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple
import httpx
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.llms.anthropic_skills import (
CreateSkillRequest,
DeleteSkillResponse,
ListSkillsParams,
ListSkillsResponse,
Skill,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
LiteLLMLoggingObj = _LiteLLMLoggingObj
else:
LiteLLMLoggingObj = Any
class BaseSkillsAPIConfig(ABC):
"""Base configuration for Skills API providers"""
def __init__(self):
pass
@property
@abstractmethod
def custom_llm_provider(self) -> LlmProviders:
pass
@abstractmethod
def validate_environment(
self, headers: dict, litellm_params: Optional[GenericLiteLLMParams]
) -> dict:
"""
Validate and update headers with provider-specific requirements
Args:
headers: Base headers dictionary
litellm_params: LiteLLM parameters
Returns:
Updated headers dictionary
"""
return headers
@abstractmethod
def get_complete_url(
self,
api_base: Optional[str],
endpoint: str,
skill_id: Optional[str] = None,
) -> str:
"""
Get the complete URL for the API request
Args:
api_base: Base API URL
endpoint: API endpoint (e.g., 'skills', 'skills/{id}')
skill_id: Optional skill ID for specific skill operations
Returns:
Complete URL
"""
if api_base is None:
raise ValueError("api_base is required")
return f"{api_base}/v1/{endpoint}"
@abstractmethod
def transform_create_skill_request(
self,
create_request: CreateSkillRequest,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Dict:
"""
Transform create skill request to provider-specific format
Args:
create_request: Skill creation parameters
litellm_params: LiteLLM parameters
headers: Request headers
Returns:
Provider-specific request body
"""
pass
@abstractmethod
def transform_create_skill_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> Skill:
"""
Transform provider response to Skill object
Args:
raw_response: Raw HTTP response
logging_obj: Logging object
Returns:
Skill object
"""
pass
@abstractmethod
def transform_list_skills_request(
self,
list_params: ListSkillsParams,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
"""
Transform list skills request parameters
Args:
list_params: List parameters (pagination, filters)
litellm_params: LiteLLM parameters
headers: Request headers
Returns:
Tuple of (url, query_params)
"""
pass
@abstractmethod
def transform_list_skills_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ListSkillsResponse:
"""
Transform provider response to ListSkillsResponse
Args:
raw_response: Raw HTTP response
logging_obj: Logging object
Returns:
ListSkillsResponse object
"""
pass
@abstractmethod
def transform_get_skill_request(
self,
skill_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
"""
Transform get skill request
Args:
skill_id: Skill ID
api_base: Base API URL
litellm_params: LiteLLM parameters
headers: Request headers
Returns:
Tuple of (url, headers)
"""
pass
@abstractmethod
def transform_get_skill_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> Skill:
"""
Transform provider response to Skill object
Args:
raw_response: Raw HTTP response
logging_obj: Logging object
Returns:
Skill object
"""
pass
@abstractmethod
def transform_delete_skill_request(
self,
skill_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
"""
Transform delete skill request
Args:
skill_id: Skill ID
api_base: Base API URL
litellm_params: LiteLLM parameters
headers: Request headers
Returns:
Tuple of (url, headers)
"""
pass
@abstractmethod
def transform_delete_skill_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> DeleteSkillResponse:
"""
Transform provider response to DeleteSkillResponse
Args:
raw_response: Raw HTTP response
logging_obj: Logging object
Returns:
DeleteSkillResponse object
"""
pass
def get_error_class(
self,
error_message: str,
status_code: int,
headers: dict,
) -> Exception:
"""Get appropriate error class for the provider."""
return BaseLLMException(
status_code=status_code,
message=error_message,
headers=headers,
)
litellm/llms/custom_httpx/llm_http_handler.py
+509 -7
View File
@@ -38,7 +38,6 @@ from litellm.llms.base_llm.google_genai.transformation import (
BaseGoogleGenAIGenerateContentConfig,
)
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
from .http_handler import get_shared_realtime_ssl_context
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
@@ -47,6 +46,7 @@ from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
from litellm.llms.base_llm.rerank.transformation import BaseRerankConfig
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.llms.base_llm.search.transformation import BaseSearchConfig, SearchResponse
from litellm.llms.base_llm.skills.transformation import BaseSkillsAPIConfig
from litellm.llms.base_llm.text_to_speech.transformation import BaseTextToSpeechConfig
from litellm.llms.base_llm.vector_store.transformation import BaseVectorStoreConfig
from litellm.llms.base_llm.vector_store_files.transformation import (
@@ -73,6 +73,11 @@ from litellm.types.containers.main import (
from litellm.types.llms.anthropic_messages.anthropic_response import (
AnthropicMessagesResponse,
)
from litellm.types.llms.anthropic_skills import (
DeleteSkillResponse,
ListSkillsResponse,
Skill,
)
from litellm.types.llms.openai import (
CreateBatchRequest,
CreateFileRequest,
@@ -90,12 +95,6 @@ from litellm.types.utils import (
LiteLLMBatch,
TranscriptionResponse,
)
from litellm.types.vector_stores import (
VectorStoreCreateOptionalRequestParams,
VectorStoreCreateResponse,
VectorStoreSearchOptionalRequestParams,
VectorStoreSearchResponse,
)
from litellm.types.vector_store_files import (
VectorStoreFileContentResponse,
VectorStoreFileCreateRequest,
@@ -105,6 +104,12 @@ from litellm.types.vector_store_files import (
VectorStoreFileObject,
VectorStoreFileUpdateRequest,
)
from litellm.types.vector_stores import (
VectorStoreCreateOptionalRequestParams,
VectorStoreCreateResponse,
VectorStoreSearchOptionalRequestParams,
VectorStoreSearchResponse,
)
from litellm.types.videos.main import VideoObject
from litellm.utils import (
CustomStreamWrapper,
@@ -113,6 +118,8 @@ from litellm.utils import (
ProviderConfigManager,
)
from .http_handler import get_shared_realtime_ssl_context
if TYPE_CHECKING:
from aiohttp import ClientSession
@@ -3554,6 +3561,7 @@ class BaseLLMHTTPHandler:
BaseVideoConfig,
BaseSearchConfig,
BaseTextToSpeechConfig,
BaseSkillsAPIConfig,
"BasePassthroughConfig",
"BaseContainerConfig",
],
@@ -7375,4 +7383,498 @@ class BaseLLMHTTPHandler:
model=model,
raw_response=response,
logging_obj=logging_obj,
)
#########################################################
########## SKILLS API HANDLERS ##########################
#########################################################
def _prepare_skill_multipart_request(
self,
request_body: Dict,
headers: dict,
) -> tuple[Optional[Dict], Optional[list]]:
"""
Helper to prepare multipart/form-data request for skills API.
Args:
request_body: Request body containing files and other fields
headers: Request headers
Returns:
Tuple of (data_dict, files_list) for multipart request, or (None, None) if no files
"""
if "files" not in request_body or not request_body["files"]:
return None, None
# Remove content-type header if present - httpx will set it automatically for multipart
if "content-type" in headers:
del headers["content-type"]
# Prepare files for multipart upload
files = []
for file_obj in request_body["files"]:
files.append(("files[]", file_obj))
# Prepare data (non-file fields)
data = {k: v for k, v in request_body.items() if k != "files"}
return data, files
def create_skill_handler(
self,
url: str,
request_body: Dict,
skills_api_provider_config: "BaseSkillsAPIConfig",
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
_is_async: bool = False,
shared_session: Optional["ClientSession"] = None,
) -> Union["Skill", Coroutine[Any, Any, "Skill"]]:
"""Create a skill"""
if _is_async:
return self.async_create_skill_handler(
url=url,
request_body=request_body,
skills_api_provider_config=skills_api_provider_config,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=logging_obj,
extra_headers=extra_headers,
timeout=timeout,
client=client,
shared_session=shared_session,
)
if client is None or not isinstance(client, HTTPHandler):
sync_httpx_client = _get_httpx_client(
params={"ssl_verify": litellm_params.get("ssl_verify", None)}
)
else:
sync_httpx_client = client
headers = extra_headers or {}
logging_obj.pre_call(
input=request_body.get("display_title", ""),
api_key="",
additional_args={
"complete_input_dict": request_body,
"api_base": url,
"headers": headers,
},
)
try:
# Check if files are present - use multipart/form-data
data, files = self._prepare_skill_multipart_request(
request_body=request_body, headers=headers
)
if files is not None:
response = sync_httpx_client.post(
url=url, headers=headers, data=data, files=files, timeout=timeout
)
else:
# No files - send as JSON
response = sync_httpx_client.post(
url=url, headers=headers, json=request_body, timeout=timeout
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=skills_api_provider_config,
)
return skills_api_provider_config.transform_create_skill_response(
raw_response=response,
logging_obj=logging_obj,
)
async def async_create_skill_handler(
self,
url: str,
request_body: Dict,
skills_api_provider_config: "BaseSkillsAPIConfig",
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
shared_session: Optional["ClientSession"] = None,
) -> "Skill":
"""Async create a skill"""
if client is None or not isinstance(client, AsyncHTTPHandler):
async_httpx_client = get_async_httpx_client(
llm_provider=litellm.LlmProviders(custom_llm_provider),
params={"ssl_verify": litellm_params.get("ssl_verify", None)},
)
else:
async_httpx_client = client
headers = extra_headers or {}
logging_obj.pre_call(
input=request_body.get("display_title", ""),
api_key="",
additional_args={
"complete_input_dict": request_body,
"api_base": url,
"headers": headers,
},
)
try:
# Check if files are present - use multipart/form-data
data, files = self._prepare_skill_multipart_request(
request_body=request_body, headers=headers
)
if files is not None:
response = await async_httpx_client.post(
url=url, headers=headers, data=data, files=files, timeout=timeout
)
else:
# No files - send as JSON
response = await async_httpx_client.post(
url=url, headers=headers, json=request_body, timeout=timeout
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=skills_api_provider_config,
)
return skills_api_provider_config.transform_create_skill_response(
raw_response=response,
logging_obj=logging_obj,
)
def list_skills_handler(
self,
url: str,
query_params: Dict,
skills_api_provider_config: "BaseSkillsAPIConfig",
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
_is_async: bool = False,
shared_session: Optional["ClientSession"] = None,
) -> Union["ListSkillsResponse", Coroutine[Any, Any, "ListSkillsResponse"]]:
"""List skills"""
if _is_async:
return self.async_list_skills_handler(
url=url,
query_params=query_params,
skills_api_provider_config=skills_api_provider_config,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=logging_obj,
extra_headers=extra_headers,
timeout=timeout,
client=client,
shared_session=shared_session,
)
if client is None or not isinstance(client, HTTPHandler):
sync_httpx_client = _get_httpx_client(
params={"ssl_verify": litellm_params.get("ssl_verify", None)}
)
else:
sync_httpx_client = client
headers = extra_headers or {}
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"complete_input_dict": query_params,
"api_base": url,
"headers": headers,
},
)
try:
response = sync_httpx_client.get(
url=url, headers=headers, params=query_params
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=skills_api_provider_config,
)
return skills_api_provider_config.transform_list_skills_response(
raw_response=response,
logging_obj=logging_obj,
)
async def async_list_skills_handler(
self,
url: str,
query_params: Dict,
skills_api_provider_config: "BaseSkillsAPIConfig",
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
shared_session: Optional["ClientSession"] = None,
) -> "ListSkillsResponse":
"""Async list skills"""
if client is None or not isinstance(client, AsyncHTTPHandler):
async_httpx_client = get_async_httpx_client(
llm_provider=litellm.LlmProviders(custom_llm_provider),
params={"ssl_verify": litellm_params.get("ssl_verify", None)},
)
else:
async_httpx_client = client
headers = extra_headers or {}
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"complete_input_dict": query_params,
"api_base": url,
"headers": headers,
},
)
try:
response = await async_httpx_client.get(
url=url, headers=headers, params=query_params
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=skills_api_provider_config,
)
return skills_api_provider_config.transform_list_skills_response(
raw_response=response,
logging_obj=logging_obj,
)
def get_skill_handler(
self,
url: str,
skills_api_provider_config: "BaseSkillsAPIConfig",
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
_is_async: bool = False,
shared_session: Optional["ClientSession"] = None,
) -> Union["Skill", Coroutine[Any, Any, "Skill"]]:
"""Get a skill"""
if _is_async:
return self.async_get_skill_handler(
url=url,
skills_api_provider_config=skills_api_provider_config,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=logging_obj,
extra_headers=extra_headers,
timeout=timeout,
client=client,
shared_session=shared_session,
)
if client is None or not isinstance(client, HTTPHandler):
sync_httpx_client = _get_httpx_client(
params={"ssl_verify": litellm_params.get("ssl_verify", None)}
)
else:
sync_httpx_client = client
headers = extra_headers or {}
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
},
)
try:
response = sync_httpx_client.get(url=url, headers=headers)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=skills_api_provider_config,
)
return skills_api_provider_config.transform_get_skill_response(
raw_response=response,
logging_obj=logging_obj,
)
async def async_get_skill_handler(
self,
url: str,
skills_api_provider_config: "BaseSkillsAPIConfig",
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
shared_session: Optional["ClientSession"] = None,
) -> "Skill":
"""Async get a skill"""
if client is None or not isinstance(client, AsyncHTTPHandler):
async_httpx_client = get_async_httpx_client(
llm_provider=litellm.LlmProviders(custom_llm_provider),
params={"ssl_verify": litellm_params.get("ssl_verify", None)},
)
else:
async_httpx_client = client
headers = extra_headers or {}
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
},
)
try:
response = await async_httpx_client.get(
url=url, headers=headers
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=skills_api_provider_config,
)
return skills_api_provider_config.transform_get_skill_response(
raw_response=response,
logging_obj=logging_obj,
)
def delete_skill_handler(
self,
url: str,
skills_api_provider_config: "BaseSkillsAPIConfig",
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
_is_async: bool = False,
shared_session: Optional["ClientSession"] = None,
) -> Union["DeleteSkillResponse", Coroutine[Any, Any, "DeleteSkillResponse"]]:
"""Delete a skill"""
if _is_async:
return self.async_delete_skill_handler(
url=url,
skills_api_provider_config=skills_api_provider_config,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=logging_obj,
extra_headers=extra_headers,
timeout=timeout,
client=client,
shared_session=shared_session,
)
if client is None or not isinstance(client, HTTPHandler):
sync_httpx_client = _get_httpx_client(
params={"ssl_verify": litellm_params.get("ssl_verify", None)}
)
else:
sync_httpx_client = client
headers = extra_headers or {}
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
},
)
try:
response = sync_httpx_client.delete(
url=url, headers=headers, timeout=timeout
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=skills_api_provider_config,
)
return skills_api_provider_config.transform_delete_skill_response(
raw_response=response,
logging_obj=logging_obj,
)
async def async_delete_skill_handler(
self,
url: str,
skills_api_provider_config: "BaseSkillsAPIConfig",
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
shared_session: Optional["ClientSession"] = None,
) -> "DeleteSkillResponse":
"""Async delete a skill"""
if client is None or not isinstance(client, AsyncHTTPHandler):
async_httpx_client = get_async_httpx_client(
llm_provider=litellm.LlmProviders(custom_llm_provider),
params={"ssl_verify": litellm_params.get("ssl_verify", None)},
)
else:
async_httpx_client = client
headers = extra_headers or {}
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
},
)
try:
response = await async_httpx_client.delete(
url=url, headers=headers, timeout=timeout
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=skills_api_provider_config,
)
return skills_api_provider_config.transform_delete_skill_response(
raw_response=response,
logging_obj=logging_obj,
)
litellm/proxy/_types.py
+27 -4
View File
@@ -380,6 +380,8 @@ class LiteLLMRoutes(enum.Enum):
anthropic_routes = [
"/v1/messages",
"/v1/messages/count_tokens",
"/v1/skills",
"/v1/skills/{skill_id}",
]
mcp_routes = [
@@ -812,7 +814,6 @@ class KeyRequestBase(GenerateRequestBase):
key: Optional[str] = None
budget_id: Optional[str] = None
tags: Optional[List[str]] = None
disable_global_guardrails: Optional[bool] = None
enforced_params: Optional[List[str]] = None
allowed_routes: Optional[list] = []
allowed_passthrough_routes: Optional[list] = None
@@ -1358,7 +1359,6 @@ class NewTeamRequest(TeamBase):
prompts: Optional[List[str]] = None
object_permission: Optional[LiteLLM_ObjectPermissionBase] = None
allowed_passthrough_routes: Optional[list] = None
disable_global_guardrails: Optional[bool] = None
model_rpm_limit: Optional[Dict[str, int]] = None
rpm_limit_type: Optional[
Literal["guaranteed_throughput", "best_effort_throughput"]
@@ -1420,7 +1420,6 @@ class UpdateTeamRequest(LiteLLMPydanticObjectBase):
model_aliases: Optional[dict] = None
guardrails: Optional[List[str]] = None
object_permission: Optional[LiteLLM_ObjectPermissionBase] = None
disable_global_guardrails: Optional[bool] = None
team_member_budget: Optional[float] = None
team_member_rpm_limit: Optional[int] = None
team_member_tpm_limit: Optional[int] = None
@@ -1686,6 +1685,27 @@ class DynamoDBArgs(LiteLLMPydanticObjectBase):
assume_role_aws_session_name: Optional[str] = None
class PassThroughGuardrailConfig(LiteLLMPydanticObjectBase):
"""
Configuration for guardrails on passthrough endpoints.
Passthrough endpoints are opt-in only for guardrails. Guardrails configured at
org/team/key levels will NOT execute unless explicitly enabled here.
"""
enabled: bool = Field(
default=False,
description="Whether to execute guardrails for this passthrough endpoint. When True, all org/team/key level guardrails will execute along with any passthrough-specific guardrails. When False (default), NO guardrails execute.",
)
specific: Optional[List[str]] = Field(
default=None,
description="Optional list of guardrail names that are specific to this passthrough endpoint. These will execute in addition to org/team/key level guardrails when enabled=True.",
)
target_fields: Optional[List[str]] = Field(
default=None,
description="Optional list of JSON paths to target specific fields for guardrail execution. Examples: 'messages[*].content', 'input', 'messages[?(@.role=='user')].content'. If not specified, guardrails execute on entire payload.",
)
class PassThroughGenericEndpoint(LiteLLMPydanticObjectBase):
id: Optional[str] = Field(
default=None,
@@ -1711,6 +1731,10 @@ class PassThroughGenericEndpoint(LiteLLMPydanticObjectBase):
default=False,
description="Whether authentication is required for the pass-through endpoint. If True, requests to the endpoint will require a valid LiteLLM API key.",
)
guardrails: Optional[PassThroughGuardrailConfig] = Field(
default=None,
description="Guardrail configuration for this passthrough endpoint. When enabled, org/team/key level guardrails will execute along with any passthrough-specific guardrails. Defaults to disabled (no guardrails execute).",
)
class PassThroughEndpointResponse(LiteLLMPydanticObjectBase):
@@ -3260,7 +3284,6 @@ LiteLLM_ManagementEndpoint_MetadataFields = [
]
LiteLLM_ManagementEndpoint_MetadataFields_Premium = [
"disable_global_guardrails",
"guardrails",
"tags",
"team_member_key_duration",
litellm/proxy/anthropic_endpoints/skills_endpoints.py
+438
View File
@@ -0,0 +1,438 @@
"""
Anthropic Skills API endpoints - /v1/skills
"""
from typing import Optional
import orjson
from fastapi import APIRouter, Depends, Request, Response
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
from litellm.proxy.common_utils.http_parsing_utils import (
convert_upload_files_to_file_data,
get_form_data,
)
from litellm.types.llms.anthropic_skills import (
DeleteSkillResponse,
ListSkillsResponse,
Skill,
)
router = APIRouter()
@router.post(
"/v1/skills",
tags=["[beta] Anthropic Skills API"],
dependencies=[Depends(user_api_key_auth)],
response_model=Skill,
)
async def create_skill(
fastapi_response: Response,
request: Request,
custom_llm_provider: Optional[str] = "anthropic",
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Create a new skill on Anthropic.
Requires `?beta=true` query parameter.
Model-based routing (for multi-account support):
- Pass model via header: `x-litellm-model: claude-account-1`
- Pass model via query: `?model=claude-account-1`
- Pass model via form field: `model=claude-account-1`
Example usage:
```bash
# Basic usage
curl -X POST "http://localhost:4000/v1/skills?beta=true" \
-H "Content-Type: multipart/form-data" \
-H "Authorization: Bearer your-key" \
-F "display_title=My Skill" \
-F "files[]=@skill.zip"
# With model-based routing
curl -X POST "http://localhost:4000/v1/skills?beta=true" \
-H "Content-Type: multipart/form-data" \
-H "Authorization: Bearer your-key" \
-H "x-litellm-model: claude-account-1" \
-F "display_title=My Skill" \
-F "files[]=@skill.zip"
```
Returns: Skill object with id, display_title, etc.
"""
from litellm.proxy.proxy_server import (
general_settings,
llm_router,
proxy_config,
proxy_logging_obj,
select_data_generator,
user_api_base,
user_max_tokens,
user_model,
user_request_timeout,
user_temperature,
version,
)
# Read form data and convert UploadFile objects to file data tuples
form_data = await get_form_data(request)
data = await convert_upload_files_to_file_data(form_data)
# Extract model for routing (header > query > body)
model = (
data.get("model")
or request.query_params.get("model")
or request.headers.get("x-litellm-model")
)
if model:
data["model"] = model
if "custom_llm_provider" not in data:
data["custom_llm_provider"] = custom_llm_provider
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="acreate_skill",
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
proxy_config=proxy_config,
select_data_generator=select_data_generator,
model=data.get("model"),
user_model=user_model,
user_temperature=user_temperature,
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
version=version,
)
except Exception as e:
raise await processor._handle_llm_api_exception(
e=e,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
@router.get(
"/v1/skills",
tags=["[beta] Anthropic Skills API"],
dependencies=[Depends(user_api_key_auth)],
response_model=ListSkillsResponse,
)
async def list_skills(
fastapi_response: Response,
request: Request,
limit: Optional[int] = 10,
after_id: Optional[str] = None,
before_id: Optional[str] = None,
custom_llm_provider: Optional[str] = "anthropic",
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
List skills on Anthropic.
Requires `?beta=true` query parameter.
Model-based routing (for multi-account support):
- Pass model via header: `x-litellm-model: claude-account-1`
- Pass model via query: `?model=claude-account-1`
- Pass model via body: `{"model": "claude-account-1"}`
Example usage:
```bash
# Basic usage
curl "http://localhost:4000/v1/skills?beta=true&limit=10" \
-H "Authorization: Bearer your-key"
# With model-based routing
curl "http://localhost:4000/v1/skills?beta=true&limit=10" \
-H "Authorization: Bearer your-key" \
-H "x-litellm-model: claude-account-1"
```
Returns: ListSkillsResponse with list of skills
"""
from litellm.proxy.proxy_server import (
general_settings,
llm_router,
proxy_config,
proxy_logging_obj,
select_data_generator,
user_api_base,
user_max_tokens,
user_model,
user_request_timeout,
user_temperature,
version,
)
# Read request body
body = await request.body()
data = orjson.loads(body) if body else {}
# Use query params if not in body
if "limit" not in data and limit is not None:
data["limit"] = limit
if "after_id" not in data and after_id is not None:
data["after_id"] = after_id
if "before_id" not in data and before_id is not None:
data["before_id"] = before_id
# Extract model for routing (header > query > body)
model = (
data.get("model")
or request.query_params.get("model")
or request.headers.get("x-litellm-model")
)
if model:
data["model"] = model
# Set custom_llm_provider: body > query param > default
if "custom_llm_provider" not in data:
data["custom_llm_provider"] = custom_llm_provider
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="alist_skills",
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
proxy_config=proxy_config,
select_data_generator=select_data_generator,
model=data.get("model"),
user_model=user_model,
user_temperature=user_temperature,
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
version=version,
)
except Exception as e:
raise await processor._handle_llm_api_exception(
e=e,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
@router.get(
"/v1/skills/{skill_id}",
tags=["[beta] Anthropic Skills API"],
dependencies=[Depends(user_api_key_auth)],
response_model=Skill,
)
async def get_skill(
skill_id: str,
fastapi_response: Response,
request: Request,
custom_llm_provider: Optional[str] = "anthropic",
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Get a specific skill by ID from Anthropic.
Requires `?beta=true` query parameter.
Model-based routing (for multi-account support):
- Pass model via header: `x-litellm-model: claude-account-1`
- Pass model via query: `?model=claude-account-1`
- Pass model via body: `{"model": "claude-account-1"}`
Example usage:
```bash
# Basic usage
curl "http://localhost:4000/v1/skills/skill_123?beta=true" \
-H "Authorization: Bearer your-key"
# With model-based routing
curl "http://localhost:4000/v1/skills/skill_123?beta=true" \
-H "Authorization: Bearer your-key" \
-H "x-litellm-model: claude-account-1"
```
Returns: Skill object
"""
from litellm.proxy.proxy_server import (
general_settings,
llm_router,
proxy_config,
proxy_logging_obj,
select_data_generator,
user_api_base,
user_max_tokens,
user_model,
user_request_timeout,
user_temperature,
version,
)
# Read request body
body = await request.body()
data = orjson.loads(body) if body else {}
# Set skill_id from path parameter
data["skill_id"] = skill_id
# Extract model for routing (header > query > body)
model = (
data.get("model")
or request.query_params.get("model")
or request.headers.get("x-litellm-model")
)
if model:
data["model"] = model
# Set custom_llm_provider: body > query param > default
if "custom_llm_provider" not in data:
data["custom_llm_provider"] = custom_llm_provider
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="aget_skill",
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
proxy_config=proxy_config,
select_data_generator=select_data_generator,
model=data.get("model"),
user_model=user_model,
user_temperature=user_temperature,
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
version=version,
)
except Exception as e:
raise await processor._handle_llm_api_exception(
e=e,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
@router.delete(
"/v1/skills/{skill_id}",
tags=["[beta] Anthropic Skills API"],
dependencies=[Depends(user_api_key_auth)],
response_model=DeleteSkillResponse,
)
async def delete_skill(
skill_id: str,
fastapi_response: Response,
request: Request,
custom_llm_provider: Optional[str] = "anthropic",
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Delete a skill by ID from Anthropic.
Requires `?beta=true` query parameter.
Note: Anthropic does not allow deleting skills with existing versions.
Model-based routing (for multi-account support):
- Pass model via header: `x-litellm-model: claude-account-1`
- Pass model via query: `?model=claude-account-1`
- Pass model via body: `{"model": "claude-account-1"}`
Example usage:
```bash
# Basic usage
curl -X DELETE "http://localhost:4000/v1/skills/skill_123?beta=true" \
-H "Authorization: Bearer your-key"
# With model-based routing
curl -X DELETE "http://localhost:4000/v1/skills/skill_123?beta=true" \
-H "Authorization: Bearer your-key" \
-H "x-litellm-model: claude-account-1"
```
Returns: DeleteSkillResponse with type="skill_deleted"
"""
from litellm.proxy.proxy_server import (
general_settings,
llm_router,
proxy_config,
proxy_logging_obj,
select_data_generator,
user_api_base,
user_max_tokens,
user_model,
user_request_timeout,
user_temperature,
version,
)
# Read request body
body = await request.body()
data = orjson.loads(body) if body else {}
# Set skill_id from path parameter
data["skill_id"] = skill_id
# Extract model for routing (header > query > body)
model = (
data.get("model")
or request.query_params.get("model")
or request.headers.get("x-litellm-model")
)
if model:
data["model"] = model
# Set custom_llm_provider: body > query param > default
if "custom_llm_provider" not in data:
data["custom_llm_provider"] = custom_llm_provider
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="adelete_skill",
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
proxy_config=proxy_config,
select_data_generator=select_data_generator,
model=data.get("model"),
user_model=user_model,
user_temperature=user_temperature,
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
version=version,
)
except Exception as e:
raise await processor._handle_llm_api_exception(
e=e,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
litellm/proxy/common_request_processing.py
+8
View File
@@ -332,6 +332,10 @@ class ProxyBaseLLMRequestProcessing:
"alist_containers",
"aretrieve_container",
"adelete_container",
"acreate_skill",
"alist_skills",
"aget_skill",
"adelete_skill",
],
version: Optional[str] = None,
user_model: Optional[str] = None,
@@ -450,6 +454,10 @@ class ProxyBaseLLMRequestProcessing:
"alist_containers",
"aretrieve_container",
"adelete_container",
"acreate_skill",
"alist_skills",
"aget_skill",
"adelete_skill",
],
proxy_logging_obj: ProxyLogging,
general_settings: dict,
litellm/proxy/common_utils/http_parsing_utils.py
+45
View File
@@ -232,6 +232,51 @@ async def get_form_data(request: Request) -> Dict[str, Any]:
return parsed_form_data
async def convert_upload_files_to_file_data(
form_data: Dict[str, Any]
) -> Dict[str, Any]:
"""
Convert FastAPI UploadFile objects to file data tuples for litellm.
Converts UploadFile objects to tuples of (filename, content, content_type)
which is the format expected by httpx and litellm's HTTP handlers.
Args:
form_data: Dictionary containing form data with potential UploadFile objects
Returns:
Dictionary with UploadFile objects converted to file data tuples
Example:
```python
form_data = await get_form_data(request)
data = await convert_upload_files_to_file_data(form_data)
# data["files"] is now [(filename, content, content_type), ...]
```
"""
data = {}
for key, value in form_data.items():
if isinstance(value, list):
# Check if it's a list of UploadFile objects
if value and hasattr(value[0], "read"):
files = []
for f in value:
file_content = await f.read()
# Create tuple: (filename, content, content_type)
files.append((f.filename, file_content, f.content_type))
data[key] = files
else:
data[key] = value
elif hasattr(value, "read"):
# Single UploadFile object - read and convert to list for consistency
file_content = await value.read()
data[key] = [(value.filename, file_content, value.content_type)]
else:
# Regular form field
data[key] = value
return data
async def get_request_body(request: Request) -> Dict[str, Any]:
"""
Read the request body and parse it as JSON.
litellm/proxy/proxy_server.py
+13 -3
View File
@@ -190,6 +190,9 @@ from litellm.proxy.analytics_endpoints.analytics_endpoints import (
router as analytics_router,
)
from litellm.proxy.anthropic_endpoints.endpoints import router as anthropic_router
from litellm.proxy.anthropic_endpoints.skills_endpoints import (
router as anthropic_skills_router,
)
from litellm.proxy.auth.auth_checks import (
ExperimentalUIJWTToken,
get_team_object,
@@ -283,7 +286,9 @@ from litellm.proxy.management_endpoints.customer_endpoints import (
from litellm.proxy.management_endpoints.internal_user_endpoints import (
router as internal_user_router,
)
from litellm.proxy.management_endpoints.internal_user_endpoints import user_update
from litellm.proxy.management_endpoints.internal_user_endpoints import (
user_update,
)
from litellm.proxy.management_endpoints.key_management_endpoints import (
delete_verification_tokens,
duration_in_seconds,
@@ -337,7 +342,9 @@ from litellm.proxy.ocr_endpoints.endpoints import router as ocr_router
from litellm.proxy.openai_files_endpoints.files_endpoints import (
router as openai_files_router,
)
from litellm.proxy.openai_files_endpoints.files_endpoints import set_files_config
from litellm.proxy.openai_files_endpoints.files_endpoints import (
set_files_config,
)
from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
passthrough_endpoint_router,
)
@@ -428,7 +435,9 @@ from litellm.types.proxy.management_endpoints.ui_sso import (
LiteLLM_UpperboundKeyGenerateParams,
)
from litellm.types.realtime import RealtimeQueryParams
from litellm.types.router import DeploymentTypedDict
from litellm.types.router import (
DeploymentTypedDict,
)
from litellm.types.router import ModelInfo as RouterModelInfo
from litellm.types.router import (
RouterGeneralSettings,
@@ -10140,6 +10149,7 @@ app.include_router(credential_router)
app.include_router(llm_passthrough_router)
app.include_router(mcp_management_router)
app.include_router(anthropic_router)
app.include_router(anthropic_skills_router)
app.include_router(google_router)
app.include_router(langfuse_router)
app.include_router(pass_through_router)
litellm/proxy/route_llm_request.py
+13 -1
View File
@@ -36,6 +36,10 @@ ROUTE_ENDPOINT_MAPPING = {
"alist_containers": "/containers",
"aretrieve_container": "/containers/{container_id}",
"adelete_container": "/containers/{container_id}",
"acreate_skill": "/skills",
"alist_skills": "/skills",
"aget_skill": "/skills/{skill_id}",
"adelete_skill": "/skills/{skill_id}",
}
@@ -126,6 +130,10 @@ async def route_request(
"alist_containers",
"aretrieve_container",
"adelete_container",
"acreate_skill",
"alist_skills",
"aget_skill",
"adelete_skill",
],
):
"""
@@ -178,8 +186,12 @@ async def route_request(
"avector_store_file_retrieve",
"avector_store_file_content",
"avector_store_file_delete",
"acreate_skill",
"alist_skills",
"aget_skill",
"adelete_skill",
] and (data.get("model") is None or data.get("model") == ""):
# These video endpoints don't need a model, use custom_llm_provider
# These endpoints don't need a model, use custom_llm_provider directly
return getattr(litellm, f"{route_type}")(**data)
team_model_name = (
litellm/router.py
+25 -1
View File
@@ -1035,14 +1035,30 @@ class Router:
delete_container, call_type="delete_container"
)
def _initialize_skills_endpoints(self):
"""Initialize Anthropic Skills API endpoints."""
self.acreate_skill = self.factory_function(
litellm.acreate_skill, call_type="acreate_skill"
)
self.alist_skills = self.factory_function(
litellm.alist_skills, call_type="alist_skills"
)
self.aget_skill = self.factory_function(
litellm.aget_skill, call_type="aget_skill"
)
self.adelete_skill = self.factory_function(
litellm.adelete_skill, call_type="adelete_skill"
)
def _initialize_specialized_endpoints(self):
"""Helper to initialize specialized router endpoints (vector store, OCR, search, video, container)."""
"""Helper to initialize specialized router endpoints (vector store, OCR, search, video, container, skills)."""
self._initialize_vector_store_endpoints()
self._initialize_vector_store_file_endpoints()
self._initialize_google_genai_endpoints()
self._initialize_ocr_search_endpoints()
self._initialize_video_endpoints()
self._initialize_container_endpoints()
self._initialize_skills_endpoints()
def initialize_router_endpoints(self):
self._initialize_core_endpoints()
@@ -3817,6 +3833,10 @@ class Router:
"retrieve_container",
"adelete_container",
"delete_container",
"acreate_skill",
"alist_skills",
"aget_skill",
"adelete_skill",
] = "assistants",
):
"""
@@ -3937,6 +3957,10 @@ class Router:
"aretrieve_container",
"adelete_container",
"acancel_batch",
"acreate_skill",
"alist_skills",
"aget_skill",
"adelete_skill",
):
return await self._ageneric_api_call_with_fallbacks(
original_function=original_function,
litellm/skills/__init__.py
+24
View File
@@ -0,0 +1,24 @@
"""Skills API integration for LiteLLM"""
from .main import (
acreate_skill,
adelete_skill,
aget_skill,
alist_skills,
create_skill,
delete_skill,
get_skill,
list_skills,
)
__all__ = [
"create_skill",
"acreate_skill",
"list_skills",
"alist_skills",
"get_skill",
"aget_skill",
"delete_skill",
"adelete_skill",
]
litellm/skills/main.py
+705
View File
@@ -0,0 +1,705 @@
"""
Main entry point for Skills API operations
Provides create, list, get, and delete operations for skills
"""
import asyncio
import contextvars
from functools import partial
from typing import Any, Coroutine, Dict, List, Optional, Union
import httpx
import litellm
from litellm.constants import request_timeout
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.skills.transformation import BaseSkillsAPIConfig
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.types.llms.anthropic_skills import (
CreateSkillRequest,
DeleteSkillResponse,
ListSkillsParams,
ListSkillsResponse,
Skill,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.utils import ProviderConfigManager, client
# Initialize HTTP handler
base_llm_http_handler = BaseLLMHTTPHandler()
DEFAULT_ANTHROPIC_API_BASE = "https://api.anthropic.com/v1"
@client
async def acreate_skill(
files: Optional[List[Any]] = None,
display_title: Optional[str] = None,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
custom_llm_provider: Optional[str] = None,
**kwargs,
) -> Skill:
"""
Async: Create a new skill
Args:
files: Files to upload for the skill. All files must be in the same top-level directory and must include a SKILL.md file at the root.
display_title: Optional display title for the skill
extra_headers: Additional headers for the request
extra_query: Additional query parameters
extra_body: Additional body parameters
timeout: Request timeout
custom_llm_provider: Provider name (e.g., 'anthropic')
**kwargs: Additional parameters
Returns:
Skill object
"""
local_vars = locals()
try:
loop = asyncio.get_event_loop()
kwargs["acreate_skill"] = True
func = partial(
create_skill,
files=files,
display_title=display_title,
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
**kwargs,
)
ctx = contextvars.copy_context()
func_with_context = partial(ctx.run, func)
init_response = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
except Exception as e:
raise litellm.exception_type(
model=None,
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
@client
def create_skill(
files: Optional[List[Any]] = None,
display_title: Optional[str] = None,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
custom_llm_provider: Optional[str] = None,
**kwargs,
) -> Union[Skill, Coroutine[Any, Any, Skill]]:
"""
Create a new skill
Args:
files: Files to upload for the skill. All files must be in the same top-level directory and must include a SKILL.md file at the root.
display_title: Optional display title for the skill
extra_headers: Additional headers for the request
extra_query: Additional query parameters
extra_body: Additional body parameters
timeout: Request timeout
custom_llm_provider: Provider name (e.g., 'anthropic')
**kwargs: Additional parameters
Returns:
Skill object
"""
local_vars = locals()
try:
litellm_logging_obj: LiteLLMLoggingObj = kwargs.get("litellm_logging_obj") # type: ignore
litellm_call_id: Optional[str] = kwargs.get("litellm_call_id", None)
_is_async = kwargs.pop("acreate_skill", False) is True
# Get LiteLLM parameters
litellm_params = GenericLiteLLMParams(**kwargs)
# Determine provider
if custom_llm_provider is None:
custom_llm_provider = "anthropic"
# Get provider config
skills_api_provider_config: Optional[BaseSkillsAPIConfig] = (
ProviderConfigManager.get_provider_skills_api_config(
provider=litellm.LlmProviders(custom_llm_provider),
)
)
if skills_api_provider_config is None:
raise ValueError(
f"CREATE skill is not supported for {custom_llm_provider}"
)
# Build create request
create_request: CreateSkillRequest = {}
if display_title is not None:
create_request["display_title"] = display_title
if files is not None:
create_request["files"] = files
# Merge extra_body if provided
if extra_body:
create_request.update(extra_body) # type: ignore
# Validate environment and get headers
headers = extra_headers or {}
headers = skills_api_provider_config.validate_environment(
headers=headers, litellm_params=litellm_params
)
# Transform request
request_body = skills_api_provider_config.transform_create_skill_request(
create_request=create_request,
litellm_params=litellm_params,
headers=headers,
)
# Get API base and URL
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
api_base = AnthropicModelInfo.get_api_base(litellm_params.api_base)
url = skills_api_provider_config.get_complete_url(
api_base=api_base, endpoint="skills"
)
# Pre-call logging
litellm_logging_obj.update_environment_variables(
model=None,
optional_params=request_body,
litellm_params={
"litellm_call_id": litellm_call_id,
},
custom_llm_provider=custom_llm_provider,
)
# Make HTTP request
response = base_llm_http_handler.create_skill_handler(
url=url,
request_body=request_body,
skills_api_provider_config=skills_api_provider_config,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
extra_headers=headers,
timeout=timeout or request_timeout,
_is_async=_is_async,
client=kwargs.get("client"),
shared_session=kwargs.get("shared_session"),
)
return response
except Exception as e:
raise litellm.exception_type(
model=None,
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
@client
async def alist_skills(
limit: Optional[int] = None,
page: Optional[str] = None,
source: Optional[str] = None,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
custom_llm_provider: Optional[str] = None,
**kwargs,
) -> ListSkillsResponse:
"""
Async: List all skills
Args:
limit: Number of results to return per page (max 100, default 20)
page: Pagination token for fetching a specific page of results
source: Filter skills by source ('custom' or 'anthropic')
extra_headers: Additional headers for the request
extra_query: Additional query parameters
timeout: Request timeout
custom_llm_provider: Provider name (e.g., 'anthropic')
**kwargs: Additional parameters
Returns:
ListSkillsResponse object
"""
local_vars = locals()
try:
loop = asyncio.get_event_loop()
kwargs["alist_skills"] = True
func = partial(
list_skills,
limit=limit,
page=page,
source=source,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
**kwargs,
)
ctx = contextvars.copy_context()
func_with_context = partial(ctx.run, func)
init_response = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
except Exception as e:
raise litellm.exception_type(
model=None,
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
@client
def list_skills(
limit: Optional[int] = None,
page: Optional[str] = None,
source: Optional[str] = None,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
custom_llm_provider: Optional[str] = None,
**kwargs,
) -> Union[ListSkillsResponse, Coroutine[Any, Any, ListSkillsResponse]]:
"""
List all skills
Args:
limit: Number of results to return per page (max 100, default 20)
page: Pagination token for fetching a specific page of results
source: Filter skills by source ('custom' or 'anthropic')
extra_headers: Additional headers for the request
extra_query: Additional query parameters
timeout: Request timeout
custom_llm_provider: Provider name (e.g., 'anthropic')
**kwargs: Additional parameters
Returns:
ListSkillsResponse object
"""
local_vars = locals()
try:
litellm_logging_obj: LiteLLMLoggingObj = kwargs.get("litellm_logging_obj") # type: ignore
litellm_call_id: Optional[str] = kwargs.get("litellm_call_id", None)
_is_async = kwargs.pop("alist_skills", False) is True
# Get LiteLLM parameters
litellm_params = GenericLiteLLMParams(**kwargs)
# Determine provider
if custom_llm_provider is None:
custom_llm_provider = "anthropic"
# Get provider config
skills_api_provider_config: Optional[BaseSkillsAPIConfig] = (
ProviderConfigManager.get_provider_skills_api_config(
provider=litellm.LlmProviders(custom_llm_provider),
)
)
if skills_api_provider_config is None:
raise ValueError(f"LIST skills is not supported for {custom_llm_provider}")
# Build list parameters
list_params: ListSkillsParams = {}
if limit is not None:
list_params["limit"] = limit
if page is not None:
list_params["page"] = page
if source is not None:
list_params["source"] = source
# Merge extra_query if provided
if extra_query:
list_params.update(extra_query) # type: ignore
# Validate environment and get headers
headers = extra_headers or {}
headers = skills_api_provider_config.validate_environment(
headers=headers, litellm_params=litellm_params
)
# Transform request
url, query_params = skills_api_provider_config.transform_list_skills_request(
list_params=list_params,
litellm_params=litellm_params,
headers=headers,
)
# Pre-call logging
litellm_logging_obj.update_environment_variables(
model=None,
optional_params=query_params,
litellm_params={
"litellm_call_id": litellm_call_id,
},
custom_llm_provider=custom_llm_provider,
)
# Make HTTP request
response = base_llm_http_handler.list_skills_handler(
url=url,
query_params=query_params,
skills_api_provider_config=skills_api_provider_config,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
extra_headers=headers,
timeout=timeout or request_timeout,
_is_async=_is_async,
client=kwargs.get("client"),
shared_session=kwargs.get("shared_session"),
)
return response
except Exception as e:
raise litellm.exception_type(
model=None,
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
@client
async def aget_skill(
skill_id: str,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
custom_llm_provider: Optional[str] = None,
**kwargs,
) -> Skill:
"""
Async: Get a skill by ID
Args:
skill_id: The ID of the skill to fetch
extra_headers: Additional headers for the request
extra_query: Additional query parameters
timeout: Request timeout
custom_llm_provider: Provider name (e.g., 'anthropic')
**kwargs: Additional parameters
Returns:
Skill object
"""
local_vars = locals()
try:
loop = asyncio.get_event_loop()
kwargs["aget_skill"] = True
func = partial(
get_skill,
skill_id=skill_id,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
**kwargs,
)
ctx = contextvars.copy_context()
func_with_context = partial(ctx.run, func)
init_response = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
except Exception as e:
raise litellm.exception_type(
model=None,
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
@client
def get_skill(
skill_id: str,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
custom_llm_provider: Optional[str] = None,
**kwargs,
) -> Union[Skill, Coroutine[Any, Any, Skill]]:
"""
Get a skill by ID
Args:
skill_id: The ID of the skill to fetch
extra_headers: Additional headers for the request
extra_query: Additional query parameters
timeout: Request timeout
custom_llm_provider: Provider name (e.g., 'anthropic')
**kwargs: Additional parameters
Returns:
Skill object
"""
local_vars = locals()
try:
litellm_logging_obj: LiteLLMLoggingObj = kwargs.get("litellm_logging_obj") # type: ignore
litellm_call_id: Optional[str] = kwargs.get("litellm_call_id", None)
_is_async = kwargs.pop("aget_skill", False) is True
# Get LiteLLM parameters
litellm_params = GenericLiteLLMParams(**kwargs)
# Determine provider
if custom_llm_provider is None:
custom_llm_provider = "anthropic"
# Get provider config
skills_api_provider_config: Optional[BaseSkillsAPIConfig] = (
ProviderConfigManager.get_provider_skills_api_config(
provider=litellm.LlmProviders(custom_llm_provider),
)
)
if skills_api_provider_config is None:
raise ValueError(f"GET skill is not supported for {custom_llm_provider}")
# Validate environment and get headers
headers = extra_headers or {}
headers = skills_api_provider_config.validate_environment(
headers=headers, litellm_params=litellm_params
)
# Get API base
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
api_base = AnthropicModelInfo.get_api_base(litellm_params.api_base)
# Transform request
url, headers = skills_api_provider_config.transform_get_skill_request(
skill_id=skill_id,
api_base=api_base or DEFAULT_ANTHROPIC_API_BASE,
litellm_params=litellm_params,
headers=headers,
)
# Pre-call logging
litellm_logging_obj.update_environment_variables(
model=None,
optional_params={"skill_id": skill_id},
litellm_params={
"litellm_call_id": litellm_call_id,
},
custom_llm_provider=custom_llm_provider,
)
# Make HTTP request
response = base_llm_http_handler.get_skill_handler(
url=url,
skills_api_provider_config=skills_api_provider_config,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
extra_headers=headers,
timeout=timeout or request_timeout,
_is_async=_is_async,
client=kwargs.get("client"),
shared_session=kwargs.get("shared_session"),
)
return response
except Exception as e:
raise litellm.exception_type(
model=None,
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
@client
async def adelete_skill(
skill_id: str,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
custom_llm_provider: Optional[str] = None,
**kwargs,
) -> DeleteSkillResponse:
"""
Async: Delete a skill by ID
Args:
skill_id: The ID of the skill to delete
extra_headers: Additional headers for the request
extra_query: Additional query parameters
timeout: Request timeout
custom_llm_provider: Provider name (e.g., 'anthropic')
**kwargs: Additional parameters
Returns:
DeleteSkillResponse object
"""
local_vars = locals()
try:
loop = asyncio.get_event_loop()
kwargs["adelete_skill"] = True
func = partial(
delete_skill,
skill_id=skill_id,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
**kwargs,
)
ctx = contextvars.copy_context()
func_with_context = partial(ctx.run, func)
init_response = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
except Exception as e:
raise litellm.exception_type(
model=None,
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
@client
def delete_skill(
skill_id: str,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
custom_llm_provider: Optional[str] = None,
**kwargs,
) -> Union[DeleteSkillResponse, Coroutine[Any, Any, DeleteSkillResponse]]:
"""
Delete a skill by ID
Args:
skill_id: The ID of the skill to delete
extra_headers: Additional headers for the request
extra_query: Additional query parameters
timeout: Request timeout
custom_llm_provider: Provider name (e.g., 'anthropic')
**kwargs: Additional parameters
Returns:
DeleteSkillResponse object
"""
local_vars = locals()
try:
litellm_logging_obj: LiteLLMLoggingObj = kwargs.get("litellm_logging_obj") # type: ignore
litellm_call_id: Optional[str] = kwargs.get("litellm_call_id", None)
_is_async = kwargs.pop("adelete_skill", False) is True
# Get LiteLLM parameters
litellm_params = GenericLiteLLMParams(**kwargs)
# Determine provider
if custom_llm_provider is None:
custom_llm_provider = "anthropic"
# Get provider config
skills_api_provider_config: Optional[BaseSkillsAPIConfig] = (
ProviderConfigManager.get_provider_skills_api_config(
provider=litellm.LlmProviders(custom_llm_provider),
)
)
if skills_api_provider_config is None:
raise ValueError(
f"DELETE skill is not supported for {custom_llm_provider}"
)
# Validate environment and get headers
headers = extra_headers or {}
headers = skills_api_provider_config.validate_environment(
headers=headers, litellm_params=litellm_params
)
# Get API base
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
api_base = AnthropicModelInfo.get_api_base(litellm_params.api_base)
# Transform request
url, headers = skills_api_provider_config.transform_delete_skill_request(
skill_id=skill_id,
api_base=api_base or DEFAULT_ANTHROPIC_API_BASE,
litellm_params=litellm_params,
headers=headers,
)
# Pre-call logging
litellm_logging_obj.update_environment_variables(
model=None,
optional_params={"skill_id": skill_id},
litellm_params={
"litellm_call_id": litellm_call_id,
},
custom_llm_provider=custom_llm_provider,
)
# Make HTTP request
response = base_llm_http_handler.delete_skill_handler(
url=url,
skills_api_provider_config=skills_api_provider_config,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
extra_headers=headers,
timeout=timeout or request_timeout,
_is_async=_is_async,
client=kwargs.get("client"),
shared_session=kwargs.get("shared_session"),
)
return response
except Exception as e:
raise litellm.exception_type(
model=None,
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
litellm/types/llms/anthropic_skills.py
+159
View File
@@ -0,0 +1,159 @@
"""
Type definitions for Anthropic Skills API
"""
from typing import Any, Dict, List, Literal, Optional, Union
from pydantic import BaseModel, Field
from typing_extensions import Required, TypedDict
# Skills API Request Types
class CreateSkillRequest(TypedDict, total=False):
"""Request parameters for creating a skill"""
display_title: Optional[str]
"""Display title for the skill (optional)"""
files: Optional[List[Any]]
"""Files to upload for the skill. All files must be in the same top-level directory and must include a SKILL.md file at the root."""
class ListSkillsParams(TypedDict, total=False):
"""Query parameters for listing skills"""
limit: Optional[int]
"""Number of results to return per page. Maximum value is 100. Defaults to 20."""
page: Optional[str]
"""Pagination token for fetching a specific page of results"""
source: Optional[str]
"""Filter skills by source ('custom' or 'anthropic')"""
# Skills API Response Types
class Skill(BaseModel):
"""Represents a skill from the Anthropic Skills API"""
id: str
"""Unique identifier for the skill"""
created_at: str
"""ISO 8601 timestamp of when the skill was created"""
display_title: Optional[str] = None
"""Display title for the skill"""
latest_version: Optional[str] = None
"""The latest version identifier for the skill"""
source: str
"""Source of the skill (custom or anthropic)"""
type: str = "skill"
"""Object type, always 'skill'"""
updated_at: str
"""ISO 8601 timestamp of when the skill was last updated"""
class ListSkillsResponse(BaseModel):
"""Response from listing skills"""
data: List[Skill]
"""List of skills"""
next_page: Optional[str] = None
"""Pagination token for the next page"""
has_more: bool = False
"""Whether there are more skills available"""
class DeleteSkillResponse(BaseModel):
"""Response from deleting a skill"""
id: str
"""The ID of the deleted skill"""
type: str = "skill_deleted"
"""Deleted object type, always 'skill_deleted'"""
# Skill Version Types
class CreateSkillVersionRequest(TypedDict, total=False):
"""Request parameters for creating a skill version"""
display_title: Optional[str]
"""Display title for this version"""
description: Optional[str]
"""Description of this version"""
instructions: Optional[str]
"""Instructions for this version"""
metadata: Optional[Dict[str, Any]]
"""Additional metadata"""
class SkillVersion(BaseModel):
"""Represents a skill version"""
id: str
"""Unique identifier for the version"""
skill_id: str
"""ID of the parent skill"""
created_at: str
"""ISO 8601 timestamp of when the version was created"""
display_title: Optional[str] = None
"""Display title for this version"""
description: Optional[str] = None
"""Description of this version"""
instructions: Optional[str] = None
"""Instructions for this version"""
metadata: Optional[Dict[str, Any]] = None
"""Additional metadata"""
type: str = "skill.version"
"""Object type"""
class ListSkillVersionsResponse(BaseModel):
"""Response from listing skill versions"""
object: str = "list"
"""Object type, always 'list'"""
data: List[SkillVersion]
"""List of skill versions"""
first_id: Optional[str] = None
"""ID of the first version in the list"""
last_id: Optional[str] = None
"""ID of the last version in the list"""
has_more: bool = False
"""Whether there are more versions available"""
class DeleteSkillVersionResponse(BaseModel):
"""Response from deleting a skill version"""
id: str
"""The ID of the deleted version"""
object: str = "skill.version.deleted"
"""Object type"""
deleted: bool
"""Whether the version was successfully deleted"""
litellm/utils.py
+18
View File
@@ -273,6 +273,7 @@ from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConf
from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
from litellm.llms.base_llm.rerank.transformation import BaseRerankConfig
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.llms.base_llm.skills.transformation import BaseSkillsAPIConfig
from litellm.llms.base_llm.vector_store.transformation import BaseVectorStoreConfig
from litellm.llms.base_llm.vector_store_files.transformation import (
BaseVectorStoreFilesConfig,
@@ -7398,6 +7399,23 @@ class ProviderConfigManager:
return litellm.LiteLLMProxyResponsesAPIConfig()
return None
@staticmethod
def get_provider_skills_api_config(
provider: LlmProviders,
) -> Optional["BaseSkillsAPIConfig"]:
"""
Get provider-specific Skills API configuration
Args:
provider: The LLM provider
Returns:
Provider-specific Skills API config or None
"""
if litellm.LlmProviders.ANTHROPIC == provider:
return litellm.AnthropicSkillsConfig()
return None
@staticmethod
def get_provider_text_completion_config(
model: str,
tests/llm_translation/test-skill/SKILL.md
+8
View File
@@ -0,0 +1,8 @@
---
name: test-skill
description: A minimal test skill for API testing
---
# Test Skill
A minimal test skill for API testing.
tests/llm_translation/test_skills_api.py
+266
View File
@@ -0,0 +1,266 @@
"""
Tests for Skills API operations across providers
"""
import os
import sys
import zipfile
from abc import ABC, abstractmethod
from contextlib import contextmanager
from pathlib import Path
from typing import Optional
import pytest
sys.path.insert(0, os.path.abspath("../.."))
import litellm
from litellm.types.llms.anthropic_skills import (
DeleteSkillResponse,
ListSkillsResponse,
Skill,
)
@contextmanager
def create_skill_zip(skill_name: str):
"""
Helper context manager to create a zip file for a skill.
Args:
skill_name: Name of the skill directory in test_skills_data/
Yields:
File handle to the zip file
The zip file is automatically cleaned up after use.
"""
test_dir = Path(__file__).parent / "test_skills_data"
skill_dir = test_dir / skill_name
# Create a zip file containing the skill directory
zip_path = test_dir / f"{skill_name}.zip"
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zip_file:
zip_file.write(skill_dir, arcname=skill_name)
zip_file.write(skill_dir / "SKILL.md", arcname=f"{skill_name}/SKILL.md")
try:
with open(zip_path, "rb") as f:
yield f
finally:
# Clean up zip file
if zip_path.exists():
zip_path.unlink()
class BaseSkillsAPITest(ABC):
"""
Base test class for Skills API operations.
Tests create, list, get, and delete operations.
"""
@abstractmethod
def get_custom_llm_provider(self) -> str:
"""Return the provider name (e.g., 'anthropic')"""
pass
@abstractmethod
def get_api_key(self) -> Optional[str]:
"""Return the API key for the provider"""
pass
@abstractmethod
def get_api_base(self) -> Optional[str]:
"""Return the API base URL for the provider"""
pass
def test_create_skill(self):
"""
Test creating a skill.
Note: This test creates a skill but does not clean it up,
as we want to verify it was created successfully.
The test_delete_skill test will handle cleanup.
"""
import time
custom_llm_provider = self.get_custom_llm_provider()
api_key = self.get_api_key()
api_base = self.get_api_base()
if not api_key:
pytest.skip(f"No API key provided for {custom_llm_provider}")
litellm.set_verbose = True
litellm._turn_on_debug()
# Use helper to create skill zip
skill_name = "test-skill-litellm"
# Use unique title to avoid conflicts with previous test runs
unique_title = f"Test Skill {int(time.time())}"
# Upload the skill with the zip file
with create_skill_zip(skill_name) as zip_file:
response = litellm.create_skill(
display_title=unique_title,
files=[zip_file],
custom_llm_provider=custom_llm_provider,
api_key=api_key,
api_base=api_base,
)
assert response is not None
assert isinstance(response, Skill)
assert response.id is not None
print(f"Created skill: {response}")
print(f"Skill ID: {response.id}")
def test_list_skills(self):
"""
Test listing skills.
"""
import os
custom_llm_provider = self.get_custom_llm_provider()
api_key = self.get_api_key()
api_base = self.get_api_base()
if not api_key:
pytest.skip(f"No API key provided for {custom_llm_provider}")
# Enable debug logging
os.environ["LITELLM_LOG"] = "DEBUG"
litellm.set_verbose = True
print(f"\n=== Testing list_skills ===")
print("API Key: [REDACTED]")
print(f"API Base: {api_base}")
response = litellm.list_skills(
limit=10,
custom_llm_provider=custom_llm_provider,
api_key=api_key,
api_base=api_base,
)
assert response is not None
assert isinstance(response, ListSkillsResponse)
assert hasattr(response, "data")
print(f"Listed skills: {response}")
def test_get_skill(self):
"""
Test getting a specific skill by ID.
"""
custom_llm_provider = self.get_custom_llm_provider()
api_key = self.get_api_key()
api_base = self.get_api_base()
if not api_key:
pytest.skip(f"No API key provided for {custom_llm_provider}")
litellm.set_verbose = True
# First list existing skills to see if any exist
list_response = litellm.list_skills(
limit=1,
custom_llm_provider=custom_llm_provider,
api_key=api_key,
api_base=api_base,
)
# Type assertion for linter
assert isinstance(list_response, ListSkillsResponse)
print(f"List response: {list_response}")
# If there are existing skills, use the first one
if list_response.data and len(list_response.data) > 0:
skill_id = list_response.data[0].id
should_cleanup = False
print(f"Using existing skill: {skill_id}")
# Now get the skill
response = litellm.get_skill(
skill_id=skill_id,
custom_llm_provider=custom_llm_provider,
api_key=api_key,
api_base=api_base,
)
assert response is not None
assert isinstance(response, Skill)
assert response.id == skill_id
print(f"GET - Retrieved skill: {response}")
def test_delete_skill(self):
"""
Test deleting a skill.
Note: Anthropic requires deleting all skill versions before deleting the skill itself.
This test is currently skipped as it would require additional API calls to delete versions.
"""
import time
custom_llm_provider = self.get_custom_llm_provider()
api_key = self.get_api_key()
api_base = self.get_api_base()
if not api_key:
pytest.skip(f"No API key provided for {custom_llm_provider}")
pytest.skip("Anthropic requires deleting all skill versions first - skipping for now")
litellm.set_verbose = True
# Use helper to create skill zip
skill_name = "test-delete-skill"
# Use unique title to avoid conflicts
unique_title = f"Test Delete Skill {int(time.time())}"
# Create a skill specifically to delete
with create_skill_zip(skill_name) as zip_file:
created_skill = litellm.create_skill(
display_title=unique_title,
files=[zip_file],
custom_llm_provider=custom_llm_provider,
api_key=api_key,
api_base=api_base,
)
# Type assertion for linter
assert isinstance(created_skill, Skill)
skill_id = created_skill.id
print(f"Created skill to delete: {skill_id}")
# Now delete the skill
response = litellm.delete_skill(
skill_id=skill_id,
custom_llm_provider=custom_llm_provider,
api_key=api_key,
api_base=api_base,
)
assert response is not None
assert isinstance(response, DeleteSkillResponse)
assert response.type == "skill_deleted"
print(f"Deleted skill response: {response}")
class TestAnthropicSkillsAPI(BaseSkillsAPITest):
"""
Test Anthropic Skills API implementation.
"""
def get_custom_llm_provider(self) -> str:
return "anthropic"
def get_api_key(self) -> Optional[str]:
return os.environ.get("ANTHROPIC_API_KEY")
def get_api_base(self) -> Optional[str]:
return os.environ.get("ANTHROPIC_API_BASE")
tests/llm_translation/test_skills_data/test-delete-skill/SKILL.md
+9
View File
@@ -0,0 +1,9 @@
---
name: test-delete-skill
description: A test skill created specifically for deletion testing
---
# Test Delete Skill
This skill is created specifically to test the delete functionality.
tests/llm_translation/test_skills_data/test-skill-litellm/SKILL.md
+9
View File
@@ -0,0 +1,9 @@
---
name: test-skill-litellm
description: A test skill created by LiteLLM automated tests
---
# Test Skill
This is a minimal test skill created for automated testing purposes.
tests/llm_translation/test_skills_data/test-skill.zip
BIN
View File
Binary file not shown.
tests/router_unit_tests/test_router_endpoints.py
+34
View File
@@ -867,6 +867,10 @@ def test_initialize_specialized_endpoints():
"retrieve_container",
"adelete_container",
"delete_container",
"acreate_skill",
"alist_skills",
"aget_skill",
"adelete_skill",
]
for endpoint in specialized_endpoints:
@@ -1070,3 +1074,33 @@ def test_initialize_container_endpoints():
for endpoint in container_endpoints:
assert hasattr(router, endpoint)
assert callable(getattr(router, endpoint))
def test_initialize_skills_endpoints():
"""
Test that _initialize_skills_endpoints correctly sets up skills endpoints.
"""
router = Router(
model_list=[
{
"model_name": "test-model",
"litellm_params": {
"model": "anthropic/test-model",
"api_key": "fake-api-key",
},
}
]
)
router._initialize_skills_endpoints()
skills_endpoints = [
"acreate_skill",
"alist_skills",
"aget_skill",
"adelete_skill",
]
for endpoint in skills_endpoints:
assert hasattr(router, endpoint)
assert callable(getattr(router, endpoint))