Appearance
Code Execution Examples
This directory contains examples of executing code in various programming languages using the AGB SDK.
Supported Languages
- Python: Full Python 3 environment with standard libraries.
- JavaScript: Node.js environment.
- Java: Java execution environment (supports snippets).
- R: R statistical computing environment.
Examples
Basic Execution (main.py)
Comprehensive example demonstrating execution in all supported languages.
py
"""
AGB Code Execution Example
This example demonstrates how to execute code in various languages (Python, JavaScript, Java, R)
using the AGB SDK.
"""
import os
import sys
from agb import AGB
from agb.session_params import CreateSessionParams
def main():
# 1. Initialize AGB client
api_key = os.getenv("AGB_API_KEY")
if not api_key:
print("Error: AGB_API_KEY environment variable is not set")
return
agb = AGB(api_key=api_key)
# 2. Create a session
print("Creating session...")
# Note: For code execution, we recommend using the 'agb-code-space-1' image
params = CreateSessionParams(image_id="agb-code-space-1")
result = agb.create(params)
if not result.success:
print(f"Failed to create session: {result.error_message}")
return
session = result.session
print(f"Session created: {session.session_id}")
try:
# 3. Python Execution
print("\n=== Executing Python Code ===")
python_code = """
import datetime
import math
print(f"Current time: {datetime.datetime.now()}")
print(f"Pi is approximately: {math.pi:.4f}")
numbers = [x**2 for x in range(5)]
print(f"Squares: {numbers}")
"""
py_result = session.code.run_code(python_code, "python")
if py_result.success:
print("Output:")
print(py_result.result)
else:
print(f"Error: {py_result.error_message}")
# 4. JavaScript Execution
print("\n=== Executing JavaScript Code ===")
js_code = """
const numbers = [1, 2, 3, 4, 5];
const sum = numbers.reduce((a, b) => a + b, 0);
console.log(`Sum of [${numbers}] is ${sum}`);
console.log(`User Agent: ${navigator.userAgent}`);
"""
js_result = session.code.run_code(js_code, "javascript")
if js_result.success:
print("Output:")
print(js_result.result)
else:
print(f"Error: {js_result.error_message}")
# 5. Java Execution
# Note: Java support allows running snippets without boilerplate class definitions
print("\n=== Executing Java Code ===")
java_code = """
String message = "Hello from Java!";
System.out.println(message);
System.out.println("String length: " + message.length());
int[] arr = {10, 20, 30};
int sum = 0;
for(int i : arr) sum += i;
System.out.println("Array sum: " + sum);
"""
java_result = session.code.run_code(java_code, "java")
if java_result.success:
print("Output:")
print(java_result.result)
else:
print(f"Error: {java_result.error_message}")
# 6. R Execution
print("\n=== Executing R Code ===")
r_code = """
data <- c(10, 20, 30, 40, 50)
cat("Data:", data, "\\n")
cat("Mean:", mean(data), "\\n")
cat("SD:", sd(data), "\\n")
"""
r_result = session.code.run_code(r_code, "r")
if r_result.success:
print("Output:")
print(r_result.result)
else:
print(f"Error: {r_result.error_message}")
finally:
# 7. Cleanup
print("\nCleaning up...")
agb.delete(session)
print("Session deleted")
if __name__ == "__main__":
main()Caching Results (caching.py)
Demonstrates how to implement client-side caching for deterministic code execution results to save costs and reduce latency.
py
"""
AGB Code Execution Caching Example
This example demonstrates how to implement client-side caching for code execution results.
Caching deterministic operations can significantly reduce latency and API costs.
"""
import hashlib
import os
from typing import Any, Dict
from agb import AGB
from agb.session_params import CreateSessionParams
class CachedAGBClient:
"""AGB client with result caching for expensive operations"""
def __init__(self, api_key: str, cache_size: int = 100):
self.agb = AGB(api_key=api_key)
self.cache: Dict[str, Any] = {}
self.cache_size = cache_size
self.cache_order = [] # For LRU eviction
def execute_with_cache(self, code: str, language: str = "python") -> Any:
"""Execute code with caching based on content hash"""
# Create cache key from code content
cache_key = self._get_cache_key(code, language)
# Check cache first
if cache_key in self.cache:
print(f"⚡ Cache hit for operation: {language}")
self._update_cache_order(cache_key)
return self.cache[cache_key]
print(f"🔄 Cache miss. Executing on cloud...")
# Execute operation
# In a real app, you might want to reuse a session instead of creating one every time
params = CreateSessionParams(image_id="agb-code-space-1")
result = self.agb.create(params)
if not result.success:
raise Exception(f"Session creation failed: {result.error_message}")
session = result.session
try:
code_result = session.code.run_code(code, language)
# Cache successful results
if code_result.success:
self._add_to_cache(cache_key, code_result)
return code_result
finally:
self.agb.delete(session)
def _get_cache_key(self, code: str, language: str) -> str:
"""Generate cache key from code content"""
content = f"{language}:{code}"
return hashlib.md5(content.encode()).hexdigest()
def _add_to_cache(self, key: str, result: Any):
"""Add result to cache with LRU eviction"""
# Remove oldest entry if cache is full
if len(self.cache) >= self.cache_size and key not in self.cache:
oldest_key = self.cache_order.pop(0)
del self.cache[oldest_key]
self.cache[key] = result
self._update_cache_order(key)
def _update_cache_order(self, key: str):
"""Update LRU order"""
if key in self.cache_order:
self.cache_order.remove(key)
self.cache_order.append(key)
def main():
api_key = os.getenv("AGB_API_KEY")
if not api_key:
print("Error: AGB_API_KEY environment variable is not set")
return
client = CachedAGBClient(api_key=api_key)
# Operation 1: Expensive calculation
expensive_code = "import time; time.sleep(1); print('Calculation complete: 42')"
print("--- First execution (slow) ---")
result1 = client.execute_with_cache(expensive_code)
print(f"Result: {result1.result.strip()}")
print("\n--- Second execution (fast) ---")
result2 = client.execute_with_cache(expensive_code)
print(f"Result: {result2.result.strip()}")
if __name__ == "__main__":
main()Concurrency (concurrency.py)
Demonstrates how to execute multiple code tasks in parallel using threading for high-throughput scenarios.
py
"""
AGB Concurrent Code Execution Example
This example demonstrates how to execute multiple code snippets in parallel using threading.
Concurrent execution is essential for high-throughput applications like data processing pipelines.
"""
import concurrent.futures
import json
import os
import time
from typing import Callable, List
from agb import AGB
from agb.session_params import CreateSessionParams
class ConcurrentAGBProcessor:
def __init__(self, api_key: str, max_workers: int = 3):
self.max_workers = max_workers
self.agb = AGB(api_key=api_key)
def process_tasks_concurrently(self, tasks: List[dict], processor: Callable):
"""Process multiple tasks concurrently"""
results = []
start_time = time.time()
print(f"🚀 Starting processing of {len(tasks)} tasks with {self.max_workers} workers...")
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
# Submit all tasks
future_to_task = {
executor.submit(self._process_single_task, task, processor): task
for task in tasks
}
# Collect results as they complete
for future in concurrent.futures.as_completed(future_to_task):
task = future_to_task[future]
try:
result = future.result()
results.append({
'task_id': task.get('id'),
'success': True,
'result': result
})
print(f"✅ Task {task.get('id')} completed")
except Exception as e:
results.append({
'task_id': task.get('id'),
'success': False,
'error': str(e)
})
print(f"❌ Task {task.get('id')} failed: {e}")
duration = time.time() - start_time
print(f"🏁 All tasks finished in {duration:.2f} seconds")
return results
def _process_single_task(self, task: dict, processor: Callable):
"""Process a single task with its own session"""
# Create a dedicated session for this task
params = CreateSessionParams(image_id="agb-code-space-1")
result = self.agb.create(params)
if not result.success:
raise Exception(f"Failed to create session: {result.error_message}")
session = result.session
try:
return processor(session, task)
finally:
self.agb.delete(session)
def data_processing_task(session, task):
"""The actual logic to run in the cloud"""
data = task['data']
operation = task['operation']
# Simulate some heavy computation
code = f"""
import json
import time
# Simulate work
time.sleep(1)
data = {json.dumps(data)}
result = []
for item in data:
if '{operation}' == 'double':
result.append(item * 2)
elif '{operation}' == 'square':
result.append(item ** 2)
else:
result.append(item)
print(json.dumps(result))
"""
code_result = session.code.run_code(code, "python")
if code_result.success:
# Parse the last line of output as JSON result
return json.loads(code_result.result.strip().split('\n')[-1])
else:
raise Exception(code_result.error_message)
def main():
api_key = os.getenv("AGB_API_KEY")
if not api_key:
print("Error: AGB_API_KEY environment variable is not set")
return
processor = ConcurrentAGBProcessor(api_key=api_key, max_workers=3)
# Define a batch of tasks
tasks = [
{'id': 1, 'data': [1, 2, 3, 4], 'operation': 'double'},
{'id': 2, 'data': [2, 4, 6, 8], 'operation': 'square'},
{'id': 3, 'data': [10, 20, 30], 'operation': 'double'},
{'id': 4, 'data': [5, 5, 5, 5], 'operation': 'square'},
]
results = processor.process_tasks_concurrently(tasks, data_processing_task)
print("\n--- Final Results ---")
for res in results:
status = "Success" if res['success'] else "Failed"
output = res['result'] if res['success'] else res['error']
print(f"Task {res['task_id']}: {status} -> {output}")
if __name__ == "__main__":
main()Security Best Practices (security.py)
Demonstrates security best practices, including input validation and sanitization for untrusted code.
py
"""
AGB Secure Code Execution Example
This example demonstrates how to validate and sanitize user-provided code before execution.
While the AGB sandbox is secure, preventing malicious code execution saves resources and time.
"""
import os
import re
from agb import AGB
from agb.session_params import CreateSessionParams
class SecureAGBClient:
"""AGB client with security validations"""
DANGEROUS_PATTERNS = [
r'import\s+os',
r'import\s+subprocess',
r'__import__',
r'eval\s*\(',
r'exec\s*\(',
r'open\s*\(',
r'file\s*\(',
]
def __init__(self, api_key: str):
self.agb = AGB(api_key=api_key)
def safe_execute_code(self, code: str, language: str = "python"):
"""Execute code with security validations"""
# Validate input
if not self._validate_code_safety(code):
raise ValueError("Code contains potentially dangerous operations")
if len(code) > 10000: # 10KB limit
raise ValueError("Code too large - potential DoS attempt")
# Execute with timeout
params = CreateSessionParams(image_id="agb-code-space-1")
result = self.agb.create(params)
if not result.success:
raise Exception(f"Session creation failed: {result.error_message}")
session = result.session
try:
# Use shorter timeout for security
print("🔒 Executing validated code...")
code_result = session.code.run_code(code, language, timeout_s=30)
return code_result
finally:
self.agb.delete(session)
def _validate_code_safety(self, code: str) -> bool:
"""Check code for dangerous patterns"""
code_lower = code.lower()
for pattern in self.DANGEROUS_PATTERNS:
if re.search(pattern, code_lower):
print(f"⚠️ Dangerous pattern detected: {pattern}")
return False
return True
def execute_trusted_code(self, code: str, language: str = "python"):
"""Execute code from trusted sources without validation"""
params = CreateSessionParams(image_id="agb-code-space-1")
result = self.agb.create(params)
if not result.success:
raise Exception(f"Session creation failed: {result.error_message}")
session = result.session
try:
return session.code.run_code(code, language)
finally:
self.agb.delete(session)
def main():
api_key = os.getenv("AGB_API_KEY")
if not api_key:
print("Error: AGB_API_KEY environment variable is not set")
return
secure_client = SecureAGBClient(api_key=api_key)
# 1. Safe code
safe_code = """
x = 10
y = 20
result = x + y
print(f"Result: {result}")
"""
try:
result = secure_client.safe_execute_code(safe_code)
print(f"✅ Safe execution result: {result.result.strip()}")
except Exception as e:
print(f"Execution failed: {e}")
# 2. Dangerous code (will be rejected)
dangerous_code = """
import os
os.system("rm -rf /")
"""
print("\nAttempting to run dangerous code...")
try:
secure_client.safe_execute_code(dangerous_code)
except ValueError as e:
print(f"🛡️ Security validation passed: {e}")
if __name__ == "__main__":
main()How to Run
Set API Key:
bashexport AGB_API_KEY="your_api_key"Run Script:
bashpython docs/examples/code_execution/main.py