Process

class Process()

Handles process and code execution within a Sandbox.

Process.init

def __init__(code_toolbox: SandboxPythonCodeToolbox, api_client: ProcessApi,
             ensure_toolbox_url: Callable[[], None])

Initialize a new Process instance.

Arguments:

code_toolbox SandboxPythonCodeToolbox - Language-specific code execution toolbox.
api_client ProcessApi - API client for process operations.
ensure_toolbox_url Callable[[], None] - Ensures the toolbox API URL is initialized. Must be called before invoking any private methods on the API client.

Process.exec

@intercept_errors(message_prefix="Failed to execute command: ")
def exec(command: str,
         cwd: Optional[str] = None,
         env: Optional[Dict[str, str]] = None,
         timeout: Optional[int] = None) -> ExecuteResponse

Execute a shell command in the Sandbox.

Arguments:

command str - Shell command to execute.
cwd Optional[str] - Working directory for command execution. If not specified, uses the sandbox working directory.
env Optional[Dict[str, str]] - Environment variables to set for the command.
timeout Optional[int] - Maximum time in seconds to wait for the command to complete. 0 means wait indefinitely.

Returns:

ExecuteResponse - Command execution results containing:
- exit_code: The command’s exit status
- result: Standard output from the command
- artifacts: ExecutionArtifacts object containing stdout (same as result) and charts (matplotlib charts metadata)

Example:

# Simple command
response = sandbox.process.exec("echo 'Hello'")
print(response.artifacts.stdout)  # Prints: Hello

# Command with working directory
result = sandbox.process.exec("ls", cwd="workspace/src")

# Command with timeout
result = sandbox.process.exec("sleep 10", timeout=5)

Process.code_run

def code_run(code: str,
             params: Optional[CodeRunParams] = None,
             timeout: Optional[int] = None) -> ExecuteResponse

Executes code in the Sandbox using the appropriate language runtime.

Arguments:

code str - Code to execute.
params Optional[CodeRunParams] - Parameters for code execution.
timeout Optional[int] - Maximum time in seconds to wait for the code to complete. 0 means wait indefinitely.

Returns:

ExecuteResponse - Code execution result containing:
- exit_code: The execution’s exit status
- result: Standard output from the code
- artifacts: ExecutionArtifacts object containing stdout (same as result) and charts (matplotlib charts metadata)

Example:

# Run Python code
response = sandbox.process.code_run('''
    x = 10
    y = 20
    print(f"Sum: {x + y}")
''')
print(response.artifacts.stdout)  # Prints: Sum: 30

Matplotlib charts are automatically detected and returned in the charts field of the ExecutionArtifacts object.

code = '''
import matplotlib.pyplot as plt
import numpy as np

x = np.linspace(0, 10, 30)
y = np.sin(x)

plt.figure(figsize=(8, 5))
plt.plot(x, y, 'b-', linewidth=2)
plt.title('Line Chart')
plt.xlabel('X-axis (seconds)')
plt.ylabel('Y-axis (amplitude)')
plt.grid(True)
plt.show()
'''

response = sandbox.process.code_run(code)
chart = response.artifacts.charts[0]

print(f"Type: {chart.type}")
print(f"Title: {chart.title}")
if chart.type == ChartType.LINE and isinstance(chart, LineChart):
    print(f"X Label: {chart.x_label}")
    print(f"Y Label: {chart.y_label}")
    print(f"X Ticks: {chart.x_ticks}")
    print(f"X Tick Labels: {chart.x_tick_labels}")
    print(f"X Scale: {chart.x_scale}")
    print(f"Y Ticks: {chart.y_ticks}")
    print(f"Y Tick Labels: {chart.y_tick_labels}")
    print(f"Y Scale: {chart.y_scale}")
    print("Elements:")
    for element in chart.elements:
        print(f"Label: {element.label}")
        print(f"Points: {element.points}")

Process.create_session

@intercept_errors(message_prefix="Failed to create session: ")
def create_session(session_id: str) -> None

Creates a new long-running background session in the Sandbox.

Sessions are background processes that maintain state between commands, making them ideal for scenarios requiring multiple related commands or persistent environment setup. You can run long-running commands and monitor process status.

Arguments:

