Daytona MCP Python Interpreter

Verified
Apache 2.0
Overview InspectNew Schema Related Servers Reviews Score
daytona-mcp-interpreter
└── docs
    └── python-sdk
        ├── daytona.mdx
        ├── errors.mdx
        ├── file-system.mdx
        ├── git.mdx
        ├── lsp-server.mdx
        ├── process.mdx
        └── sandbox.mdx


/docs/python-sdk/daytona.mdx:
--------------------------------------------------------------------------------
| ---
| title: Sandbox Management
| ---
|
| Sandboxes are isolated development environments managed by Daytona.
| This guide covers how to create, manage, and remove Sandboxes using the SDK.
|
| **Examples**:
|
|   Basic usage with environment variables:
| ```python
| from daytona_sdk import Daytona
| # Initialize using environment variables
| daytona = Daytona()  # Uses env vars DAYTONA_API_KEY, DAYTONA_SERVER_URL, DAYTONA_TARGET
|
| # Create a default Python workspace with custom environment variables
| workspace = daytona.create(CreateWorkspaceParams(
|     language="python",
|     env_vars={"PYTHON_ENV": "development"}
| ))
|
| # Execute commands in the workspace
| response = workspace.process.execute_command('echo "Hello, World!"')
| print(response.result)
|
| # Run Python code securely inside the workspace
| response = workspace.process.code_run('print("Hello from Python!")')
| print(response.result)
|
| # Remove the workspace after use
| daytona.remove(workspace)
| ```
|
|   Usage with explicit configuration:
| ```python
| from daytona_sdk import Daytona, DaytonaConfig, CreateWorkspaceParams, WorkspaceResources
|
| # Initialize with explicit configuration
| config = DaytonaConfig(
|     api_key="your-api-key",
|     server_url="https://your-server.com",
|     target="us"
| )
| daytona = Daytona(config)
|
| # Create a custom workspace with specific resources and settings
| workspace = daytona.create(CreateWorkspaceParams(
|     language="python",
|     image="python:3.11",
|     resources=WorkspaceResources(
|         cpu=2,
|         memory=4,  # 4GB RAM
|         disk=20    # 20GB disk
|     ),
|     env_vars={"PYTHON_ENV": "development"},
|     auto_stop_interval=60  # Auto-stop after 1 hour of inactivity
| ))
|
| # Use workspace features
| workspace.git.clone("https://github.com/user/repo.git")
| workspace.process.execute_command("python -m pytest")
| ```
|
| <a id="daytona_sdk.daytona.Daytona"></a>
| ## Daytona
|
| ```python
| class Daytona()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L198)
|
| Main class for interacting with Daytona Server API.
|
| This class provides methods to create, manage, and interact with Daytona Sandboxes.
| It can be initialized either with explicit configuration or using environment variables.
|
| **Attributes**:
|
| - `api_key` _str_ - API key for authentication.
| - `server_url` _str_ - URL of the Daytona server.
| - `target` _str_ - Default target location for Sandboxes.
|
|
| **Example**:
|
|   Using environment variables:
| ```python
| daytona = Daytona()  # Uses DAYTONA_API_KEY, DAYTONA_SERVER_URL
| ```
|
|   Using explicit configuration:
| ```python
| config = DaytonaConfig(
|     api_key="your-api-key",
|     server_url="https://your-server.com",
|     target="us"
| )
| daytona = Daytona(config)
| ```
|
|
| #### Daytona.\_\_init\_\_
|
| ```python
| def __init__(config: Optional[DaytonaConfig] = None)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L226)
|
| Initializes Daytona instance with optional configuration.
|
| If no config is provided, reads from environment variables:
| - `DAYTONA_API_KEY`: Required API key for authentication
| - `DAYTONA_SERVER_URL`: Required server URL
| - `DAYTONA_TARGET`: Optional target environment (defaults to WorkspaceTargetRegion.US)
|
| **Arguments**:
|
| - `config` _Optional[DaytonaConfig]_ - Object containing api_key, server_url, and target.
|
|
| **Raises**:
|
| - `DaytonaError` - If API key or Server URL is not provided either through config or environment variables
|
|
| **Example**:
|
| ```python
| from daytona_sdk import Daytona, DaytonaConfig
| # Using environment variables
| daytona1 = Daytona()
| # Using explicit configuration
| config = DaytonaConfig(
|     api_key="your-api-key",
|     server_url="https://your-server.com",
|     target="us"
| )
| daytona2 = Daytona(config)
| ```
|
|
| #### Daytona.create
|
| ```python
| @intercept_errors(message_prefix="Failed to create workspace: ")
| def create(params: Optional[CreateWorkspaceParams] = None,
|            timeout: Optional[float] = 60) -> Workspace
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L285)
|
| Creates Sandboxes with default or custom configurations. You can specify various parameters,
| including language, image, resources, environment variables, and volumes for the Sandbox.
|
| **Arguments**:
|
| - `params` _Optional[CreateWorkspaceParams]_ - Parameters for Sandbox creation. If not provided,
|   defaults to Python language.
| - `timeout` _Optional[float]_ - Timeout (in seconds) for workspace creation. 0 means no timeout. Default is 60 seconds.
|
|
| **Returns**:
|
| - `Workspace` - The created Sandbox instance.
|
|
| **Raises**:
|
| - `DaytonaError` - If timeout or auto_stop_interval is negative; If workspace fails to start or times out
|
|
| **Example**:
|
|   Create a default Python Sandbox:
| ```python
| workspace = daytona.create()
| ```
|
|   Create a custom Sandbox:
| ```python
| params = CreateWorkspaceParams(
|     language="python",
|     name="my-workspace",
|     image="debian:12.9",
|     env_vars={"DEBUG": "true"},
|     resources=WorkspaceResources(cpu=2, memory=4096),
|     auto_stop_interval=0
| )
| workspace = daytona.create(params, 40)
| ```
|
|
| #### Daytona.remove
|
| ```python
| @intercept_errors(message_prefix="Failed to remove workspace: ")
| def remove(workspace: Workspace, timeout: Optional[float] = 60) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L432)
|
| Removes a Sandbox.
|
| **Arguments**:
|
| - `workspace` _Workspace_ - The Sandbox instance to remove.
| - `timeout` _Optional[float]_ - Timeout (in seconds) for workspace removal. 0 means no timeout. Default is 60 seconds.
|
|
| **Raises**:
|
| - `DaytonaError` - If workspace fails to remove or times out
|
|
| **Example**:
|
| ```python
| workspace = daytona.create()
| # ... use workspace ...
| daytona.remove(workspace)  # Clean up when done
| ```
|
|
| #### Daytona.get\_current\_workspace
|
| ```python
| @intercept_errors(message_prefix="Failed to get workspace: ")
| def get_current_workspace(workspace_id: str) -> Workspace
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L452)
|
| Get a Sandbox by its ID.
|
| **Arguments**:
|
| - `workspace_id` _str_ - The ID of the Sandbox to retrieve.
|
|
| **Returns**:
|
| - `Workspace` - The Sandbox instance.
|
|
| **Raises**:
|
| - `DaytonaError` - If workspace_id is not provided.
|
|
| **Example**:
|
| ```python
| workspace = daytona.get_current_workspace("my-workspace-id")
| print(workspace.status)
| ```
|
|
| #### Daytona.list
|
| ```python
| @intercept_errors(message_prefix="Failed to list workspaces: ")
| def list() -> List[Workspace]
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L490)
|
| Lists all Sandboxes.
|
| **Returns**:
|
| - `List[Workspace]` - List of all available Sandbox instances.
|
|
| **Example**:
|
| ```python
| workspaces = daytona.list()
| for workspace in workspaces:
|     print(f"{workspace.id}: {workspace.status}")
| ```
|
|
| #### Daytona.start
|
| ```python
| def start(workspace: Workspace, timeout: Optional[float] = 60) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L555)
|
| Starts a Sandbox and waits for it to be ready.
|
| **Arguments**:
|
| - `workspace` _Workspace_ - The Sandbox to start.
| - `timeout` _Optional[float]_ - Optional timeout in seconds to wait for the Sandbox to start. 0 means no timeout. Default is 60 seconds.
|
|
| **Raises**:
|
| - `DaytonaError` - If timeout is negative; If Sandbox fails to start or times out
|
|
| #### Daytona.stop
|
| ```python
| def stop(workspace: Workspace, timeout: Optional[float] = 60) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L567)
|
| Stops a Sandbox and waits for it to be stopped.
|
| **Arguments**:
|
| - `workspace` _Workspace_ - The workspace to stop
| - `timeout` _Optional[float]_ - Optional timeout (in seconds) for workspace stop. 0 means no timeout. Default is 60 seconds.
|
|
| **Raises**:
|
| - `DaytonaError` - If timeout is negative; If Sandbox fails to stop or times out
|
|
| Sandboxes are isolated development environments managed by Daytona.
| This guide covers how to create, manage, and remove Sandboxes using the SDK.
|
| **Examples**:
|
|   Basic usage with environment variables:
| ```python
| from daytona_sdk import Daytona
| # Initialize using environment variables
| daytona = Daytona()  # Uses env vars DAYTONA_API_KEY, DAYTONA_SERVER_URL, DAYTONA_TARGET
|
| # Create a default Python workspace with custom environment variables
| workspace = daytona.create(CreateWorkspaceParams(
|     language="python",
|     env_vars={"PYTHON_ENV": "development"}
| ))
|
| # Execute commands in the workspace
| response = workspace.process.execute_command('echo "Hello, World!"')
| print(response.result)
|
| # Run Python code securely inside the workspace
| response = workspace.process.code_run('print("Hello from Python!")')
| print(response.result)
|
| # Remove the workspace after use
| daytona.remove(workspace)
| ```
|
|   Usage with explicit configuration:
| ```python
| from daytona_sdk import Daytona, DaytonaConfig, CreateWorkspaceParams, WorkspaceResources
|
| # Initialize with explicit configuration
| config = DaytonaConfig(
|     api_key="your-api-key",
|     server_url="https://your-server.com",
|     target="us"
| )
| daytona = Daytona(config)
|
| # Create a custom workspace with specific resources and settings
| workspace = daytona.create(CreateWorkspaceParams(
|     language="python",
|     image="python:3.11",
|     resources=WorkspaceResources(
|         cpu=2,
|         memory=4,  # 4GB RAM
|         disk=20    # 20GB disk
|     ),
|     env_vars={"PYTHON_ENV": "development"},
|     auto_stop_interval=60  # Auto-stop after 1 hour of inactivity
| ))
|
| # Use workspace features
| workspace.git.clone("https://github.com/user/repo.git")
| workspace.process.execute_command("python -m pytest")
| ```
|
|
| <a id="daytona_sdk.daytona.CodeLanguage"></a>
| ## CodeLanguage
|
| ```python
| @dataclass
| class CodeLanguage(Enum)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L85)
|
| Programming languages supported by Daytona
|
|
| <a id="daytona_sdk.daytona.DaytonaConfig"></a>
| ## DaytonaConfig
|
| ```python
| @dataclass
| class DaytonaConfig()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L101)
|
| Configuration options for initializing the Daytona client.
|
| **Attributes**:
|
| - `api_key` _str_ - API key for authentication with Daytona server.
| - `server_url` _str_ - URL of the Daytona server.
| - `target` _str_ - Target environment for Sandbox.
|
|
| **Example**:
|
| ```python
| config = DaytonaConfig(
|     api_key="your-api-key",
|     server_url="https://your-server.com",
|     target="us"
| )
| daytona = Daytona(config)
| ```
|
|
| <a id="daytona_sdk.daytona.WorkspaceResources"></a>
| ## WorkspaceResources
|
| ```python
| @dataclass
| class WorkspaceResources()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L125)
|
| Resources configuration for Sandbox.
|
| **Attributes**:
|
| - `cpu` _Optional[int]_ - Number of CPU cores to allocate.
| - `memory` _Optional[int]_ - Amount of memory in GB to allocate.
| - `disk` _Optional[int]_ - Amount of disk space in GB to allocate.
| - `gpu` _Optional[int]_ - Number of GPUs to allocate.
|
|
| **Example**:
|
| ```python
| resources = WorkspaceResources(
|     cpu=2,
|     memory=4,  # 4GB RAM
|     disk=20,   # 20GB disk
|     gpu=1
| )
| params = CreateWorkspaceParams(
|     language="python",
|     resources=resources
| )
| ```
|
|
| <a id="daytona_sdk.daytona.CreateWorkspaceParams"></a>
| ## CreateWorkspaceParams
|
| ```python
| class CreateWorkspaceParams(BaseModel)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/daytona.py#L154)
|
| Parameters for creating a new Sandbox.
|
| **Attributes**:
|
| - `language` _CodeLanguage_ - Programming language for the Sandbox ("python", "javascript", "typescript").
| - `id` _Optional[str]_ - Custom identifier for the Sandbox. If not provided, a random ID will be generated.
| - `name` _Optional[str]_ - Display name for the Sandbox. Defaults to Sandbox ID if not provided.
| - `image` _Optional[str]_ - Custom Docker image to use for the Sandbox.
| - `os_user` _Optional[str]_ - OS user for the Sandbox. Defaults to "daytona".
| - `env_vars` _Optional[Dict[str, str]]_ - Environment variables to set in the Sandbox.
| - `labels` _Optional[Dict[str, str]]_ - Custom labels for the Sandbox.
| - `public` _Optional[bool]_ - Whether the Sandbox should be public.
| - `target` _Optional[str]_ - Target location for the Sandbox. Can be "us", "eu", or "asia".
| - `resources` _Optional[WorkspaceResources]_ - Resource configuration for the Sandbox.
| - `timeout` _Optional[float]_ - Timeout in seconds for Sandbox to be created and started.
| - `auto_stop_interval` _Optional[int]_ - Interval in minutes after which Sandbox will automatically stop if no Sandbox event occurs during that time. Default is 15 minutes. 0 means no auto-stop.
|
|
| **Example**:
|
| ```python
| params = CreateWorkspaceParams(
|     language="python",
|     name="my-workspace",
|     env_vars={"DEBUG": "true"},
|     resources=WorkspaceResources(cpu=2, memory=4),
|     auto_stop_interval=20
| )
| workspace = daytona.create(params, 50)
| ```
|
|
|


