Skip to content

Daytona API

DaytonaSandbox

pydantic_ai_backends.backends.daytona.DaytonaSandbox

Bases: BaseSandbox

Daytona cloud sandbox backend.

Creates an ephemeral Daytona sandbox for running commands and managing files in an isolated cloud environment. Uses native Daytona file APIs for read_bytes and write; all other file helpers are inherited from :class:BaseSandbox (shell-based).

Source code in src/pydantic_ai_backends/backends/daytona.py
Python
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class DaytonaSandbox(BaseSandbox):
    """Daytona cloud sandbox backend.

    Creates an ephemeral Daytona sandbox for running commands and managing
    files in an isolated cloud environment.  Uses native Daytona file APIs
    for `read_bytes` and `write`; all other file helpers are inherited
    from :class:`BaseSandbox` (shell-based).
    """

    def __init__(
        self,
        api_key: str | None = None,
        sandbox_id: str | None = None,
        work_dir: str = "/home/daytona",
        startup_timeout: int = 180,
    ):
        """Initialise and start a Daytona sandbox.

        Args:
            api_key: Daytona API key.  Falls back to `DAYTONA_API_KEY` env var.
            sandbox_id: Optional identifier (generated by Daytona if omitted).
            work_dir: Working directory inside the sandbox.
            startup_timeout: Seconds to wait for the sandbox to become ready.

        Raises:
            ValueError: If no API key is provided or found in environment.
            RuntimeError: If the sandbox fails to start within *startup_timeout*.
        """
        from daytona import Daytona, DaytonaConfig

        resolved_key = api_key or os.environ.get("DAYTONA_API_KEY")
        if not resolved_key:
            raise ValueError("Daytona API key is required. Pass api_key= or set DAYTONA_API_KEY.")

        self._client: Any = Daytona(DaytonaConfig(api_key=resolved_key))
        self._sandbox: Any = self._client.create()
        self._work_dir = work_dir

        # Wait until sandbox is responsive
        self._wait_until_ready(startup_timeout)

        super().__init__(sandbox_id or self._sandbox.id)

    def _wait_until_ready(self, timeout: int) -> None:
        """Poll until the sandbox responds to a simple command."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            try:
                result = self._sandbox.process.exec("echo ready", timeout=5)
                if result.exit_code == 0:
                    return
            except Exception:
                pass
            time.sleep(2)

        # Clean up on failure
        try:
            self._client.delete(self._sandbox)
        finally:
            raise RuntimeError(f"Daytona sandbox failed to start within {timeout} seconds")

    def start(self) -> None:
        """No-op — Daytona sandboxes start automatically on creation."""

    def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Execute a command inside the Daytona sandbox.

        Args:
            command: Shell command string.
            timeout: Maximum execution time in seconds (`None` → 30 min).

        Returns:
            :class:`ExecuteResponse` with output, exit code, and truncation flag.
        """
        self.touch()
        effective_timeout = timeout if timeout is not None else DEFAULT_EXEC_TIMEOUT
        try:
            result = self._sandbox.process.exec(
                command, cwd=self._work_dir, timeout=effective_timeout
            )
            output = result.result
            truncated = len(output) > MAX_EXECUTE_OUTPUT_BYTES
            if truncated:
                output = output[:MAX_EXECUTE_OUTPUT_BYTES]
            return ExecuteResponse(
                output=output,
                exit_code=result.exit_code,
                truncated=truncated,
            )
        except Exception as e:
            return ExecuteResponse(output=f"Error: {e}", exit_code=1, truncated=False)

    def exists(self, path: str) -> bool:
        """Check existence via Daytona's native file API.

        Falls back to `False` for any error (missing file, permission
        denied, transient API failure) — mirroring the broad exception
        handling in :meth:`read_bytes`.
        """
        try:
            info = self._sandbox.fs.get_file_info(path)
        except Exception:
            return False
        return not bool(getattr(info, "is_dir", False))

    def read_bytes(self, path: str) -> bytes:
        """Download file bytes via Daytona's native file API."""
        from daytona import FileDownloadRequest

        try:
            responses = self._sandbox.fs.download_files([FileDownloadRequest(source=path)])
            data = responses[0].result
            return data.encode() if isinstance(data, str) else data
        except Exception:
            # Return empty bytes on failure so callers cannot mistake an error
            # message for real file content.
            return b""

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Upload a file via Daytona's native file API.

        Creates parent directories automatically.

        Args:
            path: Absolute path inside the sandbox.
            content: File content (str or bytes).

        Returns:
            :class:`WriteResult` with path on success or error message.
        """
        from daytona import FileUpload

        try:
            # Ensure parent directory
            parent = str(PurePosixPath(path).parent)
            self.execute(f"mkdir -p {shlex.quote(parent)}")

            payload = content if isinstance(content, bytes) else content.encode()
            self._sandbox.fs.upload_files([FileUpload(source=payload, destination=path)])
            return WriteResult(path=path)
        except Exception as e:
            return WriteResult(error=f"Failed to write file: {e}")

    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit a file by downloading it, replacing in Python and uploading it back."""
        try:
            if not self.exists(path):
                return EditResult(error=f"File not found: {path}")

            content = self.read_bytes(path).decode("utf-8", errors="replace")
            outcome = replace_in_content(content, old_string, new_string, replace_all)
            if not isinstance(outcome, Replacement):
                return EditResult(error=outcome)

            written = self.write(path, outcome.content)
            if written.error:
                return EditResult(error=written.error)
            return EditResult(path=path, occurrences=outcome.occurrences)
        except Exception as e:
            return EditResult(error=f"Failed to edit file: {e}")

    def is_alive(self) -> bool:
        """Check if the sandbox is responsive."""
        try:
            result = self.execute("echo ok", timeout=10)
            return result.exit_code == 0
        except Exception:
            return False

    def stop(self, purge: bool = False) -> None:
        """Delete the Daytona sandbox.

        Args:
            purge: Accepted for one signature across every sandbox, and it makes
                no difference here. Daytona has no "end it but keep the files"
                state this library can reattach to, so stopping *is* deleting -
                which is why the parameter is documented rather than quietly
                ignored.
        """
        client = getattr(self, "_client", None)
        sandbox = getattr(self, "_sandbox", None)
        if client is not None and sandbox is not None:
            with contextlib.suppress(Exception):
                client.delete(sandbox)
            self._sandbox = None

    def __del__(self) -> None:
        """Cleanup sandbox on garbage collection."""
        if hasattr(self, "_sandbox"):
            self.stop()