session_id str - Unique identifier for the new session.

Example:

# Create a new session
session_id = "my-session"
sandbox.process.create_session(session_id)
session = sandbox.process.get_session(session_id)
# Do work...
sandbox.process.delete_session(session_id)

Process.get_session

@intercept_errors(message_prefix="Failed to get session: ")
def get_session(session_id: str) -> Session

Gets a session in the Sandbox.

Arguments:

session_id str - Unique identifier of the session to retrieve.

Returns:

Session - Session information including:
- session_id: The session’s unique identifier
- commands: List of commands executed in the session

Example:

session = sandbox.process.get_session("my-session")
for cmd in session.commands:
    print(f"Command: {cmd.command}")

Process.get_session_command

@intercept_errors(message_prefix="Failed to get session command: ")
def get_session_command(session_id: str, command_id: str) -> Command

Gets information about a specific command executed in a session.

Arguments:

session_id str - Unique identifier of the session.
command_id str - Unique identifier of the command.

Returns:

Command - Command information including:
- id: The command’s unique identifier
- command: The executed command string
- exit_code: Command’s exit status (if completed)

Example:

cmd = sandbox.process.get_session_command("my-session", "cmd-123")
if cmd.exit_code == 0:
    print(f"Command {cmd.command} completed successfully")

Process.execute_session_command

@intercept_errors(message_prefix="Failed to execute session command: ")
def execute_session_command(
        session_id: str,
        req: SessionExecuteRequest,
        timeout: Optional[int] = None) -> SessionExecuteResponse

Executes a command in the session.

Arguments:

session_id str - Unique identifier of the session to use.
req SessionExecuteRequest - Command execution request containing:
- command: The command to execute
- run_async: Whether to execute asynchronously

Returns:

SessionExecuteResponse - Command execution results containing:
- cmd_id: Unique identifier for the executed command
- output: Combined command output (stdout and stderr) (if synchronous execution)
- stdout: Standard output from the command
- stderr: Standard error from the command
- exit_code: Command exit status (if synchronous execution)

Example:

# Execute commands in sequence, maintaining state
session_id = "my-session"

# Change directory
req = SessionExecuteRequest(command="cd /workspace")
sandbox.process.execute_session_command(session_id, req)

# Create a file
req = SessionExecuteRequest(command="echo 'Hello' > test.txt")
sandbox.process.execute_session_command(session_id, req)

# Read the file
req = SessionExecuteRequest(command="cat test.txt")
result = sandbox.process.execute_session_command(session_id, req)
print(f"Command stdout: {result.stdout}")
print(f"Command stderr: {result.stderr}")

Process.get_session_command_logs

@intercept_errors(message_prefix="Failed to get session command logs: ")
def get_session_command_logs(session_id: str,
                             command_id: str) -> SessionCommandLogsResponse

Get the logs for a command executed in a session.

Arguments:

session_id str - Unique identifier of the session.
command_id str - Unique identifier of the command.

Returns:

SessionCommandLogsResponse - Command logs including:
- output: Combined command output (stdout and stderr)
- stdout: Standard output from the command
- stderr: Standard error from the command

Example:

logs = sandbox.process.get_session_command_logs(
    "my-session",
    "cmd-123"
)
print(f"Command stdout: {logs.stdout}")
print(f"Command stderr: {logs.stderr}")

Process.get_session_command_logs_async

@intercept_errors(message_prefix="Failed to get session command logs: ")
async def get_session_command_logs_async(
        session_id: str, command_id: str, on_stdout: Callable[[str], None],
        on_stderr: Callable[[str], None]) -> None

Asynchronously retrieves and processes the logs for a command executed in a session as they become available.

Arguments:

session_id str - Unique identifier of the session.
command_id str - Unique identifier of the command.
on_stdout Callable[[str], None] - Callback function to handle stdout log chunks as they arrive.
on_stderr Callable[[str], None] - Callback function to handle stderr log chunks as they arrive.

Example:

await sandbox.process.get_session_command_logs_async(
    "my-session",
    "cmd-123",
    lambda log: print(f"[STDOUT]: {log}"),
    lambda log: print(f"[STDERR]: {log}"),
)

Process.list_sessions

