AsyncProcess | Daytona

AsyncProcess

class AsyncProcess()

Handles process and code execution within a Sandbox.

AsyncProcess.init

def __init__(language: str, api_client: ProcessApi)

Initialize a new Process instance.

Arguments:

AsyncProcess.exec

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

Execute a shell command in the Sandbox.

The command is run by the first available of zsh, bash or sh in the sandbox, which is zsh in the default snapshots. Quote arguments that contain glob characters, for example pip install -e '.[test]', or wrap bash-specific scripts in bash -c '...'.

The call returns when the command finishes. To start a long-running process such as a server, run it in the background with its output redirected, e.g. python3 -m http.server 8000 > server.log 2>&1 &, or use a session with run_async=True (see execute_session_command). A background process that keeps the command’s output open (cmd & without a redirect) can block the call until that process exits.

Tools in the sandbox may report the host’s CPU count and memory (for example nproc or os.cpu_count()) rather than the sandbox’s own limits. Size worker pools from the sandbox’s cpu and memory instead, e.g. pytest -n 2 rather than pytest -n auto.

Arguments:

Returns:

Example:

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

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

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

# Start a server in the background
await sandbox.process.exec("python3 -m http.server 8000 > server.log 2>&1 &")

AsyncProcess.code_run

@with_instrumentation()
async def code_run(code: str,
                   params: CodeRunParams | None = None,
                   timeout: int | None = None) -> ExecuteResponse

Executes code in the Sandbox using the appropriate language runtime.

Arguments:

Returns:

Example:

# Run Python code
response = await 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 = await 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}")

AsyncProcess.create_session

@intercept_errors(message_prefix="Failed to create session: ")
@with_instrumentation()
async def create_session(session_id: str,
                         request_timeout: float | None = None) -> 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:

Example:

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

AsyncProcess.get_session

@intercept_errors(message_prefix="Failed to get session: ")
async def get_session(session_id: str,
                      request_timeout: float | None = None) -> Session

Gets a session in the Sandbox.

Arguments:

Returns:

Example:

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

AsyncProcess.get_entrypoint_session

@intercept_errors(message_prefix="Failed to get sandbox entrypoint session: ")
async def get_entrypoint_session(
        request_timeout: float | None = None) -> Session

Gets the sandbox entrypoint session.

Arguments:

Returns:

Example:

session = await sandbox.process.get_entrypoint_session()
for cmd in session.commands:
    print(f"Command: {cmd.command}")

AsyncProcess.get_session_command

@intercept_errors(message_prefix="Failed to get session command: ")
@with_instrumentation()
async def get_session_command(session_id: str,
                              command_id: str,
                              request_timeout: float | None = None) -> Command

Gets information about a specific command executed in a session.

Arguments:

Returns:

Example:

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

AsyncProcess.execute_session_command

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

Executes a command in the session.

Arguments:

Returns:

Example:

# Execute commands in sequence, maintaining state
session_id = "my-session"
# Change directory
req = SessionExecuteRequest(command="cd /workspace")
await sandbox.process.execute_session_command(session_id, req)
# Create a file
req = SessionExecuteRequest(command="echo 'Hello' > test.txt")
await sandbox.process.execute_session_command(session_id, req)
# Read the file
req = SessionExecuteRequest(command="cat test.txt")
result = await sandbox.process.execute_session_command(session_id, req)
print(f"Command stdout: {result.stdout}")
print(f"Command stderr: {result.stderr}")

AsyncProcess.get_session_command_logs

@intercept_errors(message_prefix="Failed to get session command logs: ")
@with_instrumentation()
async def get_session_command_logs(
        session_id: str,
        command_id: str,
        request_timeout: float | None = None) -> SessionCommandLogsResponse

Get the logs for a command executed in a session.

Arguments:

Returns:

Example:

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

AsyncProcess.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: OutputHandler[str],
        on_stderr: OutputHandler[str]) -> None

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

Accepts both sync and async callbacks. Async callbacks are awaited. Blocking synchronous operations inside callbacks may cause WebSocket disconnections — use async callbacks and async libraries to avoid this.

Arguments:

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}"),
)

AsyncProcess.get_entrypoint_logs

@intercept_errors(message_prefix="Failed to get entrypoint logs: ")
@with_instrumentation()
async def get_entrypoint_logs(
        request_timeout: float | None = None) -> SessionCommandLogsResponse

Get the logs for the entrypoint session.

Arguments:

Returns:

Example:

logs = await sandbox.process.get_entrypoint_logs()
print(f"Command stdout: {logs.stdout}")
print(f"Command stderr: {logs.stderr}")

AsyncProcess.get_entrypoint_logs_async

@intercept_errors(message_prefix="Failed to get entrypoint logs: ")
async def get_entrypoint_logs_async(on_stdout: OutputHandler[str],
                                    on_stderr: OutputHandler[str]) -> None

