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