@intercept_errors(message_prefix="Failed to list sessions: ")
def list_sessions() -> List[Session]

Lists all sessions in the Sandbox.

Returns:

List[Session] - List of all sessions in the Sandbox.

Example:

sessions = sandbox.process.list_sessions()
for session in sessions:
    print(f"Session {session.session_id}:")
    print(f"  Commands: {len(session.commands)}")

Process.delete_session

@intercept_errors(message_prefix="Failed to delete session: ")
def delete_session(session_id: str) -> None

Terminates and removes a session from the Sandbox, cleaning up any resources associated with it.

Arguments:

session_id str - Unique identifier of the session to delete.

Example:

# Create and use a session
sandbox.process.create_session("temp-session")
# ... use the session ...

# Clean up when done
sandbox.process.delete_session("temp-session")

Process.create_pty_session

@intercept_errors(message_prefix="Failed to create PTY session: ")
def create_pty_session(id: str,
                       cwd: Optional[str] = None,
                       envs: Optional[Dict[str, str]] = None,
                       pty_size: Optional[PtySize] = None) -> PtyHandle

Creates a new PTY (pseudo-terminal) session in the Sandbox.

Creates an interactive terminal session that can execute commands and handle user input. The PTY session behaves like a real terminal, supporting features like command history.

Arguments:

id - Unique identifier for the PTY session. Must be unique within the Sandbox.
cwd - Working directory for the PTY session. Defaults to the sandbox’s working directory.
env - Environment variables to set in the PTY session. These will be merged with the Sandbox’s default environment variables.
pty_size - Terminal size configuration. Defaults to 80x24 if not specified.

Returns:

PtyHandle - Handle for managing the created PTY session. Use this to send input, receive output, resize the terminal, and manage the session lifecycle.

Raises:

DaytonaError - If the PTY session creation fails or the session ID is already in use.

Process.connect_pty_session

@intercept_errors(message_prefix="Failed to connect PTY session: ")
def connect_pty_session(session_id: str) -> PtyHandle

Connects to an existing PTY session in the Sandbox.

Establishes a WebSocket connection to an existing PTY session, allowing you to interact with a previously created terminal session.

Arguments:

session_id - Unique identifier of the PTY session to connect to.

Returns:

PtyHandle - Handle for managing the connected PTY session.

Raises:

DaytonaError - If the PTY session doesn’t exist or connection fails.

Process.list_pty_sessions

@intercept_errors(message_prefix="Failed to list PTY sessions: ")
def list_pty_sessions() -> List[PtySessionInfo]

Lists all PTY sessions in the Sandbox.

Retrieves information about all PTY sessions in this Sandbox.

Returns:

List[PtySessionInfo] - List of PTY session information objects containing details about each session’s state, creation time, and configuration.

Example:

# List all PTY sessions
sessions = sandbox.process.list_pty_sessions()

for session in sessions:
    print(f"Session ID: {session.id}")
    print(f"Active: {session.active}")
    print(f"Created: {session.created_at}")

Process.get_pty_session_info

@intercept_errors(message_prefix="Failed to get PTY session info: ")
def get_pty_session_info(session_id: str) -> PtySessionInfo

Gets detailed information about a specific PTY session.

Retrieves comprehensive information about a PTY session including its current state, configuration, and metadata.

Arguments:

session_id - Unique identifier of the PTY session to retrieve information for.

Returns:

PtySessionInfo - Detailed information about the PTY session including ID, state, creation time, working directory, environment variables, and more.

Raises:

DaytonaError - If the PTY session doesn’t exist.

Example:

# Get details about a specific PTY session
session_info = sandbox.process.get_pty_session_info("my-session")

print(f"Session ID: {session_info.id}")
print(f"Active: {session_info.active}")
print(f"Working Directory: {session_info.cwd}")
print(f"Terminal Size: {session_info.cols}x{session_info.rows}")

Process.kill_pty_session

@intercept_errors(message_prefix="Failed to kill PTY session: ")
def kill_pty_session(session_id: str) -> None

Kills a PTY session and terminates its associated process.

Forcefully terminates the PTY session and cleans up all associated resources. This will close any active connections and kill the underlying shell process. This operation is irreversible. Any unsaved work in the terminal session will be lost.

