Skip to main content
Watch filesystem for real-time changes. Get notified immediately when files are created, modified, deleted, or renamed in your sandbox.

​
Overview

File watching is perfect for:
  • Real-time monitoring of file changes
  • Building reactive applications
  • Detecting when outputs are generated
  • Implementing file-based workflows
File watching uses WebSocket connections for real-time communication. It requires async/await support and the websockets library (Python).

​
Basic File Watching

Watch a directory for changes:
  • Python
  • JavaScript
from hopx_ai import Sandbox
import asyncio

async def watch_files():
    sandbox = Sandbox.create(template="code-interpreter")
    
    # Start watching
    print("👀 Watching /workspace for changes...")
    event_count = 0
    
    async for event in sandbox.files.watch("/workspace"):
        event_type = event.get('event', 'unknown')
        path = event.get('path', '')
        
        print(f"{event_type}: {path}")
        
        event_count += 1
        if event_count >= 10:
            print("Stopping after 10 events")
            break
    
    sandbox.kill()

asyncio.run(watch_files())

​
Event Types

Handle different file system events:
  • Python
  • JavaScript
from hopx_ai import Sandbox
import asyncio

async def handle_all_events():
    sandbox = Sandbox.create(template="code-interpreter")
    
    async for event in sandbox.files.watch("/workspace"):
        event_type = event.get('event', 'unknown')
        path = event.get('path', '')
        timestamp = event.get('timestamp', '')
        
        if event_type == 'created':
            print(f"📄 Created: {path}")
        elif event_type == 'modified':
            print(f"✏️  Modified: {path}")
        elif event_type == 'deleted':
            print(f"🗑️  Deleted: {path}")
        elif event_type == 'renamed':
            old_path = event.get('old_path', '')
            print(f"📝 Renamed: {old_path} → {path}")
        else:
            print(f"❓ Unknown event: {event_type} - {path}")

asyncio.run(handle_all_events())

​
Monitoring File Generation

Watch for generated output files:
  • Python
  • JavaScript
from hopx_ai import Sandbox
import asyncio

async def monitor_outputs():
    sandbox = Sandbox.create(template="code-interpreter")
    
    # Start background task that generates files
    sandbox.run_code_background('''
import time
import json

for i in range(5):
    data = {"step": i, "value": i * 2}
    with open(f'/workspace/output_{i}.json', 'w') as f:
        json.dump(data, f)
    time.sleep(2)
''', name='file-generator')
    
    # Watch for new files
    print("👀 Watching for output files...")
    output_files = []
    
    async for event in sandbox.files.watch("/workspace"):
        if event.get('event') == 'created':
            path = event.get('path', '')
            if path.startswith('/workspace/output_'):
                output_files.append(path)
                print(f"✅ New output: {path}")
                
                if len(output_files) >= 5:
                    print("All outputs generated!")
                    break
    
    print(f"\nGenerated {len(output_files)} files")
    sandbox.kill()

asyncio.run(monitor_outputs())

​
Filtering Events

Filter events by type or path:
  • Python
  • JavaScript
from hopx_ai import Sandbox
import asyncio

async def filter_events():
    sandbox = Sandbox.create(template="code-interpreter")
    
    # Only watch for Python files
    async for event in sandbox.files.watch("/workspace"):
        event_type = event.get('event', '')
        path = event.get('path', '')
        
        # Only process Python files
        if path.endswith('.py'):
            if event_type == 'created':
                print(f"📄 New Python file: {path}")
            elif event_type == 'modified':
                print(f"✏️  Python file modified: {path}")
            elif event_type == 'deleted':
                print(f"🗑️  Python file deleted: {path}")

asyncio.run(filter_events())

​
Complete Example

Here’s a complete file watching workflow:
  • Python
  • JavaScript
from hopx_ai import Sandbox
import asyncio

async def complete_watch_example():
    sandbox = Sandbox.create(template="code-interpreter")
    
    # Start a task that creates/modifies files
    sandbox.run_code_background('''
import time
import json

# Create initial file
with open('/workspace/status.json', 'w') as f:
    json.dump({"status": "starting"}, f)

time.sleep(1)

# Modify file
with open('/workspace/status.json', 'w') as f:
    json.dump({"status": "processing"}, f)

time.sleep(1)

# Create output file
with open('/workspace/result.txt', 'w') as f:
    f.write("Processing complete")

time.sleep(1)

# Delete status file
import os
os.remove('/workspace/status.json')
''', name='file-operations')
    
    # Watch for all events
    print("👀 Watching filesystem...\n")
    events_received = []
    
    async for event in sandbox.files.watch("/workspace"):
        event_type = event.get('event', 'unknown')
        path = event.get('path', '')
        timestamp = event.get('timestamp', '')
        
        events_received.append({
            'type': event_type,
            'path': path,
            'timestamp': timestamp
        })
        
        # Display event
        icons = {
            'created': '📄',
            'modified': '✏️',
            'deleted': '🗑️',
            'renamed': '📝'
        }
        icon = icons.get(event_type, '❓')
        print(f"{icon} {event_type.upper()}: {path}")
        
        # Stop after receiving all expected events
        if len(events_received) >= 4:
            print("\n✅ All events received")
            break
    
    # Summary
    print(f"\n📊 Summary: Received {len(events_received)} events")
    for e in events_received:
        print(f"  - {e['type']}: {e['path']}")
    
    sandbox.kill()

asyncio.run(complete_watch_example())

​
Best Practices

1

1. Use Async/Await

File watching requires async/await. Make sure your code uses async functions and async for loops.
2

2. Filter Events

Filter events by type or path pattern to process only relevant changes.
3

3. Handle All Event Types

Process created, modified, deleted, and renamed events appropriately for your use case.
4

4. Set Timeout Limits

Consider setting timeouts or event count limits to avoid infinite watching.
5

5. Clean Up Connections

Always break the watch loop and clean up connections when done.

​
Next Steps