--------------------------------------------------------------------------------
/docs/python-sdk/errors.mdx:
--------------------------------------------------------------------------------
| ---
| title: Errors
| ---
|
| <a id="daytona_sdk.common.errors.DaytonaError"></a>
| ## DaytonaError
|
| ```python
| class DaytonaError(Exception)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/common/errors.py#L1)
|
| Base error for Daytona SDK.
|
|
|
|


--------------------------------------------------------------------------------
/docs/python-sdk/file-system.mdx:
--------------------------------------------------------------------------------
| ---
| title: File System Operations
| ---
|
| The Daytona SDK provides comprehensive file system operations through the `fs` module in Sandboxes.
| You can perform various operations like listing files, creating directories, reading and writing files, and more.
| This guide covers all available file system operations and best practices.
|
| **Examples**:
|
|   Basic file operations:
| ```python
|     workspace = daytona.create()
|
|     # Create a directory
|     workspace.fs.create_folder("/workspace/data", "755")
|
|     # Upload a file
|     with open("local_file.txt", "rb") as f:
|         content = f.read()
|     workspace.fs.upload_file("/workspace/data/file.txt", content)
|
|     # List directory contents
|     files = workspace.fs.list_files("/workspace")
|     for file in files:
|         print(f"Name: {file.name}")
|         print(f"Is directory: {file.is_dir}")
|         print(f"Size: {file.size}")
|         print(f"Modified: {file.mod_time}")
|
|     # Search file contents
|     matches = workspace.fs.find_files(
|         path="/workspace/src",
|         pattern="text-of-interest"
|     )
|     for match in matches:
|         print(f"Absolute file path: {match.file}")
|         print(f"Line number: {match.line}")
|         print(f"Line content: {match.content}")
|         print("
| ")
| ```
|
|   File manipulation:
| ```python
| # Move files
| workspace.fs.move_files(
|     "/workspace/data/old.txt",
|     "/workspace/data/new.txt"
| )
|
| # Replace text in files
| results = workspace.fs.replace_in_files(
|     files=["/workspace/data/new.txt"],
|     pattern="old_version",
|     new_value="new_version"
| )
|
| # Set permissions
| workspace.fs.set_file_permissions(
|     path="/workspace/data/script.sh",
|     mode="755",
|     owner="daytona"
| )
| ```
|
|
| **Notes**:
|
|   All paths should be absolute paths within the Sandbox if not explicitly
|   stated otherwise.
|
| <a id="daytona_sdk.filesystem.FileSystem"></a>
| ## FileSystem
|
| ```python
| class FileSystem()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L80)
|
| Provides file system operations within a Sandbox.
|
| This class implements a high-level interface to file system operations that can
| be performed within a Daytona Sandbox. It supports common operations like
| creating, deleting, and moving files, as well as searching file contents and
| managing permissions.
|
| **Attributes**:
|
| - `instance` _WorkspaceInstance_ - The Sandbox instance this file system belongs to.
|
|
| #### FileSystem.\_\_init\_\_
|
| ```python
| def __init__(instance: WorkspaceInstance, toolbox_api: ToolboxApi)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L92)
|
| Initializes a new FileSystem instance.
|
| **Arguments**:
|
| - `instance` _WorkspaceInstance_ - The Sandbox instance this file system belongs to.
| - `toolbox_api` _ToolboxApi_ - API client for Sandbox operations.
|
|
| #### FileSystem.create\_folder
|
| ```python
| @intercept_errors(message_prefix="Failed to create folder: ")
| def create_folder(path: str, mode: str) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L103)
|
| Creates a new directory in the Sandbox.
|
| This method creates a new directory at the specified path with the given
| permissions.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path where the folder should be created.
| - `mode` _str_ - Folder permissions in octal format (e.g., "755" for rwxr-xr-x).
|
|
| **Example**:
|
| ```python
| # Create a directory with standard permissions
| workspace.fs.create_folder("/workspace/data", "755")
|
| # Create a private directory
| workspace.fs.create_folder("/workspace/secrets", "700")
| ```
|
|
| #### FileSystem.delete\_file
|
| ```python
| @intercept_errors(message_prefix="Failed to delete file: ")
| def delete_file(path: str) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L127)
|
| Deletes a file from the Sandbox.
|
| This method permanently deletes a file from the Sandbox.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the file to delete.
|
|
| **Example**:
|
| ```python
| # Delete a file
| workspace.fs.delete_file("/workspace/data/old_file.txt")
| ```
|
|
| #### FileSystem.download\_file
|
| ```python
| @intercept_errors(message_prefix="Failed to download file: ")
| def download_file(path: str) -> bytes
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L146)
|
| Downloads a file from the Sandbox.
|
| This method retrieves the contents of a file from the Sandbox.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the file to download.
|
|
| **Returns**:
|
| - `bytes` - The file contents as a bytes object.
|
|
| **Example**:
|
| ```python
| # Download and save a file locally
| content = workspace.fs.download_file("/workspace/data/file.txt")
| with open("local_copy.txt", "wb") as f:
|     f.write(content)
|
| # Download and process text content
| content = workspace.fs.download_file("/workspace/data/config.json")
| config = json.loads(content.decode('utf-8'))
| ```
|
|
| #### FileSystem.find\_files
|
| ```python
| @intercept_errors(message_prefix="Failed to find files: ")
| def find_files(path: str, pattern: str) -> List[Match]
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L174)
|
| Searches for files containing a pattern.
|
| This method searches file contents for a specified pattern, similar to
| the grep command.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the file or directory to search. If the path is a directory, the search will be performed recursively.
| - `pattern` _str_ - Search pattern to match against file contents.
|
|
| **Returns**:
|
| - `List[Match]` - List of matches found in files. Each Match object includes:
|   - file: Path to the file containing the match
|   - line: The line number where the match was found
|   - content: The matching line content
|
|
| **Example**:
|
| ```python
| # Search for TODOs in Python files
| matches = workspace.fs.find_files("/workspace/src", "TODO:")
| for match in matches:
|     print(f"{match.file}:{match.line}: {match.content.strip()}")
| ```
|
|
| #### FileSystem.get\_file\_info
|
| ```python
| @intercept_errors(message_prefix="Failed to get file info: ")
| def get_file_info(path: str) -> FileInfo
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L203)
|
| Gets detailed information about a file.
|
| This method retrieves metadata about a file or directory, including its
| size, permissions, and timestamps.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the file or directory.
|
|
| **Returns**:
|
| - `FileInfo` - Detailed file information including:
|   - name: File name
|   - is_dir: Whether the path is a directory
|   - size: File size in bytes
|   - mode: File permissions
|   - mod_time: Last modification timestamp
|   - permissions: File permissions in octal format
|   - owner: File owner
|   - group: File group
|
|
| **Example**:
|
| ```python
| # Get file metadata
| info = workspace.fs.get_file_info("/workspace/data/file.txt")
| print(f"Size: {info.size} bytes")
| print(f"Modified: {info.mod_time}")
| print(f"Mode: {info.mode}")
|
| # Check if path is a directory
| info = workspace.fs.get_file_info("/workspace/data")
| if info.is_dir:
|     print("Path is a directory")
| ```
|
|
| #### FileSystem.list\_files
|
| ```python
| @intercept_errors(message_prefix="Failed to list files: ")
| def list_files(path: str) -> List[FileInfo]
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L242)
|
| Lists files and directories in a given path.
|
| This method returns information about all files and directories in the
| specified directory, similar to the ls -l command.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the directory to list contents from.
|
|
| **Returns**:
|
| - `List[FileInfo]` - List of file and directory information. Each FileInfo
|   object includes the same fields as described in get_file_info().
|
|
| **Example**:
|
| ```python
| # List directory contents
| files = workspace.fs.list_files("/workspace/data")
|
| # Print files and their sizes
| for file in files:
|     if not file.is_dir:
|         print(f"{file.name}: {file.size} bytes")
|
| # List only directories
| dirs = [f for f in files if f.is_dir]
| print("Subdirectories:", ", ".join(d.name for d in dirs))
| ```
|
|
| #### FileSystem.move\_files
|
| ```python
| @intercept_errors(message_prefix="Failed to move files: ")
| def move_files(source: str, destination: str) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L275)
|
| Moves files from one location to another.
|
| This method moves or renames a file or directory. The parent directory
| of the destination must exist.
|
| **Arguments**:
|
| - `source` _str_ - Absolute path to the source file or directory.
| - `destination` _str_ - Absolute path to the destination.
|
|
| **Example**:
|
| ```python
| # Rename a file
| workspace.fs.move_files(
|     "/workspace/data/old_name.txt",
|     "/workspace/data/new_name.txt"
| )
|
| # Move a file to a different directory
| workspace.fs.move_files(
|     "/workspace/data/file.txt",
|     "/workspace/archive/file.txt"
| )
|
| # Move a directory
| workspace.fs.move_files(
|     "/workspace/old_dir",
|     "/workspace/new_dir"
| )
| ```
|
|
| #### FileSystem.replace\_in\_files
|
| ```python
| @intercept_errors(message_prefix="Failed to replace in files: ")
| def replace_in_files(files: List[str], pattern: str,
|                      new_value: str) -> List[ReplaceResult]
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L313)
|
| Replaces text in multiple files.
|
| This method performs search and replace operations across multiple files.
|
| **Arguments**:
|
| - `files` _List[str]_ - List of absolute file paths to perform replacements in.
| - `pattern` _str_ - Pattern to search for.
| - `new_value` _str_ - Text to replace matches with.
|
|
| **Returns**:
|
| - `List[ReplaceResult]` - List of results indicating replacements made in
|   each file. Each ReplaceResult includes:
|   - file: Path to the modified file
|   - success: Whether the operation was successful
|   - error: Error message if the operation failed
|
|
| **Example**:
|
| ```python
| # Replace in specific files
| results = workspace.fs.replace_in_files(
|     files=["/workspace/src/file1.py", "/workspace/src/file2.py"],
|     pattern="old_function",
|     new_value="new_function"
| )
|
| # Print results
| for result in results:
|     if result.success:
|         print(f"{result.file}: {result.success}")
|     else:
|         print(f"{result.file}: {result.error}")
| ```
|
|
| #### FileSystem.search\_files
|
| ```python
| @intercept_errors(message_prefix="Failed to search files: ")
| def search_files(path: str, pattern: str) -> SearchFilesResponse
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L358)
|
| Searches for files and directories matching a pattern in their names.
|
| This method searches for files and directories whose names match the
| specified pattern. The pattern can be a simple string or a glob pattern.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the root directory to start search from.
| - `pattern` _str_ - Pattern to match against file names. Supports glob
|   patterns (e.g., "*.py" for Python files).
|
|
| **Returns**:
|
| - `SearchFilesResponse` - Search results containing:
|   - files: List of matching file and directory paths
|
|
| **Example**:
|
| ```python
| # Find all Python files
| result = workspace.fs.search_files("/workspace", "*.py")
| for file in result.files:
|     print(file)
|
| # Find files with specific prefix
| result = workspace.fs.search_files("/workspace/data", "test_*")
| print(f"Found {len(result.files)} test files")
| ```
|
|
| #### FileSystem.set\_file\_permissions
|
| ```python
| @intercept_errors(message_prefix="Failed to set file permissions: ")
| def set_file_permissions(path: str,
|                          mode: str = None,
|                          owner: str = None,
|                          group: str = None) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L390)
|
| Sets permissions and ownership for a file or directory.
|
| This method allows changing the permissions and ownership of a file or
| directory. Any of the parameters can be None to leave that attribute
| unchanged.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the file or directory.
| - `mode` _Optional[str]_ - File mode/permissions in octal format
|   (e.g., "644" for rw-r--r--).
| - `owner` _Optional[str]_ - User owner of the file.
| - `group` _Optional[str]_ - Group owner of the file.
|
|
| **Example**:
|
| ```python
| # Make a file executable
| workspace.fs.set_file_permissions(
|     path="/workspace/scripts/run.sh",
|     mode="755"  # rwxr-xr-x
| )
|
| # Change file owner
| workspace.fs.set_file_permissions(
|     path="/workspace/data/file.txt",
|     owner="daytona",
|     group="daytona"
| )
| ```
|
|
| #### FileSystem.upload\_file
|
| ```python
| @intercept_errors(message_prefix="Failed to upload file: ")
| def upload_file(path: str, file: bytes) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/filesystem.py#L431)
|
| Uploads a file to the Sandbox.
|
| This method uploads a file to the specified path in the Sandbox. The
| parent directory must exist. If a file already exists at the destination
| path, it will be overwritten.
|
| **Arguments**:
|
| - `path` _str_ - Absolute destination path in the Sandbox.
| - `file` _bytes_ - File contents as a bytes object.
|
|
| **Example**:
|
| ```python
| # Upload a text file
| content = b"Hello, World!"
| workspace.fs.upload_file("/workspace/data/hello.txt", content)
|
| # Upload a local file
| with open("local_file.txt", "rb") as f:
|     content = f.read()
| workspace.fs.upload_file("/workspace/data/file.txt", content)
|
| # Upload binary data
| import json
| data = {"key": "value"}
| content = json.dumps(data).encode('utf-8')
| workspace.fs.upload_file("/workspace/data/config.json", content)
| ```
|
|
| The Daytona SDK provides comprehensive file system operations through the `fs` module in Sandboxes.
| You can perform various operations like listing files, creating directories, reading and writing files, and more.
| This guide covers all available file system operations and best practices.
|
| **Examples**:
|
|   Basic file operations:
| ```python
|     workspace = daytona.create()
|
|     # Create a directory
|     workspace.fs.create_folder("/workspace/data", "755")
|
|     # Upload a file
|     with open("local_file.txt", "rb") as f:
|         content = f.read()
|     workspace.fs.upload_file("/workspace/data/file.txt", content)
|
|     # List directory contents
|     files = workspace.fs.list_files("/workspace")
|     for file in files:
|         print(f"Name: {file.name}")
|         print(f"Is directory: {file.is_dir}")
|         print(f"Size: {file.size}")
|         print(f"Modified: {file.mod_time}")
|
|     # Search file contents
|     matches = workspace.fs.find_files(
|         path="/workspace/src",
|         pattern="text-of-interest"
|     )
|     for match in matches:
|         print(f"Absolute file path: {match.file}")
|         print(f"Line number: {match.line}")
|         print(f"Line content: {match.content}")
|         print("
| ")
| ```
|
|   File manipulation:
| ```python
| # Move files
| workspace.fs.move_files(
|     "/workspace/data/old.txt",
|     "/workspace/data/new.txt"
| )
|
| # Replace text in files
| results = workspace.fs.replace_in_files(
|     files=["/workspace/data/new.txt"],
|     pattern="old_version",
|     new_value="new_version"
| )
|
| # Set permissions
| workspace.fs.set_file_permissions(
|     path="/workspace/data/script.sh",
|     mode="755",
|     owner="daytona"
| )
| ```
|
|
| **Notes**:
|
|   All paths should be absolute paths within the Sandbox if not explicitly
|   stated otherwise.
|
|
|


--------------------------------------------------------------------------------
/docs/python-sdk/git.mdx:
--------------------------------------------------------------------------------
| ---
| title: Git Operations
| ---
|
| The Daytona SDK provides built-in Git support. This guide covers all available Git
| operations and best practices. Daytona SDK provides an option to clone, check status,
| and manage Git repositories in Sandboxes. You can interact with Git repositories using
| the `git` module.
|
| **Example**:
|
|   Basic Git workflow:
| ```python
| workspace = daytona.create()
|
| # Clone a repository
| workspace.git.clone(
|     url="https://github.com/user/repo.git",
|     path="/workspace/repo"
| )
|
| # Make some changes
| workspace.fs.upload_file("/workspace/repo/test.txt", "Hello, World!")
|
| # Stage and commit changes
| workspace.git.add("/workspace/repo", ["test.txt"])
| workspace.git.commit(
|     path="/workspace/repo",
|     message="Add test file",
|     author="John Doe",
|     email="john@example.com"
| )
|
| # Push changes (with authentication)
| workspace.git.push(
|     path="/workspace/repo",
|     username="user",
|     password="token"
| )
| ```
|
|
| **Notes**:
|
|   All paths should be absolute paths within the Sandbox if not explicitly
|   stated otherwise.
|
| <a id="daytona_sdk.git.Git"></a>
| ## Git
|
| ```python
| class Git()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L60)
|
| Provides Git operations within a Sandbox.
|
| This class implements a high-level interface to Git operations that can be
| performed within a Daytona Sandbox. It supports common Git operations like
| cloning repositories, staging and committing changes, pushing and pulling
| changes, and checking repository status.
|
| **Attributes**:
|
| - `workspace` _Workspace_ - The parent Sandbox instance.
| - `instance` _WorkspaceInstance_ - The Sandbox instance this Git handler belongs to.
|
|
| **Example**:
|
| ```python
| # Clone a repository
| workspace.git.clone(
|     url="https://github.com/user/repo.git",
|     path="/workspace/repo"
| )
|
| # Check repository status
| status = workspace.git.status("/workspace/repo")
| print(f"Modified files: {status.modified}")
|
| # Stage and commit changes
| workspace.git.add("/workspace/repo", ["file.txt"])
| workspace.git.commit(
|     path="/workspace/repo",
|     message="Update file",
|     author="John Doe",
|     email="john@example.com"
| )
| ```
|
|
| #### Git.\_\_init\_\_
|
| ```python
| def __init__(workspace: "Workspace", toolbox_api: ToolboxApi,
|              instance: WorkspaceInstance)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L95)
|
| Initializes a new Git handler instance.
|
| **Arguments**:
|
| - `workspace` _Workspace_ - The parent Sandbox instance.
| - `toolbox_api` _ToolboxApi_ - API client for Sandbox operations.
| - `instance` _WorkspaceInstance_ - The Sandbox instance this Git handler belongs to.
|
|
| #### Git.add
|
| ```python
| @intercept_errors(message_prefix="Failed to add files: ")
| def add(path: str, files: List[str]) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L113)
|
| Stages files for commit.
|
| This method stages the specified files for the next commit, similar to
| running 'git add' on the command line.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the Git repository root.
| - `files` _List[str]_ - List of file paths or directories to stage, relative to the repository root.
|
|
| **Example**:
|
| ```python
| # Stage a single file
| workspace.git.add("/workspace/repo", ["file.txt"])
|
| # Stage multiple files
| workspace.git.add("/workspace/repo", [
|     "src/main.py",
|     "tests/test_main.py",
|     "README.md"
| ])
| ```
|
|
| #### Git.branches
|
| ```python
| @intercept_errors(message_prefix="Failed to list branches: ")
| def branches(path: str) -> ListBranchResponse
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L145)
|
| Lists branches in the repository.
|
| This method returns information about all branches in the repository.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the Git repository root.
|
|
| **Returns**:
|
| - `ListBranchResponse` - List of branches in the repository.
|
|
| **Example**:
|
| ```python
| response = workspace.git.branches("/workspace/repo")
| print(f"Branches: {response.branches}")
| ```
|
|
| #### Git.clone
|
| ```python
| @intercept_errors(message_prefix="Failed to clone repository: ")
| def clone(url: str,
|           path: str,
|           branch: Optional[str] = None,
|           commit_id: Optional[str] = None,
|           username: Optional[str] = None,
|           password: Optional[str] = None) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L168)
|
| Clones a Git repository.
|
| This method clones a Git repository into the specified path. It supports
| cloning specific branches or commits, and can authenticate with the remote
| repository if credentials are provided.
|
| **Arguments**:
|
| - `url` _str_ - Repository URL to clone from.
| - `path` _str_ - Absolute path where the repository should be cloned.
| - `branch` _Optional[str]_ - Specific branch to clone. If not specified,
|   clones the default branch.
| - `commit_id` _Optional[str]_ - Specific commit to clone. If specified,
|   the repository will be left in a detached HEAD state at this commit.
| - `username` _Optional[str]_ - Git username for authentication.
| - `password` _Optional[str]_ - Git password or token for authentication.
|
|
| **Example**:
|
| ```python
| # Clone the default branch
| workspace.git.clone(
|     url="https://github.com/user/repo.git",
|     path="/workspace/repo"
| )
|
| # Clone a specific branch with authentication
| workspace.git.clone(
|     url="https://github.com/user/private-repo.git",
|     path="/workspace/private",
|     branch="develop",
|     username="user",
|     password="token"
| )
|
| # Clone a specific commit
| workspace.git.clone(
|     url="https://github.com/user/repo.git",
|     path="/workspace/repo-old",
|     commit_id="abc123"
| )
| ```
|
|
| #### Git.commit
|
| ```python
| @intercept_errors(message_prefix="Failed to commit changes: ")
| def commit(path: str, message: str, author: str, email: str) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L231)
|
| Commits staged changes.
|
| This method creates a new commit with the staged changes. Make sure to stage
| changes using the add() method before committing.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the Git repository root.
| - `message` _str_ - Commit message describing the changes.
| - `author` _str_ - Name of the commit author.
| - `email` _str_ - Email address of the commit author.
|
|
| **Example**:
|
| ```python
| # Stage and commit changes
| workspace.git.add("/workspace/repo", ["README.md"])
| workspace.git.commit(
|     path="/workspace/repo",
|     message="Update documentation",
|     author="John Doe",
|     email="john@example.com"
| )
| ```
|
|
| #### Git.push
|
| ```python
| @intercept_errors(message_prefix="Failed to push changes: ")
| def push(path: str,
|          username: Optional[str] = None,
|          password: Optional[str] = None) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L266)
|
| Pushes local commits to the remote repository.
|
| This method pushes all local commits on the current branch to the remote
| repository. If the remote repository requires authentication, provide
| username and password/token.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the Git repository root.
| - `username` _Optional[str]_ - Git username for authentication.
| - `password` _Optional[str]_ - Git password or token for authentication.
|
|
| **Example**:
|
| ```python
| # Push without authentication (for public repos or SSH)
| workspace.git.push("/workspace/repo")
|
| # Push with authentication
| workspace.git.push(
|     path="/workspace/repo",
|     username="user",
|     password="github_token"
| )
| ```
|
|
| #### Git.pull
|
| ```python
| @intercept_errors(message_prefix="Failed to pull changes: ")
| def pull(path: str,
|          username: Optional[str] = None,
|          password: Optional[str] = None) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L303)
|
| Pulls changes from the remote repository.
|
| This method fetches and merges changes from the remote repository into
| the current branch. If the remote repository requires authentication,
| provide username and password/token.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the Git repository root.
| - `username` _Optional[str]_ - Git username for authentication.
| - `password` _Optional[str]_ - Git password or token for authentication.
|
|
| **Example**:
|
| ```python
| # Pull without authentication
| workspace.git.pull("/workspace/repo")
|
| # Pull with authentication
| workspace.git.pull(
|     path="/workspace/repo",
|     username="user",
|     password="github_token"
| )
| ```
|
|
| #### Git.status
|
| ```python
| @intercept_errors(message_prefix="Failed to get status: ")
| def status(path: str) -> GitStatus
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/git.py#L340)
|
| Gets the current Git repository status.
|
| This method returns detailed information about the current state of the
| repository, including staged, unstaged, and untracked files.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the Git repository root.
|
|
| **Returns**:
|
| - `GitStatus` - Repository status information including:
|   - current_branch: Current branch name
|   - file_status: List of file statuses
|   - ahead: Number of local commits not pushed to remote
|   - behind: Number of remote commits not pulled locally
|   - branch_published: Whether the branch has been published to the remote repository
|
|
| **Example**:
|
| ```python
| status = workspace.git.status("/workspace/repo")
| print(f"On branch: {status.current_branch}")
| print(f"Commits ahead: {status.ahead}")
| print(f"Commits behind: {status.behind}")
| ```
|
|
| The Daytona SDK provides built-in Git support. This guide covers all available Git
| operations and best practices. Daytona SDK provides an option to clone, check status,
| and manage Git repositories in Sandboxes. You can interact with Git repositories using
| the `git` module.
|
| **Example**:
|
|   Basic Git workflow:
| ```python
| workspace = daytona.create()
|
| # Clone a repository
| workspace.git.clone(
|     url="https://github.com/user/repo.git",
|     path="/workspace/repo"
| )
|
| # Make some changes
| workspace.fs.upload_file("/workspace/repo/test.txt", "Hello, World!")
|
| # Stage and commit changes
| workspace.git.add("/workspace/repo", ["test.txt"])
| workspace.git.commit(
|     path="/workspace/repo",
|     message="Add test file",
|     author="John Doe",
|     email="john@example.com"
| )
|
| # Push changes (with authentication)
| workspace.git.push(
|     path="/workspace/repo",
|     username="user",
|     password="token"
| )
| ```
|
|
| **Notes**:
|
|   All paths should be absolute paths within the Sandbox if not explicitly
|   stated otherwise.
|
|
|


--------------------------------------------------------------------------------
/docs/python-sdk/lsp-server.mdx:
--------------------------------------------------------------------------------
| ---
| title: Language Server Protocol
| ---
|
| The Daytona SDK provides Language Server Protocol (LSP) support through Sandbox instances.
| This enables advanced language features like code completion, diagnostics, and more.
|
| **Example**:
|
|   Basic LSP server usage:
| ```python
| workspace = daytona.create()
|
| # Create and start LSP server
| lsp = workspace.create_lsp_server("typescript", "/workspace/project")
| lsp.start()
|
| # Open a file for editing
| lsp.did_open("/workspace/project/src/index.ts")
|
| # Get completions at a position
| pos = Position(line=10, character=15)
| completions = lsp.completions("/workspace/project/src/index.ts", pos)
| print(f"Completions: {completions}")
|
| # Get document symbols
| symbols = lsp.document_symbols("/workspace/project/src/index.ts")
| for symbol in symbols:
|     print(f"{symbol.name}: {symbol.kind}")
|
| # Clean up
| lsp.did_close("/workspace/project/src/index.ts")
| lsp.stop()
| ```
|
|
| **Notes**:
|
|   The LSP server must be started with start() before using any other methods,
|   and should be stopped with stop() when no longer needed to free resources.
|
| <a id="daytona_sdk.lsp_server.LspServer"></a>
| ## LspServer
|
| ```python
| class LspServer()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L86)
|
| Provides Language Server Protocol functionality for code intelligence.
|
| This class implements a subset of the Language Server Protocol (LSP) to provide
| IDE-like features such as code completion, symbol search, and more.
|
| **Attributes**:
|
| - `language_id` _LspLanguageId_ - The language server type (e.g., "python", "typescript").
| - `path_to_project` _str_ - Absolute path to the project root directory.
| - `instance` _WorkspaceInstance_ - The Sandbox instance this server belongs to.
|
|
| #### LspServer.\_\_init\_\_
|
| ```python
| def __init__(language_id: LspLanguageId, path_to_project: str,
|              toolbox_api: ToolboxApi, instance: WorkspaceInstance)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L98)
|
| Initializes a new LSP server instance.
|
| **Arguments**:
|
| - `language_id` _LspLanguageId_ - The language server type (e.g., LspLanguageId.TYPESCRIPT).
| - `path_to_project` _str_ - Absolute path to the project root directory.
| - `toolbox_api` _ToolboxApi_ - API client for Sandbox operations.
| - `instance` _WorkspaceInstance_ - The Sandbox instance this server belongs to.
|
|
| #### LspServer.start
|
| ```python
| @intercept_errors(message_prefix="Failed to start LSP server: ")
| def start() -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L119)
|
| Starts the language server.
|
| This method must be called before using any other LSP functionality.
| It initializes the language server for the specified language and project.
|
| **Example**:
|
| ```python
| lsp = workspace.create_lsp_server("typescript", "/workspace/project")
| lsp.start()  # Initialize the server
| # Now ready for LSP operations
| ```
|
|
| #### LspServer.stop
|
| ```python
| @intercept_errors(message_prefix="Failed to stop LSP server: ")
| def stop() -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L141)
|
| Stops the language server.
|
| This method should be called when the LSP server is no longer needed to
| free up system resources.
|
| **Example**:
|
| ```python
| # When done with LSP features
| lsp.stop()  # Clean up resources
| ```
|
|
| #### LspServer.did\_open
|
| ```python
| @intercept_errors(message_prefix="Failed to open file: ")
| def did_open(path: str) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L162)
|
| Notifies the language server that a file has been opened.
|
| This method should be called when a file is opened in the editor to enable
| language features like diagnostics and completions for that file. The server
| will begin tracking the file's contents and providing language features.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the opened file.
|
|
| **Example**:
|
| ```python
| # When opening a file for editing
| lsp.did_open("/workspace/project/src/index.ts")
| # Now can get completions, symbols, etc. for this file
| ```
|
|
| #### LspServer.did\_close
|
| ```python
| @intercept_errors(message_prefix="Failed to close file: ")
| def did_close(path: str) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L189)
|
| Notify the language server that a file has been closed.
|
| This method should be called when a file is closed in the editor to allow
| the language server to clean up any resources associated with that file.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the closed file.
|
|
| **Example**:
|
| ```python
| # When done editing a file
| lsp.did_close("/workspace/project/src/index.ts")
| ```
|
|
| #### LspServer.document\_symbols
|
| ```python
| @intercept_errors(message_prefix="Failed to get symbols from document: ")
| def document_symbols(path: str) -> List[LspSymbol]
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L213)
|
| Gets symbol information from a document.
|
| This method returns information about all symbols (functions, classes,
| variables, etc.) defined in the specified document.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the file to get symbols from.
|
|
| **Returns**:
|
| - `List[LspSymbol]` - List of symbols in the document. Each symbol includes:
|   - name: The symbol's name
|   - kind: The symbol's kind (function, class, variable, etc.)
|   - location: The location of the symbol in the file
|
|
| **Example**:
|
| ```python
| # Get all symbols in a file
| symbols = lsp.document_symbols("/workspace/project/src/index.ts")
| for symbol in symbols:
|     print(f"{symbol.kind} {symbol.name}: {symbol.location}")
| ```
|
|
| #### LspServer.workspace\_symbols
|
| ```python
| @intercept_errors(message_prefix="Failed to get symbols from workspace: ")
| def workspace_symbols(query: str) -> List[LspSymbol]
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L244)
|
| Searches for symbols across the entire Sandbox.
|
| This method searches for symbols matching the query string across all files
| in the Sandbox. It's useful for finding declarations and definitions
| without knowing which file they're in.
|
| **Arguments**:
|
| - `query` _str_ - Search query to match against symbol names.
|
|
| **Returns**:
|
| - `List[LspSymbol]` - List of matching symbols from all files. Each symbol
|   includes:
|   - name: The symbol's name
|   - kind: The symbol's kind (function, class, variable, etc.)
|   - location: The location of the symbol in the file
|
|
| **Example**:
|
| ```python
| # Search for all symbols containing "User"
| symbols = lsp.workspace_symbols("User")
| for symbol in symbols:
|     print(f"{symbol.name} in {symbol.location}")
| ```
|
|
| #### LspServer.completions
|
| ```python
| @intercept_errors(message_prefix="Failed to get completions: ")
| def completions(path: str, position: Position) -> CompletionList
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L277)
|
| Gets completion suggestions at a position in a file.
|
| **Arguments**:
|
| - `path` _str_ - Absolute path to the file.
| - `position` _Position_ - Cursor position to get completions for.
|
|
| **Returns**:
|
| - `CompletionList` - List of completion suggestions. The list includes:
|   - isIncomplete: Whether more items might be available
|   - items: List of completion items, each containing:
|   - label: The text to insert
|   - kind: The kind of completion
|   - detail: Additional details about the item
|   - documentation: Documentation for the item
|   - sortText: Text used to sort the item in the list
|   - filterText: Text used to filter the item
|   - insertText: The actual text to insert (if different from label)
|
|
| **Example**:
|
| ```python
| # Get completions at a specific position
| pos = Position(line=10, character=15)
| completions = lsp.completions("/workspace/project/src/index.ts", pos)
| for item in completions.items:
|     print(f"{item.label} ({item.kind}): {item.detail}")
| ```
|
|
| The Daytona SDK provides Language Server Protocol (LSP) support through Sandbox instances.
| This enables advanced language features like code completion, diagnostics, and more.
|
| **Example**:
|
|   Basic LSP server usage:
| ```python
| workspace = daytona.create()
|
| # Create and start LSP server
| lsp = workspace.create_lsp_server("typescript", "/workspace/project")
| lsp.start()
|
| # Open a file for editing
| lsp.did_open("/workspace/project/src/index.ts")
|
| # Get completions at a position
| pos = Position(line=10, character=15)
| completions = lsp.completions("/workspace/project/src/index.ts", pos)
| print(f"Completions: {completions}")
|
| # Get document symbols
| symbols = lsp.document_symbols("/workspace/project/src/index.ts")
| for symbol in symbols:
|     print(f"{symbol.name}: {symbol.kind}")
|
| # Clean up
| lsp.did_close("/workspace/project/src/index.ts")
| lsp.stop()
| ```
|
|
| **Notes**:
|
|   The LSP server must be started with start() before using any other methods,
|   and should be stopped with stop() when no longer needed to free resources.
|
|
| <a id="daytona_sdk.lsp_server.Position"></a>
| ## Position
|
| ```python
| class Position()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L64)
|
| Represents a position in a text document.
|
| This class represents a zero-based position within a text document,
| specified by line number and character offset.
|
| **Attributes**:
|
| - `line` _int_ - Zero-based line number in the document.
| - `character` _int_ - Zero-based character offset on the line.
|
|
| #### Position.\_\_init\_\_
|
| ```python
| def __init__(line: int, character: int)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/lsp_server.py#L75)
|
| Initialize a new Position instance.
|
| **Arguments**:
|
| - `line` _int_ - Zero-based line number in the document.
| - `character` _int_ - Zero-based character offset on the line.
|
|
|


