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:
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())
import { Sandbox } from '@hopx-ai/sdk';
async function watchFiles() {
const sandbox = await Sandbox.create({ template: 'code-interpreter' });
// Start watching
console.log('👀 Watching /workspace for changes...');
let eventCount = 0;
for await (const event of sandbox.files.watch('/workspace')) {
const eventType = event.event || 'unknown';
const path = event.path || '';
console.log(`${eventType}: ${path}`);
eventCount++;
if (eventCount >= 10) {
console.log('Stopping after 10 events');
break;
}
}
await sandbox.kill();
}
watchFiles();
Event Types
Handle different file system events:
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())
import { Sandbox } from '@hopx-ai/sdk';
async function handleAllEvents() {
const sandbox = await Sandbox.create({ template: 'code-interpreter' });
for await (const event of sandbox.files.watch('/workspace')) {
const eventType = event.event || 'unknown';
const path = event.path || '';
const timestamp = event.timestamp || '';
if (eventType === 'created') {
console.log(`📄 Created: ${path}`);
} else if (eventType === 'modified') {
console.log(`✏️ Modified: ${path}`);
} else if (eventType === 'deleted') {
console.log(`🗑️ Deleted: ${path}`);
} else if (eventType === 'renamed') {
const oldPath = event.old_path || '';
console.log(`📝 Renamed: ${oldPath} → ${path}`);
} else {
console.log(`❓ Unknown event: ${eventType} - ${path}`);
}
}
await sandbox.kill();
}
handleAllEvents();
Monitoring File Generation
Watch for generated output files:
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())
import { Sandbox } from '@hopx-ai/sdk';
async function monitorOutputs() {
const sandbox = await Sandbox.create({ template: 'code-interpreter' });
// Start background task that generates files
await sandbox.runCodeBackground(`
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
console.log('👀 Watching for output files...');
const outputFiles = [];
for await (const event of sandbox.files.watch('/workspace')) {
if (event.event === 'created') {
const path = event.path || '';
if (path.startsWith('/workspace/output_')) {
outputFiles.push(path);
console.log(`✅ New output: ${path}`);
if (outputFiles.length >= 5) {
console.log('All outputs generated!');
break;
}
}
}
}
console.log(`\nGenerated ${outputFiles.length} files`);
await sandbox.kill();
}
monitorOutputs();
Filtering Events
Filter events by type or path:
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())
import { Sandbox } from '@hopx-ai/sdk';
async function filterEvents() {
const sandbox = await Sandbox.create({ template: 'code-interpreter' });
// Only watch for Python files
for await (const event of sandbox.files.watch('/workspace')) {
const eventType = event.event || '';
const path = event.path || '';
// Only process Python files
if (path.endsWith('.py')) {
if (eventType === 'created') {
console.log(`📄 New Python file: ${path}`);
} else if (eventType === 'modified') {
console.log(`✏️ Python file modified: ${path}`);
} else if (eventType === 'deleted') {
console.log(`🗑️ Python file deleted: ${path}`);
}
}
}
await sandbox.kill();
}
filterEvents();
Complete Example
Here’s a complete file watching workflow:
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())
import { Sandbox } from '@hopx-ai/sdk';
async function completeWatchExample() {
const sandbox = await Sandbox.create({ template: 'code-interpreter' });
// Start a task that creates/modifies files
await sandbox.runCodeBackground(`
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
console.log('👀 Watching filesystem...\n');
const eventsReceived = [];
for await (const event of sandbox.files.watch('/workspace')) {
const eventType = event.event || 'unknown';
const path = event.path || '';
const timestamp = event.timestamp || '';
eventsReceived.push({
type: eventType,
path: path,
timestamp: timestamp
});
// Display event
const icons = {
'created': '📄',
'modified': '✏️',
'deleted': '🗑️',
'renamed': '📝'
};
const icon = icons[eventType] || '❓';
console.log(`${icon} ${eventType.toUpperCase()}: ${path}`);
// Stop after receiving all expected events
if (eventsReceived.length >= 4) {
console.log('\n✅ All events received');
break;
}
}
// Summary
console.log(`\n📊 Summary: Received ${eventsReceived.length} events`);
for (const e of eventsReceived) {
console.log(` - ${e.type}: ${e.path}`);
}
await sandbox.kill();
}
completeWatchExample();
Best Practices
1. Use Async/Await
File watching requires async/await. Make sure your code uses async functions and async for loops.
2. Filter Events
Filter events by type or path pattern to process only relevant changes.
3. Handle All Event Types
Process created, modified, deleted, and renamed events appropriately for your use case.
4. Set Timeout Limits
Consider setting timeouts or event count limits to avoid infinite watching.
5. Clean Up Connections
Always break the watch loop and clean up connections when done.
Next Steps
