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
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-2")
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:")
# Display stdout logs
for stderr_item in py_result.logs.stderr:
print(f" Execution result:\n{stderr_item}")
else:
if py_result.error:
print(f"Error: {py_result.error.name}: {py_result.error.value}")
else:
print("Execution failed with unknown error")
# 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:")
# Display stdout logs
for stdout_line in js_result.logs.stdout:
print(stdout_line)
else:
if js_result.error_message:
print(f"Error: {js_result.error_message}")
else:
print("Execution failed with unknown error")
# 6. 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:")
# Display stdout logs
for stdout_line in java_result.logs.stdout:
print(stdout_line)
else:
if java_result.error:
print(f"Error: {java_result.error.name}: {java_result.error.value}")
else:
print("Execution failed with unknown error")
# 7. 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:")
# Display stdout logs
for stdout_line in r_result.logs.stdout:
print(stdout_line)
else:
if r_result.error:
print(f"Error: {r_result.error.name}: {r_result.error.value}")
else:
print("Execution failed with unknown error")
# 8. Rich Media Output Example (Python with HTML/Images)
print("\n=== Rich Media Output Example ===")
rich_code = """
import matplotlib.pyplot as plt
import numpy as np
from IPython.display import HTML, display
# Create a simple plot
x = np.linspace(0, 10, 100)
y = np.sin(x)
plt.figure(figsize=(8, 4))
plt.plot(x, y, 'b-', linewidth=2)
plt.title('Sine Wave')
plt.xlabel('x')
plt.ylabel('sin(x)')
plt.grid(True)
plt.show()
# Display HTML content
html_content = '''
<div style="background-color: #f0f8ff; padding: 10px; border-radius: 5px;">
<h3 style="color: #4169e1;">Rich HTML Output</h3>
<p>This is an example of <strong>HTML content</strong> in code execution results.</p>
<ul>
<li>Supports <em>formatted text</em></li>
<li>Can include <span style="color: red;">colored elements</span></li>
<li>Tables, lists, and more!</li>
</ul>
</div>
'''
display(HTML(html_content))
print("Rich media example completed!")
"""
rich_result = session.code.run_code(rich_code, "python")
if rich_result.success:
print("Output:")
# Display stdout logs
for stdout_line in rich_result.logs.stdout:
print(stdout_line)
# Display different types of results
for i, result in enumerate(rich_result.results):
print(f"\n--- Result {i+1} ---")
if result.text:
print(f"Text: {result.text}")
if result.html:
print(f"HTML Content: {len(result.html)} characters")
print(
"HTML Preview:",
(
result.html[:200] + "..."
if len(result.html) > 200
else result.html
),
)
if result.png:
print(f"PNG Image: {len(result.png)} bytes (base64 encoded)")
if result.jpeg:
print(f"JPEG Image: {len(result.jpeg)} bytes (base64 encoded)")
if result.svg:
print(f"SVG Image: {len(result.svg)} characters")
if result.chart:
print(f"Chart Data: {type(result.chart)}")
if result.markdown:
print(f"Markdown: {result.markdown}")
if result.latex:
print(f"LaTeX: {result.latex}")
print(f"Is Main Result: {result.is_main_result}")
# Display execution metadata
print(f"\nExecution Metadata:")
print(f" Execution Count: {rich_result.execution_count}")
print(f" Execution Time: {rich_result.execution_time}s")
print(f" Request ID: {rich_result.request_id}")
else:
if rich_result.error:
print(f"Error: {rich_result.error.name}: {rich_result.error.value}")
else:
print("Execution failed with unknown error")
finally:
# 9. Cleanup
print("\nCleaning up...")
agb.delete(session)
print("Session deleted")
if __name__ == "__main__":
main()Caching Results
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)
if result1.success and result1.logs.stdout and result1.logs.stdout[0]:
print(f"✅ execution result: {result1.logs.stdout[0]}")
else:
print(f"Execution failed: {result1.error_message}")
print("\n--- Second execution (fast) ---")
result2 = client.execute_with_cache(expensive_code)
if result2.success and result2.logs.stdout and result2.logs.stdout[0]:
print(f"✅ execution result: {result2.logs.stdout[0]}")
else:
print(f"Execution failed: {result2.error_message}")
if __name__ == "__main__":
main()Concurrency
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
# First try to get from results
if code_result.results and len(code_result.results) > 0:
for result in code_result.results:
if result.text and result.text.strip():
output_lines = result.text.strip().split("\n")
# Find the last line that looks like JSON
for line in reversed(output_lines):
line = line.strip()
if line.startswith("[") or line.startswith("{"):
try:
return json.loads(line)
except json.JSONDecodeError:
continue
# Fallback to stdout logs
if code_result.logs and code_result.logs.stdout:
for line in reversed(code_result.logs.stdout):
line = line.strip()
if line.startswith("[") or line.startswith("{"):
try:
return json.loads(line)
except json.JSONDecodeError:
continue
raise Exception("No valid JSON output found in code execution results")
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
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)
if result.success and result.logs.stdout and result.logs.stdout[0]:
print(f"✅ Safe execution result: {result.logs.stdout[0]}")
else:
print(f"Execution failed: {result.error_message}")
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