--------------------------------------------------------------------------------
/docs/python-sdk/process.mdx:
--------------------------------------------------------------------------------
| ---
| title: Process and Code Execution
| ---
|
| <a id="daytona_sdk.process.Process"></a>
| ## Process
|
| ```python
| class Process()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L57)
|
| Handles process and code execution within a Sandbox.
|
| This class provides methods for executing shell commands and running code in
| the Sandbox environment.
|
| **Attributes**:
|
| - `code_toolbox` _WorkspacePythonCodeToolbox_ - Language-specific code execution toolbox.
| - `toolbox_api` _ToolboxApi_ - API client for Sandbox operations.
| - `instance` _WorkspaceInstance_ - The Sandbox instance this process belongs to.
|
|
| #### Process.\_\_init\_\_
|
| ```python
| def __init__(code_toolbox: WorkspacePythonCodeToolbox, toolbox_api: ToolboxApi,
|              instance: WorkspaceInstance)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L69)
|
| Initialize a new Process instance.
|
| **Arguments**:
|
| - `code_toolbox` _WorkspacePythonCodeToolbox_ - Language-specific code execution toolbox.
| - `toolbox_api` _ToolboxApi_ - API client for Sandbox operations.
| - `instance` _WorkspaceInstance_ - The Sandbox instance this process belongs to.
|
|
| #### Process.exec
|
| ```python
| @intercept_errors(message_prefix="Failed to execute command: ")
| def exec(command: str,
|          cwd: Optional[str] = None,
|          timeout: Optional[int] = None) -> ExecuteResponse
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L87)
|
| 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 root directory.
| - `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
|
|
| **Example**:
|
| ```python
| # Simple command
| response = workspace.process.exec("echo 'Hello'")
| print(response.result)  # Prints: Hello
|
| # Command with working directory
| result = workspace.process.exec("ls", cwd="/workspace/src")
|
| # Command with timeout
| result = workspace.process.exec("sleep 10", timeout=5)
| ```
|
|
| #### Process.code\_run
|
| ```python
| def code_run(code: str,
|              params: Optional[CodeRunParams] = None,
|              timeout: Optional[int] = None) -> ExecuteResponse
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L126)
|
| 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
|
|
| **Example**:
|
| ```python
| # Run Python code
| response = workspace.process.code_run('''
|     x = 10
|     y = 20
|     print(f"Sum: {x + y}")
| ''')
| print(response.result)  # Prints: Sum: 30
| ```
|
|
| #### Process.create\_session
|
| ```python
| @intercept_errors(message_prefix="Failed to create session: ")
| def create_session(session_id: str) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L155)
|
| Create 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**:
|
| ```python
| # Create a new session
| session_id = "my-session"
| workspace.process.create_session(session_id)
| session = workspace.process.get_session(session_id)
| # Do work...
| workspace.process.delete_session(session_id)
| ```
|
|
| #### Process.get\_session
|
| ```python
| @intercept_errors(message_prefix="Failed to get session: ")
| def get_session(session_id: str) -> Session
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L182)
|
| Get 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**:
|
| ```python
| session = workspace.process.get_session("my-session")
| for cmd in session.commands:
|     print(f"Command: {cmd.command}")
| ```
|
|
| #### Process.get\_session\_command
|
| ```python
| @intercept_errors(message_prefix="Failed to get session command: ")
| def get_session_command(session_id: str, command_id: str) -> Command
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L206)
|
| Get 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**:
|
| ```python
| cmd = workspace.process.get_session_command("my-session", "cmd-123")
| if cmd.exit_code == 0:
|     print(f"Command {cmd.command} completed successfully")
| ```
|
|
| #### Process.execute\_session\_command
|
| ```python
| @intercept_errors(message_prefix="Failed to execute session command: ")
| def execute_session_command(
|         session_id: str,
|         req: SessionExecuteRequest,
|         timeout: Optional[int] = None) -> SessionExecuteResponse
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L233)
|
| 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
|   - var_async: Whether to execute asynchronously
|
|
| **Returns**:
|
| - `SessionExecuteResponse` - Command execution results containing:
|   - cmd_id: Unique identifier for the executed command
|   - output: Command output (if synchronous execution)
|   - exit_code: Command exit status (if synchronous execution)
|
|
| **Example**:
|
| ```python
| # Execute commands in sequence, maintaining state
| session_id = "my-session"
|
| # Change directory
| req = SessionExecuteRequest(command="cd /workspace")
| workspace.process.execute_session_command(session_id, req)
|
| # Create a file
| req = SessionExecuteRequest(command="echo 'Hello' > test.txt")
| workspace.process.execute_session_command(session_id, req)
|
| # Read the file
| req = SessionExecuteRequest(command="cat test.txt")
| result = workspace.process.execute_session_command(session_id, req)
| print(result.output)  # Prints: Hello
| ```
|
|
| #### Process.get\_session\_command\_logs
|
| ```python
| @intercept_errors(message_prefix="Failed to get session command logs: ")
| def get_session_command_logs(session_id: str, command_id: str) -> str
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L275)
|
| Get the logs for a command executed in a session.
|
| This method retrieves the complete output (stdout and stderr) from a
| command executed in a session. It's particularly useful for checking
| the output of asynchronous commands.
|
| **Arguments**:
|
| - `session_id` _str_ - Unique identifier of the session.
| - `command_id` _str_ - Unique identifier of the command.
|
|
| **Returns**:
|
| - `str` - Complete command output including both stdout and stderr.
|
|
| **Example**:
|
| ```python
| # Execute a long-running command asynchronously
| req = SessionExecuteRequest(
|     command="sleep 5; echo 'Done'",
|     var_async=True
| )
| response = workspace.process.execute_session_command("my-session", req)
|
| # Wait a bit, then get the logs
| import time
| time.sleep(6)
| logs = workspace.process.get_session_command_logs(
|     "my-session",
|     response.command_id
| )
| print(logs)  # Prints: Done
| ```
|
|
| #### Process.list\_sessions
|
| ```python
| @intercept_errors(message_prefix="Failed to list sessions: ")
| def list_sessions() -> List[Session]
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L315)
|
| List all sessions in the Sandbox.
|
| **Returns**:
|
| - `List[Session]` - List of all sessions in the Sandbox.
|
|
| **Example**:
|
| ```python
| sessions = workspace.process.list_sessions()
| for session in sessions:
|     print(f"Session {session.session_id}:")
|     print(f"  Commands: {len(session.commands)}")
| ```
|
|
| #### Process.delete\_session
|
| ```python
| @intercept_errors(message_prefix="Failed to delete session: ")
| def delete_session(session_id: str) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/process.py#L334)
|
| Delete an interactive session from the Sandbox.
|
| This method terminates and removes a session, cleaning up any resources
| associated with it.
|
| **Arguments**:
|
| - `session_id` _str_ - Unique identifier of the session to delete.
|
|
| **Example**:
|
| ```python
| # Create and use a session
| workspace.process.create_session("temp-session")
| # ... use the session ...
|
| # Clean up when done
| workspace.process.delete_session("temp-session")
| ```
|
|
|
| <a id="daytona_sdk.common.code_run_params.CodeRunParams"></a>
| ## CodeRunParams
|
| ```python
| @dataclass
| class CodeRunParams()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/common/code_run_params.py#L5)
|
| Parameters for code execution.
|
| The Daytona SDK provides powerful process and code execution capabilities through
| the `process` module in Sandboxes. This guide covers all available process operations
| and best practices.
|
| **Example**:
|
|   Basic command execution:
| ```python
| workspace = daytona.create()
|
| # Execute a shell command
| response = workspace.process.exec("ls -la")
| print(response.result)
|
| # Run Python code
| response = workspace.process.code_run("print('Hello, World!')")
| print(response.result)
| ```
|
|   Using interactive sessions:
| ```python
| # Create a new session
| session_id = "my-session"
| workspace.process.create_session(session_id)
|
| # Execute commands in the session
| req = SessionExecuteRequest(command="cd /workspace", var_async=False)
| workspace.process.execute_session_command(session_id, req)
|
| req = SessionExecuteRequest(command="pwd", var_async=False)
| response = workspace.process.execute_session_command(session_id, req)
| print(response.result)  # Should print "/workspace"
|
| # Clean up
| workspace.process.delete_session(session_id)
| ```
|
|
|


--------------------------------------------------------------------------------
/docs/python-sdk/sandbox.mdx:
--------------------------------------------------------------------------------
| ---
| title: Sandbox
| ---
|
| The Daytona SDK core Sandbox functionality.
|
| Provides the main Workspace class representing a Daytona Sandbox that coordinates file system,
| Git, process execution, and LSP functionality. It serves as the central point
| for interacting with Daytona Sandboxes.
|
| **Example**:
|
|   Basic Sandbox operations:
| ```python
| from daytona_sdk import Daytona
| daytona = Daytona()
| workspace = daytona.create()
|
| # File operations
| workspace.fs.upload_file("/workspace/test.txt", b"Hello, World!")
| content = workspace.fs.download_file("/workspace/test.txt")
|
| # Git operations
| workspace.git.clone("https://github.com/user/repo.git")
|
| # Process execution
| response = workspace.process.exec("ls -la")
| print(response.result)
|
| # LSP functionality
| lsp = workspace.create_lsp_server("python", "/workspace/project")
| lsp.did_open("/workspace/project/src/index.ts")
| completions = lsp.completions("/workspace/project/src/index.ts", Position(line=10, character=15))
| print(completions)
| ```
|
|
| **Notes**:
|
|   The Sandbox must be in a 'started' state before performing operations.
|
| <a id="daytona_sdk.workspace.Workspace"></a>
| ## Workspace
|
| ```python
| class Workspace()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L181)
|
| Represents a Daytona Sandbox.
|
| A Sandbox provides file system operations, Git operations, process execution,
| and LSP functionality. It serves as the main interface for interacting with
| a Daytona Sandbox.
|
| **Attributes**:
|
| - `id` _str_ - Unique identifier for the Sandbox.
| - `instance` _WorkspaceInstance_ - The underlying Sandbox instance.
| - `code_toolbox` _WorkspaceCodeToolbox_ - Language-specific toolbox implementation.
| - `fs` _FileSystem_ - File system operations interface.
| - `git` _Git_ - Git operations interface.
| - `process` _Process_ - Process execution interface.
|
|
| #### Workspace.\_\_init\_\_
|
| ```python
| def __init__(id: str, instance: WorkspaceInstance, workspace_api: WorkspaceApi,
|              toolbox_api: ToolboxApi, code_toolbox: WorkspaceCodeToolbox)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L197)
|
| Initialize a new Workspace instance.
|
| **Arguments**:
|
| - `id` _str_ - Unique identifier for the Sandbox.
| - `instance` _WorkspaceInstance_ - The underlying Sandbox instance.
| - `workspace_api` _WorkspaceApi_ - API client for Sandbox operations.
| - `toolbox_api` _ToolboxApi_ - API client for toolbox operations.
| - `code_toolbox` _WorkspaceCodeToolbox_ - Language-specific toolbox implementation.
|
|
| #### Workspace.info
|
| ```python
| def info() -> WorkspaceInfo
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L227)
|
| Gets structured information about the Sandbox.
|
| **Returns**:
|
| - `WorkspaceInfo` - Detailed information about the Sandbox including its
|   configuration, resources, and current state.
|
|
| **Example**:
|
| ```python
| info = workspace.info()
| print(f"Workspace {info.name}:")
| print(f"State: {info.state}")
| print(f"Resources: {info.resources.cpu} CPU, {info.resources.memory} RAM")
| ```
|
|
| #### Workspace.get\_workspace\_root\_dir
|
| ```python
| @intercept_errors(message_prefix="Failed to get workspace root directory: ")
| def get_workspace_root_dir() -> str
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L246)
|
| Gets the root directory path of the Sandbox.
|
| **Returns**:
|
| - `str` - The absolute path to the Sandbox root directory.
|
|
| **Example**:
|
| ```python
| root_dir = workspace.get_workspace_root_dir()
| print(f"Workspace root: {root_dir}")
| ```
|
|
| #### Workspace.create\_lsp\_server
|
| ```python
| def create_lsp_server(language_id: LspLanguageId,
|                       path_to_project: str) -> LspServer
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L263)
|
| Creates a new Language Server Protocol (LSP) server instance.
|
| The LSP server provides language-specific features like code completion,
| diagnostics, and more.
|
| **Arguments**:
|
| - `language_id` _LspLanguageId_ - The language server type (e.g., LspLanguageId.PYTHON).
| - `path_to_project` _str_ - Absolute path to the project root directory.
|
|
| **Returns**:
|
| - `LspServer` - A new LSP server instance configured for the specified language.
|
|
| **Example**:
|
| ```python
| lsp = workspace.create_lsp_server("python", "/workspace/project")
| ```
|
|
| #### Workspace.set\_labels
|
| ```python
| @intercept_errors(message_prefix="Failed to set labels: ")
| def set_labels(labels: Dict[str, str]) -> Dict[str, str]
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L286)
|
| Sets labels for the Sandbox.
|
| Labels are key-value pairs that can be used to organize and identify Sandboxes.
|
| **Arguments**:
|
| - `labels` _Dict[str, str]_ - Dictionary of key-value pairs representing Sandbox labels.
|
|
| **Returns**:
|
|   Dict[str, str]: Dictionary containing the updated Sandbox labels.
|
|
| **Example**:
|
| ```python
| new_labels = workspace.set_labels({
|     "project": "my-project",
|     "environment": "development",
|     "team": "backend"
| })
| print(f"Updated labels: {new_labels}")
| ```
|
|
| #### Workspace.start
|
| ```python
| @intercept_errors(message_prefix="Failed to start workspace: ")
| @with_timeout(
|     error_message=lambda self, timeout:
|     f"Workspace {self.id} failed to start within the {timeout} seconds timeout period"
| )
| def start(timeout: Optional[float] = 60)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L315)
|
| Starts the Sandbox.
|
| This method starts the Sandbox and waits for it to be ready.
|
| **Arguments**:
|
| - `timeout` _Optional[float]_ - Maximum time to wait in seconds. 0 means no timeout. Default is 60 seconds.
|
|
| **Raises**:
|
| - `DaytonaError` - If timeout is negative. If workspace fails to start or times out.
|
|
| **Example**:
|
| ```python
| workspace = daytona.get_current_workspace("my-workspace")
| workspace.start(timeout=40)  # Wait up to 40 seconds
| print("Workspace started successfully")
| ```
|
|
| #### Workspace.stop
|
| ```python
| @intercept_errors(message_prefix="Failed to stop workspace: ")
| @with_timeout(
|     error_message=lambda self, timeout:
|     f"Workspace {self.id} failed to stop within the {timeout} seconds timeout period"
| )
| def stop(timeout: Optional[float] = 60)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L338)
|
| Stops the Sandbox.
|
| This method stops the Sandbox and waits for it to be fully stopped.
|
| **Arguments**:
|
| - `timeout` _Optional[float]_ - Maximum time to wait in seconds. 0 means no timeout. Default is 60 seconds.
|
|
| **Raises**:
|
| - `DaytonaError` - If timeout is negative; If workspace fails to stop or times out
|
|
| **Example**:
|
| ```python
| workspace = daytona.get_current_workspace("my-workspace")
| workspace.stop()
| print("Workspace stopped successfully")
| ```
|
|
| #### Workspace.wait\_for\_workspace\_start
|
| ```python
| @intercept_errors(
|     message_prefix="Failure during waiting for workspace to start: ")
| @with_timeout(
|     error_message=lambda self, timeout:
|     f"Workspace {self.id} failed to become ready within the {timeout} seconds timeout period"
| )
| def wait_for_workspace_start(timeout: Optional[float] = 60) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L361)
|
| Waits for the Sandbox to reach the 'started' state.
|
| This method polls the Sandbox status until it reaches the 'started' state
| or encounters an error.
|
| **Arguments**:
|
| - `timeout` _Optional[float]_ - Maximum time to wait in seconds. 0 means no timeout. Default is 60 seconds.
|
|
| **Raises**:
|
| - `DaytonaError` - If timeout is negative; If workspace fails to start or times out
|
|
| #### Workspace.wait\_for\_workspace\_stop
|
| ```python
| @intercept_errors(
|     message_prefix="Failure during waiting for workspace to stop: ")
| @with_timeout(
|     error_message=lambda self, timeout:
|     f"Workspace {self.id} failed to become stopped within the {timeout} seconds timeout period"
| )
| def wait_for_workspace_stop(timeout: Optional[float] = 60) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L387)
|
| Waits for the Sandbox to reach the 'stopped' state.
|
| This method polls the Sandbox status until it reaches the 'stopped' state
| or encounters an error. It will wait up to 60 seconds for the Sandbox to stop.
|
| **Arguments**:
|
| - `timeout` _Optional[float]_ - Maximum time to wait in seconds. 0 means no timeout. Default is 60 seconds.
|
|
| **Raises**:
|
| - `DaytonaError` - If timeout is negative. If Sandbox fails to stop or times out.
|
|
| #### Workspace.set\_autostop\_interval
|
| ```python
| @intercept_errors(message_prefix="Failed to set auto-stop interval: ")
| def set_autostop_interval(interval: int) -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L418)
|
| Sets the auto-stop interval for the Sandbox.
|
| The Sandbox will automatically stop after being idle (no new events) for the specified interval.
| Events include any state changes or interactions with the Sandbox through the SDK.
| Interactions using Sandbox Previews are not included.
|
| **Arguments**:
|
| - `interval` _int_ - Number of minutes of inactivity before auto-stopping.
|   Set to 0 to disable auto-stop. Defaults to 15.
|
|
| **Raises**:
|
| - `DaytonaError` - If interval is negative
|
|
| **Example**:
|
| ```python
| # Auto-stop after 1 hour
| workspace.set_autostop_interval(60)
| # Or disable auto-stop
| workspace.set_autostop_interval(0)
| ```
|
|
| #### Workspace.get\_preview\_link
|
| ```python
| @intercept_errors(message_prefix="Failed to get preview link: ")
| def get_preview_link(port: int) -> str
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L448)
|
| Gets the preview link for the workspace at a specific port. If the port is not open, it will open it and return the link.
|
| **Arguments**:
|
| - `port` _int_ - The port to open the preview link on
|
|
| **Returns**:
|
|   The preview link for the workspace at the specified port
|
|
| #### Workspace.archive
|
| ```python
| @intercept_errors(message_prefix="Failed to archive workspace: ")
| def archive() -> None
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L466)
|
| Archives the workspace, making it inactive and preserving its state. When sandboxes are archived, the entire filesystem
| state is moved to cost-effective object storage, making it possible to keep sandboxes available for an extended period.
| The tradeoff between archived and stopped states is that starting an archived sandbox takes more time, depending on its size.
| Workspace must be stopped before archiving.
|
|
| The Daytona SDK core Sandbox functionality.
|
| Provides the main Workspace class representing a Daytona Sandbox that coordinates file system,
| Git, process execution, and LSP functionality. It serves as the central point
| for interacting with Daytona Sandboxes.
|
| **Example**:
|
|   Basic Sandbox operations:
| ```python
| from daytona_sdk import Daytona
| daytona = Daytona()
| workspace = daytona.create()
|
| # File operations
| workspace.fs.upload_file("/workspace/test.txt", b"Hello, World!")
| content = workspace.fs.download_file("/workspace/test.txt")
|
| # Git operations
| workspace.git.clone("https://github.com/user/repo.git")
|
| # Process execution
| response = workspace.process.exec("ls -la")
| print(response.result)
|
| # LSP functionality
| lsp = workspace.create_lsp_server("python", "/workspace/project")
| lsp.did_open("/workspace/project/src/index.ts")
| completions = lsp.completions("/workspace/project/src/index.ts", Position(line=10, character=15))
| print(completions)
| ```
|
|
| **Notes**:
|
|   The Sandbox must be in a 'started' state before performing operations.
|
|
| <a id="daytona_sdk.workspace.WorkspaceTargetRegion"></a>
| ## WorkspaceTargetRegion
|
| ```python
| @dataclass
| class WorkspaceTargetRegion(Enum)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L58)
|
| Target regions for workspaces
|
|
| <a id="daytona_sdk.workspace.WorkspaceResources"></a>
| ## WorkspaceResources
|
| ```python
| @dataclass
| class WorkspaceResources()
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L74)
|
| Resources allocated to a Sandbox.
|
| **Attributes**:
|
| - `cpu` _str_ - Number of CPU cores allocated (e.g., "1", "2").
| - `gpu` _Optional[str]_ - Number of GPUs allocated (e.g., "1") or None if no GPU.
| - `memory` _str_ - Amount of memory allocated with unit (e.g., "2Gi", "4Gi").
| - `disk` _str_ - Amount of disk space allocated with unit (e.g., "10Gi", "20Gi").
|
|
| **Example**:
|
| ```python
| resources = WorkspaceResources(
|     cpu="2",
|     gpu="1",
|     memory="4Gi",
|     disk="20Gi"
| )
| ```
|
|
| <a id="daytona_sdk.workspace.WorkspaceState"></a>
| ## WorkspaceState
|
| ```python
| @dataclass
| class WorkspaceState(Enum)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L100)
|
| States of a Sandbox.
|
|
| <a id="daytona_sdk.workspace.WorkspaceInfo"></a>
| ## WorkspaceInfo
|
| ```python
| class WorkspaceInfo(ApiWorkspaceInfo)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L124)
|
| Structured information about a Sandbox.
|
| This class provides detailed information about a Sandbox's configuration,
| resources, and current state.
|
| **Attributes**:
|
| - `id` _str_ - Unique identifier for the Sandbox.
| - `name` _str_ - Display name of the Sandbox.
| - `image` _str_ - Docker image used for the Sandbox.
| - `user` _str_ - OS user running in the Sandbox.
| - `env` _Dict[str, str]_ - Environment variables set in the Sandbox.
| - `labels` _Dict[str, str]_ - Custom labels attached to the Sandbox.
| - `public` _bool_ - Whether the Sandbox is publicly accessible.
| - `target` _str_ - Target environment where the Sandbox runs.
| - `resources` _WorkspaceResources_ - Resource allocations for the Sandbox.
| - `state` _str_ - Current state of the Sandbox (e.g., "started", "stopped").
| - `error_reason` _Optional[str]_ - Error message if Sandbox is in error state.
| - `snapshot_state` _Optional[str]_ - Current state of Sandbox snapshot.
| - `snapshot_state_created_at` _Optional[datetime]_ - When the snapshot state was created.
|
|
| **Example**:
|
| ```python
| workspace = daytona.create()
| info = workspace.info()
| print(f"Workspace {info.name} is {info.state}")
| print(f"Resources: {info.resources.cpu} CPU, {info.resources.memory} RAM")
| ```
|
|
| <a id="daytona_sdk.workspace.WorkspaceInstance"></a>
| ## WorkspaceInstance
|
| ```python
| class WorkspaceInstance(ApiWorkspace)
| ```
|
| [[view_source]](https://github.com/daytonaio/sdk/blob/e2c391f367e740a14945617b9e5c7b965ba4d7d9/packages/python/src/daytona_sdk/workspace.py#L176)
|
| Represents a Daytona workspace instance.
|
|
|


--------------------------------------------------------------------------------

├── examples
    └── python
    │   ├── exec-command
    │       ├── exec-session.py
    │       └── exec.py
    │   ├── file-operations
    │       └── main.py
    │   ├── git-lsp
    │       └── main.py
    │   └── lifecycle
    │       └── lifecycle.py
└── packages
    └── python
        ├── setup.py
        └── src
            └── daytona_sdk
                ├── __init__.py
                ├── _utils
                    ├── __init__.py
                    ├── enum.py
                    ├── errors.py
                    └── timeout.py
                ├── code_toolbox
                    ├── __init__.py
                    ├── workspace_python_code_toolbox.py
                    └── workspace_ts_code_toolbox.py
                ├── common
                    ├── __init__.py
                    ├── code_run_params.py
                    └── errors.py
                ├── daytona.py
                ├── filesystem.py
                ├── git.py
                ├── lsp_server.py
                ├── process.py
                ├── protocols.py
                └── workspace.py


/examples/python/exec-command/exec-session.py:
--------------------------------------------------------------------------------
| from daytona_sdk import Daytona, CreateWorkspaceParams, SessionExecuteRequest
|
| daytona = Daytona()
| workspace = daytona.create()
|
| exec_session_id = "exec-session-1"
| workspace.process.create_session(exec_session_id)
|
| # Get the session details any time
| session = workspace.process.get_session(exec_session_id)
| print(session)
|
| # Execute a first command in the session
| execCommand1 = workspace.process.execute_session_command(exec_session_id, SessionExecuteRequest(
|     command="export FOO=BAR"
| ))
| if execCommand1.exit_code != 0:
|     print(f"Error: {execCommand1.exit_code} {execCommand1.output}")
|
| # Get the session details again to see the command has been executed
| session = workspace.process.get_session(exec_session_id)
| print(session)
|
| # Get the command details
| session_command = workspace.process.get_session_command(exec_session_id, execCommand1.cmd_id)
| print(session_command)
|
| # Execute a second command in the session and see that the environment variable is set
| execCommand2 = workspace.process.execute_session_command(exec_session_id, SessionExecuteRequest(
|     command="echo $FOO"
| ))
| if execCommand2.exit_code != 0:
|     print(f"Error: {execCommand2.exit_code} {execCommand2.output}")
| else:
|     print(execCommand2.output)
|
| print("Now getting logs for the second command")
| logs = workspace.process.get_session_command_logs(exec_session_id, execCommand2.cmd_id)
| print(logs)
|
| # You can also list all active sessions
| sessions = workspace.process.list_sessions()
| print(sessions)
|
| # And of course you can delete the session at any time
| workspace.process.delete_session(exec_session_id)
|
| daytona.remove(workspace)
|


--------------------------------------------------------------------------------
/examples/python/exec-command/exec.py:
--------------------------------------------------------------------------------
| from daytona_sdk import Daytona, CreateWorkspaceParams
|
| daytona = Daytona()
|
| params = CreateWorkspaceParams(
|     language="python",
| )
| workspace = daytona.create(params)
|
| # Run the code securely inside the workspace
| response = workspace.process.code_run('print("Hello World!")')
| if response.exit_code != 0:
|     print(f"Error: {response.exit_code} {response.result}")
| else:
|     print(response.result)
|
| # Execute an os command in the workspace
| response = workspace.process.exec('echo "Hello World from exec!"', cwd="/home/daytona", timeout=10)
| if response.exit_code != 0:
|     print(f"Error: {response.exit_code} {response.result}")
| else:
|     print(response.result)
|
| daytona.remove(workspace)
|


--------------------------------------------------------------------------------
/examples/python/file-operations/main.py:
--------------------------------------------------------------------------------
| import os
| from daytona_sdk import Daytona, CreateWorkspaceParams
|
| daytona = Daytona()
| params = CreateWorkspaceParams(
|   language="python",
| )
|
| # First, create a workspace
| workspace = daytona.create(params)
|
| # Get workspace root directory
| root_dir = workspace.get_workspace_root_dir()
|
| # List files in the workspace
| files = workspace.fs.list_files(root_dir)
| print("Files:", files)
|
| # Create a new directory in the workspace
| new_dir = os.path.join(root_dir, "new-dir")
| workspace.fs.create_folder(new_dir, "755")
|
| file_path = os.path.join(new_dir, "data.txt")
|
| # Add a new file to the workspace
| file_content = b"Hello, World!"
| workspace.fs.upload_file(file_path, file_content)
|
| # Search for the file we just added
| matches = workspace.fs.find_files(root_dir, "World!")
| print("Matches:", matches)
|
| # Replace the contents of the file
| workspace.fs.replace_in_files([file_path], "Hello, World!", "Goodbye, World!")
|
| # Read the file
| downloaded_file = workspace.fs.download_file(file_path)
| print("File content:", downloaded_file.decode("utf-8"))
|
| # Change the file permissions
| workspace.fs.set_file_permissions(file_path, mode="777")
|
| # Get file info
| file_info = workspace.fs.get_file_info(file_path)
| print("File info:", file_info)  # Should show the new permissions
|
| # Move the file to the new location
| new_file_path = os.path.join(root_dir, "moved-data.txt")
| workspace.fs.move_files(file_path, new_file_path)
|
| # Find the file in the new location
| search_results = workspace.fs.search_files(root_dir, "moved-data.txt")
| print("Search results:", search_results)
|
| # Delete the file
| workspace.fs.delete_file(new_file_path)
|
| # daytona.remove(workspace)
|


--------------------------------------------------------------------------------
/examples/python/git-lsp/main.py:
--------------------------------------------------------------------------------
| from daytona_sdk import Daytona
| import os
|
|
| def main():
|     daytona = Daytona()
|
|     workspace = daytona.create()
|
|     try:
|         root_dir = workspace.get_workspace_root_dir()
|         project_dir = os.path.join(root_dir, "learn-typescript")
|
|         # Clone the repository
|         workspace.git.clone(
|             "https://github.com/panaverse/learn-typescript", project_dir, "master"
|         )
|
|         workspace.git.pull(project_dir)
|
|         # Search for the file we want to work on
|         matches = workspace.fs.find_files(project_dir, "var obj1 = new Base();")
|         print("Matches:", matches)
|
|         # Start the language server
|         lsp = workspace.create_lsp_server("typescript", project_dir)
|         lsp.start()
|
|         # Notify the language server of the document we want to work on
|         lsp.did_open(matches[0].file)
|
|         # Get symbols in the document
|         symbols = lsp.document_symbols(matches[0].file)
|         print("Symbols:", symbols)
|
|         # Fix the error in the document
|         workspace.fs.replace_in_files(
|             [matches[0].file], "var obj1 = new Base();", "var obj1 = new E();"
|         )
|
|         # Notify the language server of the document change
|         lsp.did_close(matches[0].file)
|         lsp.did_open(matches[0].file)
|
|         # Get completions at a specific position
|         completions = lsp.completions(matches[0].file, {"line": 12, "character": 18})
|         print("Completions:", completions)
|
|     except Exception as error:
|         print("Error creating workspace:", error)
|     finally:
|         # Cleanup
|         daytona.remove(workspace)
|
|
| if __name__ == "__main__":
|     main()
|


--------------------------------------------------------------------------------
/examples/python/lifecycle/lifecycle.py:
--------------------------------------------------------------------------------
| from daytona_sdk import Daytona
| from pprint import pprint
|
| daytona = Daytona()
|
| print("Creating workspace")
| workspace = daytona.create()
| print("Workspace created")
|
| workspace.set_labels({
|     "public": True,
| })
|
| print("Stopping workspace")
| daytona.stop(workspace)
| print("Workspace stopped")
|
| print("Starting workspace")
| daytona.start(workspace)
| print("Workspace started")
|
| print("Getting existing workspace")
| existing_workspace = daytona.get_current_workspace(workspace.id)
| print("Get existing workspace")
|
| response = existing_workspace.process.exec('echo "Hello World from exec!"', cwd="/home/daytona", timeout=10)
| if response.exit_code != 0:
|     print(f"Error: {response.exit_code} {response.result}")
| else:
|     print(response.result)
|
| workspaces = daytona.list()
| print("Total workspaces count:" , len(workspaces))
| pprint(vars(workspaces[0].info()))  # This will show all attributes of the first workspace
|
| print("Removing workspace")
| daytona.remove(workspace)
| print("Workspace removed")
|


--------------------------------------------------------------------------------
/packages/python/setup.py:
--------------------------------------------------------------------------------
| from setuptools import setup, find_packages
|
| setup(
|     name="daytona_sdk",
|     version="0.1.4",
|     packages=find_packages(where="src"),
|     package_dir={"": "src"},
|     install_requires=[
|         # Add dependencies here
|     ],
|     # Include both packages
|     package_data={
|         "daytona_sdk": ["*"]
|     },
| )
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/__init__.py:
--------------------------------------------------------------------------------
| from .daytona import (
|     Daytona,
|     DaytonaConfig,
|     CreateWorkspaceParams,
|     CodeLanguage,
|     Workspace,
|     SessionExecuteRequest,
|     SessionExecuteResponse,
|     DaytonaError,
|     WorkspaceTargetRegion,
| )
| from .lsp_server import LspLanguageId
| from .workspace import WorkspaceState
| from .common.code_run_params import CodeRunParams
|
| __all__ = [
|     "Daytona",
|     "DaytonaConfig",
|     "CreateWorkspaceParams",
|     "CodeLanguage",
|     "Workspace",
|     "SessionExecuteRequest",
|     "SessionExecuteResponse",
|     "DaytonaError",
|     "LspLanguageId",
|     "WorkspaceTargetRegion",
|     "WorkspaceState",
|     "CodeRunParams"
| ]
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/_utils/__init__.py:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/daytonaio/sdk/dcf0ddc88a191f3b23a037f3539d78a53a2652b7/packages/python/src/daytona_sdk/_utils/__init__.py


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/_utils/enum.py:
--------------------------------------------------------------------------------
| from enum import Enum
| from typing import Optional
|
|
| def to_enum(enum_class: type, value: str) -> Optional[Enum]:
|     """Convert a string to an enum.
|
|     Args:
|         enum_class (type): The enum class to convert to.
|         value (str): The value to convert to an enum.
|
|     Returns:
|         The enum value, or None if the value is not a valid enum.
|     """
|     if isinstance(value, enum_class):
|         return value
|     str_value = str(value)
|     if str_value in enum_class._value2member_map_:
|         return enum_class(str_value)
|     return None
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/_utils/errors.py:
--------------------------------------------------------------------------------
| import json
| import functools
| from typing import Callable, Optional, TypeVar, Any, ParamSpec
| from daytona_api_client.exceptions import OpenApiException
| from daytona_sdk.common.errors import DaytonaError
|
|
| P = ParamSpec('P')
| T = TypeVar('T')
|
|
| def intercept_errors(message_prefix: str = "") -> Callable[[Callable[P, T]], Callable[P, T]]:
|     """Decorator to intercept errors, process them, and optionally add a message prefix.
|     If the error is an OpenApiException, it will be processed to extract the most meaningful error message.
|
|     Args:
|         message_prefix (str): Custom message prefix for the error.
|     """
|     def decorator(func: Callable[P, T]) -> Callable[P, T]:
|         @functools.wraps(func)
|         def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
|             try:
|                 return func(*args, **kwargs)
|             except OpenApiException as e:
|                 message = _get_open_api_exception_message(e)
|
|                 raise DaytonaError(f"{message_prefix}{message}") from None
|             except Exception as e:
|                 if message_prefix:
|                     message = f"{message_prefix}{str(e)}"
|                     raise DaytonaError(message)
|                 raise DaytonaError(str(e))
|
|         return wrapper
|     return decorator
|
|
| def _get_open_api_exception_message(exception: OpenApiException) -> str:
|     """Process API exceptions to extract the most meaningful error message.
|
|     This method examines the exception's body attribute and attempts to extract
|     the most informative error message using the following logic:
|     1. If the body is missing or empty, returns the original exception
|     2. If the body contains valid JSON with a 'message' field, uses that message
|     3. If the body is not valid JSON or does not contain a 'message' field, uses the raw body string
|
|     Args:
|         exception (OpenApiException): The OpenApiException to process
|
|     Returns:
|         Processed message
|     """
|     if not hasattr(exception, 'body') or not exception.body:
|         return str(exception)
|
|     body_str = str(exception.body)
|     try:
|         data = json.loads(body_str)
|         message = data.get('message', body_str) if isinstance(
|             data, dict) else body_str
|     except json.JSONDecodeError:
|         message = body_str
|
|     return message
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/_utils/timeout.py:
--------------------------------------------------------------------------------
| import functools
| import concurrent.futures
| from typing import Callable, Optional, Any, TypeVar, ParamSpec
| from daytona_sdk._utils.errors import DaytonaError
|
|
| P = ParamSpec('P')
| T = TypeVar('T')
|
|
| def with_timeout(error_message: Optional[Callable[[Any, float], str]] = None) -> Callable[[Callable[P, T]], Callable[P, T]]:
|     """Decorator to add a timeout mechanism with an optional custom error message.
|
|     Args:
|         error_message (Optional[Callable[[Any, float], str]]): A callable that accepts `self` and `timeout`,
|                                                                and returns a string error message.
|     """
|     def decorator(func: Callable[P, T]) -> Callable[P, T]:
|         @functools.wraps(func)
|         def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
|             # Get function argument names
|             arg_names = func.__code__.co_varnames[:func.__code__.co_argcount]
|             arg_dict = dict(zip(arg_names, args))
|
|             # Extract self if method is bound
|             self_instance = args[0] if args else None
|
|             # Check for 'timeout' in kwargs first, then in positional arguments
|             timeout = kwargs.get('timeout', arg_dict.get('timeout', None))
|
|             if timeout is None or timeout == 0:
|                 # If timeout is None or 0, run the function normally
|                 return func(*args, **kwargs)
|
|             if timeout < 0:
|                 raise DaytonaError(
|                     "Timeout must be a non-negative number or None.")
|
|             with concurrent.futures.ThreadPoolExecutor() as executor:
|                 future = executor.submit(func, *args, **kwargs)
|                 try:
|                     return future.result(timeout=timeout)
|                 except concurrent.futures.TimeoutError:
|                     # Use custom error message if provided, otherwise default
|                     msg = error_message(
|                         self_instance, timeout) if error_message else f"Function '{func.__name__}' exceeded timeout of {timeout} seconds."
|                     raise TimeoutError(msg)
|         return wrapper
|     return decorator
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/code_toolbox/__init__.py:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/daytonaio/sdk/dcf0ddc88a191f3b23a037f3539d78a53a2652b7/packages/python/src/daytona_sdk/code_toolbox/__init__.py


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/code_toolbox/workspace_python_code_toolbox.py:
--------------------------------------------------------------------------------
| import base64
| from typing import Optional
| from ..common.code_run_params import CodeRunParams
|
|
| class WorkspacePythonCodeToolbox:
|     def get_run_command(self, code: str, params: Optional[CodeRunParams] = None) -> str:
|         # Encode the provided code in base64
|         base64_code = base64.b64encode(code.encode()).decode()
|
|         # Build environment variables string
|         env_vars = ""
|         if params and params.env:
|             env_vars = ' '.join(f"{key}='{value}'" for key, value in params.env.items())
|
|         # Build command-line arguments string
|         argv = ""
|         if params and params.argv:
|             argv = ' '.join(params.argv)
|
|         # Combine everything into the final command
|         return f""" sh -c '{env_vars} python3 -c "exec(__import__(\\\"base64\\\").b64decode(\\\"{base64_code}\\\").decode())" {argv}' """
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/code_toolbox/workspace_ts_code_toolbox.py:
--------------------------------------------------------------------------------
| import base64
| from typing import Optional
| from ..common.code_run_params import CodeRunParams
|
|
| class WorkspaceTsCodeToolbox:
|     def get_run_command(self, code: str, params: Optional[CodeRunParams] = None) -> str:
|         # Encode the provided code in base64
|         base64_code = base64.b64encode(code.encode()).decode()
|
|         # Build environment variables string
|         env_vars = ""
|         if params and params.env:
|             env_vars = ' '.join(f"{key}='{value}'" for key, value in params.env.items())
|
|         # Build command-line arguments string
|         argv = ""
|         if params and params.argv:
|             argv = ' '.join(params.argv)
|
|         # Combine everything into the final command for TypeScript
|         return f""" sh -c 'echo {base64_code} | base64 --decode | {env_vars} npx ts-node -O "{{\\\"module\\\":\\\"CommonJS\\\"}}" -e "$(cat)" x {argv} 2>&1 | grep -vE "npm notice|npm warn exec"' """
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/common/__init__.py:
--------------------------------------------------------------------------------
https://raw.githubusercontent.com/daytonaio/sdk/dcf0ddc88a191f3b23a037f3539d78a53a2652b7/packages/python/src/daytona_sdk/common/__init__.py


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/common/code_run_params.py:
--------------------------------------------------------------------------------
| from dataclasses import dataclass
| from typing import Optional, List, Dict
|
| @dataclass
| class CodeRunParams:
|     """Parameters for code execution."""
|     argv: Optional[List[str]] = None
|     env: Optional[Dict[str, str]] = None


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/common/errors.py:
--------------------------------------------------------------------------------
| class DaytonaError(Exception):
|     """Base error for Daytona SDK."""
|     pass
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/daytona.py:
--------------------------------------------------------------------------------
| """
| Sandboxes are isolated development environments managed by Daytona.
| This guide covers how to create, manage, and remove Sandboxes using the SDK.
|
| Examples:
|     Basic usage with environment variables:
|     ```python
|     from daytona_sdk import Daytona
|     # Initialize using environment variables
|     daytona = Daytona()  # Uses env vars DAYTONA_API_KEY, DAYTONA_SERVER_URL, DAYTONA_TARGET
|
|     # Create a default Python workspace with custom environment variables
|     workspace = daytona.create(CreateWorkspaceParams(
|         language="python",
|         env_vars={"PYTHON_ENV": "development"}
|     ))
|
|     # Execute commands in the workspace
|     response = workspace.process.execute_command('echo "Hello, World!"')
|     print(response.result)
|
|     # Run Python code securely inside the workspace
|     response = workspace.process.code_run('print("Hello from Python!")')
|     print(response.result)
|
|     # Remove the workspace after use
|     daytona.remove(workspace)
|     ```
|
|     Usage with explicit configuration:
|     ```python
|     from daytona_sdk import Daytona, DaytonaConfig, CreateWorkspaceParams, WorkspaceResources
|
|     # Initialize with explicit configuration
|     config = DaytonaConfig(
|         api_key="your-api-key",
|         server_url="https://your-server.com",
|         target="us"
|     )
|     daytona = Daytona(config)
|
|     # Create a custom workspace with specific resources and settings
|     workspace = daytona.create(CreateWorkspaceParams(
|         language="python",
|         image="python:3.11",
|         resources=WorkspaceResources(
|             cpu=2,
|             memory=4,  # 4GB RAM
|             disk=20    # 20GB disk
|         ),
|         env_vars={"PYTHON_ENV": "development"},
|         auto_stop_interval=60  # Auto-stop after 1 hour of inactivity
|     ))
|
|     # Use workspace features
|     workspace.git.clone("https://github.com/user/repo.git")
|     workspace.process.execute_command("python -m pytest")
|     ```
| """
|
| from enum import Enum
| import uuid
| from typing import Optional, Dict, List, Annotated
| from pydantic import BaseModel, Field
| from dataclasses import dataclass
| from environs import Env
| from daytona_api_client import (
|     Configuration,
|     WorkspaceApi,
|     ToolboxApi,
|     ApiClient,
|     CreateWorkspace,
|     SessionExecuteRequest,
|     SessionExecuteResponse
| )
| from daytona_sdk._utils.errors import intercept_errors, DaytonaError
| from .code_toolbox.workspace_python_code_toolbox import WorkspacePythonCodeToolbox
| from .code_toolbox.workspace_ts_code_toolbox import WorkspaceTsCodeToolbox
| from ._utils.enum import to_enum
| from .workspace import Workspace, WorkspaceTargetRegion
| from ._utils.timeout import with_timeout
|
|
| @dataclass
| class CodeLanguage(Enum):
|     """Programming languages supported by Daytona"""
|     PYTHON = "python"
|     TYPESCRIPT = "typescript"
|     JAVASCRIPT = "javascript"
|
|     def __str__(self):
|         return self.value
|
|     def __eq__(self, other):
|         if isinstance(other, str):
|             return self.value == other
|         return super().__eq__(other)
|
|
| @dataclass
| class DaytonaConfig:
|     """Configuration options for initializing the Daytona client.
|
|     Attributes:
|         api_key (str): API key for authentication with Daytona server.
|         server_url (str): URL of the Daytona server.
|         target (str): Target environment for Sandbox.
|
|     Example:
|         ```python
|         config = DaytonaConfig(
|             api_key="your-api-key",
|             server_url="https://your-server.com",
|             target="us"
|         )
|         daytona = Daytona(config)
|         ```
|     """
|     api_key: str
|     server_url: str
|     target: WorkspaceTargetRegion
|
|
| @dataclass
| class WorkspaceResources:
|     """Resources configuration for Sandbox.
|
|     Attributes:
|         cpu (Optional[int]): Number of CPU cores to allocate.
|         memory (Optional[int]): Amount of memory in GB to allocate.
|         disk (Optional[int]): Amount of disk space in GB to allocate.
|         gpu (Optional[int]): Number of GPUs to allocate.
|
|     Example:
|         ```python
|         resources = WorkspaceResources(
|             cpu=2,
|             memory=4,  # 4GB RAM
|             disk=20,   # 20GB disk
|             gpu=1
|         )
|         params = CreateWorkspaceParams(
|             language="python",
|             resources=resources
|         )
|         ```
|     """
|     cpu: Optional[int] = None
|     memory: Optional[int] = None
|     disk: Optional[int] = None
|     gpu: Optional[int] = None
|
|
| class CreateWorkspaceParams(BaseModel):
|     """Parameters for creating a new Sandbox.
|
|     Attributes:
|         language (CodeLanguage): Programming language for the Sandbox ("python", "javascript", "typescript").
|         id (Optional[str]): Custom identifier for the Sandbox. If not provided, a random ID will be generated.
|         name (Optional[str]): Display name for the Sandbox. Defaults to Sandbox ID if not provided.
|         image (Optional[str]): Custom Docker image to use for the Sandbox.
|         os_user (Optional[str]): OS user for the Sandbox.
|         env_vars (Optional[Dict[str, str]]): Environment variables to set in the Sandbox.
|         labels (Optional[Dict[str, str]]): Custom labels for the Sandbox.
|         public (Optional[bool]): Whether the Sandbox should be public.
|         target (Optional[str]): Target location for the Sandbox. Can be "us", "eu", or "asia".
|         resources (Optional[WorkspaceResources]): Resource configuration for the Sandbox.
|         timeout (Optional[float]): Timeout in seconds for Sandbox to be created and started.
|         auto_stop_interval (Optional[int]): Interval in minutes after which Sandbox will automatically stop if no Sandbox event occurs during that time. Default is 15 minutes. 0 means no auto-stop.
|
|     Example:
|         ```python
|         params = CreateWorkspaceParams(
|             language="python",
|             name="my-workspace",
|             env_vars={"DEBUG": "true"},
|             resources=WorkspaceResources(cpu=2, memory=4),
|             auto_stop_interval=20
|         )
|         workspace = daytona.create(params, 50)
|         ```
|     """
|     language: CodeLanguage
|     id: Optional[str] = None
|     name: Optional[str] = None
|     image: Optional[str] = None
|     os_user: Optional[str] = None
|     env_vars: Optional[Dict[str, str]] = None
|     labels: Optional[Dict[str, str]] = None
|     public: Optional[bool] = None
|     target: Optional[WorkspaceTargetRegion] = None
|     resources: Optional[WorkspaceResources] = None
|     timeout: Annotated[Optional[float], Field(
|         default=None, deprecated='The `timeout` field is deprecated and will be removed in future versions. Use `timeout` argument in method calls instead.')]
|     auto_stop_interval: Optional[int] = None
|
|
| class Daytona:
|     """Main class for interacting with Daytona Server API.
|
|     This class provides methods to create, manage, and interact with Daytona Sandboxes.
|     It can be initialized either with explicit configuration or using environment variables.
|
|     Attributes:
|         api_key (str): API key for authentication.
|         server_url (str): URL of the Daytona server.
|         target (str): Default target location for Sandboxes.
|
|     Example:
|         Using environment variables:
|         ```python
|         daytona = Daytona()  # Uses DAYTONA_API_KEY, DAYTONA_SERVER_URL
|         ```
|
|         Using explicit configuration:
|         ```python
|         config = DaytonaConfig(
|             api_key="your-api-key",
|             server_url="https://your-server.com",
|             target="us"
|         )
|         daytona = Daytona(config)
|         ```
|     """
|
|     def __init__(self, config: Optional[DaytonaConfig] = None):
|         """Initializes Daytona instance with optional configuration.
|
|         If no config is provided, reads from environment variables:
|         - `DAYTONA_API_KEY`: Required API key for authentication
|         - `DAYTONA_SERVER_URL`: Required server URL
|         - `DAYTONA_TARGET`: Optional target environment (defaults to WorkspaceTargetRegion.US)
|
|         Args:
|             config (Optional[DaytonaConfig]): Object containing api_key, server_url, and target.
|
|         Raises:
|             DaytonaError: If API key or Server URL is not provided either through config or environment variables
|
|         Example:
|             ```python
|             from daytona_sdk import Daytona, DaytonaConfig
|             # Using environment variables
|             daytona1 = Daytona()
|             # Using explicit configuration
|             config = DaytonaConfig(
|                 api_key="your-api-key",
|                 server_url="https://your-server.com",
|                 target="us"
|             )
|             daytona2 = Daytona(config)
|             ```
|         """
|         if config is None:
|             # Initialize env - it automatically reads from .env and .env.local
|             env = Env()
|             env.read_env()  # reads .env
|             # reads .env.local and overrides values
|             env.read_env(".env.local", override=True)
|
|             self.api_key = env.str("DAYTONA_API_KEY")
|             self.server_url = env.str("DAYTONA_SERVER_URL")
|             self.target = env.str("DAYTONA_TARGET", WorkspaceTargetRegion.US)
|         else:
|             self.api_key = config.api_key
|             self.server_url = config.server_url
|             self.target = config.target
|
|         if not self.api_key:
|             raise DaytonaError("API key is required")
|
|         if not self.server_url:
|             raise DaytonaError("Server URL is required")
|
|         if not self.target:
|             self.target = WorkspaceTargetRegion.US
|
|         # Create API configuration without api_key
|         configuration = Configuration(host=self.server_url)
|         api_client = ApiClient(configuration)
|         api_client.default_headers["Authorization"] = f"Bearer {self.api_key}"
|
|         # Initialize API clients with the api_client instance
|         self.workspace_api = WorkspaceApi(api_client)
|         self.toolbox_api = ToolboxApi(api_client)
|
|     @intercept_errors(message_prefix="Failed to create workspace: ")
|     def create(self, params: Optional[CreateWorkspaceParams] = None, timeout: Optional[float] = 60) -> Workspace:
|         """Creates Sandboxes with default or custom configurations. You can specify various parameters,
|         including language, image, resources, environment variables, and volumes for the Sandbox.
|
|         Args:
|             params (Optional[CreateWorkspaceParams]): Parameters for Sandbox creation. If not provided,
|                    defaults to Python language.
|             timeout (Optional[float]): Timeout (in seconds) for workspace creation. 0 means no timeout. Default is 60 seconds.
|
|         Returns:
|             Workspace: The created Sandbox instance.
|
|         Raises:
|             DaytonaError: If timeout or auto_stop_interval is negative; If workspace fails to start or times out
|
|         Example:
|             Create a default Python Sandbox:
|             ```python
|             workspace = daytona.create()
|             ```
|
|             Create a custom Sandbox:
|             ```python
|             params = CreateWorkspaceParams(
|                 language="python",
|                 name="my-workspace",
|                 image="debian:12.9",
|                 env_vars={"DEBUG": "true"},
|                 resources=WorkspaceResources(cpu=2, memory=4096),
|                 auto_stop_interval=0
|             )
|             workspace = daytona.create(params, 40)
|             ```
|         """
|         # If no params provided, create default params for Python
|         if params is None:
|             params = CreateWorkspaceParams(language="python")
|
|         params.id = params.id if params.id else f"sandbox-{str(uuid.uuid4())[:8]}"
|
|         effective_timeout = params.timeout if params.timeout else timeout
|
|         try:
|             return self._create(params, effective_timeout)
|         except Exception as e:
|             try:
|                 self.workspace_api.delete_workspace(
|                     workspace_id=params.id, force=True)
|             except Exception:
|                 pass
|             raise e
|
|     @with_timeout(error_message=lambda self, timeout: f"Failed to create and start workspace within {timeout} seconds timeout period.")
|     def _create(self, params: Optional[CreateWorkspaceParams] = None, timeout: Optional[float] = 60) -> Workspace:
|         """Creates a new Sandbox and waits for it to start.
|
|         Args:
|             params (Optional[CreateWorkspaceParams]): Parameters for Sandbox creation. If not provided,
|                    defaults to Python language.
|             timeout (Optional[float]): Timeout (in seconds) for workspace creation. 0 means no timeout. Default is 60 seconds.
|
|         Returns:
|             Workspace: The created Sandbox instance.
|
|         Raises:
|             DaytonaError: If timeout or auto_stop_interval is negative; If workspace fails to start or times out
|         """
|         code_toolbox = self._get_code_toolbox(params)
|
|         if timeout < 0:
|             raise DaytonaError("Timeout must be a non-negative number")
|
|         if params.auto_stop_interval is not None and params.auto_stop_interval < 0:
|             raise DaytonaError(
|                 "auto_stop_interval must be a non-negative integer")
|
|         target = params.target if params.target else self.target
|
|         # Create workspace using dictionary
|         workspace_data = CreateWorkspace(
|             id=params.id,
|             name=params.name if params.name else params.id,
|             image=params.image,
|             user=params.os_user,
|             env=params.env_vars if params.env_vars else {},
|             labels=params.labels,
|             public=params.public,
|             target=str(target) if target else None,
|             auto_stop_interval=params.auto_stop_interval
|         )
|
|         if params.resources:
|             workspace_data.cpu = params.resources.cpu
|             workspace_data.memory = params.resources.memory
|             workspace_data.disk = params.resources.disk
|             workspace_data.gpu = params.resources.gpu
|
|         response = self.workspace_api.create_workspace(
|             create_workspace=workspace_data, _request_timeout=timeout or None)
|         workspace_info = Workspace._to_workspace_info(response)
|         response.info = workspace_info
|
|         workspace = Workspace(
|             params.id,
|             response,
|             self.workspace_api,
|             self.toolbox_api,
|             code_toolbox
|         )
|
|         # Wait for workspace to start
|         try:
|             workspace.wait_for_workspace_start()
|         finally:
|             # If not Daytona SaaS, we don't need to handle pulling image state
|             pass
|
|         return workspace
|
|     def _get_code_toolbox(self, params: Optional[CreateWorkspaceParams] = None):
|         """Helper method to get the appropriate code toolbox based on language.
|
|         Args:
|             params (Optional[CreateWorkspaceParams]): Sandbox parameters. If not provided, defaults to Python toolbox.
|
|         Returns:
|             The appropriate code toolbox instance for the specified language.
|
|         Raises:
|             DaytonaError: If an unsupported language is specified.
|         """
|         if not params:
|             return WorkspacePythonCodeToolbox()
|
|         enum_language = to_enum(CodeLanguage, params.language)
|         if enum_language is None:
|             raise DaytonaError(f"Unsupported language: {params.language}")
|         else:
|             params.language = enum_language
|
|         match params.language:
|             case CodeLanguage.JAVASCRIPT | CodeLanguage.TYPESCRIPT:
|                 return WorkspaceTsCodeToolbox()
|             case CodeLanguage.PYTHON:
|                 return WorkspacePythonCodeToolbox()
|             case _:
|                 raise DaytonaError(f"Unsupported language: {params.language}")
|
|     @intercept_errors(message_prefix="Failed to remove workspace: ")
|     def remove(self, workspace: Workspace, timeout: Optional[float] = 60) -> None:
|         """Removes a Sandbox.
|
|         Args:
|             workspace (Workspace): The Sandbox instance to remove.
|             timeout (Optional[float]): Timeout (in seconds) for workspace removal. 0 means no timeout. Default is 60 seconds.
|
|         Raises:
|             DaytonaError: If workspace fails to remove or times out
|
|         Example:
|             ```python
|             workspace = daytona.create()
|             # ... use workspace ...
|             daytona.remove(workspace)  # Clean up when done
|             ```
|         """
|         return self.workspace_api.delete_workspace(workspace_id=workspace.id, force=True, _request_timeout=timeout or None)
|
|     @intercept_errors(message_prefix="Failed to get workspace: ")
|     def get_current_workspace(self, workspace_id: str) -> Workspace:
|         """Get a Sandbox by its ID.
|
|         Args:
|             workspace_id (str): The ID of the Sandbox to retrieve.
|
|         Returns:
|             Workspace: The Sandbox instance.
|
|         Raises:
|             DaytonaError: If workspace_id is not provided.
|
|         Example:
|             ```python
|             workspace = daytona.get_current_workspace("my-workspace-id")
|             print(workspace.status)
|             ```
|         """
|         if not workspace_id:
|             raise DaytonaError("workspace_id is required")
|
|         # Get the workspace instance
|         workspace_instance = self.workspace_api.get_workspace(
|             workspace_id=workspace_id)
|         workspace_info = Workspace._to_workspace_info(workspace_instance)
|         workspace_instance.info = workspace_info
|
|         # Create and return workspace with Python code toolbox as default
|         code_toolbox = WorkspacePythonCodeToolbox()
|         return Workspace(
|             workspace_id,
|             workspace_instance,
|             self.workspace_api,
|             self.toolbox_api,
|             code_toolbox
|         )
|
|     @intercept_errors(message_prefix="Failed to list workspaces: ")
|     def list(self) -> List[Workspace]:
|         """Lists all Sandboxes.
|
|         Returns:
|             List[Workspace]: List of all available Sandbox instances.
|
|         Example:
|             ```python
|             workspaces = daytona.list()
|             for workspace in workspaces:
|                 print(f"{workspace.id}: {workspace.status}")
|             ```
|         """
|         workspaces = self.workspace_api.list_workspaces()
|
|         for workspace in workspaces:
|             workspace_info = Workspace._to_workspace_info(workspace)
|             workspace.info = workspace_info
|
|         return [
|             Workspace(
|                 workspace.id,
|                 workspace,
|                 self.workspace_api,
|                 self.toolbox_api,
|                 self._get_code_toolbox(
|                     CreateWorkspaceParams(
|                         language=self._validate_language_label(
|                             workspace.labels.get("code-toolbox-language"))
|                     )
|                 )
|             )
|             for workspace in workspaces
|         ]
|
|     def _validate_language_label(self, language: Optional[str]) -> CodeLanguage:
|         """Validates and normalizes the language label.
|
|         Args:
|             language (Optional[str]): The language label to validate.
|
|         Returns:
|             CodeLanguage: The validated language, defaults to "python" if None
|
|         Raises:
|             DaytonaError: If the language is not supported.
|         """
|         if not language:
|             return CodeLanguage.PYTHON
|
|         enum_language = to_enum(CodeLanguage, language)
|         if enum_language is None:
|             raise DaytonaError(f"Invalid code-toolbox-language: {language}")
|         else:
|             return enum_language
|
|     # def resize(self, workspace: Workspace, resources: WorkspaceResources) -> None:
|     #     """Resizes a workspace.
|
|     #     Args:
|     #         workspace: The workspace to resize
|     #         resources: The new resources to set
|     #     """
|     #     self.workspace_api. (workspace_id=workspace.id, resources=resources)
|
|     def start(self, workspace: Workspace, timeout: Optional[float] = 60) -> None:
|         """Starts a Sandbox and waits for it to be ready.
|
|         Args:
|             workspace (Workspace): The Sandbox to start.
|             timeout (Optional[float]): Optional timeout in seconds to wait for the Sandbox to start. 0 means no timeout. Default is 60 seconds.
|
|         Raises:
|             DaytonaError: If timeout is negative; If Sandbox fails to start or times out
|         """
|         workspace.start(timeout)
|
|     def stop(self, workspace: Workspace, timeout: Optional[float] = 60) -> None:
|         """Stops a Sandbox and waits for it to be stopped.
|
|         Args:
|             workspace (Workspace): The workspace to stop
|             timeout (Optional[float]): Optional timeout (in seconds) for workspace stop. 0 means no timeout. Default is 60 seconds.
|
|         Raises:
|             DaytonaError: If timeout is negative; If Sandbox fails to stop or times out
|         """
|         workspace.stop(timeout)
|
|
| # Export these at module level
| __all__ = [
|     "Daytona",
|     "DaytonaConfig",
|     "CreateWorkspaceParams",
|     "CodeLanguage",
|     "Workspace",
|     "SessionExecuteRequest",
|     "SessionExecuteResponse"
| ]
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/filesystem.py:
--------------------------------------------------------------------------------
| """
| The Daytona SDK provides comprehensive file system operations through the `fs` module in Sandboxes.
| You can perform various operations like listing files, creating directories, reading and writing files, and more.
| This guide covers all available file system operations and best practices.
|
| Examples:
|     Basic file operations:
|     ```python
|     workspace = daytona.create()
|
|     # Create a directory
|     workspace.fs.create_folder("/workspace/data", "755")
|
|     # Upload a file
|     with open("local_file.txt", "rb") as f:
|         content = f.read()
|     workspace.fs.upload_file("/workspace/data/file.txt", content)
|
|     # List directory contents
|     files = workspace.fs.list_files("/workspace")
|     for file in files:
|         print(f"Name: {file.name}")
|         print(f"Is directory: {file.is_dir}")
|         print(f"Size: {file.size}")
|         print(f"Modified: {file.mod_time}")
|
|     # Search file contents
|     matches = workspace.fs.find_files(
|         path="/workspace/src",
|         pattern="text-of-interest"
|     )
|     for match in matches:
|         print(f"Absolute file path: {match.file}")
|         print(f"Line number: {match.line}")
|         print(f"Line content: {match.content}")
|         print("\n")
|     ```
|
|     File manipulation:
|     ```python
|     # Move files
|     workspace.fs.move_files(
|         "/workspace/data/old.txt",
|         "/workspace/data/new.txt"
|     )
|
|     # Replace text in files
|     results = workspace.fs.replace_in_files(
|         files=["/workspace/data/new.txt"],
|         pattern="old_version",
|         new_value="new_version"
|     )
|
|     # Set permissions
|     workspace.fs.set_file_permissions(
|         path="/workspace/data/script.sh",
|         mode="755",
|         owner="daytona"
|     )
|     ```
|
| Note:
|     All paths should be absolute paths within the Sandbox if not explicitly
|     stated otherwise.
| """
|
| from typing import List
| from daytona_api_client import (
|     FileInfo,
|     Match,
|     ReplaceRequest,
|     ReplaceResult,
|     SearchFilesResponse,
|     ToolboxApi,
| )
| from daytona_sdk._utils.errors import intercept_errors
| from .protocols import WorkspaceInstance
|
|
| class FileSystem:
|     """Provides file system operations within a Sandbox.
|
|     This class implements a high-level interface to file system operations that can
|     be performed within a Daytona Sandbox. It supports common operations like
|     creating, deleting, and moving files, as well as searching file contents and
|     managing permissions.
|
|     Attributes:
|         instance (WorkspaceInstance): The Sandbox instance this file system belongs to.
|     """
|
|     def __init__(self, instance: WorkspaceInstance, toolbox_api: ToolboxApi):
|         """Initializes a new FileSystem instance.
|
|         Args:
|             instance (WorkspaceInstance): The Sandbox instance this file system belongs to.
|             toolbox_api (ToolboxApi): API client for Sandbox operations.
|         """
|         self.instance = instance
|         self.toolbox_api = toolbox_api
|
|     @intercept_errors(message_prefix="Failed to create folder: ")
|     def create_folder(self, path: str, mode: str) -> None:
|         """Creates a new directory in the Sandbox.
|
|         This method creates a new directory at the specified path with the given
|         permissions.
|
|         Args:
|             path (str): Absolute path where the folder should be created.
|             mode (str): Folder permissions in octal format (e.g., "755" for rwxr-xr-x).
|
|         Example:
|             ```python
|             # Create a directory with standard permissions
|             workspace.fs.create_folder("/workspace/data", "755")
|
|             # Create a private directory
|             workspace.fs.create_folder("/workspace/secrets", "700")
|             ```
|         """
|         self.toolbox_api.create_folder(
|             workspace_id=self.instance.id, path=path, mode=mode
|         )
|
|     @intercept_errors(message_prefix="Failed to delete file: ")
|     def delete_file(self, path: str) -> None:
|         """Deletes a file from the Sandbox.
|
|         This method permanently deletes a file from the Sandbox.
|
|         Args:
|             path (str): Absolute path to the file to delete.
|
|         Example:
|             ```python
|             # Delete a file
|             workspace.fs.delete_file("/workspace/data/old_file.txt")
|             ```
|         """
|         self.toolbox_api.delete_file(
|             workspace_id=self.instance.id, path=path
|         )
|
|     @intercept_errors(message_prefix="Failed to download file: ")
|     def download_file(self, path: str) -> bytes:
|         """Downloads a file from the Sandbox.
|
|         This method retrieves the contents of a file from the Sandbox.
|
|         Args:
|             path (str): Absolute path to the file to download.
|
|         Returns:
|             bytes: The file contents as a bytes object.
|
|         Example:
|             ```python
|             # Download and save a file locally
|             content = workspace.fs.download_file("/workspace/data/file.txt")
|             with open("local_copy.txt", "wb") as f:
|                 f.write(content)
|
|             # Download and process text content
|             content = workspace.fs.download_file("/workspace/data/config.json")
|             config = json.loads(content.decode('utf-8'))
|             ```
|         """
|         return self.toolbox_api.download_file(
|             workspace_id=self.instance.id, path=path
|         )
|
|     @intercept_errors(message_prefix="Failed to find files: ")
|     def find_files(self, path: str, pattern: str) -> List[Match]:
|         """Searches for files containing a pattern.
|
|         This method searches file contents for a specified pattern, similar to
|         the grep command.
|
|         Args:
|             path (str): Absolute path to the file or directory to search. If the path is a directory, the search will be performed recursively.
|             pattern (str): Search pattern to match against file contents.
|
|         Returns:
|             List[Match]: List of matches found in files. Each Match object includes:
|                 - file: Path to the file containing the match
|                 - line: The line number where the match was found
|                 - content: The matching line content
|
|         Example:
|             ```python
|             # Search for TODOs in Python files
|             matches = workspace.fs.find_files("/workspace/src", "TODO:")
|             for match in matches:
|                 print(f"{match.file}:{match.line}: {match.content.strip()}")
|             ```
|         """
|         return self.toolbox_api.find_in_files(
|             workspace_id=self.instance.id, path=path, pattern=pattern
|         )
|
|     @intercept_errors(message_prefix="Failed to get file info: ")
|     def get_file_info(self, path: str) -> FileInfo:
|         """Gets detailed information about a file.
|
|         This method retrieves metadata about a file or directory, including its
|         size, permissions, and timestamps.
|
|         Args:
|             path (str): Absolute path to the file or directory.
|
|         Returns:
|             FileInfo: Detailed file information including:
|                 - name: File name
|                 - is_dir: Whether the path is a directory
|                 - size: File size in bytes
|                 - mode: File permissions
|                 - mod_time: Last modification timestamp
|                 - permissions: File permissions in octal format
|                 - owner: File owner
|                 - group: File group
|
|         Example:
|             ```python
|             # Get file metadata
|             info = workspace.fs.get_file_info("/workspace/data/file.txt")
|             print(f"Size: {info.size} bytes")
|             print(f"Modified: {info.mod_time}")
|             print(f"Mode: {info.mode}")
|
|             # Check if path is a directory
|             info = workspace.fs.get_file_info("/workspace/data")
|             if info.is_dir:
|                 print("Path is a directory")
|             ```
|         """
|         return self.toolbox_api.get_file_info(
|             workspace_id=self.instance.id, path=path
|         )
|
|     @intercept_errors(message_prefix="Failed to list files: ")
|     def list_files(self, path: str) -> List[FileInfo]:
|         """Lists files and directories in a given path.
|
|         This method returns information about all files and directories in the
|         specified directory, similar to the ls -l command.
|
|         Args:
|             path (str): Absolute path to the directory to list contents from.
|
|         Returns:
|             List[FileInfo]: List of file and directory information. Each FileInfo
|                 object includes the same fields as described in get_file_info().
|
|         Example:
|             ```python
|             # List directory contents
|             files = workspace.fs.list_files("/workspace/data")
|
|             # Print files and their sizes
|             for file in files:
|                 if not file.is_dir:
|                     print(f"{file.name}: {file.size} bytes")
|
|             # List only directories
|             dirs = [f for f in files if f.is_dir]
|             print("Subdirectories:", ", ".join(d.name for d in dirs))
|             ```
|         """
|         return self.toolbox_api.list_files(
|             workspace_id=self.instance.id, path=path
|         )
|
|     @intercept_errors(message_prefix="Failed to move files: ")
|     def move_files(self, source: str, destination: str) -> None:
|         """Moves files from one location to another.
|
|         This method moves or renames a file or directory. The parent directory
|         of the destination must exist.
|
|         Args:
|             source (str): Absolute path to the source file or directory.
|             destination (str): Absolute path to the destination.
|
|         Example:
|             ```python
|             # Rename a file
|             workspace.fs.move_files(
|                 "/workspace/data/old_name.txt",
|                 "/workspace/data/new_name.txt"
|             )
|
|             # Move a file to a different directory
|             workspace.fs.move_files(
|                 "/workspace/data/file.txt",
|                 "/workspace/archive/file.txt"
|             )
|
|             # Move a directory
|             workspace.fs.move_files(
|                 "/workspace/old_dir",
|                 "/workspace/new_dir"
|             )
|             ```
|         """
|         self.toolbox_api.move_file(
|             workspace_id=self.instance.id,
|             source=source,
|             destination=destination,
|         )
|
|     @intercept_errors(message_prefix="Failed to replace in files: ")
|     def replace_in_files(
|         self, files: List[str], pattern: str, new_value: str
|     ) -> List[ReplaceResult]:
|         """Replaces text in multiple files.
|
|         This method performs search and replace operations across multiple files.
|
|         Args:
|             files (List[str]): List of absolute file paths to perform replacements in.
|             pattern (str): Pattern to search for.
|             new_value (str): Text to replace matches with.
|
|         Returns:
|             List[ReplaceResult]: List of results indicating replacements made in
|                 each file. Each ReplaceResult includes:
|                 - file: Path to the modified file
|                 - success: Whether the operation was successful
|                 - error: Error message if the operation failed
|
|         Example:
|             ```python
|             # Replace in specific files
|             results = workspace.fs.replace_in_files(
|                 files=["/workspace/src/file1.py", "/workspace/src/file2.py"],
|                 pattern="old_function",
|                 new_value="new_function"
|             )
|
|             # Print results
|             for result in results:
|                 if result.success:
|                     print(f"{result.file}: {result.success}")
|                 else:
|                     print(f"{result.file}: {result.error}")
|             ```
|         """
|         replace_request = ReplaceRequest(
|             files=files, new_value=new_value, pattern=pattern
|         )
|
|         return self.toolbox_api.replace_in_files(
|             workspace_id=self.instance.id, replace_request=replace_request
|         )
|
|     @intercept_errors(message_prefix="Failed to search files: ")
|     def search_files(self, path: str, pattern: str) -> SearchFilesResponse:
|         """Searches for files and directories matching a pattern in their names.
|
|         This method searches for files and directories whose names match the
|         specified pattern. The pattern can be a simple string or a glob pattern.
|
|         Args:
|             path (str): Absolute path to the root directory to start search from.
|             pattern (str): Pattern to match against file names. Supports glob
|                 patterns (e.g., "*.py" for Python files).
|
|         Returns:
|             SearchFilesResponse: Search results containing:
|                 - files: List of matching file and directory paths
|
|         Example:
|             ```python
|             # Find all Python files
|             result = workspace.fs.search_files("/workspace", "*.py")
|             for file in result.files:
|                 print(file)
|
|             # Find files with specific prefix
|             result = workspace.fs.search_files("/workspace/data", "test_*")
|             print(f"Found {len(result.files)} test files")
|             ```
|         """
|         return self.toolbox_api.search_files(
|             workspace_id=self.instance.id, path=path, pattern=pattern
|         )
|
|     @intercept_errors(message_prefix="Failed to set file permissions: ")
|     def set_file_permissions(
|         self, path: str, mode: str = None, owner: str = None, group: str = None
|     ) -> None:
|         """Sets permissions and ownership for a file or directory.
|
|         This method allows changing the permissions and ownership of a file or
|         directory. Any of the parameters can be None to leave that attribute
|         unchanged.
|
|         Args:
|             path (str): Absolute path to the file or directory.
|             mode (Optional[str]): File mode/permissions in octal format
|                 (e.g., "644" for rw-r--r--).
|             owner (Optional[str]): User owner of the file.
|             group (Optional[str]): Group owner of the file.
|
|         Example:
|             ```python
|             # Make a file executable
|             workspace.fs.set_file_permissions(
|                 path="/workspace/scripts/run.sh",
|                 mode="755"  # rwxr-xr-x
|             )
|
|             # Change file owner
|             workspace.fs.set_file_permissions(
|                 path="/workspace/data/file.txt",
|                 owner="daytona",
|                 group="daytona"
|             )
|             ```
|         """
|         self.toolbox_api.set_file_permissions(
|             workspace_id=self.instance.id,
|             path=path,
|             mode=mode,
|             owner=owner,
|             group=group,
|         )
|
|     @intercept_errors(message_prefix="Failed to upload file: ")
|     def upload_file(self, path: str, file: bytes) -> None:
|         """Uploads a file to the Sandbox.
|
|         This method uploads a file to the specified path in the Sandbox. The
|         parent directory must exist. If a file already exists at the destination
|         path, it will be overwritten.
|
|         Args:
|             path (str): Absolute destination path in the Sandbox.
|             file (bytes): File contents as a bytes object.
|
|         Example:
|             ```python
|             # Upload a text file
|             content = b"Hello, World!"
|             workspace.fs.upload_file("/workspace/data/hello.txt", content)
|
|             # Upload a local file
|             with open("local_file.txt", "rb") as f:
|                 content = f.read()
|             workspace.fs.upload_file("/workspace/data/file.txt", content)
|
|             # Upload binary data
|             import json
|             data = {"key": "value"}
|             content = json.dumps(data).encode('utf-8')
|             workspace.fs.upload_file("/workspace/data/config.json", content)
|             ```
|         """
|         self.toolbox_api.upload_file(
|             workspace_id=self.instance.id, path=path, file=file
|         )
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/git.py:
--------------------------------------------------------------------------------
| """
| The Daytona SDK provides built-in Git support. This guide covers all available Git
| operations and best practices. Daytona SDK provides an option to clone, check status,
| and manage Git repositories in Sandboxes. You can interact with Git repositories using
| the `git` module.
|
| Example:
|     Basic Git workflow:
|     ```python
|     workspace = daytona.create()
|
|     # Clone a repository
|     workspace.git.clone(
|         url="https://github.com/user/repo.git",
|         path="/workspace/repo"
|     )
|
|     # Make some changes
|     workspace.fs.upload_file("/workspace/repo/test.txt", "Hello, World!")
|
|     # Stage and commit changes
|     workspace.git.add("/workspace/repo", ["test.txt"])
|     workspace.git.commit(
|         path="/workspace/repo",
|         message="Add test file",
|         author="John Doe",
|         email="john@example.com"
|     )
|
|     # Push changes (with authentication)
|     workspace.git.push(
|         path="/workspace/repo",
|         username="user",
|         password="token"
|     )
|     ```
|
| Note:
|     All paths should be absolute paths within the Sandbox if not explicitly
|     stated otherwise.
| """
|
| from typing import List, Optional, TYPE_CHECKING
| from daytona_api_client import (
|     GitStatus,
|     ListBranchResponse,
|     ToolboxApi,
|     GitAddRequest,
|     GitCloneRequest,
|     GitCommitRequest,
|     GitRepoRequest,
| )
| from daytona_sdk._utils.errors import intercept_errors
| from .protocols import WorkspaceInstance
|
| if TYPE_CHECKING:
|     from .workspace import Workspace
|
|
| class Git:
|     """Provides Git operations within a Sandbox.
|
|     This class implements a high-level interface to Git operations that can be
|     performed within a Daytona Sandbox. It supports common Git operations like
|     cloning repositories, staging and committing changes, pushing and pulling
|     changes, and checking repository status.
|
|     Attributes:
|         workspace (Workspace): The parent Sandbox instance.
|         instance (WorkspaceInstance): The Sandbox instance this Git handler belongs to.
|
|     Example:
|         ```python
|         # Clone a repository
|         workspace.git.clone(
|             url="https://github.com/user/repo.git",
|             path="/workspace/repo"
|         )
|
|         # Check repository status
|         status = workspace.git.status("/workspace/repo")
|         print(f"Modified files: {status.modified}")
|
|         # Stage and commit changes
|         workspace.git.add("/workspace/repo", ["file.txt"])
|         workspace.git.commit(
|             path="/workspace/repo",
|             message="Update file",
|             author="John Doe",
|             email="john@example.com"
|         )
|         ```
|     """
|
|     def __init__(
|         self,
|         workspace: "Workspace",
|         toolbox_api: ToolboxApi,
|         instance: WorkspaceInstance,
|     ):
|         """Initializes a new Git handler instance.
|
|         Args:
|             workspace (Workspace): The parent Sandbox instance.
|             toolbox_api (ToolboxApi): API client for Sandbox operations.
|             instance (WorkspaceInstance): The Sandbox instance this Git handler belongs to.
|         """
|         self.workspace = workspace
|         self.toolbox_api = toolbox_api
|         self.instance = instance
|
|     @intercept_errors(message_prefix="Failed to add files: ")
|     def add(self, path: str, files: List[str]) -> None:
|         """Stages files for commit.
|
|         This method stages the specified files for the next commit, similar to
|         running 'git add' on the command line.
|
|         Args:
|             path (str): Absolute path to the Git repository root.
|             files (List[str]): List of file paths or directories to stage, relative to the repository root.
|
|         Example:
|             ```python
|             # Stage a single file
|             workspace.git.add("/workspace/repo", ["file.txt"])
|
|             # Stage multiple files
|             workspace.git.add("/workspace/repo", [
|                 "src/main.py",
|                 "tests/test_main.py",
|                 "README.md"
|             ])
|             ```
|         """
|         self.toolbox_api.git_add_files(
|             workspace_id=self.instance.id,
|             git_add_request=GitAddRequest(
|                 path=path,
|                 files=files
|             ),
|         )
|
|     @intercept_errors(message_prefix="Failed to list branches: ")
|     def branches(self, path: str) -> ListBranchResponse:
|         """Lists branches in the repository.
|
|         This method returns information about all branches in the repository.
|
|         Args:
|             path (str): Absolute path to the Git repository root.
|
|         Returns:
|             ListBranchResponse: List of branches in the repository.
|
|         Example:
|             ```python
|             response = workspace.git.branches("/workspace/repo")
|             print(f"Branches: {response.branches}")
|             ```
|         """
|         return self.toolbox_api.git_list_branches(
|             workspace_id=self.instance.id,
|             path=path,
|         )
|
|     @intercept_errors(message_prefix="Failed to clone repository: ")
|     def clone(
|         self,
|         url: str,
|         path: str,
|         branch: Optional[str] = None,
|         commit_id: Optional[str] = None,
|         username: Optional[str] = None,
|         password: Optional[str] = None,
|     ) -> None:
|         """Clones a Git repository.
|
|         This method clones a Git repository into the specified path. It supports
|         cloning specific branches or commits, and can authenticate with the remote
|         repository if credentials are provided.
|
|         Args:
|             url (str): Repository URL to clone from.
|             path (str): Absolute path where the repository should be cloned.
|             branch (Optional[str]): Specific branch to clone. If not specified,
|                 clones the default branch.
|             commit_id (Optional[str]): Specific commit to clone. If specified,
|                 the repository will be left in a detached HEAD state at this commit.
|             username (Optional[str]): Git username for authentication.
|             password (Optional[str]): Git password or token for authentication.
|
|         Example:
|             ```python
|             # Clone the default branch
|             workspace.git.clone(
|                 url="https://github.com/user/repo.git",
|                 path="/workspace/repo"
|             )
|
|             # Clone a specific branch with authentication
|             workspace.git.clone(
|                 url="https://github.com/user/private-repo.git",
|                 path="/workspace/private",
|                 branch="develop",
|                 username="user",
|                 password="token"
|             )
|
|             # Clone a specific commit
|             workspace.git.clone(
|                 url="https://github.com/user/repo.git",
|                 path="/workspace/repo-old",
|                 commit_id="abc123"
|             )
|             ```
|         """
|         self.toolbox_api.git_clone_repository(
|             workspace_id=self.instance.id,
|             git_clone_request=GitCloneRequest(
|                 url=url,
|                 branch=branch,
|                 path=path,
|                 username=username,
|                 password=password,
|                 commitId=commit_id,
|             )
|         )
|
|     @intercept_errors(message_prefix="Failed to commit changes: ")
|     def commit(self, path: str, message: str, author: str, email: str) -> None:
|         """Commits staged changes.
|
|         This method creates a new commit with the staged changes. Make sure to stage
|         changes using the add() method before committing.
|
|         Args:
|             path (str): Absolute path to the Git repository root.
|             message (str): Commit message describing the changes.
|             author (str): Name of the commit author.
|             email (str): Email address of the commit author.
|
|         Example:
|             ```python
|             # Stage and commit changes
|             workspace.git.add("/workspace/repo", ["README.md"])
|             workspace.git.commit(
|                 path="/workspace/repo",
|                 message="Update documentation",
|                 author="John Doe",
|                 email="john@example.com"
|             )
|             ```
|         """
|         self.toolbox_api.git_commit_changes(
|             workspace_id=self.instance.id,
|             git_commit_request=GitCommitRequest(
|                 path=path,
|                 message=message,
|                 author=author,
|                 email=email
|             ),
|         )
|
|     @intercept_errors(message_prefix="Failed to push changes: ")
|     def push(
|         self, path: str, username: Optional[str] = None, password: Optional[str] = None
|     ) -> None:
|         """Pushes local commits to the remote repository.
|
|         This method pushes all local commits on the current branch to the remote
|         repository. If the remote repository requires authentication, provide
|         username and password/token.
|
|         Args:
|             path (str): Absolute path to the Git repository root.
|             username (Optional[str]): Git username for authentication.
|             password (Optional[str]): Git password or token for authentication.
|
|         Example:
|             ```python
|             # Push without authentication (for public repos or SSH)
|             workspace.git.push("/workspace/repo")
|
|             # Push with authentication
|             workspace.git.push(
|                 path="/workspace/repo",
|                 username="user",
|                 password="github_token"
|             )
|             ```
|         """
|         self.toolbox_api.git_push_changes(
|             workspace_id=self.instance.id,
|             git_repo_request=GitRepoRequest(
|                 path=path,
|                 username=username,
|                 password=password
|             ),
|         )
|
|     @intercept_errors(message_prefix="Failed to pull changes: ")
|     def pull(
|         self, path: str, username: Optional[str] = None, password: Optional[str] = None
|     ) -> None:
|         """Pulls changes from the remote repository.
|
|         This method fetches and merges changes from the remote repository into
|         the current branch. If the remote repository requires authentication,
|         provide username and password/token.
|
|         Args:
|             path (str): Absolute path to the Git repository root.
|             username (Optional[str]): Git username for authentication.
|             password (Optional[str]): Git password or token for authentication.
|
|         Example:
|             ```python
|             # Pull without authentication
|             workspace.git.pull("/workspace/repo")
|
|             # Pull with authentication
|             workspace.git.pull(
|                 path="/workspace/repo",
|                 username="user",
|                 password="github_token"
|             )
|             ```
|         """
|         self.toolbox_api.git_pull_changes(
|             workspace_id=self.instance.id,
|             git_repo_request=GitRepoRequest(
|                 path=path,
|                 username=username,
|                 password=password
|             ),
|         )
|
|     @intercept_errors(message_prefix="Failed to get status: ")
|     def status(self, path: str) -> GitStatus:
|         """Gets the current Git repository status.
|
|         This method returns detailed information about the current state of the
|         repository, including staged, unstaged, and untracked files.
|
|         Args:
|             path (str): Absolute path to the Git repository root.
|
|         Returns:
|             GitStatus: Repository status information including:
|                 - current_branch: Current branch name
|                 - file_status: List of file statuses
|                 - ahead: Number of local commits not pushed to remote
|                 - behind: Number of remote commits not pulled locally
|                 - branch_published: Whether the branch has been published to the remote repository
|
|         Example:
|             ```python
|             status = workspace.git.status("/workspace/repo")
|             print(f"On branch: {status.current_branch}")
|             print(f"Commits ahead: {status.ahead}")
|             print(f"Commits behind: {status.behind}")
|             ```
|         """
|         return self.toolbox_api.git_get_status(
|             workspace_id=self.instance.id,
|             path=path,
|         )
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/lsp_server.py:
--------------------------------------------------------------------------------
| """
| The Daytona SDK provides Language Server Protocol (LSP) support through Sandbox instances.
| This enables advanced language features like code completion, diagnostics, and more.
|
| Example:
|     Basic LSP server usage:
|     ```python
|     workspace = daytona.create()
|
|     # Create and start LSP server
|     lsp = workspace.create_lsp_server("typescript", "/workspace/project")
|     lsp.start()
|
|     # Open a file for editing
|     lsp.did_open("/workspace/project/src/index.ts")
|
|     # Get completions at a position
|     pos = Position(line=10, character=15)
|     completions = lsp.completions("/workspace/project/src/index.ts", pos)
|     print(f"Completions: {completions}")
|
|     # Get document symbols
|     symbols = lsp.document_symbols("/workspace/project/src/index.ts")
|     for symbol in symbols:
|         print(f"{symbol.name}: {symbol.kind}")
|
|     # Clean up
|     lsp.did_close("/workspace/project/src/index.ts")
|     lsp.stop()
|     ```
|
| Note:
|     The LSP server must be started with start() before using any other methods,
|     and should be stopped with stop() when no longer needed to free resources.
| """
| from enum import Enum
| from typing import List
| from daytona_api_client import (
|     CompletionList,
|     LspSymbol,
|     ToolboxApi,
|     LspServerRequest,
|     LspDocumentRequest,
|     LspCompletionParams
| )
| from daytona_sdk._utils.errors import intercept_errors
| from .protocols import WorkspaceInstance
|
|
| class LspLanguageId(Enum):
|     PYTHON = "python"
|     TYPESCRIPT = "typescript"
|     JAVASCRIPT = "javascript"
|
|     def __str__(self):
|         return self.value
|
|     def __eq__(self, other):
|         if isinstance(other, str):
|             return self.value == other
|         return super().__eq__(other)
|
|
| class Position:
|     """Represents a position in a text document.
|
|     This class represents a zero-based position within a text document,
|     specified by line number and character offset.
|
|     Attributes:
|         line (int): Zero-based line number in the document.
|         character (int): Zero-based character offset on the line.
|     """
|
|     def __init__(self, line: int, character: int):
|         """Initialize a new Position instance.
|
|         Args:
|             line (int): Zero-based line number in the document.
|             character (int): Zero-based character offset on the line.
|         """
|         self.line = line
|         self.character = character
|
|
| class LspServer:
|     """Provides Language Server Protocol functionality for code intelligence.
|
|     This class implements a subset of the Language Server Protocol (LSP) to provide
|     IDE-like features such as code completion, symbol search, and more.
|
|     Attributes:
|         language_id (LspLanguageId): The language server type (e.g., "python", "typescript").
|         path_to_project (str): Absolute path to the project root directory.
|         instance (WorkspaceInstance): The Sandbox instance this server belongs to.
|     """
|
|     def __init__(
|         self,
|         language_id: LspLanguageId,
|         path_to_project: str,
|         toolbox_api: ToolboxApi,
|         instance: WorkspaceInstance,
|     ):
|         """Initializes a new LSP server instance.
|
|         Args:
|             language_id (LspLanguageId): The language server type (e.g., LspLanguageId.TYPESCRIPT).
|             path_to_project (str): Absolute path to the project root directory.
|             toolbox_api (ToolboxApi): API client for Sandbox operations.
|             instance (WorkspaceInstance): The Sandbox instance this server belongs to.
|         """
|         self.language_id = str(language_id)
|         self.path_to_project = path_to_project
|         self.toolbox_api = toolbox_api
|         self.instance = instance
|
|     @intercept_errors(message_prefix="Failed to start LSP server: ")
|     def start(self) -> None:
|         """Starts the language server.
|
|         This method must be called before using any other LSP functionality.
|         It initializes the language server for the specified language and project.
|
|         Example:
|             ```python
|             lsp = workspace.create_lsp_server("typescript", "/workspace/project")
|             lsp.start()  # Initialize the server
|             # Now ready for LSP operations
|             ```
|         """
|         self.toolbox_api.lsp_start(
|             workspace_id=self.instance.id,
|             lsp_server_request=LspServerRequest(
|                 language_id=self.language_id,
|                 path_to_project=self.path_to_project,
|             ),
|         )
|
|     @intercept_errors(message_prefix="Failed to stop LSP server: ")
|     def stop(self) -> None:
|         """Stops the language server.
|
|         This method should be called when the LSP server is no longer needed to
|         free up system resources.
|
|         Example:
|             ```python
|             # When done with LSP features
|             lsp.stop()  # Clean up resources
|             ```
|         """
|         self.toolbox_api.lsp_stop(
|             workspace_id=self.instance.id,
|             lsp_server_request=LspServerRequest(
|                 language_id=self.language_id,
|                 path_to_project=self.path_to_project,
|             ),
|         )
|
|     @intercept_errors(message_prefix="Failed to open file: ")
|     def did_open(self, path: str) -> None:
|         """Notifies the language server that a file has been opened.
|
|         This method should be called when a file is opened in the editor to enable
|         language features like diagnostics and completions for that file. The server
|         will begin tracking the file's contents and providing language features.
|
|         Args:
|             path (str): Absolute path to the opened file.
|
|         Example:
|             ```python
|             # When opening a file for editing
|             lsp.did_open("/workspace/project/src/index.ts")
|             # Now can get completions, symbols, etc. for this file
|             ```
|         """
|         self.toolbox_api.lsp_did_open(
|             workspace_id=self.instance.id,
|             lsp_document_request=LspDocumentRequest(
|                 language_id=self.language_id,
|                 path_to_project=self.path_to_project,
|                 uri=f"file://{path}",
|             ),
|         )
|
|     @intercept_errors(message_prefix="Failed to close file: ")
|     def did_close(self, path: str) -> None:
|         """Notify the language server that a file has been closed.
|
|         This method should be called when a file is closed in the editor to allow
|         the language server to clean up any resources associated with that file.
|         Args:
|             path (str): Absolute path to the closed file.
|
|         Example:
|             ```python
|             # When done editing a file
|             lsp.did_close("/workspace/project/src/index.ts")
|             ```
|         """
|         self.toolbox_api.lsp_did_close(
|             workspace_id=self.instance.id,
|             lsp_document_request=LspDocumentRequest(
|                 language_id=self.language_id,
|                 path_to_project=self.path_to_project,
|                 uri=f"file://{path}",
|             ),
|         )
|
|     @intercept_errors(message_prefix="Failed to get symbols from document: ")
|     def document_symbols(self, path: str) -> List[LspSymbol]:
|         """Gets symbol information from a document.
|
|         This method returns information about all symbols (functions, classes,
|         variables, etc.) defined in the specified document.
|
|         Args:
|             path (str): Absolute path to the file to get symbols from.
|
|         Returns:
|             List[LspSymbol]: List of symbols in the document. Each symbol includes:
|                 - name: The symbol's name
|                 - kind: The symbol's kind (function, class, variable, etc.)
|                 - location: The location of the symbol in the file
|
|         Example:
|             ```python
|             # Get all symbols in a file
|             symbols = lsp.document_symbols("/workspace/project/src/index.ts")
|             for symbol in symbols:
|                 print(f"{symbol.kind} {symbol.name}: {symbol.location}")
|             ```
|         """
|         return self.toolbox_api.lsp_document_symbols(
|             workspace_id=self.instance.id,
|             language_id=self.language_id,
|             path_to_project=self.path_to_project,
|             uri=f"file://{path}",
|         )
|
|     @intercept_errors(message_prefix="Failed to get symbols from workspace: ")
|     def workspace_symbols(self, query: str) -> List[LspSymbol]:
|         """Searches for symbols across the entire Sandbox.
|
|         This method searches for symbols matching the query string across all files
|         in the Sandbox. It's useful for finding declarations and definitions
|         without knowing which file they're in.
|
|         Args:
|             query (str): Search query to match against symbol names.
|
|         Returns:
|             List[LspSymbol]: List of matching symbols from all files. Each symbol
|                 includes:
|                 - name: The symbol's name
|                 - kind: The symbol's kind (function, class, variable, etc.)
|                 - location: The location of the symbol in the file
|
|         Example:
|             ```python
|             # Search for all symbols containing "User"
|             symbols = lsp.workspace_symbols("User")
|             for symbol in symbols:
|                 print(f"{symbol.name} in {symbol.location}")
|             ```
|         """
|         return self.toolbox_api.lsp_workspace_symbols(
|             workspace_id=self.instance.id,
|             language_id=self.language_id,
|             path_to_project=self.path_to_project,
|             query=query,
|         )
|
|     @intercept_errors(message_prefix="Failed to get completions: ")
|     def completions(self, path: str, position: Position) -> CompletionList:
|         """Gets completion suggestions at a position in a file.
|
|         Args:
|             path (str): Absolute path to the file.
|             position (Position): Cursor position to get completions for.
|
|         Returns:
|             CompletionList: List of completion suggestions. The list includes:
|                 - isIncomplete: Whether more items might be available
|                 - items: List of completion items, each containing:
|                     - label: The text to insert
|                     - kind: The kind of completion
|                     - detail: Additional details about the item
|                     - documentation: Documentation for the item
|                     - sortText: Text used to sort the item in the list
|                     - filterText: Text used to filter the item
|                     - insertText: The actual text to insert (if different from label)
|
|         Example:
|             ```python
|             # Get completions at a specific position
|             pos = Position(line=10, character=15)
|             completions = lsp.completions("/workspace/project/src/index.ts", pos)
|             for item in completions.items:
|                 print(f"{item.label} ({item.kind}): {item.detail}")
|             ```
|         """
|         return self.toolbox_api.lsp_completions(
|             workspace_id=self.instance.id,
|             lsp_completion_params=LspCompletionParams(
|                 language_id=self.language_id,
|                 path_to_project=self.path_to_project,
|                 uri=f"file://{path}",
|                 position=position,
|             ),
|         )
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/process.py:
--------------------------------------------------------------------------------
| """
| The Daytona SDK provides powerful process and code execution capabilities through
| the `process` module in Sandboxes. This guide covers all available process operations
| and best practices.
|
| Example:
|     Basic command execution:
|     ```python
|     workspace = daytona.create()
|
|     # Execute a shell command
|     response = workspace.process.exec("ls -la")
|     print(response.result)
|
|     # Run Python code
|     response = workspace.process.code_run("print('Hello, World!')")
|     print(response.result)
|     ```
|
|     Using interactive sessions:
|     ```python
|     # Create a new session
|     session_id = "my-session"
|     workspace.process.create_session(session_id)
|
|     # Execute commands in the session
|     req = SessionExecuteRequest(command="cd /workspace", var_async=False)
|     workspace.process.execute_session_command(session_id, req)
|
|     req = SessionExecuteRequest(command="pwd", var_async=False)
|     response = workspace.process.execute_session_command(session_id, req)
|     print(response.result)  # Should print "/workspace"
|
|     # Clean up
|     workspace.process.delete_session(session_id)
|     ```
| """
|
| from typing import Optional, List, Dict
| from daytona_api_client import (
|     ToolboxApi,
|     ExecuteResponse,
|     ExecuteRequest,
|     Session,
|     SessionExecuteRequest,
|     SessionExecuteResponse,
|     CreateSessionRequest,
|     Command
| )
|
| from daytona_sdk._utils.errors import intercept_errors
| from .code_toolbox.workspace_python_code_toolbox import WorkspacePythonCodeToolbox
| from .protocols import WorkspaceInstance
| from .common.code_run_params import CodeRunParams
|
|
| class Process:
|     """Handles process and code execution within a Sandbox.
|
|     This class provides methods for executing shell commands and running code in
|     the Sandbox environment.
|
|     Attributes:
|         code_toolbox (WorkspacePythonCodeToolbox): Language-specific code execution toolbox.
|         toolbox_api (ToolboxApi): API client for Sandbox operations.
|         instance (WorkspaceInstance): The Sandbox instance this process belongs to.
|     """
|
|     def __init__(
|         self,
|         code_toolbox: WorkspacePythonCodeToolbox,
|         toolbox_api: ToolboxApi,
|         instance: WorkspaceInstance,
|     ):
|         """Initialize a new Process instance.
|
|         Args:
|             code_toolbox (WorkspacePythonCodeToolbox): Language-specific code execution toolbox.
|             toolbox_api (ToolboxApi): API client for Sandbox operations.
|             instance (WorkspaceInstance): The Sandbox instance this process belongs to.
|         """
|         self.code_toolbox = code_toolbox
|         self.toolbox_api = toolbox_api
|         self.instance = instance
|
|     @intercept_errors(message_prefix="Failed to execute command: ")
|     def exec(self, command: str, cwd: Optional[str] = None, timeout: Optional[int] = None) -> ExecuteResponse:
|         """Execute a shell command in the Sandbox.
|
|         Args:
|             command (str): Shell command to execute.
|             cwd (Optional[str]): Working directory for command execution. If not
|                 specified, uses the Sandbox root directory.
|             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
|
|         Example:
|             ```python
|             # Simple command
|             response = workspace.process.exec("echo 'Hello'")
|             print(response.result)  # Prints: Hello
|
|             # Command with working directory
|             result = workspace.process.exec("ls", cwd="/workspace/src")
|
|             # Command with timeout
|             result = workspace.process.exec("sleep 10", timeout=5)
|             ```
|         """
|         execute_request = ExecuteRequest(
|             command=command,
|             cwd=cwd,
|             timeout=timeout
|         )
|
|         return self.toolbox_api.execute_command(
|             workspace_id=self.instance.id,
|             execute_request=execute_request
|         )
|
|     def code_run(self, code: str, params: Optional[CodeRunParams] = None, timeout: Optional[int] = None) -> ExecuteResponse:
|         """Executes code in the Sandbox using the appropriate language runtime.
|
|         Args:
|             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
|
|         Example:
|             ```python
|             # Run Python code
|             response = workspace.process.code_run('''
|                 x = 10
|                 y = 20
|                 print(f"Sum: {x + y}")
|             ''')
|             print(response.result)  # Prints: Sum: 30
|             ```
|         """
|         command = self.code_toolbox.get_run_command(code, params)
|         return self.exec(command, timeout=timeout)
|
|     @intercept_errors(message_prefix="Failed to create session: ")
|     def create_session(self, session_id: str) -> None:
|         """Create 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.
|
|         Args:
|             session_id (str): Unique identifier for the new session.
|
|         Example:
|             ```python
|             # Create a new session
|             session_id = "my-session"
|             workspace.process.create_session(session_id)
|             session = workspace.process.get_session(session_id)
|             # Do work...
|             workspace.process.delete_session(session_id)
|             ```
|         """
|         request = CreateSessionRequest(sessionId=session_id)
|         self.toolbox_api.create_session(
|             workspace_id=self.instance.id,
|             create_session_request=request
|         )
|
|     @intercept_errors(message_prefix="Failed to get session: ")
|     def get_session(self, session_id: str) -> Session:
|         """Get a session in the Sandbox.
|
|         Args:
|             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:
|             ```python
|             session = workspace.process.get_session("my-session")
|             for cmd in session.commands:
|                 print(f"Command: {cmd.command}")
|             ```
|         """
|         return self.toolbox_api.get_session(
|             workspace_id=self.instance.id,
|             session_id=session_id
|         )
|
|     @intercept_errors(message_prefix="Failed to get session command: ")
|     def get_session_command(self, session_id: str, command_id: str) -> Command:
|         """Get information about a specific command executed in a session.
|
|         Args:
|             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:
|             ```python
|             cmd = workspace.process.get_session_command("my-session", "cmd-123")
|             if cmd.exit_code == 0:
|                 print(f"Command {cmd.command} completed successfully")
|             ```
|         """
|         return self.toolbox_api.get_session_command(
|             workspace_id=self.instance.id,
|             session_id=session_id,
|             command_id=command_id
|         )
|
|     @intercept_errors(message_prefix="Failed to execute session command: ")
|     def execute_session_command(self, session_id: str, req: SessionExecuteRequest, timeout: Optional[int] = None) -> SessionExecuteResponse:
|         """Executes a command in the session.
|
|         Args:
|             session_id (str): Unique identifier of the session to use.
|             req (SessionExecuteRequest): Command execution request containing:
|                 - command: The command to execute
|                 - var_async: Whether to execute asynchronously
|
|         Returns:
|             SessionExecuteResponse: Command execution results containing:
|                 - cmd_id: Unique identifier for the executed command
|                 - output: Command output (if synchronous execution)
|                 - exit_code: Command exit status (if synchronous execution)
|
|         Example:
|             ```python
|             # Execute commands in sequence, maintaining state
|             session_id = "my-session"
|
|             # Change directory
|             req = SessionExecuteRequest(command="cd /workspace")
|             workspace.process.execute_session_command(session_id, req)
|
|             # Create a file
|             req = SessionExecuteRequest(command="echo 'Hello' > test.txt")
|             workspace.process.execute_session_command(session_id, req)
|
|             # Read the file
|             req = SessionExecuteRequest(command="cat test.txt")
|             result = workspace.process.execute_session_command(session_id, req)
|             print(result.output)  # Prints: Hello
|             ```
|         """
|         return self.toolbox_api.execute_session_command(
|             workspace_id=self.instance.id,
|             session_id=session_id,
|             session_execute_request=req,
|             _request_timeout=timeout or None
|         )
|
|     @intercept_errors(message_prefix="Failed to get session command logs: ")
|     def get_session_command_logs(self, session_id: str, command_id: str) -> str:
|         """Get the logs for a command executed in a session.
|
|         This method retrieves the complete output (stdout and stderr) from a
|         command executed in a session. It's particularly useful for checking
|         the output of asynchronous commands.
|
|         Args:
|             session_id (str): Unique identifier of the session.
|             command_id (str): Unique identifier of the command.
|
|         Returns:
|             str: Complete command output including both stdout and stderr.
|
|         Example:
|             ```python
|             # Execute a long-running command asynchronously
|             req = SessionExecuteRequest(
|                 command="sleep 5; echo 'Done'",
|                 var_async=True
|             )
|             response = workspace.process.execute_session_command("my-session", req)
|
|             # Wait a bit, then get the logs
|             import time
|             time.sleep(6)
|             logs = workspace.process.get_session_command_logs(
|                 "my-session",
|                 response.command_id
|             )
|             print(logs)  # Prints: Done
|             ```
|         """
|         return self.toolbox_api.get_session_command_logs(
|             workspace_id=self.instance.id,
|             session_id=session_id,
|             command_id=command_id
|         )
|
|     @intercept_errors(message_prefix="Failed to list sessions: ")
|     def list_sessions(self) -> List[Session]:
|         """List all sessions in the Sandbox.
|
|         Returns:
|             List[Session]: List of all sessions in the Sandbox.
|
|         Example:
|             ```python
|             sessions = workspace.process.list_sessions()
|             for session in sessions:
|                 print(f"Session {session.session_id}:")
|                 print(f"  Commands: {len(session.commands)}")
|             ```
|         """
|         return self.toolbox_api.list_sessions(
|             workspace_id=self.instance.id
|         )
|
|     @intercept_errors(message_prefix="Failed to delete session: ")
|     def delete_session(self, session_id: str) -> None:
|         """Delete an interactive session from the Sandbox.
|
|         This method terminates and removes a session, cleaning up any resources
|         associated with it.
|
|         Args:
|             session_id (str): Unique identifier of the session to delete.
|
|         Example:
|             ```python
|             # Create and use a session
|             workspace.process.create_session("temp-session")
|             # ... use the session ...
|
|             # Clean up when done
|             workspace.process.delete_session("temp-session")
|             ```
|         """
|         self.toolbox_api.delete_session(
|             workspace_id=self.instance.id,
|             session_id=session_id
|         )
|


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/protocols.py:
--------------------------------------------------------------------------------
| from typing import Protocol, Dict, Any
|
| class WorkspaceCodeToolbox(Protocol):
|     def get_default_image(self) -> str: ...
|     def get_code_run_command(self, code: str) -> str: ...
|     def get_code_run_args(self) -> list[str]: ...
|     # ... other protocol methods
|
| class WorkspaceInstance(Protocol):
|     id: str


--------------------------------------------------------------------------------
/packages/python/src/daytona_sdk/workspace.py:
--------------------------------------------------------------------------------
| """
| The Daytona SDK core Sandbox functionality.
|
| Provides the main Workspace class representing a Daytona Sandbox that coordinates file system,
| Git, process execution, and LSP functionality. It serves as the central point
| for interacting with Daytona Sandboxes.
|
| Example:
|     Basic Sandbox operations:
|     ```python
|     from daytona_sdk import Daytona
|     daytona = Daytona()
|     workspace = daytona.create()
|
|     # File operations
|     workspace.fs.upload_file("/workspace/test.txt", b"Hello, World!")
|     content = workspace.fs.download_file("/workspace/test.txt")
|
|     # Git operations
|     workspace.git.clone("https://github.com/user/repo.git")
|
|     # Process execution
|     response = workspace.process.exec("ls -la")
|     print(response.result)
|
|     # LSP functionality
|     lsp = workspace.create_lsp_server("python", "/workspace/project")
|     lsp.did_open("/workspace/project/src/index.ts")
|     completions = lsp.completions("/workspace/project/src/index.ts", Position(line=10, character=15))
|     print(completions)
|     ```
|
| Note:
|     The Sandbox must be in a 'started' state before performing operations.
| """
|
| import json
| import time
| from typing import Dict, Optional
| from daytona_sdk._utils.errors import intercept_errors
| from .filesystem import FileSystem
| from .git import Git
| from .process import Process
| from .lsp_server import LspServer, LspLanguageId
| from daytona_api_client import Workspace as ApiWorkspace, ToolboxApi, WorkspaceApi, WorkspaceInfo as ApiWorkspaceInfo
| from .protocols import WorkspaceCodeToolbox
| from dataclasses import dataclass
| from datetime import datetime
| from daytona_sdk._utils.errors import DaytonaError
| from enum import Enum
| from pydantic import Field
| from typing_extensions import Annotated
| from ._utils.enum import to_enum
| from ._utils.timeout import with_timeout
|
|
| @dataclass
| class WorkspaceTargetRegion(Enum):
|     """Target regions for workspaces"""
|     EU = "eu"
|     US = "us"
|     ASIA = "asia"
|
|     def __str__(self):
|         return self.value
|
|     def __eq__(self, other):
|         if isinstance(other, str):
|             return self.value == other
|         return super().__eq__(other)
|
|
| @dataclass
| class WorkspaceResources:
|     """Resources allocated to a Sandbox.
|
|     Attributes:
|         cpu (str): Number of CPU cores allocated (e.g., "1", "2").
|         gpu (Optional[str]): Number of GPUs allocated (e.g., "1") or None if no GPU.
|         memory (str): Amount of memory allocated with unit (e.g., "2Gi", "4Gi").
|         disk (str): Amount of disk space allocated with unit (e.g., "10Gi", "20Gi").
|
|     Example:
|         ```python
|         resources = WorkspaceResources(
|             cpu="2",
|             gpu="1",
|             memory="4Gi",
|             disk="20Gi"
|         )
|         ```
|     """
|     cpu: str
|     gpu: Optional[str]
|     memory: str
|     disk: str
|
|
| @dataclass
| class WorkspaceState(Enum):
|     """States of a Sandbox."""
|     CREATING = "creating"
|     RESTORING = "restoring"
|     DESTROYED = "destroyed"
|     DESTROYING = "destroying"
|     STARTED = "started"
|     STOPPED = "stopped"
|     STARTING = "starting"
|     STOPPING = "stopping"
|     RESIZING = "resizing"
|     ERROR = "error"
|     UNKNOWN = "unknown"
|     PULLING_IMAGE = "pulling_image"
|
|     def __str__(self):
|         return self.value
|
|     def __eq__(self, other):
|         if isinstance(other, str):
|             return self.value == other
|         return super().__eq__(other)
|
|
| class WorkspaceInfo(ApiWorkspaceInfo):
|     """Structured information about a Sandbox.
|
|     This class provides detailed information about a Sandbox's configuration,
|     resources, and current state.
|
|     Attributes:
|         id (str): Unique identifier for the Sandbox.
|         name (str): Display name of the Sandbox.
|         image (str): Docker image used for the Sandbox.
|         user (str): OS user running in the Sandbox.
|         env (Dict[str, str]): Environment variables set in the Sandbox.
|         labels (Dict[str, str]): Custom labels attached to the Sandbox.
|         public (bool): Whether the Sandbox is publicly accessible.
|         target (str): Target environment where the Sandbox runs.
|         resources (WorkspaceResources): Resource allocations for the Sandbox.
|         state (str): Current state of the Sandbox (e.g., "started", "stopped").
|         error_reason (Optional[str]): Error message if Sandbox is in error state.
|         snapshot_state (Optional[str]): Current state of Sandbox snapshot.
|         snapshot_state_created_at (Optional[datetime]): When the snapshot state was created.
|
|     Example:
|         ```python
|         workspace = daytona.create()
|         info = workspace.info()
|         print(f"Workspace {info.name} is {info.state}")
|         print(f"Resources: {info.resources.cpu} CPU, {info.resources.memory} RAM")
|         ```
|     """
|     id: str
|     name: str
|     image: str
|     user: str
|     env: Dict[str, str]
|     labels: Dict[str, str]
|     public: bool
|     target: WorkspaceTargetRegion
|     resources: WorkspaceResources
|     state: WorkspaceState
|     error_reason: Optional[str]
|     snapshot_state: Optional[str]
|     snapshot_state_created_at: Optional[datetime]
|     node_domain: str
|     region: str
|     class_name: str
|     updated_at: str
|     last_snapshot: Optional[str]
|     auto_stop_interval: int
|     provider_metadata: Annotated[Optional[str], Field(
|         deprecated='The `provider_metadata` field is deprecated. Use `state`, `node_domain`, `region`, `class_name`, `updated_at`, `last_snapshot`, `resources`, `auto_stop_interval` instead.')]
|
|
| class WorkspaceInstance(ApiWorkspace):
|     """Represents a Daytona workspace instance."""
|     info: Optional[WorkspaceInfo]
|
|
| class Workspace:
|     """Represents a Daytona Sandbox.
|
|     A Sandbox provides file system operations, Git operations, process execution,
|     and LSP functionality. It serves as the main interface for interacting with
|     a Daytona Sandbox.
|
|     Attributes:
|         id (str): Unique identifier for the Sandbox.
|         instance (WorkspaceInstance): The underlying Sandbox instance.
|         code_toolbox (WorkspaceCodeToolbox): Language-specific toolbox implementation.
|         fs (FileSystem): File system operations interface.
|         git (Git): Git operations interface.
|         process (Process): Process execution interface.
|     """
|
|     def __init__(
|         self,
|         id: str,
|         instance: WorkspaceInstance,
|         workspace_api: WorkspaceApi,
|         toolbox_api: ToolboxApi,
|         code_toolbox: WorkspaceCodeToolbox,
|     ):
|         """Initialize a new Workspace instance.
|
|         Args:
|             id (str): Unique identifier for the Sandbox.
|             instance (WorkspaceInstance): The underlying Sandbox instance.
|             workspace_api (WorkspaceApi): API client for Sandbox operations.
|             toolbox_api (ToolboxApi): API client for toolbox operations.
|             code_toolbox (WorkspaceCodeToolbox): Language-specific toolbox implementation.
|         """
|         self.id = id
|         self.instance = instance
|         self.workspace_api = workspace_api
|         self.toolbox_api = toolbox_api
|         self.code_toolbox = code_toolbox
|
|         # Initialize components
|         # File system operations
|         self.fs = FileSystem(instance, self.toolbox_api)
|         self.git = Git(self, self.toolbox_api, instance)  # Git operations
|         self.process = Process(
|             self.code_toolbox, self.toolbox_api, instance)  # Process execution
|
|     def info(self) -> WorkspaceInfo:
|         """Gets structured information about the Sandbox.
|
|         Returns:
|             WorkspaceInfo: Detailed information about the Sandbox including its
|                 configuration, resources, and current state.
|
|         Example:
|             ```python
|             info = workspace.info()
|             print(f"Workspace {info.name}:")
|             print(f"State: {info.state}")
|             print(f"Resources: {info.resources.cpu} CPU, {info.resources.memory} RAM")
|             ```
|         """
|         instance = self.workspace_api.get_workspace(self.id)
|         return Workspace._to_workspace_info(instance)
|
|     @intercept_errors(message_prefix="Failed to get workspace root directory: ")
|     def get_workspace_root_dir(self) -> str:
|         """Gets the root directory path of the Sandbox.
|
|         Returns:
|             str: The absolute path to the Sandbox root directory.
|
|         Example:
|             ```python
|             root_dir = workspace.get_workspace_root_dir()
|             print(f"Workspace root: {root_dir}")
|             ```
|         """
|         response = self.toolbox_api.get_project_dir(
|             workspace_id=self.instance.id
|         )
|         return response.dir
|
|     def create_lsp_server(
|         self, language_id: LspLanguageId, path_to_project: str
|     ) -> LspServer:
|         """Creates a new Language Server Protocol (LSP) server instance.
|
|         The LSP server provides language-specific features like code completion,
|         diagnostics, and more.
|
|         Args:
|             language_id (LspLanguageId): The language server type (e.g., LspLanguageId.PYTHON).
|             path_to_project (str): Absolute path to the project root directory.
|
|         Returns:
|             LspServer: A new LSP server instance configured for the specified language.
|
|         Example:
|             ```python
|             lsp = workspace.create_lsp_server("python", "/workspace/project")
|             ```
|         """
|         return LspServer(language_id, path_to_project, self.toolbox_api, self.instance)
|
|     @intercept_errors(message_prefix="Failed to set labels: ")
|     def set_labels(self, labels: Dict[str, str]) -> Dict[str, str]:
|         """Sets labels for the Sandbox.
|
|         Labels are key-value pairs that can be used to organize and identify Sandboxes.
|
|         Args:
|             labels (Dict[str, str]): Dictionary of key-value pairs representing Sandbox labels.
|
|         Returns:
|             Dict[str, str]: Dictionary containing the updated Sandbox labels.
|
|         Example:
|             ```python
|             new_labels = workspace.set_labels({
|                 "project": "my-project",
|                 "environment": "development",
|                 "team": "backend"
|             })
|             print(f"Updated labels: {new_labels}")
|             ```
|         """
|         # Convert all values to strings and create the expected labels structure
|         string_labels = {k: str(v).lower() if isinstance(
|             v, bool) else str(v) for k, v in labels.items()}
|         labels_payload = {"labels": string_labels}
|         return self.workspace_api.replace_labels(self.id, labels_payload)
|
|     @intercept_errors(message_prefix="Failed to start workspace: ")
|     @with_timeout(error_message=lambda self, timeout: f"Workspace {self.id} failed to start within the {timeout} seconds timeout period")
|     def start(self, timeout: Optional[float] = 60):
|         """Starts the Sandbox.
|
|         This method starts the Sandbox and waits for it to be ready.
|
|         Args:
|             timeout (Optional[float]): Maximum time to wait in seconds. 0 means no timeout. Default is 60 seconds.
|
|         Raises:
|             DaytonaError: If timeout is negative. If workspace fails to start or times out.
|
|         Example:
|             ```python
|             workspace = daytona.get_current_workspace("my-workspace")
|             workspace.start(timeout=40)  # Wait up to 40 seconds
|             print("Workspace started successfully")
|             ```
|         """
|         self.workspace_api.start_workspace(self.id, _request_timeout=timeout or None)
|         self.wait_for_workspace_start()
|
|     @intercept_errors(message_prefix="Failed to stop workspace: ")
|     @with_timeout(error_message=lambda self, timeout: f"Workspace {self.id} failed to stop within the {timeout} seconds timeout period")
|     def stop(self, timeout: Optional[float] = 60):
|         """Stops the Sandbox.
|
|         This method stops the Sandbox and waits for it to be fully stopped.
|
|         Args:
|             timeout (Optional[float]): Maximum time to wait in seconds. 0 means no timeout. Default is 60 seconds.
|
|         Raises:
|             DaytonaError: If timeout is negative; If workspace fails to stop or times out
|
|         Example:
|             ```python
|             workspace = daytona.get_current_workspace("my-workspace")
|             workspace.stop()
|             print("Workspace stopped successfully")
|             ```
|         """
|         self.workspace_api.stop_workspace(self.id, _request_timeout=timeout or None)
|         self.wait_for_workspace_stop()
|
|     @intercept_errors(message_prefix="Failure during waiting for workspace to start: ")
|     @with_timeout(error_message=lambda self, timeout: f"Workspace {self.id} failed to become ready within the {timeout} seconds timeout period")
|     def wait_for_workspace_start(self, timeout: Optional[float] = 60) -> None:
|         """Waits for the Sandbox to reach the 'started' state.
|
|         This method polls the Sandbox status until it reaches the 'started' state
|         or encounters an error.
|
|         Args:
|             timeout (Optional[float]): Maximum time to wait in seconds. 0 means no timeout. Default is 60 seconds.
|
|         Raises:
|             DaytonaError: If timeout is negative; If workspace fails to start or times out
|         """
|         state = None
|         while state != "started":
|             response = self.workspace_api.get_workspace(self.id)
|             provider_metadata = json.loads(response.info.provider_metadata)
|             state = provider_metadata.get('state', '')
|
|             if state == "error":
|                 raise DaytonaError(
|                     f"Workspace {self.id} failed to start with state: {state}, error reason: {response.error_reason}")
|
|             time.sleep(0.1)  # Wait 100ms between checks
|
|     @intercept_errors(message_prefix="Failure during waiting for workspace to stop: ")
|     @with_timeout(error_message=lambda self, timeout: f"Workspace {self.id} failed to become stopped within the {timeout} seconds timeout period")
|     def wait_for_workspace_stop(self, timeout: Optional[float] = 60) -> None:
|         """Waits for the Sandbox to reach the 'stopped' state.
|
|         This method polls the Sandbox status until it reaches the 'stopped' state
|         or encounters an error. It will wait up to 60 seconds for the Sandbox to stop.
|
|         Args:
|             timeout (Optional[float]): Maximum time to wait in seconds. 0 means no timeout. Default is 60 seconds.
|
|         Raises:
|             DaytonaError: If timeout is negative. If Sandbox fails to stop or times out.
|         """
|         state = None
|         while state != "stopped":
|             try:
|                 workspace_check = self.workspace_api.get_workspace(self.id)
|                 provider_metadata = json.loads(
|                     workspace_check.info.provider_metadata)
|                 state = provider_metadata.get('state')
|
|                 if state == "error":
|                     raise DaytonaError(
|                         f"Workspace {self.id} failed to stop with status: {state}, error reason: {workspace_check.error_reason}")
|             except Exception as e:
|                 # If there's a validation error, continue waiting
|                 if "validation error" not in str(e):
|                     raise e
|
|             time.sleep(0.1)  # Wait 100ms between checks
|
|     @intercept_errors(message_prefix="Failed to set auto-stop interval: ")
|     def set_autostop_interval(self, interval: int) -> None:
|         """Sets the auto-stop interval for the Sandbox.
|
|         The Sandbox will automatically stop after being idle (no new events) for the specified interval.
|         Events include any state changes or interactions with the Sandbox through the SDK.
|         Interactions using Sandbox Previews are not included.
|
|         Args:
|             interval (int): Number of minutes of inactivity before auto-stopping.
|                 Set to 0 to disable auto-stop. Defaults to 15.
|
|         Raises:
|             DaytonaError: If interval is negative
|
|         Example:
|             ```python
|             # Auto-stop after 1 hour
|             workspace.set_autostop_interval(60)
|             # Or disable auto-stop
|             workspace.set_autostop_interval(0)
|             ```
|         """
|         if not isinstance(interval, int) or interval < 0:
|             raise DaytonaError(
|                 "Auto-stop interval must be a non-negative integer")
|
|         self.workspace_api.set_autostop_interval(self.id, interval)
|         self.instance.auto_stop_interval = interval
|
|     @intercept_errors(message_prefix="Failed to get preview link: ")
|     def get_preview_link(self, port: int) -> str:
|         """Gets the preview link for the workspace at a specific port. If the port is not open, it will open it and return the link.
|
|         Args:
|             port (int): The port to open the preview link on
|
|         Returns:
|             The preview link for the workspace at the specified port
|         """
|         provider_metadata = json.loads(self.instance.info.provider_metadata)
|         node_domain = provider_metadata.get('nodeDomain', '')
|         if not node_domain:
|             raise DaytonaError(
|                 "Node domain not found in provider metadata. Please contact support.")
|
|         return f"https://{port}-{self.id}.{node_domain}"
|
|     @intercept_errors(message_prefix="Failed to archive workspace: ")
|     def archive(self) -> None:
|         """Archives the workspace, making it inactive and preserving its state. When sandboxes are archived, the entire filesystem
|         state is moved to cost-effective object storage, making it possible to keep sandboxes available for an extended period.
|         The tradeoff between archived and stopped states is that starting an archived sandbox takes more time, depending on its size.
|         Workspace must be stopped before archiving.
|         """
|         self.workspace_api.archive_workspace(self.id)
|
|     @staticmethod
|     def _to_workspace_info(instance: ApiWorkspace) -> WorkspaceInfo:
|         """Converts an API workspace instance to a WorkspaceInfo object.
|
|         Args:
|             instance (ApiWorkspace): The API workspace instance to convert
|
|         Returns:
|             WorkspaceInfo: The converted WorkspaceInfo object
|         """
|         provider_metadata = json.loads(instance.info.provider_metadata or '{}')
|         resources_data = provider_metadata.get('resources', provider_metadata)
|
|         # Extract resources with defaults
|         resources = WorkspaceResources(
|             cpu=str(resources_data.get('cpu', '1')),
|             gpu=str(resources_data.get('gpu')
|                     ) if resources_data.get('gpu') else None,
|             memory=str(resources_data.get('memory', '2')) + 'Gi',
|             disk=str(resources_data.get('disk', '10')) + 'Gi'
|         )
|
|         enum_state = to_enum(
|             WorkspaceState, provider_metadata.get('state', ''))
|         enum_target = to_enum(WorkspaceTargetRegion, instance.target)
|
|         return WorkspaceInfo(
|             id=instance.id,
|             name=instance.name,
|             image=instance.image,
|             user=instance.user,
|             env=instance.env or {},
|             labels=instance.labels or {},
|             public=instance.public,
|             target=enum_target or instance.target,
|             resources=resources,
|             state=enum_state or provider_metadata.get('state', ''),
|             error_reason=instance.error_reason,
|             snapshot_state=provider_metadata.get('snapshotState'),
|             snapshot_state_created_at=datetime.fromisoformat(provider_metadata.get(
|                 'snapshotStateCreatedAt')) if provider_metadata.get('snapshotStateCreatedAt') else None,
|             node_domain=provider_metadata.get('nodeDomain', ''),
|             region=provider_metadata.get('region', ''),
|             class_name=provider_metadata.get('class', ''),
|             updated_at=provider_metadata.get('updatedAt', ''),
|             last_snapshot=provider_metadata.get('lastSnapshot'),
|             auto_stop_interval=provider_metadata.get('autoStopInterval', 0),
|             created=instance.info.created or '',
|             provider_metadata=instance.info.provider_metadata,
|         )
|


--------------------------------------------------------------------------------