Asynchronously retrieves and processes the logs for the entrypoint session as they become available.

Arguments:

Example:

await sandbox.process.get_entrypoint_logs_async(
    lambda log: print(f"[STDOUT]: {log}"),
    lambda log: print(f"[STDERR]: {log}"),
)

AsyncProcess.send_session_command_input

@intercept_errors(message_prefix="Failed to send session command input: ")
async def send_session_command_input(
        session_id: str,
        command_id: str,
        data: str,
        request_timeout: float | None = None) -> None

Sends input data to a command executed in a session.

Arguments:

AsyncProcess.list_sessions

@intercept_errors(message_prefix="Failed to list sessions: ")
@with_instrumentation()
async def list_sessions(request_timeout: float | None = None) -> list[Session]

Lists all sessions in the Sandbox.

Arguments:

Returns:

Example:

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

AsyncProcess.delete_session

@intercept_errors(message_prefix="Failed to delete session: ")
@with_instrumentation()
async def delete_session(session_id: str,
                         request_timeout: float | None = None) -> None

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

Arguments:

Example:

# Create and use a session
await sandbox.process.create_session("temp-session")
# ... use the session ...
# Clean up when done
await sandbox.process.delete_session("temp-session")

AsyncProcess.create_pty_session

@intercept_errors(message_prefix="Failed to create PTY session: ")
@with_instrumentation()
async def create_pty_session(
        id: str,
        on_data: Callable[[bytes], None] | Callable[[bytes], Awaitable[None]],
        cwd: str | None = None,
        envs: dict[str, str] | None = None,
        pty_size: PtySize | None = None) -> AsyncPtyHandle

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:

Returns:

Raises:

AsyncProcess.connect_pty_session

@intercept_errors(message_prefix="Failed to connect PTY session: ")
@with_instrumentation()
async def connect_pty_session(
    session_id: str,
    on_data: Callable[[bytes], None] | Callable[[bytes], Awaitable[None]]
) -> AsyncPtyHandle

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:

Returns:

Raises:

AsyncProcess.list_pty_sessions

@intercept_errors(message_prefix="Failed to list PTY sessions: ")
@with_instrumentation()
async def list_pty_sessions(
        request_timeout: float | None = None) -> list[PtySessionInfo]

Lists all PTY sessions in the Sandbox.

Retrieves information about all PTY sessions in this Sandbox.

Arguments:

Returns:

Example:

# List all PTY sessions
sessions = await 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}")

AsyncProcess.get_pty_session_info

@intercept_errors(message_prefix="Failed to get PTY session info: ")
@with_instrumentation()
async def get_pty_session_info(
        session_id: str,
        request_timeout: float | None = None) -> PtySessionInfo

Gets detailed information about a specific PTY session.

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

Arguments:

Returns:

Raises:

Example:

# Get details about a specific PTY session
session_info = await 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}")

AsyncProcess.kill_pty_session

@intercept_errors(message_prefix="Failed to kill PTY session: ")
@with_instrumentation()
async def kill_pty_session(session_id: str,
                           request_timeout: float | None = None) -> 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:

Raises:

Example:

# Kill a specific PTY session
await sandbox.process.kill_pty_session("my-session")
# Verify the session no longer exists
pty_sessions = await sandbox.process.list_pty_sessions()
for pty_session in pty_sessions:
    print(f"PTY session: {pty_session.id}")

AsyncProcess.resize_pty_session

@intercept_errors(message_prefix="Failed to resize PTY session: ")
@with_instrumentation()
async def resize_pty_session(
        session_id: str,
        pty_size: PtySize,
        request_timeout: float | None = None) -> 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:

Returns:

Raises:

Example:

from daytona.common.pty import PtySize
# Resize a PTY session to a larger terminal
new_size = PtySize(rows=40, cols=150)
updated_info = await 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 AsyncPtyHandle's resize method
await pty_handle.resize(new_size)

CodeRunParams

@dataclass
class CodeRunParams()

Parameters for code execution.

Attributes:

SessionExecuteRequest

class SessionExecuteRequest(ApiSessionExecuteRequest,
                            AsyncApiSessionExecuteRequest)

Contains the request for executing a command in a session.

Attributes:

ExecutionArtifacts

@dataclass
class ExecutionArtifacts()

Artifacts from the command execution.

Attributes:

ExecuteResponse

class ExecuteResponse(BaseModel)

Response from the command execution.

Attributes:

SessionExecuteResponse

class SessionExecuteResponse(ApiSessionExecuteResponse)

Response from the session command execution.

Attributes:

SessionCommandLogsResponse

@dataclass
class SessionCommandLogsResponse()

Response from the command logs.

Attributes:

OutputHandler
OutputHandler = Callable[[T], None] | Callable[[T], Awaitable[None]]

Callback type that accepts both sync and async handlers.

Blocking synchronous operations inside handlers may cause WebSocket disconnections.