Arguments:

session_id - Unique identifier of the PTY session to kill.

Raises:

DaytonaError - If the PTY session doesn’t exist or cannot be killed.

Example:

# Kill a specific PTY session
sandbox.process.kill_pty_session("my-session")

# Verify the session no longer exists
pty_sessions = sandbox.process.list_pty_sessions()
for pty_session in pty_sessions:
    print(f"PTY session: {pty_session.id}")

Process.resize_pty_session

@intercept_errors(message_prefix="Failed to resize PTY session: ")
def resize_pty_session(session_id: str, pty_size: PtySize) -> PtySessionInfo

Resizes a PTY session’s terminal dimensions.

Changes the terminal size of an active PTY session. This is useful when the client terminal is resized or when you need to adjust the display for different output requirements.

Arguments:

session_id - Unique identifier of the PTY session to resize.
pty_size - New terminal dimensions containing the desired columns and rows.

Returns:

PtySessionInfo - Updated session information reflecting the new terminal size.

Raises:

DaytonaError - If the PTY session doesn’t exist or resize operation fails.

Example:

from daytona.common.pty import PtySize

# Resize a PTY session to a larger terminal
new_size = PtySize(rows=40, cols=150)
updated_info = sandbox.process.resize_pty_session("my-session", new_size)

print(f"Terminal resized to {updated_info.cols}x{updated_info.rows}")

# You can also use the PtyHandle's resize method
pty_handle.resize(new_size)

CodeRunParams

@dataclass
class CodeRunParams()

Parameters for code execution.

Attributes:

argv Optional[List[str]] - Command line arguments
env Optional[Dict[str, str]] - Environment variables

SessionExecuteRequest

class SessionExecuteRequest(ApiSessionExecuteRequest,
                            AsyncApiSessionExecuteRequest)

Contains the request for executing a command in a session.

Attributes:

command str - The command to execute.
run_async Optional[bool] - Whether to execute the command asynchronously.
var_async Optional[bool] - Deprecated. Use run_async instead.

ExecutionArtifacts

class ExecutionArtifacts()

Artifacts from the command execution.

Attributes:

stdout str - Standard output from the command, same as result in ExecuteResponse
charts Optional[List[Chart]] - List of chart metadata from matplotlib

ExecuteResponse

class ExecuteResponse(BaseModel)

Response from the command execution.

Attributes:

exit_code int - The exit code from the command execution
result str - The output from the command execution
artifacts Optional[ExecutionArtifacts] - Artifacts from the command execution

SessionExecuteResponse

class SessionExecuteResponse(ApiSessionExecuteResponse)

Response from the session command execution.

Attributes:

output str - The output from the command execution
exit_code int - The exit code from the command execution

SessionCommandLogsResponse

class SessionCommandLogsResponse()

Response from the command logs.

Attributes:

output str - The combined output from the command
stdout str - The stdout from the command
stderr str - The stderr from the command

parse_session_command_logs

def parse_session_command_logs(data: bytes) -> SessionCommandLogsResponse

Parse combined stdout/stderr output into separate streams.

Arguments:

data - Combined log bytes with STDOUT_PREFIX and STDERR_PREFIX markers

Returns:

SessionCommandLogsResponse with separated stdout and stderr

demux_log

def demux_log(data: bytes) -> tuple[bytes, bytes]

Demultiplex combined stdout/stderr log data.

Arguments:

data - Combined log bytes with STDOUT_PREFIX and STDERR_PREFIX markers

Returns:

Tuple of (stdout_bytes, stderr_bytes)

Process

Process.__init__

Process.exec

Process.code_run

Process.create_session

Process.get_session

Process.get_session_command

Process.execute_session_command

Process.get_session_command_logs

Process.get_session_command_logs_async

Process.list_sessions

Process.delete_session

Process.create_pty_session

Process.connect_pty_session

Process.list_pty_sessions

Process.get_pty_session_info

Process.kill_pty_session

Process.resize_pty_session

CodeRunParams

SessionExecuteRequest

ExecutionArtifacts

ExecuteResponse

SessionExecuteResponse

SessionCommandLogsResponse

parse_session_command_logs

demux_log

Process.init