Session Storage Location
Claude Agent SDK stores sessions in:{project-hash}- Hash of the project directory{session-id}- Unique session identifier
Persisting and resuming Claude Agent SDK sessions
~/.claude/projects/{project-hash}/{session-id}.jsonl
{project-hash} - Hash of the project directory{session-id} - Unique session identifierimport json
from moru import Sandbox
sandbox = Sandbox.create("claude-code")
# Run some tasks to create session
sandbox.commands.run("claude -p 'Create a hello world script'")
# Find and read the session file
result = sandbox.commands.run("find ~/.claude -name '*.jsonl' -type f")
session_files = result.stdout.strip().split("\n")
sessions = {}
for path in session_files:
if path:
content = sandbox.files.read(path)
sessions[path] = content
# Save locally
with open("session_backup.json", "w") as f:
json.dump(sessions, f)
import json
from moru import Sandbox
# Load saved sessions
with open("session_backup.json") as f:
sessions = json.load(f)
# Create new sandbox
sandbox = Sandbox.create("claude-code")
# Restore session files
for path, content in sessions.items():
# Ensure directory exists
dir_path = "/".join(path.split("/")[:-1])
sandbox.files.make_dir(dir_path)
sandbox.files.write(path, content)
# Resume the session
sandbox.commands.run("claude --resume")
import tarfile
import io
from moru import Sandbox
sandbox = Sandbox.create("claude-code")
# Work with Claude...
sandbox.commands.run("claude -p 'Create a Python project'")
# Create tar archive of workspace
result = sandbox.commands.run(
"tar -czf /tmp/workspace.tar.gz -C /home/user .",
timeout=60
)
# Download the archive
archive_data = sandbox.files.read("/tmp/workspace.tar.gz", format="bytes")
with open("workspace_backup.tar.gz", "wb") as f:
f.write(archive_data)
from moru import Sandbox
sandbox = Sandbox.create("claude-code")
# Upload the archive
with open("workspace_backup.tar.gz", "rb") as f:
archive_data = f.read()
sandbox.files.write("/tmp/workspace.tar.gz", archive_data)
# Extract to home directory
sandbox.commands.run("tar -xzf /tmp/workspace.tar.gz -C /home/user")
# Continue working
sandbox.commands.run("claude --resume")
from moru import Sandbox
# Create and work
sandbox = Sandbox.create("claude-code")
sandbox.commands.run("claude -p 'Start a project'")
# Save sandbox ID for later
sandbox_id = sandbox.sandbox_id
# Pause instead of killing
sandbox.beta_pause()
# Later, resume
sandbox = Sandbox.connect(sandbox_id)
sandbox.connect() # Resume from paused state
# Session is intact, continue working
sandbox.commands.run("claude --resume")
class AgentSession:
def __init__(self, session_id: str):
self.session_id = session_id
self.sandbox = None
async def start(self):
"""Start a new session."""
self.sandbox = Sandbox.create("claude-code")
return self.sandbox.sandbox_id
async def resume(self, sandbox_id: str):
"""Resume an existing session."""
self.sandbox = Sandbox.connect(sandbox_id)
if not self.sandbox.is_running():
self.sandbox.connect() # Resume if paused
async def run_task(self, prompt: str):
"""Run a task with Claude."""
return self.sandbox.commands.run(
f"claude -p '{prompt}'",
timeout=300
)
async def save_state(self) -> dict:
"""Save session state for external storage."""
# Get session files
result = self.sandbox.commands.run(
"find ~/.claude -name '*.jsonl'"
)
state = {
"sandbox_id": self.sandbox.sandbox_id,
"sessions": {}
}
for path in result.stdout.strip().split("\n"):
if path:
content = self.sandbox.files.read(path)
state["sessions"][path] = content
return state
async def pause(self):
"""Pause the session."""
self.sandbox.beta_pause()
async def stop(self):
"""Stop and clean up."""
self.sandbox.kill()
import time
import threading
def checkpoint_thread(session, interval=300):
while session.sandbox.is_running():
state = session.save_state()
storage.save(session.session_id, state)
time.sleep(interval)
sandbox = Sandbox.create("claude-code", timeout=3600)
# Set a reminder before timeout
def save_before_timeout():
time.sleep(3500) # 50 seconds before timeout
state = save_session_state(sandbox)
storage.save(session_id, state)
threading.Thread(target=save_before_timeout, daemon=True).start()
import tempfile
import os
def save_state_atomically(session_id, state):
# Write to temp file first
with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
json.dump(state, f)
temp_path = f.name
# Atomically replace
final_path = f"sessions/{session_id}.json"
os.rename(temp_path, final_path)