__init__(api_key=None, sandbox_id=None, work_dir='/home/daytona', startup_timeout=180)

Initialise and start a Daytona sandbox.

Parameters:

Name Type Description Default
api_key str | None

Daytona API key. Falls back to DAYTONA_API_KEY env var.

 None
sandbox_id str | None

Optional identifier (generated by Daytona if omitted).

 None
work_dir str

Working directory inside the sandbox.

 '/home/daytona'
startup_timeout int

Seconds to wait for the sandbox to become ready.

 180

Raises:

Type Description
ValueError

If no API key is provided or found in environment.
RuntimeError

If the sandbox fails to start within startup_timeout.
Source code in src/pydantic_ai_backends/backends/daytona.py
Python
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __init__(
    self,
    api_key: str | None = None,
    sandbox_id: str | None = None,
    work_dir: str = "/home/daytona",
    startup_timeout: int = 180,
):
    """Initialise and start a Daytona sandbox.

    Args:
        api_key: Daytona API key.  Falls back to `DAYTONA_API_KEY` env var.
        sandbox_id: Optional identifier (generated by Daytona if omitted).
        work_dir: Working directory inside the sandbox.
        startup_timeout: Seconds to wait for the sandbox to become ready.

    Raises:
        ValueError: If no API key is provided or found in environment.
        RuntimeError: If the sandbox fails to start within *startup_timeout*.
    """
    from daytona import Daytona, DaytonaConfig

    resolved_key = api_key or os.environ.get("DAYTONA_API_KEY")
    if not resolved_key:
        raise ValueError("Daytona API key is required. Pass api_key= or set DAYTONA_API_KEY.")

    self._client: Any = Daytona(DaytonaConfig(api_key=resolved_key))
    self._sandbox: Any = self._client.create()
    self._work_dir = work_dir

    # Wait until sandbox is responsive
    self._wait_until_ready(startup_timeout)

    super().__init__(sandbox_id or self._sandbox.id)

execute(command, timeout=None)

Execute a command inside the Daytona sandbox.

Parameters:

Name Type Description Default
command str

Shell command string.

 required
timeout int | None

Maximum execution time in seconds (None → 30 min).

 None

Returns:

Type Description
ExecuteResponse

class:ExecuteResponse with output, exit code, and truncation flag.
Source code in src/pydantic_ai_backends/backends/daytona.py
Python
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
    """Execute a command inside the Daytona sandbox.

    Args:
        command: Shell command string.
        timeout: Maximum execution time in seconds (`None` → 30 min).

    Returns:
        :class:`ExecuteResponse` with output, exit code, and truncation flag.
    """
    self.touch()
    effective_timeout = timeout if timeout is not None else DEFAULT_EXEC_TIMEOUT
    try:
        result = self._sandbox.process.exec(
            command, cwd=self._work_dir, timeout=effective_timeout
        )
        output = result.result
        truncated = len(output) > MAX_EXECUTE_OUTPUT_BYTES
        if truncated:
            output = output[:MAX_EXECUTE_OUTPUT_BYTES]
        return ExecuteResponse(
            output=output,
            exit_code=result.exit_code,
            truncated=truncated,
        )
    except Exception as e:
        return ExecuteResponse(output=f"Error: {e}", exit_code=1, truncated=False)

