[Feat] Unified Skills API - works across Anthropic, Vertex, Azure, Bedrock (#18232)

* init LiteLLM_SkillsTable

* init LiteLLMSkillsTransformationHandler

* init LiteLLMSkillsTransformationHandler

* init skills

* init SkillsInjectionHook

* init litellm skills handler

* _fetch_skill_from_db

* LiteLLMSkillsTransformationHandler

* add _prisma_skill_to_litellm

* use SkillPromptInjectionHandler

* refactor skills handler

* add slack GIF

* test_skill_code_execution_via_deployment_hook

* init readme

* test_get_skill_sdk

* test skills in DB

* fix code qa

* init with containers param

* SkillsInjectionHook

* fix type

* convert_skill_to_anthropic_tool, get_litellm_code_execution_tool_anthropic

* fix messages skills handler

* test_slack_gif_skill_creates_gif

* init SkillsInjectionHook

* clean async_pre_call_hook

* fixes

* working unified skills API! yeet

* fix mypy linting

* LiteLLM_SkillsTable

* llm-sandbox==0.3.31

* fix
This commit is contained in:
Ishaan Jaff 2025-12-19 18:55:59 +05:30 • committed by GitHub
parent 3c380086b6
commit 09b5ee9428
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
33 changed files with 4727 additions and 22 deletions

View file

@ -727,4 +727,22 @@ model LiteLLM_UISettings {
ui_settings Json
created_at DateTime @default(now())
updated_at DateTime @updatedAt
}
// Skills table for storing LiteLLM-managed skills
model LiteLLM_SkillsTable {
skill_id String @id @default(uuid())
display_title String?
description String?
instructions String? // The skill instructions/prompt (from SKILL.md)
source String @default("custom") // "custom" or "anthropic"
latest_version String?
file_content Bytes? // Binary content of the skill files (zip)
file_name String? // Original filename
file_type String? // MIME type (e.g., "application/zip")
metadata Json? @default("{}")
created_at DateTime @default(now())
created_by String?
updated_at DateTime @default(now()) @updatedAt
updated_by String?
}

View file

@ -37,6 +37,7 @@ async def acreate(
tools: Optional[List[Dict]] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
container: Optional[Dict] = None,
**kwargs
) -> Union[AnthropicMessagesResponse, AsyncIterator]:
"""
@ -56,6 +57,7 @@ async def acreate(
tools (List[Dict], optional): List of tool definitions
top_k (int, optional): Top K sampling parameter
top_p (float, optional): Nucleus sampling parameter
container (Dict, optional): Container config with skills for code execution
**kwargs: Additional arguments
Returns:
@ -75,6 +77,7 @@ async def acreate(
tools=tools,
top_k=top_k,
top_p=top_p,
container=container,
**kwargs,
)
@ -93,6 +96,7 @@ def create(
tools: Optional[List[Dict]] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
container: Optional[Dict] = None,
**kwargs
) -> Union[
AnthropicMessagesResponse,
@ -135,5 +139,6 @@ def create(
tools=tools,
top_k=top_k,
top_p=top_p,
container=container,
**kwargs,
)

View file

@ -119,6 +119,7 @@ def anthropic_messages_handler(
tools: Optional[List[Dict]] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
container: Optional[Dict] = None,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
client: Optional[AsyncHTTPHandler] = None,
@ -131,6 +132,9 @@ def anthropic_messages_handler(
]:
"""
Makes Anthropic `/v1/messages` API calls In the Anthropic API Spec
Args:
container: Container config with skills for code execution
"""
from litellm.types.utils import LlmProviders

View file

@ -0,0 +1,381 @@
# LiteLLM Skills - Database-Backed Skills Storage
This module provides database-backed skills storage as an alternative to Anthropic's cloud-based Skills API. It enables using skills with **any LLM provider** (Bedrock, OpenAI, Azure, etc.) by storing skills locally and converting them to tools + system prompt injection.
## Architecture
```mermaid
flowchart TB
subgraph "Skill Creation"
A[User creates skill with ZIP file] --> B{custom_llm_provider?}
B -->|anthropic| C[Forward to Anthropic API]
B -->|litellm_proxy| D[Store in LiteLLM Database]
D --> E[Extract & store:<br/>- display_title<br/>- description<br/>- instructions<br/>- file_content ZIP]
end
subgraph "Skill Usage in Messages API"
F[Request with container.skills] --> G[SkillsInjectionHook]
G --> H{skill_id prefix?}
H -->|"litellm:skill_abc"| I[Fetch from LiteLLM DB]
H -->|"skill_xyz" no prefix| J[Pass to Anthropic as native skill]
I --> K{Model provider?}
K -->|Anthropic API| L[Convert to tools]
K -->|Bedrock/OpenAI/etc| M[Convert to tools +<br/>Inject SKILL.md into system prompt]
J --> N[Keep in container.skills]
end
subgraph "Skill Resolution for Non-Anthropic"
M --> O[Extract SKILL.md from ZIP]
O --> P[Add to system prompt:<br/># Available Skills<br/>## Skill: My Skill<br/>SKILL.md content...]
P --> Q[Create OpenAI-style tool:<br/>type: function<br/>name: skill_id<br/>description: instructions]
Q --> R[Send to LLM Provider]
end
```
## Automatic Code Execution
For skills that include executable code (Python files), LiteLLM automatically handles:
1. **Pre-call hook** (`async_pre_call_hook`): Adds `litellm_code_execution` tool, injects SKILL.md content
2. **Post-call hook** (`async_post_call_success_deployment_hook`): Detects tool calls, executes code in Docker sandbox, continues loop
3. **Returns files**: Generated files (GIFs, images, etc.) returned directly on response
```mermaid
sequenceDiagram
participant User
participant LiteLLM as LiteLLM SDK
participant PreHook as async_pre_call_hook
participant LLM as LLM Provider
participant PostHook as async_post_call_success_deployment_hook
participant Sandbox as Docker Sandbox
User->>LiteLLM: litellm.acompletion(model, messages, container={skills: [...]})
Note over LiteLLM,PreHook: PRE-CALL HOOK
LiteLLM->>PreHook: Intercept request
PreHook->>PreHook: Fetch skill from DB (litellm:skill_id)
PreHook->>PreHook: Extract SKILL.md from ZIP
PreHook->>PreHook: Inject SKILL.md into system prompt
PreHook->>PreHook: Add litellm_code_execution tool
PreHook->>PreHook: Store skill files in metadata
PreHook-->>LiteLLM: Modified request
LiteLLM->>LLM: Forward to provider (OpenAI/Bedrock/etc)
LLM-->>LiteLLM: Response with tool_calls
Note over LiteLLM,PostHook: POST-CALL HOOK (Agentic Loop)
LiteLLM->>PostHook: Check response
loop Until no more tool calls
PostHook->>PostHook: Check for litellm_code_execution tool call
alt Has code execution tool call
PostHook->>Sandbox: Execute Python code
Sandbox->>Sandbox: Copy skill files to /sandbox
Sandbox->>Sandbox: Install requirements.txt
Sandbox->>Sandbox: Run code
Sandbox-->>PostHook: Result + generated files
PostHook->>PostHook: Add tool result to messages
PostHook->>LLM: Make another LLM call
LLM-->>PostHook: New response
else No code execution
PostHook->>PostHook: Break loop
end
end
PostHook->>PostHook: Attach files to response._litellm_generated_files
PostHook-->>LiteLLM: Modified response with files
LiteLLM-->>User: Final response with generated files
```
```python
import litellm
from litellm.proxy.hooks.litellm_skills import SkillsInjectionHook
# Register the hook (done once at startup)
hook = SkillsInjectionHook()
litellm.callbacks.append(hook)
# ONE request - LiteLLM handles everything automatically
# The container parameter triggers the SkillsInjectionHook
response = await litellm.acompletion(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Create a bouncing ball GIF"}],
container={
"skills": [{"type": "custom", "skill_id": "litellm:skill_abc123"}]
},
)
# Files are attached directly to response
generated_files = response._litellm_generated_files
for f in generated_files:
print(f"Generated: {f['name']} ({f['size']} bytes)")
# f['content_base64'] contains the file data
```
This mimics Anthropic's behavior - no manual agentic loop needed!
### How it works
The `SkillsInjectionHook` uses two hooks:
1. **`async_pre_call_hook`** (proxy only): Transforms the request before LLM call
- Fetches skills from DB
- Injects SKILL.md into system prompt
- Adds `litellm_code_execution` tool
- Sets `_litellm_code_execution_enabled=True` in metadata
2. **`async_post_call_success_deployment_hook`** (SDK + proxy): Called after LLM response
- Checks if response has `litellm_code_execution` tool call
- Executes code in Docker sandbox
- Adds result to messages, makes another LLM call
- Repeats until model gives final response
- Attaches generated files to `response._litellm_generated_files`
## File Structure
```
litellm/llms/litellm_proxy/skills/
├── __init__.py # Exports all skill components
├── handler.py # LiteLLMSkillsHandler - database CRUD operations (Prisma)
├── transformation.py # LiteLLMSkillsTransformationHandler - SDK transformation layer
├── prompt_injection.py # SkillPromptInjectionHandler - SKILL.md extraction and injection
├── sandbox_executor.py # SkillsSandboxExecutor - Docker sandbox code execution
├── code_execution.py # CodeExecutionHandler - automatic agentic loop
└── README.md # This file
litellm/proxy/hooks/litellm_skills/
├── __init__.py # Re-exports from SDK + SkillsInjectionHook
└── main.py # SkillsInjectionHook - CustomLogger hook for proxy
```
## Components
### 1. `handler.py` - LiteLLMSkillsHandler
Database operations for skills CRUD:
```python
from litellm.llms.litellm_proxy.skills import LiteLLMSkillsHandler
# Create skill
skill = await LiteLLMSkillsHandler.create_skill(
data=NewSkillRequest(
display_title="My Skill",
description="A helpful skill",
instructions="Use this skill when...",
file_content=zip_bytes, # ZIP file content
file_name="my-skill.zip",
file_type="application/zip",
),
user_id="user_123"
)
# List skills
skills = await LiteLLMSkillsHandler.list_skills(limit=10, offset=0)
# Get skill
skill = await LiteLLMSkillsHandler.get_skill(skill_id="skill_abc123")
# Delete skill
await LiteLLMSkillsHandler.delete_skill(skill_id="skill_abc123")
```
### 2. `transformation.py` - LiteLLMSkillsTransformationHandler
SDK-level transformation layer that wraps handler operations:
```python
from litellm.llms.litellm_proxy.skills import LiteLLMSkillsTransformationHandler
handler = LiteLLMSkillsTransformationHandler()
# Async create
skill = await handler.create_skill_handler(
display_title="My Skill",
files=[zip_file],
_is_async=True
)
```
## Skill ZIP Format
Skills must be packaged as ZIP files with a `SKILL.md` file:
```
my-skill.zip
└── my-skill/
└── SKILL.md
```
### SKILL.md Format
```markdown
---
name: my-skill
description: A brief description of what this skill does
---
# My Skill
Detailed instructions for the LLM on how to use this skill.
## Usage
When the user asks about X, use this skill to...
## Examples
- Example 1: ...
- Example 2: ...
```
## SDK Usage
### Create Skill in LiteLLM Database
```python
import litellm
# Create skill stored in LiteLLM DB
skill = litellm.create_skill(
display_title="Data Analysis Skill",
files=[open("data-analysis.zip", "rb")],
custom_llm_provider="litellm_proxy", # Store in LiteLLM DB
)
print(f"Created skill: {skill.id}") # skill_abc123
```
### Use Skill with Any Provider
```python
import litellm
# Use LiteLLM-stored skill with Bedrock
response = litellm.completion(
model="bedrock/anthropic.claude-3-sonnet-20240229-v1:0",
messages=[{"role": "user", "content": "Analyze this data..."}],
container={
"skills": [
{"type": "custom", "skill_id": "litellm:skill_abc123"} # litellm: prefix
]
}
)
```
## How Skill Resolution Works
### Step 1: Request with Skills
```python
{
"model": "bedrock/claude-3-sonnet",
"messages": [{"role": "user", "content": "Help me analyze data"}],
"container": {
"skills": [
{"type": "custom", "skill_id": "litellm:skill_abc123"}
]
}
}
```
### Step 2: SkillsInjectionHook Processing
The hook (`litellm/proxy/hooks/litellm_skills/main.py`) intercepts the request:
1. **Detects `litellm:` prefix** → Fetches skill from database
2. **Checks model provider** → Bedrock is not Anthropic
3. **Extracts SKILL.md** from stored ZIP file
4. **Converts skill to tool** + **Injects content into system prompt**
### Step 3: Transformed Request
```python
{
"model": "bedrock/claude-3-sonnet",
"messages": [
{
"role": "system",
"content": """
---
# Available Skills
## Skill: Data Analysis Skill
# Data Analysis Skill
This skill helps with data analysis tasks...
## Usage
When the user asks about data analysis...
"""
},
{"role": "user", "content": "Help me analyze data"}
],
"tools": [
{
"type": "function",
"function": {
"name": "skill_abc123",
"description": "This skill helps with data analysis tasks...",
"parameters": {"type": "object", "properties": {}, "required": []}
}
}
]
# container is removed for non-Anthropic providers
}
```
## Database Schema
Skills are stored in `LiteLLM_SkillsTable`:
```prisma
model LiteLLM_SkillsTable {
skill_id String @id @default(uuid())
display_title String?
description String?
instructions String?
source String @default("custom")
latest_version String?
metadata Json? @default("{}")
file_content Bytes? // ZIP file binary content
file_name String? // Original filename
file_type String? // MIME type
created_at DateTime @default(now())
created_by String?
updated_at DateTime @default(now()) @updatedAt
updated_by String?
}
```
## Routing Summary
| Scenario | custom_llm_provider | skill_id Format | Behavior |
|----------|---------------------|-----------------|----------|
| Create skill on Anthropic | `anthropic` | N/A | Forward to Anthropic API |
| Create skill in LiteLLM DB | `litellm_proxy` | N/A | Store in database |
| Use Anthropic native skill | N/A | `skill_xyz` | Pass to Anthropic container.skills |
| Use LiteLLM skill on Anthropic | N/A | `litellm:skill_abc` | Convert to tools |
| Use LiteLLM skill on Bedrock/OpenAI | N/A | `litellm:skill_abc` | Convert to tools + inject SKILL.md |
## Testing
Run the tests:
```bash
pytest tests/proxy_unit_tests/test_skills_db.py -v
```
Tests cover:
- Creating skills with file content
- Listing and retrieving skills
- Deleting skills
- Hook resolution with ZIP file extraction
- System prompt injection for non-Anthropic models

View file

@ -0,0 +1,54 @@
"""
LiteLLM Proxy Skills - Database-backed skills storage and execution
This module provides:
- Database-backed skills storage (alternative to Anthropic's cloud-based skills API)
- Skill content extraction and prompt injection
- Sandboxed code execution for skills
- Automatic code execution handler
Main components:
- handler.py: LiteLLMSkillsHandler - database CRUD operations
- transformation.py: LiteLLMSkillsTransformationHandler - SDK transformation layer
- prompt_injection.py: SkillPromptInjectionHandler - SKILL.md extraction and injection
- sandbox_executor.py: SkillsSandboxExecutor - Docker sandbox execution
- code_execution.py: CodeExecutionHandler - automatic agentic loop
"""
from litellm.llms.litellm_proxy.skills.code_execution import (
LITELLM_CODE_EXECUTION_TOOL,
CodeExecutionHandler,
LiteLLMInternalTools,
add_code_execution_tool,
code_execution_handler,
get_litellm_code_execution_tool,
has_code_execution_tool,
)
from litellm.llms.litellm_proxy.skills.constants import (
DEFAULT_MAX_ITERATIONS,
DEFAULT_SANDBOX_TIMEOUT,
)
from litellm.llms.litellm_proxy.skills.handler import LiteLLMSkillsHandler
from litellm.llms.litellm_proxy.skills.prompt_injection import (
SkillPromptInjectionHandler,
)
from litellm.llms.litellm_proxy.skills.sandbox_executor import SkillsSandboxExecutor
from litellm.llms.litellm_proxy.skills.transformation import (
LiteLLMSkillsTransformationHandler,
)
__all__ = [
"LiteLLMSkillsHandler",
"LiteLLMSkillsTransformationHandler",
"SkillPromptInjectionHandler",
"SkillsSandboxExecutor",
"CodeExecutionHandler",
"LiteLLMInternalTools",
"LITELLM_CODE_EXECUTION_TOOL",
"get_litellm_code_execution_tool",
"code_execution_handler",
"has_code_execution_tool",
"add_code_execution_tool",
"DEFAULT_MAX_ITERATIONS",
"DEFAULT_SANDBOX_TIMEOUT",
]

View file

@ -0,0 +1,311 @@
"""
Automatic Code Execution Handler for LiteLLM Skills
When `litellm_code_execution` tool is present, this handler automatically:
1. Makes the LLM call
2. Executes any code the model generates
3. Continues the conversation with results
4. Returns final response with generated files inline (base64)
This mimics Anthropic's behavior where code execution happens automatically.
Generated files are returned directly in the response - no separate storage needed.
"""
import base64
import json
from enum import Enum
from typing import Any, Dict, List, Optional
from litellm._logging import verbose_logger
class LiteLLMInternalTools(str, Enum):
"""
Enum for internal LiteLLM tools that are injected into requests.
These tools are handled automatically by LiteLLM hooks and are not
passed to the underlying LLM provider directly.
"""
CODE_EXECUTION = "litellm_code_execution"
def get_litellm_code_execution_tool() -> Dict[str, Any]:
"""
Returns the litellm_code_execution tool definition in OpenAI format.
This tool enables automatic code execution in a sandboxed environment
when skills include executable Python code.
"""
return {
"type": "function",
"function": {
"name": LiteLLMInternalTools.CODE_EXECUTION.value,
"description": "Execute Python code in a sandboxed environment. Use this to run code that generates files, processes data, or performs computations. Generated files will be returned directly.",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "Python code to execute"
}
},
"required": ["code"]
}
}
}
def get_litellm_code_execution_tool_anthropic() -> Dict[str, Any]:
"""
Returns the litellm_code_execution tool definition in Anthropic/messages API format.
This tool enables automatic code execution in a sandboxed environment
when skills include executable Python code.
"""
return {
"name": LiteLLMInternalTools.CODE_EXECUTION.value,
"description": "Execute Python code in a sandboxed environment. Use this to run code that generates files, processes data, or performs computations. Generated files will be returned directly.",
"input_schema": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "Python code to execute"
}
},
"required": ["code"]
}
}
# Singleton tool definition for backwards compatibility
LITELLM_CODE_EXECUTION_TOOL = get_litellm_code_execution_tool()
class CodeExecutionHandler:
"""
Handles automatic code execution for LiteLLM skills.
When enabled, this handler intercepts LLM responses with code execution
tool calls, executes them in a sandbox, and continues the conversation
automatically until completion.
"""
def __init__(
self,
max_iterations: Optional[int] = None,
sandbox_timeout: Optional[int] = None,
):
from litellm.llms.litellm_proxy.skills.constants import (
DEFAULT_MAX_ITERATIONS,
DEFAULT_SANDBOX_TIMEOUT,
)
self.max_iterations = max_iterations or DEFAULT_MAX_ITERATIONS
self.sandbox_timeout = sandbox_timeout or DEFAULT_SANDBOX_TIMEOUT
async def execute_with_code_execution(
self,
model: str,
messages: List[Dict],
tools: List[Dict],
skill_files: Dict[str, bytes],
skill_id: Optional[str] = None,
**kwargs,
) -> Dict[str, Any]:
"""
Execute an LLM call with automatic code execution handling.
This method:
1. Makes the initial LLM call
2. If model calls litellm_code_execution, executes the code
3. Continues conversation with results
4. Repeats until model stops calling tools
5. Returns final response with generated files inline
Args:
model: Model to use
messages: Initial messages
tools: Tools including litellm_code_execution
skill_files: Dict of skill files for execution
skill_id: Optional skill ID for tracking
**kwargs: Additional args for litellm.acompletion
Returns:
Dict with:
- response: Final LLM response
- files: List of generated files with content (base64)
- execution_results: List of code execution results
"""
import litellm
from litellm.llms.litellm_proxy.skills.sandbox_executor import (
SkillsSandboxExecutor,
)
current_messages = list(messages)
generated_files: List[Dict[str, Any]] = [] # Files returned directly
execution_results: List[Dict] = []
executor = SkillsSandboxExecutor(timeout=self.sandbox_timeout)
response: Any = None # Initialize to avoid possibly unbound error
for iteration in range(self.max_iterations):
verbose_logger.debug(
f"CodeExecutionHandler: Iteration {iteration + 1}/{self.max_iterations}"
)
# Make LLM call
response = await litellm.acompletion(
model=model,
messages=current_messages,
tools=tools,
**kwargs,
)
assistant_message = response.choices[0].message # type: ignore
stop_reason = response.choices[0].finish_reason # type: ignore
# Build assistant message for conversation history
assistant_msg_dict: Dict[str, Any] = {
"role": "assistant",
"content": assistant_message.content,
}
if assistant_message.tool_calls:
assistant_msg_dict["tool_calls"] = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in assistant_message.tool_calls
]
current_messages.append(assistant_msg_dict)
# Check if we're done (no tool calls or not tool_calls finish reason)
if stop_reason != "tool_calls" or not assistant_message.tool_calls:
verbose_logger.debug(
f"CodeExecutionHandler: Completed after {iteration + 1} iterations"
)
return {
"response": response,
"files": generated_files, # Files returned directly with base64 content
"execution_results": execution_results,
"messages": current_messages,
}
# Handle tool calls
for tool_call in assistant_message.tool_calls:
tool_name = tool_call.function.name
if tool_name == LiteLLMInternalTools.CODE_EXECUTION.value:
# Execute code in sandbox
try:
args = json.loads(tool_call.function.arguments)
code = args.get("code", "")
verbose_logger.debug(
f"CodeExecutionHandler: Executing code ({len(code)} chars)"
)
exec_result = executor.execute(
code=code,
skill_files=skill_files,
)
verbose_logger.debug(
f"CodeExecutionHandler: Execution result: {exec_result}"
)
execution_results.append({
"iteration": iteration,
"success": exec_result["success"],
"output": exec_result["output"],
"error": exec_result["error"],
"files": [f["name"] for f in exec_result["files"]],
})
# Build tool result content
tool_result = exec_result["output"] or ""
# Collect generated files (returned directly, no storage)
if exec_result["files"]:
tool_result += "\n\nGenerated files:"
for f in exec_result["files"]:
file_content = base64.b64decode(f["content_base64"])
# Add to generated files list (returned in response)
generated_files.append({
"name": f["name"],
"mime_type": f["mime_type"],
"content_base64": f["content_base64"],
"size": len(file_content),
})
tool_result += f"\n- {f['name']} ({len(file_content)} bytes)"
verbose_logger.debug(
f"CodeExecutionHandler: Generated file {f['name']} ({len(file_content)} bytes)"
)
if exec_result["error"]:
tool_result += f"\n\nError:\n{exec_result['error']}"
except Exception as e:
tool_result = f"Code execution failed: {str(e)}"
execution_results.append({
"iteration": iteration,
"success": False,
"error": str(e),
})
# Add tool result to messages
current_messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_result,
})
else:
# Non-code-execution tool - pass through
# In a full implementation, this would call other tool handlers
current_messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": f"Tool '{tool_name}' not handled by code execution handler",
})
# Max iterations reached
verbose_logger.warning(
f"CodeExecutionHandler: Max iterations ({self.max_iterations}) reached"
)
return {
"response": response,
"files": generated_files,
"execution_results": execution_results,
"messages": current_messages,
"max_iterations_reached": True,
}
def has_code_execution_tool(tools: Optional[List[Dict]]) -> bool:
"""Check if litellm_code_execution tool is in the tools list."""
if not tools:
return False
for tool in tools:
func = tool.get("function", {})
if func.get("name") == LiteLLMInternalTools.CODE_EXECUTION.value:
return True
return False
def add_code_execution_tool(tools: Optional[List[Dict]]) -> List[Dict]:
"""Add litellm_code_execution tool if not already present."""
tools = tools or []
if not has_code_execution_tool(tools):
tools.append(LITELLM_CODE_EXECUTION_TOOL)
return tools
# Global handler instance
code_execution_handler = CodeExecutionHandler()

View file

@ -0,0 +1,13 @@
"""
Constants for LiteLLM Skills
Centralized constants for skills processing, code execution, and sandbox configuration.
"""
# Code execution loop settings
DEFAULT_MAX_ITERATIONS: int = 10
"""Maximum number of iterations for the automatic code execution loop."""
DEFAULT_SANDBOX_TIMEOUT: int = 120
"""Default timeout in seconds for sandbox code execution."""

View file

@ -0,0 +1,219 @@
"""
Handler for LiteLLM database-backed skills operations.
This module contains the actual database operations for skills CRUD.
Used by the transformation layer and skills injection hook.
"""
import uuid
from typing import Any, Dict, List, Optional
from litellm._logging import verbose_logger
from litellm.proxy._types import LiteLLM_SkillsTable, NewSkillRequest
def _prisma_skill_to_litellm(prisma_skill) -> LiteLLM_SkillsTable:
"""
Convert a Prisma skill record to LiteLLM_SkillsTable.
Handles Base64 decoding of file_content field.
"""
import base64
data = prisma_skill.model_dump()
# Decode Base64 file_content back to bytes
# model_dump() converts Base64 field to base64-encoded string
if data.get("file_content") is not None:
if isinstance(data["file_content"], str):
data["file_content"] = base64.b64decode(data["file_content"])
elif isinstance(data["file_content"], bytes):
# Already bytes, no conversion needed
pass
return LiteLLM_SkillsTable(**data)
class LiteLLMSkillsHandler:
"""
Handler for LiteLLM database-backed skills operations.
This class provides static methods for CRUD operations on skills
stored in the LiteLLM proxy database (LiteLLM_SkillsTable).
"""
@staticmethod
async def _get_prisma_client():
"""Get the prisma client from proxy server."""
from litellm.proxy.proxy_server import prisma_client
if prisma_client is None:
raise ValueError(
"Prisma client is not initialized. "
"Database connection required for LiteLLM skills."
)
return prisma_client
@staticmethod
async def create_skill(
data: NewSkillRequest,
user_id: Optional[str] = None,
) -> LiteLLM_SkillsTable:
"""
Create a new skill in the LiteLLM database.
Args:
data: NewSkillRequest with skill details
user_id: Optional user ID for tracking
Returns:
LiteLLM_SkillsTable record
"""
prisma_client = await LiteLLMSkillsHandler._get_prisma_client()
skill_id = f"litellm_skill_{uuid.uuid4()}"
skill_data: Dict[str, Any] = {
"skill_id": skill_id,
"display_title": data.display_title,
"description": data.description,
"instructions": data.instructions,
"source": "custom",
"created_by": user_id,
"updated_by": user_id,
}
# Handle metadata
if data.metadata is not None:
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
skill_data["metadata"] = safe_dumps(data.metadata)
# Handle file content - wrap bytes in Base64 for Prisma
if data.file_content is not None:
from prisma.fields import Base64
skill_data["file_content"] = Base64.encode(data.file_content)
if data.file_name is not None:
skill_data["file_name"] = data.file_name
if data.file_type is not None:
skill_data["file_type"] = data.file_type
verbose_logger.debug(
f"LiteLLMSkillsHandler: Creating skill {skill_id} with title={data.display_title}"
)
new_skill = await prisma_client.db.litellm_skillstable.create(data=skill_data)
return _prisma_skill_to_litellm(new_skill)
@staticmethod
async def list_skills(
limit: int = 20,
offset: int = 0,
) -> List[LiteLLM_SkillsTable]:
"""
List skills from the LiteLLM database.
Args:
limit: Maximum number of skills to return
offset: Number of skills to skip
Returns:
List of LiteLLM_SkillsTable records
"""
prisma_client = await LiteLLMSkillsHandler._get_prisma_client()
verbose_logger.debug(
f"LiteLLMSkillsHandler: Listing skills with limit={limit}, offset={offset}"
)
skills = await prisma_client.db.litellm_skillstable.find_many(
take=limit,
skip=offset,
order={"created_at": "desc"},
)
return [_prisma_skill_to_litellm(s) for s in skills]
@staticmethod
async def get_skill(skill_id: str) -> LiteLLM_SkillsTable:
"""
Get a skill by ID from the LiteLLM database.
Args:
skill_id: The skill ID to retrieve
Returns:
LiteLLM_SkillsTable record
Raises:
ValueError: If skill not found
"""
prisma_client = await LiteLLMSkillsHandler._get_prisma_client()
verbose_logger.debug(f"LiteLLMSkillsHandler: Getting skill {skill_id}")
skill = await prisma_client.db.litellm_skillstable.find_unique(
where={"skill_id": skill_id}
)
if skill is None:
raise ValueError(f"Skill not found: {skill_id}")
return _prisma_skill_to_litellm(skill)
@staticmethod
async def delete_skill(skill_id: str) -> Dict[str, str]:
"""
Delete a skill by ID from the LiteLLM database.
Args:
skill_id: The skill ID to delete
Returns:
Dict with id and type of deleted skill
Raises:
ValueError: If skill not found
"""
prisma_client = await LiteLLMSkillsHandler._get_prisma_client()
verbose_logger.debug(f"LiteLLMSkillsHandler: Deleting skill {skill_id}")
# Check if skill exists
skill = await prisma_client.db.litellm_skillstable.find_unique(
where={"skill_id": skill_id}
)
if skill is None:
raise ValueError(f"Skill not found: {skill_id}")
# Delete the skill
await prisma_client.db.litellm_skillstable.delete(where={"skill_id": skill_id})
return {"id": skill_id, "type": "skill_deleted"}
@staticmethod
async def fetch_skill_from_db(skill_id: str) -> Optional[LiteLLM_SkillsTable]:
"""
Fetch a skill from the database (used by skills injection hook).
This is a convenience method that returns None instead of raising
an exception if the skill is not found.
Args:
skill_id: The skill ID to fetch
Returns:
LiteLLM_SkillsTable or None if not found
"""
try:
return await LiteLLMSkillsHandler.get_skill(skill_id)
except ValueError:
return None
except Exception as e:
verbose_logger.warning(
f"LiteLLMSkillsHandler: Error fetching skill {skill_id}: {e}"
)
return None

View file

@ -0,0 +1,305 @@
"""
Prompt Injection Handler for LiteLLM Skills
Handles extraction of skill content (SKILL.md) from stored ZIP files
and injection into the system prompt for non-Anthropic models.
"""
import zipfile
from io import BytesIO
from typing import Any, Dict, List, Optional
from litellm._logging import verbose_logger
from litellm.proxy._types import LiteLLM_SkillsTable
class SkillPromptInjectionHandler:
"""
Handles skill content extraction and system prompt injection.
Responsibilities:
- Extract SKILL.md content from skill ZIP files
- Extract ALL files from ZIP for code execution
- Inject skill content into system message
- Create execute_code tool definition
"""
def extract_skill_content(self, skill: LiteLLM_SkillsTable) -> Optional[str]:
"""
Extract skill content from the stored zip file.
Looks for SKILL.md or README.md in the zip and returns its content.
This content describes the skill's capabilities and instructions.
Args:
skill: The skill from LiteLLM database
Returns:
The skill content as a string, or None if not available
"""
if not skill.file_content:
return skill.instructions
try:
zip_buffer = BytesIO(skill.file_content)
with zipfile.ZipFile(zip_buffer, "r") as zf:
# Look for SKILL.md first
for name in zf.namelist():
if name.endswith("SKILL.md"):
content = zf.read(name).decode("utf-8")
if content:
return f"## Skill: {skill.display_title or skill.skill_id}\n\n{content}"
# Fall back to README.md
for name in zf.namelist():
if name.endswith("README.md"):
content = zf.read(name).decode("utf-8")
if content:
return f"## Skill: {skill.display_title or skill.skill_id}\n\n{content}"
# Fall back to any .md file
for name in zf.namelist():
if name.endswith(".md"):
content = zf.read(name).decode("utf-8")
if content:
return f"## Skill: {skill.display_title or skill.skill_id}\n\n{content}"
except Exception as e:
verbose_logger.warning(
f"SkillPromptInjectionHandler: Error extracting content from skill {skill.skill_id}: {e}"
)
return skill.instructions
def extract_all_files(self, skill: LiteLLM_SkillsTable) -> Dict[str, bytes]:
"""
Extract ALL files from skill ZIP for code execution.
Returns a dict mapping file paths to their binary content.
The paths have the skill folder prefix removed (e.g., "slack-gif-creator/core/..." -> "core/...").
Args:
skill: The skill from LiteLLM database
Returns:
Dict mapping file paths to binary content
"""
files: Dict[str, bytes] = {}
if not skill.file_content:
return files
try:
zip_buffer = BytesIO(skill.file_content)
with zipfile.ZipFile(zip_buffer, "r") as zf:
for name in zf.namelist():
# Skip directories
if name.endswith("/"):
continue
# Remove skill folder prefix (first path component)
parts = name.split("/")
if len(parts) > 1:
clean_path = "/".join(parts[1:])
else:
clean_path = name
if clean_path:
files[clean_path] = zf.read(name)
except Exception as e:
verbose_logger.warning(
f"SkillPromptInjectionHandler: Error extracting files from skill {skill.skill_id}: {e}"
)
return files
def inject_skill_content_to_messages(
self, data: dict, skill_contents: List[str], use_anthropic_format: bool = False
) -> dict:
"""
Inject skill content into the system prompt.
For Anthropic messages API (use_anthropic_format=True):
- Injects into top-level 'system' parameter (not in messages array)
For OpenAI-style APIs (use_anthropic_format=False):
- Injects into messages array with role="system"
Args:
data: The request data dict
skill_contents: List of skill content strings to inject
use_anthropic_format: If True, use top-level 'system' param for Anthropic
Returns:
Modified data dict with skill content in system prompt
"""
if not skill_contents:
return data
# Build the skill injection text
skill_section = "\n\n---\n\n# Available Skills\n\n" + "\n\n---\n\n".join(skill_contents)
if use_anthropic_format:
# Anthropic messages API: use top-level 'system' parameter
current_system = data.get("system", "")
if current_system:
data["system"] = current_system + skill_section
else:
data["system"] = skill_section.strip()
return data
# OpenAI-style: inject into messages array
messages = data.get("messages", [])
if not messages:
return data
# Find or create system message
system_msg_idx = None
for i, msg in enumerate(messages):
if isinstance(msg, dict) and msg.get("role") == "system":
system_msg_idx = i
break
if system_msg_idx is not None:
# Append to existing system message
current_content = messages[system_msg_idx].get("content", "")
messages[system_msg_idx]["content"] = current_content + skill_section
else:
# Create new system message at the beginning
messages.insert(0, {"role": "system", "content": skill_section.strip()})
data["messages"] = messages
return data
def create_execute_code_tool(self, skill_modules: List[str]) -> Dict[str, Any]:
"""
Create the execute_code tool definition.
This tool allows the model to execute Python code with access
to the skill's modules (e.g., 'from core.gif_builder import GIFBuilder').
Args:
skill_modules: List of available module paths (e.g., ["core/gif_builder.py"])
Returns:
OpenAI-style tool definition
"""
# Format module list for description
module_examples = []
for mod in skill_modules[:5]: # Limit to 5 examples
if mod.endswith(".py"):
# Convert path to import: "core/gif_builder.py" -> "from core.gif_builder import ..."
import_path = mod.replace("/", ".").replace(".py", "")
module_examples.append(f"from {import_path} import ...")
module_hint = ""
if module_examples:
module_hint = f" Available modules: {', '.join(module_examples)}"
return {
"type": "function",
"function": {
"name": "execute_code",
"description": f"Execute Python code in a sandboxed environment. Generated files will be returned.{module_hint}",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "Python code to execute. You can import skill modules and use standard libraries."
}
},
"required": ["code"]
}
}
}
def convert_skill_to_tool(self, skill: LiteLLM_SkillsTable) -> Dict[str, Any]:
"""
Convert a LiteLLM skill to an OpenAI-style tool.
The skill's instructions are used as the function description,
allowing the model to understand when and how to use the skill.
Args:
skill: The skill from LiteLLM database
Returns:
OpenAI-style tool definition
"""
# Create a function name from skill_id (sanitize for function naming)
func_name = skill.skill_id.replace("-", "_").replace(" ", "_")
# Use instructions as description, fall back to description or title
description = (
skill.instructions
or skill.description
or skill.display_title
or f"Skill: {skill.skill_id}"
)
# Truncate description if too long (OpenAI has limits)
max_desc_length = 1024
if len(description) > max_desc_length:
description = description[: max_desc_length - 3] + "..."
tool: Dict[str, Any] = {
"type": "function",
"function": {
"name": func_name,
"description": description,
"parameters": {
"type": "object",
"properties": {},
"required": [],
},
},
}
# If skill has metadata with parameter definitions, use them
if skill.metadata and isinstance(skill.metadata, dict):
params = skill.metadata.get("parameters")
if params and isinstance(params, dict):
tool["function"]["parameters"] = params
return tool
def convert_skill_to_anthropic_tool(self, skill: LiteLLM_SkillsTable) -> Dict[str, Any]:
"""
Convert a LiteLLM skill to an Anthropic-style tool (messages API format).
Args:
skill: The skill from LiteLLM database
Returns:
Anthropic-style tool definition with name, description, input_schema
"""
func_name = skill.skill_id.replace("-", "_").replace(" ", "_")
description = (
skill.instructions
or skill.description
or skill.display_title
or f"Skill: {skill.skill_id}"
)
max_desc_length = 1024
if len(description) > max_desc_length:
description = description[: max_desc_length - 3] + "..."
input_schema: Dict[str, Any] = {
"type": "object",
"properties": {},
"required": [],
}
if skill.metadata and isinstance(skill.metadata, dict):
params = skill.metadata.get("parameters")
if params and isinstance(params, dict):
input_schema = params
return {
"name": func_name,
"description": description,
"input_schema": input_schema,
}

View file

@ -0,0 +1,286 @@
"""
Sandbox Executor for LiteLLM Skills
Executes skill code in a sandboxed environment using llm-sandbox.
Supports Docker, Podman, and Kubernetes backends.
"""
import base64
import os
from typing import Any, Dict, List, Optional
from litellm._logging import verbose_logger
class SkillsSandboxExecutor:
"""
Executes skill code in llm-sandbox Docker container.
Responsibilities:
- Create sandbox session with skill files
- Install requirements
- Execute model-generated code
- Collect generated files (GIFs, images, etc.)
"""
def __init__(
self,
timeout: int = 60,
backend: str = "docker",
image: Optional[str] = None,
):
"""
Initialize the sandbox executor.
Args:
timeout: Maximum execution time in seconds
backend: Sandbox backend ("docker", "podman", "kubernetes")
image: Custom Docker image (default: uses llm-sandbox default)
"""
self.timeout = timeout
self.backend = backend
self.image = image
self._session = None
def execute(
self,
code: str,
skill_files: Dict[str, bytes],
requirements: Optional[str] = None,
) -> Dict[str, Any]:
"""
Execute code with skill files in sandbox.
Args:
code: Python code to execute
skill_files: Dict mapping file paths to binary content
requirements: Optional requirements.txt content
Returns:
{
"success": bool,
"output": str,
"error": str (if failed),
"files": [{"name": str, "content_base64": str, "mime_type": str}]
}
"""
try:
from llm_sandbox import SandboxSession
except ImportError:
verbose_logger.error(
"SkillsSandboxExecutor: llm-sandbox not installed. "
"Install with: pip install llm-sandbox"
)
return {
"success": False,
"output": "",
"error": "llm-sandbox not installed. Install with: pip install llm-sandbox",
"files": [],
}
try:
# Create sandbox session
session_kwargs: Dict[str, Any] = {
"lang": "python",
"verbose": False,
}
if self.image:
session_kwargs["image"] = self.image
with SandboxSession(**session_kwargs) as session:
# 1. Copy skill files into sandbox using copy_to_runtime
import tempfile
# Create a temp directory to stage files
with tempfile.TemporaryDirectory() as tmpdir:
for path, content in skill_files.items():
# Create the file in temp directory
local_path = os.path.join(tmpdir, path)
os.makedirs(os.path.dirname(local_path), exist_ok=True)
with open(local_path, "wb") as f:
f.write(content)
# Copy to sandbox
sandbox_path = f"/sandbox/{path}"
session.copy_to_runtime(local_path, sandbox_path)
verbose_logger.debug(
f"SkillsSandboxExecutor: Copied {len(skill_files)} files to sandbox"
)
# 2. Install requirements if present
req_packages = None
if requirements:
req_packages = requirements.strip().replace("\n", " ")
elif "requirements.txt" in skill_files:
req_content = skill_files["requirements.txt"].decode("utf-8")
req_packages = req_content.strip().replace("\n", " ")
if req_packages:
# Run pip install as code
pip_code = f"""
import subprocess
subprocess.run(['pip', 'install'] + '{req_packages}'.split(), check=True)
"""
result = session.run(pip_code)
verbose_logger.debug(
"SkillsSandboxExecutor: Installed requirements"
)
# 3. Execute the code
# Wrap code to run from /sandbox directory
wrapped_code = f"""
import os
os.chdir('/sandbox')
import sys
sys.path.insert(0, '/sandbox')
{code}
"""
result = session.run(wrapped_code)
success = result.exit_code == 0
output = result.stdout or ""
error = result.stderr or ""
if success:
verbose_logger.debug(
"SkillsSandboxExecutor: Code execution succeeded"
)
else:
verbose_logger.debug(
f"SkillsSandboxExecutor: Code execution failed with exit code {result.exit_code}"
)
verbose_logger.debug(
f"SkillsSandboxExecutor: stderr: {error[:500] if error else 'No stderr'}"
)
verbose_logger.debug(
f"SkillsSandboxExecutor: stdout: {output[:500] if output else 'No stdout'}"
)
# 4. Collect generated files
generated_files = self._collect_generated_files(session, skill_files)
return {
"success": success,
"output": output,
"error": error,
"files": generated_files,
}
except Exception as e:
verbose_logger.error(
f"SkillsSandboxExecutor: Execution failed: {e}"
)
return {
"success": False,
"output": "",
"error": str(e),
"files": [],
}
def _collect_generated_files(
self,
session: Any,
original_files: Dict[str, bytes],
) -> List[Dict[str, Any]]:
"""
Collect files generated during execution.
Looks for new files in /sandbox that weren't in the original skill files.
Focuses on common output types: GIF, PNG, JPG, PDF, CSV, etc.
Args:
session: The sandbox session
original_files: Original skill files (to exclude)
Returns:
List of generated files with base64 content
"""
generated_files: List[Dict[str, Any]] = []
try:
import tempfile
# List files in /sandbox using Python code
list_code = """
import os
import json
files = []
for root, dirs, filenames in os.walk('/sandbox'):
for f in filenames:
if f.endswith(('.gif', '.png', '.jpg', '.jpeg', '.pdf', '.csv', '.json')):
files.append(os.path.join(root, f))
print(json.dumps(files))
"""
result = session.run(list_code)
if result.exit_code == 0 and result.stdout:
import json
try:
filepaths = json.loads(result.stdout.strip())
except json.JSONDecodeError:
filepaths = []
for filepath in filepaths:
if not filepath:
continue
# Get relative path
rel_path = filepath.replace("/sandbox/", "")
# Skip if it was an original file
if rel_path in original_files:
continue
# Copy file from sandbox using copy_from_runtime
with tempfile.NamedTemporaryFile(delete=False) as tmp:
tmp_path = tmp.name
try:
session.copy_from_runtime(filepath, tmp_path)
with open(tmp_path, "rb") as f:
content = f.read()
content_b64 = base64.b64encode(content).decode("utf-8")
generated_files.append({
"name": os.path.basename(filepath),
"path": rel_path,
"content_base64": content_b64,
"mime_type": self._get_mime_type(filepath),
})
verbose_logger.debug(
f"SkillsSandboxExecutor: Collected generated file: {rel_path}"
)
except Exception as e:
verbose_logger.warning(
f"SkillsSandboxExecutor: Error copying file {filepath}: {e}"
)
finally:
if os.path.exists(tmp_path):
os.unlink(tmp_path)
except Exception as e:
verbose_logger.warning(
f"SkillsSandboxExecutor: Error collecting generated files: {e}"
)
return generated_files
def _get_mime_type(self, filename: str) -> str:
"""Get MIME type for a file based on extension."""
ext = filename.lower().split(".")[-1]
return {
"gif": "image/gif",
"png": "image/png",
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
"pdf": "application/pdf",
"csv": "text/csv",
"json": "application/json",
"txt": "text/plain",
}.get(ext, "application/octet-stream")

View file

@ -0,0 +1,336 @@
"""
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",
)

View file

@ -1140,6 +1140,60 @@ class MakeMCPServersPublicRequest(LiteLLMPydanticObjectBase):
mcp_server_ids: List[str]
######## Skills API Types ########
class NewSkillRequest(LiteLLMPydanticObjectBase):
"""Request to create a new skill in LiteLLM database"""
display_title: Optional[str] = None
description: Optional[str] = None
instructions: Optional[str] = None
file_content: Optional[bytes] = None # Binary content of skill files (zip)
file_name: Optional[str] = None # Original filename
file_type: Optional[str] = None # MIME type (e.g., "application/zip")
metadata: Optional[Dict[str, Any]] = None
class UpdateSkillRequest(LiteLLMPydanticObjectBase):
"""Request to update an existing skill"""
skill_id: str
display_title: Optional[str] = None
description: Optional[str] = None
instructions: Optional[str] = None
file_content: Optional[bytes] = None # Binary content of skill files (zip)
file_name: Optional[str] = None # Original filename
file_type: Optional[str] = None # MIME type
metadata: Optional[Dict[str, Any]] = None
class LiteLLM_SkillsTable(LiteLLMPydanticObjectBase):
"""Represents a LiteLLM_SkillsTable record"""
skill_id: str
display_title: Optional[str] = None
description: Optional[str] = None
instructions: Optional[str] = None
source: str = "custom"
latest_version: Optional[str] = None
file_content: Optional[bytes] = None # Binary content of skill files (zip)
file_name: Optional[str] = None # Original filename
file_type: Optional[str] = None # MIME type
metadata: Optional[Dict[str, Any]] = None
created_at: Optional[datetime] = None
created_by: Optional[str] = None
updated_at: Optional[datetime] = None
updated_by: Optional[str] = None
class ListSkillsRequest(LiteLLMPydanticObjectBase):
"""Request to list skills from LiteLLM database"""
limit: Optional[int] = 20
offset: Optional[int] = 0
class NewUserRequestTeam(LiteLLMPydanticObjectBase):
team_id: str
max_budget_in_team: Optional[float] = None

View file

@ -3,6 +3,7 @@ from typing import Literal, Union
from . import *
from .cache_control_check import _PROXY_CacheControlCheck
from .litellm_skills import SkillsInjectionHook
from .max_budget_limiter import _PROXY_MaxBudgetLimiter
from .parallel_request_limiter import _PROXY_MaxParallelRequestsHandler
from .parallel_request_limiter_v3 import _PROXY_MaxParallelRequestsHandler_v3
@ -21,6 +22,7 @@ PROXY_HOOKS = {
"parallel_request_limiter": _PROXY_MaxParallelRequestsHandler_v3,
"cache_control_check": _PROXY_CacheControlCheck,
"responses_id_security": ResponsesIDSecurity,
"litellm_skills": SkillsInjectionHook,
}
## FEATURE FLAG HOOKS ##

View file

@ -0,0 +1,39 @@
"""
LiteLLM Skills Hook - Proxy integration for skills
This module provides the CustomLogger hook for skills processing.
The actual skill logic is in litellm/llms/litellm_proxy/skills/.
Usage:
from litellm.proxy.hooks.litellm_skills import SkillsInjectionHook
# Register hook in proxy
litellm.callbacks.append(SkillsInjectionHook())
"""
# Re-export from the SDK location for convenience
from litellm.llms.litellm_proxy.skills import (
LITELLM_CODE_EXECUTION_TOOL,
CodeExecutionHandler,
LiteLLMInternalTools,
SkillPromptInjectionHandler,
SkillsSandboxExecutor,
code_execution_handler,
get_litellm_code_execution_tool,
)
from litellm.proxy.hooks.litellm_skills.main import (
SkillsInjectionHook,
skills_injection_hook,
)
__all__ = [
"SkillsInjectionHook",
"skills_injection_hook",
"CodeExecutionHandler",
"LiteLLMInternalTools",
"LITELLM_CODE_EXECUTION_TOOL",
"get_litellm_code_execution_tool",
"code_execution_handler",
"SkillPromptInjectionHandler",
"SkillsSandboxExecutor",
]

View file

@ -0,0 +1,869 @@
"""
Skills Injection Hook for LiteLLM Proxy
Main hook that orchestrates skill processing:
- Fetches skills from LiteLLM DB
- Injects SKILL.md content into system prompt
- Adds litellm_code_execution tool for automatic code execution
- Handles agentic loop internally when litellm_code_execution is called
For non-Anthropic models (e.g., Bedrock, OpenAI, etc.):
- Skills are converted to OpenAI-style tools
- Skill file content (SKILL.md) is extracted and injected into the system prompt
- litellm_code_execution tool is added - when model calls it, LiteLLM handles
execution automatically and returns final response with file_ids
Usage:
# Simple - LiteLLM handles everything automatically via proxy
# The container parameter triggers the SkillsInjectionHook
response = await litellm.acompletion(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Create a bouncing ball GIF"}],
container={"skills": [{"skill_id": "litellm:skill_abc123"}]},
)
# Response includes file_ids for generated files
"""
import base64
import json
from typing import Any, Dict, List, Optional, Union
from litellm._logging import verbose_proxy_logger
from litellm.caching.caching import DualCache
from litellm.integrations.custom_logger import CustomLogger
from litellm.llms.litellm_proxy.skills.prompt_injection import (
SkillPromptInjectionHandler,
)
from litellm.proxy._types import LiteLLM_SkillsTable, UserAPIKeyAuth
from litellm.types.utils import CallTypes, CallTypesLiteral
class SkillsInjectionHook(CustomLogger):
"""
Pre/Post-call hook that processes skills from container.skills parameter.
Pre-call (async_pre_call_hook):
- Skills with 'litellm:' prefix are fetched from LiteLLM DB
- For Anthropic models: native skills pass through, LiteLLM skills converted to tools
- For non-Anthropic models: LiteLLM skills are converted to tools + execute_code tool
Post-call (async_post_call_success_deployment_hook):
- If response has litellm_code_execution tool call, automatically execute code
- Continue conversation loop until model gives final response
- Return response with generated files inline
This hook is called automatically by litellm during completion calls.
"""
def __init__(self, **kwargs):
from litellm.llms.litellm_proxy.skills.constants import (
DEFAULT_MAX_ITERATIONS,
DEFAULT_SANDBOX_TIMEOUT,
)
self.optional_params = kwargs
self.prompt_handler = SkillPromptInjectionHandler()
self.max_iterations = kwargs.get("max_iterations", DEFAULT_MAX_ITERATIONS)
self.sandbox_timeout = kwargs.get("sandbox_timeout", DEFAULT_SANDBOX_TIMEOUT)
super().__init__(**kwargs)
async def async_pre_call_hook(
self,
user_api_key_dict: UserAPIKeyAuth,
cache: DualCache,
data: dict,
call_type: CallTypesLiteral,
) -> Optional[Union[Exception, str, dict]]:
"""
Process skills from container.skills before the LLM call.
1. Check if container.skills exists in request
2. Separate skills by prefix (litellm: vs native)
3. Fetch LiteLLM skills from database
4. For Anthropic: keep native skills in container
5. For non-Anthropic: convert LiteLLM skills to tools, inject content, add execute_code
"""
# Only process completion-type calls
if call_type not in ["completion", "acompletion", "anthropic_messages"]:
return data
container = data.get("container")
if not container or not isinstance(container, dict):
return data
skills = container.get("skills")
if not skills or not isinstance(skills, list):
return data
verbose_proxy_logger.debug(f"SkillsInjectionHook: Processing {len(skills)} skills")
litellm_skills: List[LiteLLM_SkillsTable] = []
anthropic_skills: List[Dict[str, Any]] = []
# Separate skills by prefix
for skill in skills:
if not isinstance(skill, dict):
continue
skill_id = skill.get("skill_id", "")
if skill_id.startswith("litellm_"):
# Fetch from LiteLLM DB
db_skill = await self._fetch_skill_from_db(skill_id)
if db_skill:
litellm_skills.append(db_skill)
else:
verbose_proxy_logger.warning(
f"SkillsInjectionHook: Skill '{skill_id}' not found in LiteLLM DB"
)
else:
# Native Anthropic skill - pass through
anthropic_skills.append(skill)
# Check if using messages API spec (anthropic_messages call type)
# Messages API always uses Anthropic-style tool format
use_anthropic_format = call_type == "anthropic_messages"
if len(litellm_skills) > 0:
data = self._process_for_messages_api(
data=data,
litellm_skills=litellm_skills,
use_anthropic_format=use_anthropic_format,
)
return data
def _process_for_messages_api(
self,
data: dict,
litellm_skills: List[LiteLLM_SkillsTable],
use_anthropic_format: bool = True,
) -> dict:
"""
Process skills for messages API (Anthropic format tools).
- Converts skills to Anthropic-style tools (name, description, input_schema)
- Extracts and injects SKILL.md content into system prompt
- Adds litellm_code_execution tool for code execution
- Stores skill files in metadata for sandbox execution
"""
from litellm.llms.litellm_proxy.skills.code_execution import (
get_litellm_code_execution_tool_anthropic,
)
tools = data.get("tools", [])
skill_contents: List[str] = []
all_skill_files: Dict[str, Dict[str, bytes]] = {}
all_module_paths: List[str] = []
for skill in litellm_skills:
# Convert skill to Anthropic-style tool
tools.append(self.prompt_handler.convert_skill_to_anthropic_tool(skill))
# Extract skill content from file if available
content = self.prompt_handler.extract_skill_content(skill)
if content:
skill_contents.append(content)
# Extract all files for code execution
skill_files = self.prompt_handler.extract_all_files(skill)
if skill_files:
all_skill_files[skill.skill_id] = skill_files
for path in skill_files.keys():
if path.endswith(".py"):
all_module_paths.append(path)
if tools:
data["tools"] = tools
# Inject skill content into system prompt
# For Anthropic messages API, use top-level 'system' param instead of messages array
if skill_contents:
data = self.prompt_handler.inject_skill_content_to_messages(
data, skill_contents, use_anthropic_format=use_anthropic_format
)
# Add litellm_code_execution tool if we have skill files
if all_skill_files:
code_exec_tool = get_litellm_code_execution_tool_anthropic()
data["tools"] = data.get("tools", []) + [code_exec_tool]
# Store skill files in litellm_metadata for automatic code execution
data["litellm_metadata"] = data.get("litellm_metadata", {})
data["litellm_metadata"]["_skill_files"] = all_skill_files
data["litellm_metadata"]["_litellm_code_execution_enabled"] = True
# Remove container (not supported by underlying providers)
data.pop("container", None)
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Messages API - converted {len(litellm_skills)} skills to Anthropic tools, "
f"injected {len(skill_contents)} skill contents, "
f"added litellm_code_execution tool with {len(all_module_paths)} modules"
)
return data
def _process_non_anthropic_model(
self,
data: dict,
litellm_skills: List[LiteLLM_SkillsTable],
) -> dict:
"""
Process skills for non-Anthropic models (OpenAI format tools).
- Converts skills to OpenAI-style tools
- Extracts and injects SKILL.md content
- Adds execute_code tool for code execution
- Stores skill files in metadata for sandbox execution
"""
tools = data.get("tools", [])
skill_contents: List[str] = []
all_skill_files: Dict[str, Dict[str, bytes]] = {}
all_module_paths: List[str] = []
for skill in litellm_skills:
# Convert skill to OpenAI-style tool
tools.append(self.prompt_handler.convert_skill_to_tool(skill))
# Extract skill content from file if available
content = self.prompt_handler.extract_skill_content(skill)
if content:
skill_contents.append(content)
# Extract all files for code execution
skill_files = self.prompt_handler.extract_all_files(skill)
if skill_files:
all_skill_files[skill.skill_id] = skill_files
# Collect Python module paths
for path in skill_files.keys():
if path.endswith(".py"):
all_module_paths.append(path)
if tools:
data["tools"] = tools
# Inject skill content into system prompt
if skill_contents:
data = self.prompt_handler.inject_skill_content_to_messages(data, skill_contents)
# Add litellm_code_execution tool if we have skill files
if all_skill_files:
from litellm.llms.litellm_proxy.skills.code_execution import (
get_litellm_code_execution_tool,
)
data["tools"] = data.get("tools", []) + [get_litellm_code_execution_tool()]
# Store skill files in litellm_metadata for automatic code execution
# Using litellm_metadata instead of metadata to avoid conflicts with user metadata
data["litellm_metadata"] = data.get("litellm_metadata", {})
data["litellm_metadata"]["_skill_files"] = all_skill_files
data["litellm_metadata"]["_litellm_code_execution_enabled"] = True
# Remove container for non-Anthropic (they don't support it)
data.pop("container", None)
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Non-Anthropic model - converted {len(litellm_skills)} skills to tools, "
f"injected {len(skill_contents)} skill contents, "
f"added execute_code tool with {len(all_module_paths)} modules"
)
return data
async def _fetch_skill_from_db(self, skill_id: str) -> Optional[LiteLLM_SkillsTable]:
"""
Fetch a skill from the LiteLLM database.
Args:
skill_id: The skill ID (without 'litellm:' prefix)
Returns:
LiteLLM_SkillsTable or None if not found
"""
try:
from litellm.llms.litellm_proxy.skills.handler import LiteLLMSkillsHandler
return await LiteLLMSkillsHandler.fetch_skill_from_db(skill_id)
except Exception as e:
verbose_proxy_logger.warning(
f"SkillsInjectionHook: Error fetching skill {skill_id}: {e}"
)
return None
def _is_anthropic_model(self, model: str) -> bool:
"""
Check if the model is an Anthropic model using get_llm_provider.
Args:
model: The model name/identifier
Returns:
True if Anthropic model, False otherwise
"""
try:
from litellm.litellm_core_utils.get_llm_provider_logic import (
get_llm_provider,
)
_, custom_llm_provider, _, _ = get_llm_provider(model=model)
return custom_llm_provider == "anthropic"
except Exception:
# Fallback to simple check if get_llm_provider fails
return "claude" in model.lower() or model.lower().startswith("anthropic/")
async def async_post_call_success_deployment_hook(
self,
request_data: dict,
response: Any,
call_type: Optional[CallTypes],
) -> Optional[Any]:
"""
Post-call hook to handle automatic code execution.
Handles both OpenAI format (response.choices) and Anthropic/messages API
format (response["content"]).
If the response contains a tool call (litellm_code_execution or skill tool):
1. Execute the code in sandbox
2. Add result to messages
3. Make another LLM call
4. Repeat until model gives final response
5. Return modified response with generated files
"""
from litellm.llms.litellm_proxy.skills.code_execution import (
LiteLLMInternalTools,
)
# Check if code execution is enabled for this request
litellm_metadata = request_data.get("litellm_metadata", {})
metadata = request_data.get("metadata", {})
code_exec_enabled = (
litellm_metadata.get("_litellm_code_execution_enabled") or
metadata.get("_litellm_code_execution_enabled")
)
if not code_exec_enabled:
return None
# Get skill files
skill_files_by_id = (
litellm_metadata.get("_skill_files") or
metadata.get("_skill_files", {})
)
all_skill_files: Dict[str, bytes] = {}
for files_dict in skill_files_by_id.values():
all_skill_files.update(files_dict)
if not all_skill_files:
verbose_proxy_logger.warning(
"SkillsInjectionHook: No skill files found, cannot execute code"
)
return None
# Check for tool calls - handle both Anthropic and OpenAI formats
tool_calls = self._extract_tool_calls(response)
if not tool_calls:
return None
# Check if any tool call needs execution (litellm_code_execution or skill tool)
has_executable_tool = False
for tc in tool_calls:
tool_name = tc.get("name", "")
# Execute if it's litellm_code_execution OR a skill tool (skill_xxx)
if tool_name == LiteLLMInternalTools.CODE_EXECUTION.value or tool_name.startswith("skill_"):
has_executable_tool = True
break
if not has_executable_tool:
return None
verbose_proxy_logger.debug(
"SkillsInjectionHook: Detected tool call, starting execution loop"
)
# Start the agentic loop
return await self._execute_code_loop_messages_api(
data=request_data,
response=response,
skill_files=all_skill_files,
)
def _extract_tool_calls(self, response: Any) -> List[Dict[str, Any]]:
"""Extract tool calls from response, handling both formats."""
tool_calls = []
# Get content - handle both dict and object responses
content = None
if isinstance(response, dict):
content = response.get("content", [])
elif hasattr(response, "content"):
content = response.content
# Anthropic/messages API format: response has "content" list with tool_use blocks
if content:
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
tool_calls.append({
"id": block.get("id"),
"name": block.get("name"),
"input": block.get("input", {}),
})
elif hasattr(block, "type") and getattr(block, "type", None) == "tool_use":
tool_calls.append({
"id": getattr(block, "id", None),
"name": getattr(block, "name", None),
"input": getattr(block, "input", {}),
})
# OpenAI format: response has choices[0].message.tool_calls
if not tool_calls and hasattr(response, "choices") and response.choices: # type: ignore[union-attr]
msg = response.choices[0].message # type: ignore[union-attr]
if hasattr(msg, "tool_calls") and msg.tool_calls:
for tc in msg.tool_calls:
tool_calls.append({
"id": tc.id,
"name": tc.function.name,
"input": json.loads(tc.function.arguments) if tc.function.arguments else {},
})
return tool_calls
async def _execute_code_loop_messages_api(
self,
data: dict,
response: Any,
skill_files: Dict[str, bytes],
) -> Any:
"""
Execute the code execution loop for messages API (Anthropic format).
Returns the final response with generated files inline.
"""
import litellm
from litellm.llms.litellm_proxy.skills.code_execution import (
LiteLLMInternalTools,
)
from litellm.llms.litellm_proxy.skills.sandbox_executor import (
SkillsSandboxExecutor,
)
# Ensure response is not None
if response is None:
verbose_proxy_logger.error(
"SkillsInjectionHook: Response is None, cannot execute code loop"
)
return None
model = data.get("model", "")
messages = list(data.get("messages", []))
tools = data.get("tools", [])
max_tokens = data.get("max_tokens", 4096)
executor = SkillsSandboxExecutor(timeout=self.sandbox_timeout)
generated_files: List[Dict[str, Any]] = []
current_response = response
for iteration in range(self.max_iterations):
# Extract tool calls from current response
tool_calls = self._extract_tool_calls(current_response)
stop_reason = current_response.get("stop_reason") if isinstance(current_response, dict) else getattr(current_response, "stop_reason", None)
# Get content for assistant message - convert to plain dicts
raw_content = current_response.get("content", []) if isinstance(current_response, dict) else getattr(current_response, "content", [])
content_blocks = []
for block in raw_content or []:
if isinstance(block, dict):
content_blocks.append(block)
elif hasattr(block, "model_dump"):
content_blocks.append(block.model_dump())
elif hasattr(block, "__dict__"):
content_blocks.append(dict(block.__dict__))
else:
content_blocks.append({"type": "text", "text": str(block)})
# Build assistant message for conversation history (Anthropic format)
assistant_msg = {"role": "assistant", "content": content_blocks}
messages.append(assistant_msg)
# Check if we're done (no tool calls)
if stop_reason != "tool_use" or not tool_calls:
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Loop completed after {iteration + 1} iterations, "
f"{len(generated_files)} files generated"
)
return self._attach_files_to_response(current_response, generated_files)
# Process tool calls
tool_results = []
for tc in tool_calls:
tool_name = tc.get("name", "")
tool_id = tc.get("id", "")
tool_input = tc.get("input", {})
# Execute if it's litellm_code_execution OR a skill tool
if tool_name == LiteLLMInternalTools.CODE_EXECUTION.value:
code = tool_input.get("code", "")
result = await self._execute_code(code, skill_files, executor, generated_files)
elif tool_name.startswith("skill_"):
# Skill tool - execute the skill's code
result = await self._execute_skill_tool(tool_name, tool_input, skill_files, executor, generated_files)
else:
result = f"Tool '{tool_name}' not handled"
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_id,
"content": result,
})
# Add tool results to messages (Anthropic format)
messages.append({"role": "user", "content": tool_results})
# Make next LLM call
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Making LLM call iteration {iteration + 2}"
)
try:
current_response = await litellm.anthropic.acreate(
model=model,
messages=messages,
tools=tools,
max_tokens=max_tokens,
)
if current_response is None:
verbose_proxy_logger.error(
"SkillsInjectionHook: LLM call returned None"
)
return self._attach_files_to_response(response, generated_files)
except Exception as e:
verbose_proxy_logger.error(
f"SkillsInjectionHook: LLM call failed: {e}"
)
return self._attach_files_to_response(response, generated_files)
verbose_proxy_logger.warning(
f"SkillsInjectionHook: Max iterations ({self.max_iterations}) reached"
)
return self._attach_files_to_response(current_response, generated_files)
async def _execute_code(
self,
code: str,
skill_files: Dict[str, bytes],
executor: Any,
generated_files: List[Dict[str, Any]],
) -> str:
"""Execute code in sandbox and return result string."""
try:
verbose_proxy_logger.debug(f"SkillsInjectionHook: Executing code ({len(code)} chars)")
exec_result = executor.execute(code=code, skill_files=skill_files)
result = exec_result.get("output", "") or ""
# Collect generated files
if exec_result.get("files"):
for f in exec_result["files"]:
generated_files.append({
"name": f["name"],
"mime_type": f["mime_type"],
"content_base64": f["content_base64"],
"size": len(base64.b64decode(f["content_base64"])),
})
result += f"\n\nGenerated file: {f['name']}"
if exec_result.get("error"):
result += f"\n\nError: {exec_result['error']}"
return result or "Code executed successfully"
except Exception as e:
return f"Code execution failed: {str(e)}"
async def _execute_skill_tool(
self,
tool_name: str,
tool_input: Dict[str, Any],
skill_files: Dict[str, bytes],
executor: Any,
generated_files: List[Dict[str, Any]],
) -> str:
"""Execute a skill tool by generating and running code based on skill content."""
# Generate code based on available skill modules
# Look for Python modules in the skill
python_modules = [p for p in skill_files.keys() if p.endswith(".py") and not p.endswith("__init__.py")]
# Try to find the main builder/creator module
main_module = None
for mod in python_modules:
if "builder" in mod.lower() or "creator" in mod.lower() or "generator" in mod.lower():
main_module = mod
break
if not main_module and python_modules:
# Use first non-init module
main_module = python_modules[0]
if main_module:
# Convert path to import: "core/gif_builder.py" -> "core.gif_builder"
import_path = main_module.replace("/", ".").replace(".py", "")
# Generate code that imports and uses the module
code = f"""
# Auto-generated code to execute skill
import sys
sys.path.insert(0, '/sandbox')
from {import_path} import *
# Try to find and use a Builder/Creator class
import inspect
module = __import__('{import_path}', fromlist=[''])
for name, obj in inspect.getmembers(module):
if inspect.isclass(obj) and name != 'object':
try:
instance = obj()
# Try common methods
if hasattr(instance, 'create'):
result = instance.create()
elif hasattr(instance, 'build'):
result = instance.build()
elif hasattr(instance, 'generate'):
result = instance.generate()
elif hasattr(instance, 'save'):
instance.save('output.gif')
print(f'Used {{name}} class')
break
except Exception as e:
print(f'Error with {{name}}: {{e}}')
continue
# List generated files
import os
for f in os.listdir('.'):
if f.endswith(('.gif', '.png', '.jpg')):
print(f'Generated: {{f}}')
"""
else:
# Fallback generic code
code = """
print('No executable skill module found')
"""
return await self._execute_code(code, skill_files, executor, generated_files)
async def _execute_code_loop(
self,
data: dict,
response: Any,
skill_files: Dict[str, bytes],
) -> Any:
"""
Execute the code execution loop until model gives final response.
Returns the final response with generated files inline.
"""
import litellm
from litellm.llms.litellm_proxy.skills.code_execution import (
LiteLLMInternalTools,
)
from litellm.llms.litellm_proxy.skills.sandbox_executor import (
SkillsSandboxExecutor,
)
model = data.get("model", "")
messages = list(data.get("messages", []))
tools = data.get("tools", [])
# Keys to exclude when passing through to acompletion
# These are either handled explicitly or are internal LiteLLM fields
_EXCLUDED_ACOMPLETION_KEYS = frozenset({
"messages",
"model",
"tools",
"metadata",
"litellm_metadata",
"container",
})
kwargs = {
k: v for k, v in data.items()
if k not in _EXCLUDED_ACOMPLETION_KEYS
}
executor = SkillsSandboxExecutor(timeout=self.sandbox_timeout)
generated_files: List[Dict[str, Any]] = []
current_response: Any = response
for iteration in range(self.max_iterations):
# OpenAI format response has choices[0].message
assistant_message = current_response.choices[0].message # type: ignore[union-attr]
stop_reason = current_response.choices[0].finish_reason # type: ignore[union-attr]
# Build assistant message for conversation history
assistant_msg_dict: Dict[str, Any] = {
"role": "assistant",
"content": assistant_message.content,
}
if assistant_message.tool_calls:
assistant_msg_dict["tool_calls"] = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in assistant_message.tool_calls
]
messages.append(assistant_msg_dict)
# Check if we're done (no tool calls)
if stop_reason != "tool_calls" or not assistant_message.tool_calls:
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Code execution loop completed after "
f"{iteration + 1} iterations, {len(generated_files)} files generated"
)
# Attach generated files to response
return self._attach_files_to_response(current_response, generated_files)
# Process tool calls
for tool_call in assistant_message.tool_calls:
tool_name = tool_call.function.name
if tool_name == LiteLLMInternalTools.CODE_EXECUTION.value:
tool_result = await self._execute_code_tool(
tool_call=tool_call,
skill_files=skill_files,
executor=executor,
generated_files=generated_files,
)
else:
# Non-code-execution tool - cannot handle
tool_result = f"Tool '{tool_name}' not handled automatically"
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_result,
})
# Make next LLM call using the messages API
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Making LLM call iteration {iteration + 2}"
)
current_response = await litellm.anthropic.acreate(
model=model,
messages=messages,
tools=tools,
max_tokens=kwargs.get("max_tokens", 4096),
)
# Max iterations reached
verbose_proxy_logger.warning(
f"SkillsInjectionHook: Max iterations ({self.max_iterations}) reached"
)
return self._attach_files_to_response(current_response, generated_files)
async def _execute_code_tool(
self,
tool_call: Any,
skill_files: Dict[str, bytes],
executor: Any,
generated_files: List[Dict[str, Any]],
) -> str:
"""Execute a litellm_code_execution tool call and return result string."""
try:
args = json.loads(tool_call.function.arguments)
code = args.get("code", "")
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Executing code ({len(code)} chars)"
)
exec_result = executor.execute(
code=code,
skill_files=skill_files,
)
# Build tool result content
tool_result = exec_result.get("output", "") or ""
# Collect generated files
if exec_result.get("files"):
tool_result += "\n\nGenerated files:"
for f in exec_result["files"]:
file_content = base64.b64decode(f["content_base64"])
generated_files.append({
"name": f["name"],
"mime_type": f["mime_type"],
"content_base64": f["content_base64"],
"size": len(file_content),
})
tool_result += f"\n- {f['name']} ({len(file_content)} bytes)"
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Generated file {f['name']} "
f"({len(file_content)} bytes)"
)
if exec_result.get("error"):
tool_result += f"\n\nError:\n{exec_result['error']}"
return tool_result
except Exception as e:
verbose_proxy_logger.error(
f"SkillsInjectionHook: Code execution failed: {e}"
)
return f"Code execution failed: {str(e)}"
def _attach_files_to_response(
self,
response: Any,
generated_files: List[Dict[str, Any]],
) -> Any:
"""
Attach generated files to the response object.
Files are added to response._litellm_generated_files for easy access.
For dict responses, files are added as a key.
"""
if not generated_files:
return response
# Handle dict response (Anthropic/messages API format)
if isinstance(response, dict):
response["_litellm_generated_files"] = generated_files
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Attached {len(generated_files)} files to dict response"
)
return response
# Handle object response (OpenAI format)
try:
response._litellm_generated_files = generated_files
except AttributeError:
pass
# Also add to model_extra if available (for serialization)
if hasattr(response, "model_extra"):
if response.model_extra is None:
response.model_extra = {}
response.model_extra["_litellm_generated_files"] = generated_files
verbose_proxy_logger.debug(
f"SkillsInjectionHook: Attached {len(generated_files)} files to response"
)
return response
# Global instance for registration
skills_injection_hook = SkillsInjectionHook()
import litellm
litellm.logging_callback_manager.add_litellm_callback(skills_injection_hook)

