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-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(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(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(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(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(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, 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, "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, 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, 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()Multiple Results Scenarios
Demonstrates scenarios where code execution returns multiple results, including multiple display() calls, charts, and mixed output types.
py
#!/usr/bin/env python3
"""
Demonstrates scenarios where run returns multiple results.
Usage:
python docs/examples/code_execution/multiple_results_demo.py
"""
import os
from dotenv import load_dotenv
from agb import AGB
from agb.session_params import CreateSessionParams
load_dotenv()
def demo_multiple_display():
"""Scenario 1: Multiple display() calls"""
print("\n" + "="*60)
print("Scenario 1: Multiple display() calls")
print("="*60)
code = """
from IPython.display import display, HTML, Markdown
# First display - HTML
display(HTML("<h1>Heading 1</h1>"))
# Second display - Markdown
display(Markdown("# Markdown Heading"))
# Third display - HTML
display(HTML("<p>This is a paragraph</p>"))
# Final return value
"Processing completed"
"""
return code, "Expected 4 result items (3 displays + 1 return value)"
def demo_multiple_charts():
"""Scenario 2: Generate multiple charts"""
print("\n" + "="*60)
print("Scenario 2: Generate multiple charts")
print("="*60)
code = """
import matplotlib.pyplot as plt
# First chart
plt.figure(figsize=(6, 4))
plt.plot([1, 2, 3], [1, 2, 3])
plt.title("Line Chart")
plt.show()
# Second chart
plt.figure(figsize=(6, 4))
plt.bar([1, 2, 3], [3, 1, 4])
plt.title("Bar Chart")
plt.show()
"Chart generation completed"
"""
return code, "Expected 3 result items (2 charts + 1 return value)"
def demo_mixed_output():
"""Scenario 3: Mixed output (data + chart + text)"""
print("\n" + "="*60)
print("Scenario 3: Mixed output (data + chart + text)")
print("="*60)
code = """
import pandas as pd
from IPython.display import display, HTML
import matplotlib.pyplot as plt
# 1. Display data table
df = pd.DataFrame({
'Product': ['A', 'B', 'C'],
'Sales': [100, 200, 150]
})
display(df)
# 2. Display descriptive text
display(HTML("<p><b>Data Analysis Results:</b></p>"))
# 3. Display visualization chart
plt.figure(figsize=(8, 5))
df.plot(x='Product', y='Sales', kind='bar', legend=False)
plt.title("Product Sales Comparison")
plt.show()
# 4. Return statistical result
f"Total Sales: {df['Sales'].sum()}"
"""
return code, "Expected 4 result items (table + HTML text + chart + return value)"
def demo_single_result():
"""Scenario 4: Single result (comparison scenario)"""
print("\n" + "="*60)
print("Scenario 4: Single result (comparison scenario)")
print("="*60)
code = """
# Only print output and one return value
print("Starting calculation...")
result = 42 * 2
print(f"Calculation result: {result}")
# Final return
result
"""
return code, "Expected 1 result item (return value), print in stdout"
def demo_only_prints():
"""Scenario 5: Only print, no return value"""
print("\n" + "="*60)
print("Scenario 5: Only print, no explicit return value")
print("="*60)
code = """
print("This is the first line of output")
print("This is the second line of output")
print("This is the third line of output")
# No explicit return value
"""
return code, "Expected results may be empty or only None, output in stdout"
def analyze_results(result, description):
"""Analyze and display results"""
print(f"\n📊 {description}")
print("-" * 60)
if not result.success:
print(f"❌ Execution failed: {result.error_message}")
return
print(f"✅ Execution succeeded")
print(f"⏱️ Execution time: {result.execution_time:.3f} seconds")
# Analyze results
print(f"\n📦 results list: {len(result.results)} result items")
if result.results:
for i, item in enumerate(result.results, 1):
print(f"\n Result {i}:")
print(f" is_main_result: {item.is_main_result}")
# List contained formats
formats = []
if item.text:
formats.append(f"text({len(item.text)} chars)")
if item.html:
formats.append(f"html({len(item.html)} chars)")
if item.markdown:
formats.append(f"markdown({len(item.markdown)} chars)")
if item.png:
formats.append(f"png({len(item.png)} bytes)")
if item.jpeg:
formats.append(f"jpeg({len(item.jpeg)} bytes)")
if item.chart:
formats.append(f"chart({type(item.chart).__name__})")
print(f" Formats: {', '.join(formats) if formats else 'None'}")
# Display content preview
if item.text and len(item.text) < 100:
print(f" Preview: {item.text}")
elif item.text:
print(f" Preview: {item.text[:100]}...")
else:
print(" (empty list)")
# Analyze logs
print(f"\n📝 logs.stdout: {len(result.logs.stdout)} lines")
if result.logs.stdout:
for i, line in enumerate(result.logs.stdout[:3], 1): # Show only first 3 lines
print(f" Line {i}: {line.strip()}")
if len(result.logs.stdout) > 3:
print(f" ... {len(result.logs.stdout) - 3} more lines")
print(f"\n📝 logs.stderr: {len(result.logs.stderr)} lines")
if result.logs.stderr:
for line in result.logs.stderr:
print(f" {line.strip()}")
def main():
"""Main function"""
print("="*60)
print("🚀 Multiple Results Scenarios Demo")
print("="*60)
# Initialize
api_key = os.getenv("AGB_API_KEY")
if not api_key:
print("❌ Error: Please set AGB_API_KEY environment variable")
return
agb = AGB(api_key=api_key)
# Create session
print("\n📍 Creating session...")
params = CreateSessionParams(image_id="agb-code-space-1")
session_result = agb.create(params)
if not session_result.success:
print(f"❌ Failed to create session: {session_result.error_message}")
return
session = session_result.session
print(f"✅ Session created successfully: {session.session_id}")
try:
# Run demo scenarios
scenarios = [
demo_single_result, # Start with simple scenario
demo_only_prints, # No return value scenario
demo_multiple_display, # Multiple display calls
demo_multiple_charts, # Multiple charts
demo_mixed_output, # Mixed output
]
for scenario_func in scenarios:
code, description = scenario_func()
print(f"\n🔧 Executing code...")
print("```python")
print(code.strip())
print("```")
result = session.code.run(code, "python")
analyze_results(result, description)
input("\nPress Enter to continue to next scenario...")
print("\n" + "="*60)
print("✅ All demo scenarios completed")
print("="*60)
finally:
# Cleanup session
print("\n🧹 Cleaning up session...")
delete_result = agb.delete(session)
if delete_result.success:
print("✅ Session deleted")
else:
print(f"⚠️ Failed to delete session: {delete_result.error_message}")
if __name__ == "__main__":
main()