read_bytes(path)

Download file bytes via Daytona's native file API.

Source code in src/pydantic_ai_backends/backends/daytona.py
Python
140
141
142
143
144
145
146
147
148
149
150
151
def read_bytes(self, path: str) -> bytes:
    """Download file bytes via Daytona's native file API."""
    from daytona import FileDownloadRequest

    try:
        responses = self._sandbox.fs.download_files([FileDownloadRequest(source=path)])
        data = responses[0].result
        return data.encode() if isinstance(data, str) else data
    except Exception:
        # Return empty bytes on failure so callers cannot mistake an error
        # message for real file content.
        return b""

write(path, content)

Upload a file via Daytona's native file API.

Creates parent directories automatically.

Parameters:

Name Type Description Default
path str

Absolute path inside the sandbox.

 required
content str | bytes

File content (str or bytes).

 required

Returns:

Type Description
WriteResult

class:WriteResult with path on success or error message.
Source code in src/pydantic_ai_backends/backends/daytona.py
Python
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def write(self, path: str, content: str | bytes) -> WriteResult:
    """Upload a file via Daytona's native file API.

    Creates parent directories automatically.

    Args:
        path: Absolute path inside the sandbox.
        content: File content (str or bytes).

    Returns:
        :class:`WriteResult` with path on success or error message.
    """
    from daytona import FileUpload

    try:
        # Ensure parent directory
        parent = str(PurePosixPath(path).parent)
        self.execute(f"mkdir -p {shlex.quote(parent)}")

        payload = content if isinstance(content, bytes) else content.encode()
        self._sandbox.fs.upload_files([FileUpload(source=payload, destination=path)])
        return WriteResult(path=path)
    except Exception as e:
        return WriteResult(error=f"Failed to write file: {e}")

edit(path, old_string, new_string, replace_all=False)

Edit a file by downloading it, replacing in Python and uploading it back.

Source code in src/pydantic_ai_backends/backends/daytona.py
Python
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def edit(
    self, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult:
    """Edit a file by downloading it, replacing in Python and uploading it back."""
    try:
        if not self.exists(path):
            return EditResult(error=f"File not found: {path}")

        content = self.read_bytes(path).decode("utf-8", errors="replace")
        outcome = replace_in_content(content, old_string, new_string, replace_all)
        if not isinstance(outcome, Replacement):
            return EditResult(error=outcome)

        written = self.write(path, outcome.content)
        if written.error:
            return EditResult(error=written.error)
        return EditResult(path=path, occurrences=outcome.occurrences)
    except Exception as e:
        return EditResult(error=f"Failed to edit file: {e}")

exists(path)

Check existence via Daytona's native file API.

Falls back to False for any error (missing file, permission denied, transient API failure) — mirroring the broad exception handling in :meth:read_bytes.

Source code in src/pydantic_ai_backends/backends/daytona.py
Python
127
128
129
130
131
132
133
134
135
136
137
138
def exists(self, path: str) -> bool:
    """Check existence via Daytona's native file API.

    Falls back to `False` for any error (missing file, permission
    denied, transient API failure) — mirroring the broad exception
    handling in :meth:`read_bytes`.
    """
    try:
        info = self._sandbox.fs.get_file_info(path)
    except Exception:
        return False
    return not bool(getattr(info, "is_dir", False))

start()

No-op — Daytona sandboxes start automatically on creation.

Source code in src/pydantic_ai_backends/backends/daytona.py
Python
96
97
def start(self) -> None:
    """No-op — Daytona sandboxes start automatically on creation."""

stop(purge=False)

Delete the Daytona sandbox.

Parameters:

Name Type Description Default
purge bool

Accepted for one signature across every sandbox, and it makes no difference here. Daytona has no "end it but keep the files" state this library can reattach to, so stopping is deleting - which is why the parameter is documented rather than quietly ignored.

 False
Source code in src/pydantic_ai_backends/backends/daytona.py
Python
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def stop(self, purge: bool = False) -> None:
    """Delete the Daytona sandbox.

    Args:
        purge: Accepted for one signature across every sandbox, and it makes
            no difference here. Daytona has no "end it but keep the files"
            state this library can reattach to, so stopping *is* deleting -
            which is why the parameter is documented rather than quietly
            ignored.
    """
    client = getattr(self, "_client", None)
    sandbox = getattr(self, "_sandbox", None)
    if client is not None and sandbox is not None:
        with contextlib.suppress(Exception):
            client.delete(sandbox)
        self._sandbox = None

is_alive()

Check if the sandbox is responsive.

Source code in src/pydantic_ai_backends/backends/daytona.py
Python
198
199
200
201
202
203
204
def is_alive(self) -> bool:
    """Check if the sandbox is responsive."""
    try:
        result = self.execute("echo ok", timeout=10)
        return result.exit_code == 0
    except Exception:
        return False