View file

@ -1,10 +1,5 @@
model_list:
- model_name: gemini/*
- model_name: anthropic/*
litellm_params:
model: gemini/*
model: anthropic/*
litellm_settings:
callbacks: ["dynamic_rate_limiter_v3"]
priority_reservation:
"prod": 0.9 # 90% reserved for production
"dev": 0.1 # 10% reserved for development

View file

@ -727,4 +727,22 @@ model LiteLLM_UISettings {
ui_settings Json
created_at DateTime @default(now())
updated_at DateTime @updatedAt
}
// Skills table for storing LiteLLM-managed skills
model LiteLLM_SkillsTable {
skill_id String @id @default(uuid())
display_title String?
description String?
instructions String? // The skill instructions/prompt (from SKILL.md)
source String @default("custom") // "custom" or "anthropic"
latest_version String?
file_content Bytes? // Binary content of the skill files (zip)
file_name String? // Original filename
file_type String? // MIME type (e.g., "application/zip")
metadata Json? @default("{}")
created_at DateTime @default(now())
created_by String?
updated_at DateTime @default(now()) @updatedAt
updated_by String?
}

View file

@ -23,12 +23,27 @@ from litellm.types.llms.anthropic_skills import (
Skill,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager, client
# Initialize HTTP handler
base_llm_http_handler = BaseLLMHTTPHandler()
DEFAULT_ANTHROPIC_API_BASE = "https://api.anthropic.com/v1"
# Initialize LiteLLM skills handler (lazy - only used when custom_llm_provider="litellm")
_litellm_skills_handler = None
def _get_litellm_skills_handler():
"""Lazy initialization of LiteLLM skills handler to avoid import overhead."""
global _litellm_skills_handler
if _litellm_skills_handler is None:
from litellm.llms.litellm_proxy.skills.transformation import (
LiteLLMSkillsTransformationHandler,
)
_litellm_skills_handler = LiteLLMSkillsTransformationHandler()
return _litellm_skills_handler
@client
async def acreate_skill(
@ -133,18 +148,6 @@ def create_skill(
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:
@ -156,6 +159,30 @@ def create_skill(
if extra_body:
create_request.update(extra_body) # type: ignore
# Route to LiteLLM DB if custom_llm_provider="litellm_proxy"
if custom_llm_provider == LlmProviders.LITELLM_PROXY.value:
return _get_litellm_skills_handler().create_skill_handler(
display_title=display_title,
files=files,
metadata=extra_body.get("metadata") if extra_body else None,
user_id=kwargs.get("user_id"),
_is_async=_is_async,
logging_obj=litellm_logging_obj,
litellm_call_id=litellm_call_id,
)
# Get provider config for external providers (Anthropic, etc.)
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}"
)
# Validate environment and get headers
headers = extra_headers or {}
headers = skills_api_provider_config.validate_environment(
@ -316,7 +343,17 @@ def list_skills(
if custom_llm_provider is None:
custom_llm_provider = "anthropic"
# Get provider config
# Route to LiteLLM DB if custom_llm_provider="litellm_proxy"
if custom_llm_provider == LlmProviders.LITELLM_PROXY.value:
return _get_litellm_skills_handler().list_skills_handler(
limit=limit or 20,
offset=0,
_is_async=_is_async,
logging_obj=litellm_logging_obj,
litellm_call_id=litellm_call_id,
)
# Get provider config for external providers (Anthropic, etc.)
skills_api_provider_config: Optional[BaseSkillsAPIConfig] = (
ProviderConfigManager.get_provider_skills_api_config(
provider=litellm.LlmProviders(custom_llm_provider),
@ -481,7 +518,16 @@ def get_skill(
if custom_llm_provider is None:
custom_llm_provider = "anthropic"
# Get provider config
# Route to LiteLLM DB if custom_llm_provider="litellm_proxy"
if custom_llm_provider == LlmProviders.LITELLM_PROXY.value:
return _get_litellm_skills_handler().get_skill_handler(
skill_id=skill_id,
_is_async=_is_async,
logging_obj=litellm_logging_obj,
litellm_call_id=litellm_call_id,
)
# Get provider config for external providers (Anthropic, etc.)
skills_api_provider_config: Optional[BaseSkillsAPIConfig] = (
ProviderConfigManager.get_provider_skills_api_config(
provider=litellm.LlmProviders(custom_llm_provider),
@ -638,7 +684,16 @@ def delete_skill(
if custom_llm_provider is None:
custom_llm_provider = "anthropic"
# Get provider config
# Route to LiteLLM DB if custom_llm_provider="litellm_proxy"
if custom_llm_provider == LlmProviders.LITELLM_PROXY.value:
return _get_litellm_skills_handler().delete_skill_handler(
skill_id=skill_id,
_is_async=_is_async,
logging_obj=litellm_logging_obj,
litellm_call_id=litellm_call_id,
)
# Get provider config for external providers (Anthropic, etc.)
skills_api_provider_config: Optional[BaseSkillsAPIConfig] = (
ProviderConfigManager.get_provider_skills_api_config(
provider=litellm.LlmProviders(custom_llm_provider),

View file

@ -358,6 +358,7 @@ class AnthropicMessagesRequestOptionalParams(TypedDict, total=False):
top_p: Optional[float]
mcp_servers: Optional[List[AnthropicMcpServerTool]]
context_management: Optional[Dict[str, Any]]
container: Optional[Dict[str, Any]] # Container config with skills for code execution
class AnthropicMessagesRequest(AnthropicMessagesRequestOptionalParams, total=False):

View file

@ -903,6 +903,7 @@ class ChatCompletionRequest(TypedDict, total=False):
functions: List
user: str
metadata: dict # litellm specific param
reasoning_effort: str # OpenAI o1/o3 reasoning parameter
class ChatCompletionDeltaChunk(TypedDict, total=False):

View file

@ -48,6 +48,7 @@ detect-secrets==1.5.0 # Enterprise - secret detection / masking in LLM requests
cryptography==44.0.1
tzdata==2025.1 # IANA time zone database
litellm-proxy-extras==0.4.14 # for proxy extras - e.g. prisma migrations
llm-sandbox==0.3.31 # for skill execution in sandbox
### LITELLM PACKAGE DEPENDENCIES
python-dotenv==1.0.1 # for env
tiktoken==0.8.0 # for calculating usage

View file

@ -727,4 +727,22 @@ model LiteLLM_UISettings {
ui_settings Json
created_at DateTime @default(now())
updated_at DateTime @updatedAt
}
// Skills table for storing LiteLLM-managed skills
model LiteLLM_SkillsTable {
skill_id String @id @default(uuid())
display_title String?
description String?
instructions String? // The skill instructions/prompt (from SKILL.md)
source String @default("custom") // "custom" or "anthropic"
latest_version String?
file_content Bytes? // Binary content of the skill files (zip)
file_name String? // Original filename
file_type String? // MIME type (e.g., "application/zip")
metadata Json? @default("{}")
created_at DateTime @default(now())
created_by String?
updated_at DateTime @default(now()) @updatedAt
updated_by String?
}

View file

@ -0,0 +1,202 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

View file

@ -0,0 +1,254 @@
---
name: slack-gif-creator
description: Knowledge and utilities for creating animated GIFs optimized for Slack. Provides constraints, validation tools, and animation concepts. Use when users request animated GIFs for Slack like "make me a GIF of X doing Y for Slack."
license: Complete terms in LICENSE.txt
---
# Slack GIF Creator
A toolkit providing utilities and knowledge for creating animated GIFs optimized for Slack.
## Slack Requirements
**Dimensions:**
- Emoji GIFs: 128x128 (recommended)
- Message GIFs: 480x480
**Parameters:**
- FPS: 10-30 (lower is smaller file size)
- Colors: 48-128 (fewer = smaller file size)
- Duration: Keep under 3 seconds for emoji GIFs
## Core Workflow
```python
from core.gif_builder import GIFBuilder
from PIL import Image, ImageDraw
# 1. Create builder
builder = GIFBuilder(width=128, height=128, fps=10)
# 2. Generate frames
for i in range(12):
frame = Image.new('RGB', (128, 128), (240, 248, 255))
draw = ImageDraw.Draw(frame)
# Draw your animation using PIL primitives
# (circles, polygons, lines, etc.)
builder.add_frame(frame)
# 3. Save with optimization
builder.save('output.gif', num_colors=48, optimize_for_emoji=True)
```
## Drawing Graphics
### Working with User-Uploaded Images
If a user uploads an image, consider whether they want to:
- **Use it directly** (e.g., "animate this", "split this into frames")
- **Use it as inspiration** (e.g., "make something like this")
Load and work with images using PIL:
```python
from PIL import Image
uploaded = Image.open('file.png')
# Use directly, or just as reference for colors/style
```
### Drawing from Scratch
When drawing graphics from scratch, use PIL ImageDraw primitives:
```python
from PIL import ImageDraw
draw = ImageDraw.Draw(frame)
# Circles/ovals
draw.ellipse([x1, y1, x2, y2], fill=(r, g, b), outline=(r, g, b), width=3)
# Stars, triangles, any polygon
points = [(x1, y1), (x2, y2), (x3, y3), ...]
draw.polygon(points, fill=(r, g, b), outline=(r, g, b), width=3)
# Lines
draw.line([(x1, y1), (x2, y2)], fill=(r, g, b), width=5)
# Rectangles
draw.rectangle([x1, y1, x2, y2], fill=(r, g, b), outline=(r, g, b), width=3)
```
**Don't use:** Emoji fonts (unreliable across platforms) or assume pre-packaged graphics exist in this skill.
### Making Graphics Look Good
Graphics should look polished and creative, not basic. Here's how:
**Use thicker lines** - Always set `width=2` or higher for outlines and lines. Thin lines (width=1) look choppy and amateurish.
**Add visual depth**:
- Use gradients for backgrounds (`create_gradient_background`)
- Layer multiple shapes for complexity (e.g., a star with a smaller star inside)
**Make shapes more interesting**:
- Don't just draw a plain circle - add highlights, rings, or patterns
- Stars can have glows (draw larger, semi-transparent versions behind)
- Combine multiple shapes (stars + sparkles, circles + rings)
**Pay attention to colors**:
- Use vibrant, complementary colors
- Add contrast (dark outlines on light shapes, light outlines on dark shapes)
- Consider the overall composition
**For complex shapes** (hearts, snowflakes, etc.):
- Use combinations of polygons and ellipses
- Calculate points carefully for symmetry
- Add details (a heart can have a highlight curve, snowflakes have intricate branches)
Be creative and detailed! A good Slack GIF should look polished, not like placeholder graphics.
## Available Utilities
### GIFBuilder (`core.gif_builder`)
Assembles frames and optimizes for Slack:
```python
builder = GIFBuilder(width=128, height=128, fps=10)
builder.add_frame(frame) # Add PIL Image
builder.add_frames(frames) # Add list of frames
builder.save('out.gif', num_colors=48, optimize_for_emoji=True, remove_duplicates=True)
```
### Validators (`core.validators`)
Check if GIF meets Slack requirements:
```python
from core.validators import validate_gif, is_slack_ready
# Detailed validation
passes, info = validate_gif('my.gif', is_emoji=True, verbose=True)
# Quick check
if is_slack_ready('my.gif'):
print("Ready!")
```
### Easing Functions (`core.easing`)
Smooth motion instead of linear:
```python
from core.easing import interpolate
# Progress from 0.0 to 1.0
t = i / (num_frames - 1)
# Apply easing
y = interpolate(start=0, end=400, t=t, easing='ease_out')
# Available: linear, ease_in, ease_out, ease_in_out,
# bounce_out, elastic_out, back_out
```
### Frame Helpers (`core.frame_composer`)
Convenience functions for common needs:
```python
from core.frame_composer import (
create_blank_frame, # Solid color background
create_gradient_background, # Vertical gradient
draw_circle, # Helper for circles
draw_text, # Simple text rendering
draw_star # 5-pointed star
)
```
## Animation Concepts
### Shake/Vibrate
Offset object position with oscillation:
- Use `math.sin()` or `math.cos()` with frame index
- Add small random variations for natural feel
- Apply to x and/or y position
### Pulse/Heartbeat
Scale object size rhythmically:
- Use `math.sin(t * frequency * 2 * math.pi)` for smooth pulse
- For heartbeat: two quick pulses then pause (adjust sine wave)
- Scale between 0.8 and 1.2 of base size
### Bounce
Object falls and bounces:
- Use `interpolate()` with `easing='bounce_out'` for landing
- Use `easing='ease_in'` for falling (accelerating)
- Apply gravity by increasing y velocity each frame
### Spin/Rotate
Rotate object around center:
- PIL: `image.rotate(angle, resample=Image.BICUBIC)`
- For wobble: use sine wave for angle instead of linear
### Fade In/Out
Gradually appear or disappear:
- Create RGBA image, adjust alpha channel
- Or use `Image.blend(image1, image2, alpha)`
- Fade in: alpha from 0 to 1
- Fade out: alpha from 1 to 0
### Slide
Move object from off-screen to position:
- Start position: outside frame bounds
- End position: target location
- Use `interpolate()` with `easing='ease_out'` for smooth stop
- For overshoot: use `easing='back_out'`
### Zoom
Scale and position for zoom effect:
- Zoom in: scale from 0.1 to 2.0, crop center
- Zoom out: scale from 2.0 to 1.0
- Can add motion blur for drama (PIL filter)
### Explode/Particle Burst
Create particles radiating outward:
- Generate particles with random angles and velocities
- Update each particle: `x += vx`, `y += vy`
- Add gravity: `vy += gravity_constant`
- Fade out particles over time (reduce alpha)
## Optimization Strategies
Only when asked to make the file size smaller, implement a few of the following methods:
1. **Fewer frames** - Lower FPS (10 instead of 20) or shorter duration
2. **Fewer colors** - `num_colors=48` instead of 128
3. **Smaller dimensions** - 128x128 instead of 480x480
4. **Remove duplicates** - `remove_duplicates=True` in save()
5. **Emoji mode** - `optimize_for_emoji=True` auto-optimizes
```python
# Maximum optimization for emoji
builder.save(
'emoji.gif',
num_colors=48,
optimize_for_emoji=True,
remove_duplicates=True
)
```
## Philosophy
This skill provides:
- **Knowledge**: Slack's requirements and animation concepts
- **Utilities**: GIFBuilder, validators, easing functions
- **Flexibility**: Create the animation logic using PIL primitives
It does NOT provide:
- Rigid animation templates or pre-made functions
- Emoji font rendering (unreliable across platforms)
- A library of pre-packaged graphics built into the skill
**Note on user uploads**: This skill doesn't include pre-built graphics, but if a user uploads an image, use PIL to load and work with it - interpret based on their request whether they want it used directly or just as inspiration.
Be creative! Combine concepts (bouncing + rotating, pulsing + sliding, etc.) and use PIL's full capabilities.
## Dependencies
```bash
pip install pillow imageio numpy
```

View file

@ -0,0 +1,234 @@
#!/usr/bin/env python3
"""
Easing Functions - Timing functions for smooth animations.
Provides various easing functions for natural motion and timing.
All functions take a value t (0.0 to 1.0) and return eased value (0.0 to 1.0).
"""
import math
def linear(t: float) -> float:
"""Linear interpolation (no easing)."""
return t
def ease_in_quad(t: float) -> float:
"""Quadratic ease-in (slow start, accelerating)."""
return t * t
def ease_out_quad(t: float) -> float:
"""Quadratic ease-out (fast start, decelerating)."""
return t * (2 - t)
def ease_in_out_quad(t: float) -> float:
"""Quadratic ease-in-out (slow start and end)."""
if t < 0.5:
return 2 * t * t
return -1 + (4 - 2 * t) * t
def ease_in_cubic(t: float) -> float:
"""Cubic ease-in (slow start)."""
return t * t * t
def ease_out_cubic(t: float) -> float:
"""Cubic ease-out (fast start)."""
return (t - 1) * (t - 1) * (t - 1) + 1
def ease_in_out_cubic(t: float) -> float:
"""Cubic ease-in-out."""
if t < 0.5:
return 4 * t * t * t
return (t - 1) * (2 * t - 2) * (2 * t - 2) + 1
def ease_in_bounce(t: float) -> float:
"""Bounce ease-in (bouncy start)."""
return 1 - ease_out_bounce(1 - t)
def ease_out_bounce(t: float) -> float:
"""Bounce ease-out (bouncy end)."""
if t < 1 / 2.75:
return 7.5625 * t * t
elif t < 2 / 2.75:
t -= 1.5 / 2.75
return 7.5625 * t * t + 0.75
elif t < 2.5 / 2.75:
t -= 2.25 / 2.75
return 7.5625 * t * t + 0.9375
else:
t -= 2.625 / 2.75
return 7.5625 * t * t + 0.984375
def ease_in_out_bounce(t: float) -> float:
"""Bounce ease-in-out."""
if t < 0.5:
return ease_in_bounce(t * 2) * 0.5
return ease_out_bounce(t * 2 - 1) * 0.5 + 0.5
def ease_in_elastic(t: float) -> float:
"""Elastic ease-in (spring effect)."""
if t == 0 or t == 1:
return t
return -math.pow(2, 10 * (t - 1)) * math.sin((t - 1.1) * 5 * math.pi)
def ease_out_elastic(t: float) -> float:
"""Elastic ease-out (spring effect)."""
if t == 0 or t == 1:
return t
return math.pow(2, -10 * t) * math.sin((t - 0.1) * 5 * math.pi) + 1
def ease_in_out_elastic(t: float) -> float:
"""Elastic ease-in-out."""
if t == 0 or t == 1:
return t
t = t * 2 - 1
if t < 0:
return -0.5 * math.pow(2, 10 * t) * math.sin((t - 0.1) * 5 * math.pi)
return math.pow(2, -10 * t) * math.sin((t - 0.1) * 5 * math.pi) * 0.5 + 1
# Convenience mapping
EASING_FUNCTIONS = {
"linear": linear,
"ease_in": ease_in_quad,
"ease_out": ease_out_quad,
"ease_in_out": ease_in_out_quad,
"bounce_in": ease_in_bounce,
"bounce_out": ease_out_bounce,
"bounce": ease_in_out_bounce,
"elastic_in": ease_in_elastic,
"elastic_out": ease_out_elastic,
"elastic": ease_in_out_elastic,
}
def get_easing(name: str = "linear"):
"""Get easing function by name."""
return EASING_FUNCTIONS.get(name, linear)
def interpolate(start: float, end: float, t: float, easing: str = "linear") -> float:
"""
Interpolate between two values with easing.
Args:
start: Start value
end: End value
t: Progress from 0.0 to 1.0
easing: Name of easing function
Returns:
Interpolated value
"""
ease_func = get_easing(easing)
eased_t = ease_func(t)
return start + (end - start) * eased_t
def ease_back_in(t: float) -> float:
"""Back ease-in (slight overshoot backward before forward motion)."""
c1 = 1.70158
c3 = c1 + 1
return c3 * t * t * t - c1 * t * t
def ease_back_out(t: float) -> float:
"""Back ease-out (overshoot forward then settle back)."""
c1 = 1.70158
c3 = c1 + 1
return 1 + c3 * pow(t - 1, 3) + c1 * pow(t - 1, 2)
def ease_back_in_out(t: float) -> float:
"""Back ease-in-out (overshoot at both ends)."""
c1 = 1.70158
c2 = c1 * 1.525
if t < 0.5:
return (pow(2 * t, 2) * ((c2 + 1) * 2 * t - c2)) / 2
return (pow(2 * t - 2, 2) * ((c2 + 1) * (t * 2 - 2) + c2) + 2) / 2
def apply_squash_stretch(
base_scale: tuple[float, float], intensity: float, direction: str = "vertical"
) -> tuple[float, float]:
"""
Calculate squash and stretch scales for more dynamic animation.
Args:
base_scale: (width_scale, height_scale) base scales
intensity: Squash/stretch intensity (0.0-1.0)
direction: 'vertical', 'horizontal', or 'both'
Returns:
(width_scale, height_scale) with squash/stretch applied
"""
width_scale, height_scale = base_scale
if direction == "vertical":
# Compress vertically, expand horizontally (preserve volume)
height_scale *= 1 - intensity * 0.5
width_scale *= 1 + intensity * 0.5
elif direction == "horizontal":
# Compress horizontally, expand vertically
width_scale *= 1 - intensity * 0.5
height_scale *= 1 + intensity * 0.5
elif direction == "both":
# General squash (both dimensions)
width_scale *= 1 - intensity * 0.3
height_scale *= 1 - intensity * 0.3
return (width_scale, height_scale)
def calculate_arc_motion(
start: tuple[float, float], end: tuple[float, float], height: float, t: float
) -> tuple[float, float]:
"""
Calculate position along a parabolic arc (natural motion path).
Args:
start: (x, y) starting position
end: (x, y) ending position
height: Arc height at midpoint (positive = upward)
t: Progress (0.0-1.0)
Returns:
(x, y) position along arc
"""
x1, y1 = start
x2, y2 = end
# Linear interpolation for x
x = x1 + (x2 - x1) * t
# Parabolic interpolation for y
# y = start + progress * (end - start) + arc_offset
# Arc offset peaks at t=0.5
arc_offset = 4 * height * t * (1 - t)
y = y1 + (y2 - y1) * t - arc_offset
return (x, y)
# Add new easing functions to the convenience mapping
EASING_FUNCTIONS.update(
{
"back_in": ease_back_in,
"back_out": ease_back_out,
"back_in_out": ease_back_in_out,
"anticipate": ease_back_in, # Alias
"overshoot": ease_back_out, # Alias
}
)

View file

@ -0,0 +1,176 @@
#!/usr/bin/env python3
"""
Frame Composer - Utilities for composing visual elements into frames.
Provides functions for drawing shapes, text, emojis, and compositing elements
together to create animation frames.
"""
from typing import Optional
import numpy as np
from PIL import Image, ImageDraw, ImageFont
def create_blank_frame(
width: int, height: int, color: tuple[int, int, int] = (255, 255, 255)
) -> Image.Image:
"""
Create a blank frame with solid color background.
Args:
width: Frame width
height: Frame height
color: RGB color tuple (default: white)
Returns:
PIL Image
"""
return Image.new("RGB", (width, height), color)
def draw_circle(
frame: Image.Image,
center: tuple[int, int],
radius: int,
fill_color: Optional[tuple[int, int, int]] = None,
outline_color: Optional[tuple[int, int, int]] = None,
outline_width: int = 1,
) -> Image.Image:
"""
Draw a circle on a frame.
Args:
frame: PIL Image to draw on
center: (x, y) center position
radius: Circle radius
fill_color: RGB fill color (None for no fill)
outline_color: RGB outline color (None for no outline)
outline_width: Outline width in pixels
Returns:
Modified frame
"""
draw = ImageDraw.Draw(frame)
x, y = center
bbox = [x - radius, y - radius, x + radius, y + radius]
draw.ellipse(bbox, fill=fill_color, outline=outline_color, width=outline_width)
return frame
def draw_text(
frame: Image.Image,
text: str,
position: tuple[int, int],
color: tuple[int, int, int] = (0, 0, 0),
centered: bool = False,
) -> Image.Image:
"""
Draw text on a frame.
Args:
frame: PIL Image to draw on
text: Text to draw
position: (x, y) position (top-left unless centered=True)
color: RGB text color
centered: If True, center text at position
Returns:
Modified frame
"""
draw = ImageDraw.Draw(frame)
# Uses Pillow's default font.
# If the font should be changed for the emoji, add additional logic here.
font = ImageFont.load_default()
if centered:
bbox = draw.textbbox((0, 0), text, font=font)
text_width = bbox[2] - bbox[0]
text_height = bbox[3] - bbox[1]
x = position[0] - text_width // 2
y = position[1] - text_height // 2
position = (x, y)
draw.text(position, text, fill=color, font=font)
return frame
def create_gradient_background(
width: int,
height: int,
top_color: tuple[int, int, int],
bottom_color: tuple[int, int, int],
) -> Image.Image:
"""
Create a vertical gradient background.
Args:
width: Frame width
height: Frame height
top_color: RGB color at top
bottom_color: RGB color at bottom
Returns:
PIL Image with gradient
"""
frame = Image.new("RGB", (width, height))
draw = ImageDraw.Draw(frame)
# Calculate color step for each row
r1, g1, b1 = top_color
r2, g2, b2 = bottom_color
for y in range(height):
# Interpolate color
ratio = y / height
r = int(r1 * (1 - ratio) + r2 * ratio)
g = int(g1 * (1 - ratio) + g2 * ratio)
b = int(b1 * (1 - ratio) + b2 * ratio)
# Draw horizontal line
draw.line([(0, y), (width, y)], fill=(r, g, b))
return frame
def draw_star(
frame: Image.Image,
center: tuple[int, int],
size: int,
fill_color: tuple[int, int, int],
outline_color: Optional[tuple[int, int, int]] = None,
outline_width: int = 1,
) -> Image.Image:
"""
Draw a 5-pointed star.
Args:
frame: PIL Image to draw on
center: (x, y) center position
size: Star size (outer radius)
fill_color: RGB fill color
outline_color: RGB outline color (None for no outline)
outline_width: Outline width
Returns:
Modified frame
"""
import math
draw = ImageDraw.Draw(frame)
x, y = center
# Calculate star points
points = []
for i in range(10):
angle = (i * 36 - 90) * math.pi / 180 # 36 degrees per point, start at top
radius = size if i % 2 == 0 else size * 0.4 # Alternate between outer and inner
px = x + radius * math.cos(angle)
py = y + radius * math.sin(angle)
points.append((px, py))
# Draw star
draw.polygon(points, fill=fill_color, outline=outline_color, width=outline_width)
return frame

View file

@ -0,0 +1,269 @@
#!/usr/bin/env python3
"""
GIF Builder - Core module for assembling frames into GIFs optimized for Slack.
This module provides the main interface for creating GIFs from programmatically
generated frames, with automatic optimization for Slack's requirements.
"""
from pathlib import Path
from typing import Optional
import imageio.v3 as imageio
import numpy as np
from PIL import Image
class GIFBuilder:
"""Builder for creating optimized GIFs from frames."""
def __init__(self, width: int = 480, height: int = 480, fps: int = 15):
"""
Initialize GIF builder.
Args:
width: Frame width in pixels
height: Frame height in pixels
fps: Frames per second
"""
self.width = width
self.height = height
self.fps = fps
self.frames: list[np.ndarray] = []
def add_frame(self, frame: np.ndarray | Image.Image):
"""
Add a frame to the GIF.
Args:
frame: Frame as numpy array or PIL Image (will be converted to RGB)
"""
if isinstance(frame, Image.Image):
frame = np.array(frame.convert("RGB"))
# Ensure frame is correct size
if frame.shape[:2] != (self.height, self.width):
pil_frame = Image.fromarray(frame)
pil_frame = pil_frame.resize(
(self.width, self.height), Image.Resampling.LANCZOS
)
frame = np.array(pil_frame)
self.frames.append(frame)
def add_frames(self, frames: list[np.ndarray | Image.Image]):
"""Add multiple frames at once."""
for frame in frames:
self.add_frame(frame)
def optimize_colors(
self, num_colors: int = 128, use_global_palette: bool = True
) -> list[np.ndarray]:
"""
Reduce colors in all frames using quantization.
Args:
num_colors: Target number of colors (8-256)
use_global_palette: Use a single palette for all frames (better compression)
Returns:
List of color-optimized frames
"""
optimized = []
if use_global_palette and len(self.frames) > 1:
# Create a global palette from all frames
# Sample frames to build palette
sample_size = min(5, len(self.frames))
sample_indices = [
int(i * len(self.frames) / sample_size) for i in range(sample_size)
]
sample_frames = [self.frames[i] for i in sample_indices]
# Combine sample frames into a single image for palette generation
# Flatten each frame to get all pixels, then stack them
all_pixels = np.vstack(
[f.reshape(-1, 3) for f in sample_frames]
) # (total_pixels, 3)
# Create a properly-shaped RGB image from the pixel data
# We'll make a roughly square image from all the pixels
total_pixels = len(all_pixels)
width = min(512, int(np.sqrt(total_pixels))) # Reasonable width, max 512
height = (total_pixels + width - 1) // width # Ceiling division
# Pad if necessary to fill the rectangle
pixels_needed = width * height
if pixels_needed > total_pixels:
padding = np.zeros((pixels_needed - total_pixels, 3), dtype=np.uint8)
all_pixels = np.vstack([all_pixels, padding])
# Reshape to proper RGB image format (H, W, 3)
img_array = (
all_pixels[:pixels_needed].reshape(height, width, 3).astype(np.uint8)
)
combined_img = Image.fromarray(img_array, mode="RGB")
# Generate global palette
global_palette = combined_img.quantize(colors=num_colors, method=2)
# Apply global palette to all frames
for frame in self.frames:
pil_frame = Image.fromarray(frame)
quantized = pil_frame.quantize(palette=global_palette, dither=1)
optimized.append(np.array(quantized.convert("RGB")))
else:
# Use per-frame quantization
for frame in self.frames:
pil_frame = Image.fromarray(frame)
quantized = pil_frame.quantize(colors=num_colors, method=2, dither=1)
optimized.append(np.array(quantized.convert("RGB")))
return optimized
def deduplicate_frames(self, threshold: float = 0.9995) -> int:
"""
Remove duplicate or near-duplicate consecutive frames.
Args:
threshold: Similarity threshold (0.0-1.0). Higher = more strict (0.9995 = nearly identical).
Use 0.9995+ to preserve subtle animations, 0.98 for aggressive removal.
Returns:
Number of frames removed
"""
if len(self.frames) < 2:
return 0
deduplicated = [self.frames[0]]
removed_count = 0
for i in range(1, len(self.frames)):
# Compare with previous frame
prev_frame = np.array(deduplicated[-1], dtype=np.float32)
curr_frame = np.array(self.frames[i], dtype=np.float32)
# Calculate similarity (normalized)
diff = np.abs(prev_frame - curr_frame)
similarity = 1.0 - (np.mean(diff) / 255.0)
# Keep frame if sufficiently different
# High threshold (0.9995+) means only remove nearly identical frames
if similarity < threshold:
deduplicated.append(self.frames[i])
else:
removed_count += 1
self.frames = deduplicated
return removed_count
def save(
self,
output_path: str | Path,
num_colors: int = 128,
optimize_for_emoji: bool = False,
remove_duplicates: bool = False,
) -> dict:
"""
Save frames as optimized GIF for Slack.
Args:
output_path: Where to save the GIF
num_colors: Number of colors to use (fewer = smaller file)
optimize_for_emoji: If True, optimize for emoji size (128x128, fewer colors)
remove_duplicates: If True, remove duplicate consecutive frames (opt-in)
Returns:
Dictionary with file info (path, size, dimensions, frame_count)
"""
if not self.frames:
raise ValueError("No frames to save. Add frames with add_frame() first.")
output_path = Path(output_path)
# Remove duplicate frames to reduce file size
if remove_duplicates:
removed = self.deduplicate_frames(threshold=0.9995)
if removed > 0:
print(
f" Removed {removed} nearly identical frames (preserved subtle animations)"
)
# Optimize for emoji if requested
if optimize_for_emoji:
if self.width > 128 or self.height > 128:
print(
f" Resizing from {self.width}x{self.height} to 128x128 for emoji"
)
self.width = 128
self.height = 128
# Resize all frames
resized_frames = []
for frame in self.frames:
pil_frame = Image.fromarray(frame)
pil_frame = pil_frame.resize((128, 128), Image.Resampling.LANCZOS)
resized_frames.append(np.array(pil_frame))
self.frames = resized_frames
num_colors = min(num_colors, 48) # More aggressive color limit for emoji
# More aggressive FPS reduction for emoji
if len(self.frames) > 12:
print(
f" Reducing frames from {len(self.frames)} to ~12 for emoji size"
)
# Keep every nth frame to get close to 12 frames
keep_every = max(1, len(self.frames) // 12)
self.frames = [
self.frames[i] for i in range(0, len(self.frames), keep_every)
]
# Optimize colors with global palette
optimized_frames = self.optimize_colors(num_colors, use_global_palette=True)
# Calculate frame duration in milliseconds
frame_duration = 1000 / self.fps
# Save GIF
imageio.imwrite(
output_path,
optimized_frames,
duration=frame_duration,
loop=0, # Infinite loop
)
# Get file info
file_size_kb = output_path.stat().st_size / 1024
file_size_mb = file_size_kb / 1024
info = {
"path": str(output_path),
"size_kb": file_size_kb,
"size_mb": file_size_mb,
"dimensions": f"{self.width}x{self.height}",
"frame_count": len(optimized_frames),
"fps": self.fps,
"duration_seconds": len(optimized_frames) / self.fps,
"colors": num_colors,
}
# Print info
print(f"\n✓ GIF created successfully!")
print(f" Path: {output_path}")
print(f" Size: {file_size_kb:.1f} KB ({file_size_mb:.2f} MB)")
print(f" Dimensions: {self.width}x{self.height}")
print(f" Frames: {len(optimized_frames)} @ {self.fps} fps")
print(f" Duration: {info['duration_seconds']:.1f}s")
print(f" Colors: {num_colors}")
# Size info
if optimize_for_emoji:
print(f" Optimized for emoji (128x128, reduced colors)")
if file_size_mb > 1.0:
print(f"\n Note: Large file size ({file_size_kb:.1f} KB)")
print(" Consider: fewer frames, smaller dimensions, or fewer colors")
return info
def clear(self):
"""Clear all frames (useful for creating multiple GIFs)."""
self.frames = []

View file

@ -0,0 +1,136 @@
#!/usr/bin/env python3
"""
Validators - Check if GIFs meet Slack's requirements.
These validators help ensure your GIFs meet Slack's size and dimension constraints.
"""
from pathlib import Path
def validate_gif(
gif_path: str | Path, is_emoji: bool = True, verbose: bool = True
) -> tuple[bool, dict]:
"""
Validate GIF for Slack (dimensions, size, frame count).
Args:
gif_path: Path to GIF file
is_emoji: True for emoji (128x128 recommended), False for message GIF
verbose: Print validation details
Returns:
Tuple of (passes: bool, results: dict with all details)
"""
from PIL import Image
gif_path = Path(gif_path)
if not gif_path.exists():
return False, {"error": f"File not found: {gif_path}"}
# Get file size
size_bytes = gif_path.stat().st_size
size_kb = size_bytes / 1024
size_mb = size_kb / 1024
# Get dimensions and frame info
try:
with Image.open(gif_path) as img:
width, height = img.size
# Count frames
frame_count = 0
try:
while True:
img.seek(frame_count)
frame_count += 1
except EOFError:
pass
# Get duration
try:
duration_ms = img.info.get("duration", 100)
total_duration = (duration_ms * frame_count) / 1000
fps = frame_count / total_duration if total_duration > 0 else 0
except:
total_duration = None
fps = None
except Exception as e:
return False, {"error": f"Failed to read GIF: {e}"}
# Validate dimensions
if is_emoji:
optimal = width == height == 128
acceptable = width == height and 64 <= width <= 128
dim_pass = acceptable
else:
aspect_ratio = (
max(width, height) / min(width, height)
if min(width, height) > 0
else float("inf")
)
dim_pass = aspect_ratio <= 2.0 and 320 <= min(width, height) <= 640
results = {
"file": str(gif_path),
"passes": dim_pass,
"width": width,
"height": height,
"size_kb": size_kb,
"size_mb": size_mb,
"frame_count": frame_count,
"duration_seconds": total_duration,
"fps": fps,
"is_emoji": is_emoji,
"optimal": optimal if is_emoji else None,
}
# Print if verbose
if verbose:
print(f"\nValidating {gif_path.name}:")
print(
f" Dimensions: {width}x{height}"
+ (
f" ({'optimal' if optimal else 'acceptable'})"
if is_emoji and acceptable
else ""
)
)
print(
f" Size: {size_kb:.1f} KB"
+ (f" ({size_mb:.2f} MB)" if size_mb >= 1.0 else "")
)
print(
f" Frames: {frame_count}"
+ (f" @ {fps:.1f} fps ({total_duration:.1f}s)" if fps else "")
)
if not dim_pass:
print(
f" Note: {'Emoji should be 128x128' if is_emoji else 'Unusual dimensions for Slack'}"
)
if size_mb > 5.0:
print(f" Note: Large file size - consider fewer frames/colors")
return dim_pass, results
def is_slack_ready(
gif_path: str | Path, is_emoji: bool = True, verbose: bool = True
) -> bool:
"""
Quick check if GIF is ready for Slack.
Args:
gif_path: Path to GIF file
is_emoji: True for emoji GIF, False for message GIF
verbose: Print feedback
Returns:
True if dimensions are acceptable
"""
passes, _ = validate_gif(gif_path, is_emoji, verbose)
return passes

View file

@ -0,0 +1,4 @@
pillow>=10.0.0
imageio>=2.31.0
imageio-ffmpeg>=0.4.9
numpy>=1.24.0

View file

@ -0,0 +1,187 @@
"""
End-to-end test for LiteLLM Skills with Messages API.
Tests the slack-gif-creator skill with GPT-4o via messages API
to verify skills work correctly and can generate a GIF.
"""
import os
import sys
import zipfile
from io import BytesIO
from pathlib import Path
import pytest
sys.path.insert(0, os.path.abspath("../.."))
import litellm
import litellm.proxy.proxy_server
from litellm.caching.caching import DualCache
from litellm.proxy._types import NewSkillRequest, UserAPIKeyAuth
from litellm.proxy.utils import PrismaClient, ProxyLogging
proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
def create_skill_zip_from_folder(skill_name: str) -> bytes:
"""Create a ZIP file from a skill folder in test_skills_data."""
test_dir = Path(__file__).parent / "test_skills_data"
skill_dir = test_dir / skill_name
zip_buffer = BytesIO()
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
for file_path in skill_dir.rglob("*"):
if file_path.is_file():
arcname = f"{skill_name}/{file_path.relative_to(skill_dir)}"
zf.write(file_path, arcname=arcname)
return zip_buffer.getvalue()
@pytest.fixture
def prisma_client():
"""Set up prisma client for tests."""
from litellm.proxy.proxy_cli import append_query_params
params = {"connection_limit": 100, "pool_timeout": 60}
database_url = os.getenv("DATABASE_URL")
if not database_url:
pytest.skip("DATABASE_URL not set")
modified_url = append_query_params(database_url, params)
os.environ["DATABASE_URL"] = modified_url
prisma_client = PrismaClient(
database_url=os.environ["DATABASE_URL"], proxy_logging_obj=proxy_logging_obj
)
return prisma_client
@pytest.mark.asyncio
async def test_slack_gif_skill_creates_gif(prisma_client):
"""
Test slack-gif-creator skill generates a GIF using GPT-4o via messages API.
Flow:
1. Store skill in LiteLLM DB
2. Hook resolves skill, adds litellm_code_execution tool, injects SKILL.md
3. Make GPT-4o call via messages API
4. Hook handles code execution loop
5. Verify GIF is generated
"""
litellm._turn_on_debug()
if not os.getenv("OPENAI_API_KEY"):
pytest.skip("OPENAI_API_KEY not set")
setattr(litellm.proxy.proxy_server, "prisma_client", prisma_client)
await litellm.proxy.proxy_server.prisma_client.connect()
from litellm.llms.litellm_proxy.skills.handler import LiteLLMSkillsHandler
from litellm.proxy.hooks.litellm_skills import SkillsInjectionHook
from litellm.types.utils import CallTypes
# 1. Store skill in DB
skill_name = "slack-gif-creator"
zip_content = create_skill_zip_from_folder(skill_name)
skill_request = NewSkillRequest(
display_title="Slack GIF Creator",
description="Create animated GIFs optimized for Slack",
instructions="Use this skill to create animated GIFs for Slack emoji",
file_content=zip_content,
file_name=f"{skill_name}.zip",
file_type="application/zip",
)
created_skill = await LiteLLMSkillsHandler.create_skill(
data=skill_request,
user_id="test_user",
)
print(f"\nCreated skill: {created_skill.skill_id}")
hook = SkillsInjectionHook()
try:
# 2. Build request with container.skills (messages API spec)
request_data = {
"model": "claude-sonnet-4-5",
"max_tokens": 4096,
"messages": [
{
"role": "user",
"content": "Create a simple bouncing red ball GIF for Slack emoji."
}
],
"container": {
"skills": [
{"type": "custom", "skill_id": f"litellm:{created_skill.skill_id}"}
]
},
}
# 3. Pre-call hook resolves skill
user_api_key_dict = UserAPIKeyAuth(api_key="test-key")
cache = DualCache()
transformed = await hook.async_pre_call_hook(
user_api_key_dict=user_api_key_dict,
cache=cache,
data=request_data,
call_type="anthropic_messages",
)
assert isinstance(transformed, dict)
# Hook returns Anthropic-format tools for messages API
tool_names = [t.get('name') for t in transformed.get('tools', [])]
print(f"\nTools after hook: {tool_names}")
assert "litellm_code_execution" in tool_names, "Should have litellm_code_execution tool"
# 4. Make GPT-4o call via messages API (tools already in Anthropic format)
print("\n--- Making GPT-4o call via messages API ---")
response = await litellm.anthropic.acreate(
model=transformed["model"],
max_tokens=transformed.get("max_tokens", 4096),
messages=transformed["messages"],
tools=transformed.get("tools"),
)
print(f"Initial response: {response}")
# 5. Post-call hook handles code execution loop
final_response = await hook.async_post_call_success_deployment_hook(
request_data=transformed,
response=response,
call_type=CallTypes.anthropic_messages,
)
if final_response:
response = final_response
print("Code execution completed!")
# 6. Check for generated files (handle both dict and object response)
if isinstance(response, dict):
generated_files = response.get("_litellm_generated_files", [])
else:
generated_files = getattr(response, "_litellm_generated_files", [])
print(f"\nGenerated files: {len(generated_files)}")
if generated_files:
import base64
for f in generated_files:
print(f" - {f['name']} ({f['size']} bytes)")
if f['name'].endswith('.gif'):
content = base64.b64decode(f['content_base64'])
assert content[:6] in [b'GIF89a', b'GIF87a'], "Should be valid GIF"
print(" Valid GIF!")
print("\nSUCCESS - GIF generated!")
else:
# Print response for debugging
if hasattr(response, "choices"):
print(f"\nResponse: {response.choices[0].message}")
else:
print(f"\nResponse: {response}")
finally:
await LiteLLMSkillsHandler.delete_skill(skill_id=created_skill.skill_id)

View file

@ -0,0 +1,257 @@
"""
Test LiteLLM Skills SDK with custom_llm_provider=litellm_proxy
Tests the SDK-level skills methods when using the LiteLLM database backend:
1. Create a skill using SDK and verify it was stored correctly
2. List skills using SDK
3. Get a skill by ID using SDK
4. Delete a skill using SDK
5. Skills injection hook correctly resolves skills from database
"""
import os
import sys
import zipfile
from contextlib import contextmanager
from io import BytesIO
from pathlib import Path
import pytest
sys.path.insert(0, os.path.abspath("../.."))
import litellm
from litellm.caching.caching import DualCache
from litellm.proxy import proxy_server
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.utils import PrismaClient, ProxyLogging
from litellm.types.utils import LlmProviders
proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
@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:
Tuple of (file handle, file content bytes)
The zip file is automatically cleaned up after use.
"""
test_dir = Path(__file__).parent.parent / "llm_translation" / "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:
content = f.read()
f.seek(0)
yield f, content
finally:
# Clean up zip file
if zip_path.exists():
zip_path.unlink()
@pytest.fixture
def prisma_client():
"""Set up prisma client for tests."""
from litellm.proxy.proxy_cli import append_query_params
params = {"connection_limit": 100, "pool_timeout": 60}
database_url = os.getenv("DATABASE_URL")
modified_url = append_query_params(database_url, params)
os.environ["DATABASE_URL"] = modified_url
prisma_client = PrismaClient(
database_url=os.environ["DATABASE_URL"], proxy_logging_obj=proxy_logging_obj
)
return prisma_client
@pytest.mark.asyncio
async def test_create_skill_sdk(prisma_client):
"""
Test creating a skill using SDK with custom_llm_provider=litellm_proxy.
Verifies that:
- Skill is created with correct display_title
- Skill ID is generated and returned
- Skill response has correct type
"""
setattr(proxy_server, "prisma_client", prisma_client)
await proxy_server.prisma_client.connect()
from litellm.skills.main import acreate_skill, adelete_skill
# Create a skill using SDK
skill = await acreate_skill(
display_title="SDK Test Skill",
extra_body={
"description": "A test skill created via SDK",
"instructions": "Use this skill for SDK testing",
},
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
# Verify skill was created correctly
assert skill is not None
assert skill.id is not None
assert skill.id.startswith("skill_")
assert skill.display_title == "SDK Test Skill"
assert skill.type == "skill"
assert skill.source == "custom"
# Clean up
await adelete_skill(
skill_id=skill.id,
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
@pytest.mark.asyncio
async def test_list_skills_sdk(prisma_client):
"""
Test listing skills using SDK with custom_llm_provider=litellm_proxy.
Verifies that:
- Multiple skills can be created
- List returns the created skills
"""
setattr(proxy_server, "prisma_client", prisma_client)
await proxy_server.prisma_client.connect()
from litellm.skills.main import acreate_skill, adelete_skill, alist_skills
# Create multiple skills
created_skill_ids = []
for i in range(3):
skill = await acreate_skill(
display_title=f"List Test Skill {i}",
extra_body={
"description": f"Test skill {i} for list test",
},
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
created_skill_ids.append(skill.id)
# List skills using SDK
response = await alist_skills(
limit=10,
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
# Verify we got skills back
assert response is not None
assert response.data is not None
assert len(response.data) >= 3
# Verify our created skills are in the list
skill_ids_in_list = [s.id for s in response.data]
for created_id in created_skill_ids:
assert created_id in skill_ids_in_list
# Clean up
for skill_id in created_skill_ids:
await adelete_skill(
skill_id=skill_id,
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
@pytest.mark.asyncio
async def test_get_skill_sdk(prisma_client):
"""
Test getting a skill by ID using SDK with custom_llm_provider=litellm_proxy.
Verifies that:
- Skill can be retrieved by ID
- Retrieved skill has correct data
"""
setattr(proxy_server, "prisma_client", prisma_client)
await proxy_server.prisma_client.connect()
from litellm.skills.main import acreate_skill, adelete_skill, aget_skill
# Create a skill
created_skill = await acreate_skill(
display_title="Get Test Skill",
extra_body={
"description": "A skill for get test",
},
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
# Get the skill by ID using SDK
retrieved_skill = await aget_skill(
skill_id=created_skill.id,
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
# Verify retrieved skill matches created skill
assert retrieved_skill is not None
assert retrieved_skill.id == created_skill.id
assert retrieved_skill.display_title == "Get Test Skill"
# Clean up
await adelete_skill(
skill_id=created_skill.id,
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
@pytest.mark.asyncio
async def test_delete_skill_sdk(prisma_client):
"""
Test deleting a skill using SDK with custom_llm_provider=litellm_proxy.
Verifies that:
- Skill can be deleted by ID
- Deleted skill cannot be retrieved
"""
setattr(proxy_server, "prisma_client", prisma_client)
await proxy_server.prisma_client.connect()
from litellm.skills.main import acreate_skill, adelete_skill, aget_skill
# Create a skill
created_skill = await acreate_skill(
display_title="Delete Test Skill",
extra_body={
"description": "A skill to be deleted",
},
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
# Verify skill exists
retrieved = await aget_skill(
skill_id=created_skill.id,
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
assert retrieved is not None
# Delete the skill using SDK
result = await adelete_skill(
skill_id=created_skill.id,
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)
assert result.id == created_skill.id
assert result.type == "skill_deleted"
# Verify skill no longer exists
with pytest.raises(Exception):
await aget_skill(
skill_id=created_skill.id,
custom_llm_provider=LlmProviders.LITELLM_PROXY.value,
)