Skip to main content
The Hopx SDK provides process management capabilities for monitoring and controlling background code executions and processes.

​
Background Code Execution

​
run_code_background()

Execute code in the background and return immediately. 
process = sandbox.run_code_background(
    '''
import time
for i in range(100):
    print(f"Step {i}")
    time.sleep(1)
''',
    name="long_task"
)

print(f"Process ID: {process['process_id']}")
print(f"Status: {process['status']}")
Method Signature: 
def run_code_background(
    self,
    code: str,
    *,
    language: str = "python",
    timeout: int = 300,
    env: Optional[Dict[str, str]] = None,
    working_dir: str = "/workspace",
    name: Optional[str] = None,
) -> Dict[str, Any]
Parameters:
  • code (str): Code to execute
  • language (str): Programming language (default: “python”)
  • timeout (int): Execution timeout in seconds (default: 300)
  • env (Dict[str, str], optional): Environment variables
  • working_dir (str): Working directory (default: “/workspace”)
  • name (str, optional): Process name for identification
Returns: Dictionary with process_id and status Example: 
# Start long-running task
process = sandbox.run_code_background(
    '''
import pandas as pd
import time

# Process large dataset
df = pd.read_csv('/workspace/large_data.csv')
df = df.groupby('category').sum()
df.to_csv('/workspace/result.csv')

print("Processing complete!")
''',
    name="data_processing",
    timeout=1800  # 30 minutes
)

process_id = process['process_id']
print(f"Started process: {process_id}")

​
Listing Processes

​
list_processes()

List all background execution processes. 
processes = sandbox.list_processes()
for proc in processes:
    print(f"{proc['name']}: {proc['status']} (PID: {proc['process_id']})")
Method Signature: 
def list_processes(self) -> List[Dict[str, Any]]
Returns: List of process dictionaries Process Dictionary Fields:
  • process_id (str): Process identifier
  • name (str): Process name (if provided)
  • status (str): Process status (“running”, “completed”, “failed”, “killed”)
  • started_at (str): Start timestamp
  • language (str): Programming language
  • exit_code (int, optional): Exit code (if completed)
Example: 
# List all processes
processes = sandbox.list_processes()

# Filter running processes
running = [p for p in processes if p['status'] == 'running']
print(f"Running processes: {len(running)}")

# Filter by name
data_proc = [p for p in processes if p.get('name') == 'data_processing']

​
Killing Processes

​
kill_process()

Terminate a running background process. 
sandbox.kill_process("proc_abc123")
Method Signature: 
def kill_process(self, process_id: str) -> Dict[str, Any]
Parameters:
  • process_id (str): Process ID to kill
Returns: Dictionary with confirmation message Raises:
  • NotFoundError: If process not found
  • APIError: If kill operation fails
Example: 
# List processes
processes = sandbox.list_processes()

# Kill specific process
for proc in processes:
    if proc['status'] == 'running' and proc.get('name') == 'old_task':
        sandbox.kill_process(proc['process_id'])
        print(f"Killed process: {proc['process_id']}")

​
Process Status Monitoring

Monitor process status over time: 
import time

# Start background process
process = sandbox.run_code_background(
    'import time; time.sleep(60); print("Done")',
    name="monitored_task"
)

process_id = process['process_id']

# Monitor until completion
while True:
    processes = sandbox.list_processes()
    proc = next((p for p in processes if p['process_id'] == process_id), None)
    
    if not proc:
        print("Process not found")
        break
    
    status = proc['status']
    print(f"Status: {status}")
    
    if status in ('completed', 'failed', 'killed'):
        print(f"Process finished with status: {status}")
        if 'exit_code' in proc:
            print(f"Exit code: {proc['exit_code']}")
        break
    
    time.sleep(5)  # Check every 5 seconds

​
Background Commands

Run shell commands in the background: 
# Run command in background
result = sandbox.commands.run(
    'python long_script.py',
    background=True
)

# The command returns immediately with process info
print(f"Background command started: {result.stdout}")

​
Best Practices

​
1. Name Your Processes

Use descriptive names for easier identification: 
process = sandbox.run_code_background(
    code,
    name="data_analysis_task"
)

​
2. Monitor Long-Running Processes

Regularly check process status: 
def wait_for_process(sandbox, process_id, timeout=3600):
    start_time = time.time()
    while time.time() - start_time < timeout:
        processes = sandbox.list_processes()
        proc = next((p for p in processes if p['process_id'] == process_id), None)
        
        if not proc or proc['status'] in ('completed', 'failed', 'killed'):
            return proc
        
        time.sleep(10)
    
    # Timeout - kill process
    sandbox.kill_process(process_id)
    return None

​
3. Clean Up Completed Processes

Periodically clean up old processes: 
# List all processes
processes = sandbox.list_processes()

# Kill old completed processes
for proc in processes:
    if proc['status'] == 'completed':
        # Optionally log or archive before cleanup
        pass

​
4. Handle Process Failures

Check exit codes for failed processes: 
processes = sandbox.list_processes()
for proc in processes:
    if proc['status'] == 'failed':
        print(f"Process {proc['process_id']} failed")
        if 'exit_code' in proc:
            print(f"Exit code: {proc['exit_code']}")

​
Examples

​
Data Processing Pipeline

# Start data processing
process = sandbox.run_code_background(
    '''
import pandas as pd
import time

# Load and process data
df = pd.read_csv('/workspace/raw_data.csv')
df = df.groupby('category').agg({
    'value': 'sum',
    'count': 'count'
})
df.to_csv('/workspace/processed_data.csv')

print("Processing complete!")
''',
    name="data_processing",
    timeout=1800
)

# Monitor progress
process_id = process['process_id']
while True:
    processes = sandbox.list_processes()
    proc = next((p for p in processes if p['process_id'] == process_id), None)
    
    if proc and proc['status'] == 'completed':
        print("Processing complete!")
        break
    elif proc and proc['status'] == 'failed':
        print("Processing failed!")
        break
    
    time.sleep(10)

​
Parallel Task Execution

# Start multiple background tasks
tasks = []
for i in range(5):
    process = sandbox.run_code_background(
        f'''
import time
time.sleep(10)
print("Task {i} complete")
''',
        name=f"task_{i}"
    )
    tasks.append(process['process_id'])

# Wait for all tasks
while tasks:
    processes = sandbox.list_processes()
    for process_id in tasks[:]:
        proc = next((p for p in processes if p['process_id'] == process_id), None)
        if proc and proc['status'] in ('completed', 'failed'):
            print(f"Task {process_id} finished: {proc['status']}")
            tasks.remove(process_id)
    
    if tasks:
        time.sleep(5)