Env

README.md

@@ -51,7 +51,7 @@ sandbox = client.sandboxes.create()
 
 try:
     result = sandbox.process.execute(command="echo hello from leap0")
-    print(result.result.strip())
+    print(result.stdout.strip())
 finally:
     sandbox.delete()
     client.close()
@@ -65,7 +65,7 @@ from leap0 import Leap0Client
 with Leap0Client(api_key="your-api-key") as client:
     sandbox = client.sandboxes.create()
     result = sandbox.process.execute(command="echo hello from the sandbox")
-    print(result.result.strip())
+    print(result.stdout.strip())
     sandbox.delete()
 ```
 
@@ -117,7 +117,14 @@ Execute one-shot shell commands inside a running sandbox.
 
 ```python
 result = sandbox.process.execute(command="ls -la /workspace")
-print(result.result)
+print(result.stdout)
+
+interpolated = sandbox.process.execute(
+    command="echo $NAME from ${PLACE}",
+    cwd="/workspace/$NAME",
+    env={"NAME": "leap0", "PLACE": "sandbox"},
+)
+print(interpolated.stdout)
 ```
 
 ### Interactive Terminal (PTY)

examples/async_code_interpreter_stream.py

@@ -6,7 +6,7 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 
-from leap0 import AsyncLeap0Client, AsyncSandbox, DEFAULT_CODE_INTERPRETER_TEMPLATE_NAME, StreamEvent
+from leap0 import AsyncLeap0Client, AsyncSandbox, CodeLanguage, DEFAULT_CODE_INTERPRETER_TEMPLATE_NAME, StreamEvent
 
 
 async def main() -> None:
@@ -16,7 +16,7 @@ async def main() -> None:
         try:
             async for event in sandbox.code_interpreter.execute_stream(
                 code="import time\nfor i in range(3):\n    print(f'async step {i}')\n    time.sleep(1)",
-                language="python",
+                language=CodeLanguage.PYTHON,
                 timeout_ms=10_000,
             ):
                 typed_event: StreamEvent = event

examples/code_interpreter_stream.py

@@ -4,7 +4,7 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 
-from leap0 import DEFAULT_CODE_INTERPRETER_TEMPLATE_NAME, Leap0Client, Sandbox, StreamEvent
+from leap0 import CodeLanguage, DEFAULT_CODE_INTERPRETER_TEMPLATE_NAME, Leap0Client, Sandbox, StreamEvent
 
 
 def main() -> None:
@@ -15,7 +15,7 @@ def main() -> None:
         event: StreamEvent
         for event in sandbox.code_interpreter.execute_stream(
             code="import time\nfor i in range(3):\n    print(f'step {i}')\n    time.sleep(1)",
-            language="python",
+            language=CodeLanguage.PYTHON,
             timeout_ms=10_000,
         ):
             print(event)

leap0/__init__.py

@@ -19,12 +19,14 @@
     from ._sync.client import Leap0, Leap0Client
     from ._sync.code_interpreter import CodeInterpreterClient
     from .models.code_interpreter import (
+        CodeLanguage,
         CodeContext,
         CodeExecutionError,
         CodeExecutionOutput,
         CodeExecutionResult,
         ExecutionLogs,
         StreamEvent,
+        StreamEventType,
     )
     from .models.config import (
         DEFAULT_CODE_INTERPRETER_TEMPLATE_NAME,
@@ -133,6 +135,7 @@
     "AzureRegistryCredentialsDict",
     "BasicRegistryCredentials",
     "BasicRegistryCredentialsDict",
+    "CodeLanguage",
     "CodeContext",
     "CodeExecutionError",
     "CodeExecutionOutput",
@@ -207,6 +210,7 @@
     "SshClient",
     "SshValidation",
     "StreamEvent",
+    "StreamEventType",
     "sandbox_id_of",
     "Template",
     "TemplatesClient",
@@ -300,12 +304,14 @@
     "RegistryCredentialsInput": (".models.template", "RegistryCredentialsInput"),
     "Template": (".models.template", "Template"),
     "RenameTemplateParams": (".models.template", "RenameTemplateParams"),
+    "CodeLanguage": (".models.code_interpreter", "CodeLanguage"),
     "CodeContext": (".models.code_interpreter", "CodeContext"),
     "CodeExecutionError": (".models.code_interpreter", "CodeExecutionError"),
     "CodeExecutionOutput": (".models.code_interpreter", "CodeExecutionOutput"),
     "CodeExecutionResult": (".models.code_interpreter", "CodeExecutionResult"),
     "ExecutionLogs": (".models.code_interpreter", "ExecutionLogs"),
     "StreamEvent": (".models.code_interpreter", "StreamEvent"),
+    "StreamEventType": (".models.code_interpreter", "StreamEventType"),
     "DesktopDisplayInfo": (".models.desktop", "DesktopDisplayInfo"),
     "DesktopHealth": (".models.desktop", "DesktopHealth"),
     "DesktopPointerPosition": (".models.desktop", "DesktopPointerPosition"),

leap0/_async/process.py

@@ -2,11 +2,11 @@
 
 from typing import cast
 
-from .._internal.types import JsonObject
 from ..models.process import ProcessResult
 from ..models.sandbox import SandboxRef, sandbox_id_of
 from .._schemas.process import ProcessResultDict
 from .._utils.errors import intercept_errors
+from .._utils.process import build_command_payload
 from ._transport import AsyncTransport
 
 
@@ -24,7 +24,7 @@ def __init__(self, transport: AsyncTransport):
         self._transport = transport
 
     @intercept_errors("Failed to execute command: ")
-    async def execute(self, sandbox: SandboxRef, *, command: str, cwd: str | None = None, timeout: int | None = None) -> ProcessResult:
+    async def execute(self, sandbox: SandboxRef, *, command: str, cwd: str | None = None, timeout: int | None = None, env: dict[str, str] | None = None) -> ProcessResult:
         """Run a shell command and wait for the result.
 
         The command runs inside ``/bin/sh -c``.
@@ -34,6 +34,7 @@ async def execute(self, sandbox: SandboxRef, *, command: str, cwd: str | None =
             command: Shell command to execute.
             cwd: Working directory.
             timeout: Timeout in seconds (default 30).
+            env: Optional local values used to expand ``$NAME`` and ``${NAME}`` in string fields before sending the request.
 
         Returns:
             ProcessResult: Command result including exit code, stdout, and stderr.
@@ -47,10 +48,6 @@ async def execute(self, sandbox: SandboxRef, *, command: str, cwd: str | None =
             print(result.stderr)
             ```
         """
-        payload: JsonObject = {"command": command}
-        if cwd is not None:
-            payload["cwd"] = cwd
-        if timeout is not None:
-            payload["timeout"] = timeout
+        payload = build_command_payload(command=command, cwd=cwd, timeout=timeout, env=env)
         data = cast(ProcessResultDict, await self._transport.request_json("POST", f"/v1/sandbox/{sandbox_id_of(sandbox)}/process/execute", json=payload))
         return ProcessResult.from_dict(data)

leap0/_schemas/process.py

@@ -7,4 +7,3 @@ class ProcessResultDict(TypedDict, total=False):
     exit_code: int
     stdout: str
     stderr: str
-    result: object

leap0/_sync/process.py

@@ -2,11 +2,11 @@
 
 from typing import cast
 
-from .._internal.types import JsonObject
 from ..models.process import ProcessResult
 from ..models.sandbox import SandboxRef, sandbox_id_of
 from .._schemas.process import ProcessResultDict
 from .._utils.errors import intercept_errors
+from .._utils.process import build_command_payload
 from ._transport import Transport
 
 
@@ -24,7 +24,7 @@ def __init__(self, transport: Transport):
         self._transport = transport
 
     @intercept_errors("Failed to execute command: ")
-    def execute(self, sandbox: SandboxRef, *, command: str, cwd: str | None = None, timeout: int | None = None, http_timeout: int | None = None) -> ProcessResult:
+    def execute(self, sandbox: SandboxRef, *, command: str, cwd: str | None = None, timeout: int | None = None, env: dict[str, str] | None = None, http_timeout: int | None = None) -> ProcessResult:
         """Run a shell command and wait for the result.
 
         The command runs inside ``/bin/sh -c``.
@@ -34,6 +34,7 @@ def execute(self, sandbox: SandboxRef, *, command: str, cwd: str | None = None,
             command: Shell command to execute.
             cwd: Working directory.
             timeout: Timeout in seconds. If omitted, the server-side default is used.
+            env: Optional local values used to expand ``$NAME`` and ``${NAME}`` in string fields before sending the request.
             http_timeout: Optional HTTP request timeout in seconds for this SDK call.
 
         Returns:
@@ -48,10 +49,6 @@ def execute(self, sandbox: SandboxRef, *, command: str, cwd: str | None = None,
             print(result.stderr)
             ```
         """
-        payload: JsonObject = {"command": command}
-        if cwd is not None:
-            payload["cwd"] = cwd
-        if timeout is not None:
-            payload["timeout"] = timeout
+        payload = build_command_payload(command=command, cwd=cwd, timeout=timeout, env=env)
         data = cast(ProcessResultDict, self._transport.request_json("POST", f"/v1/sandbox/{sandbox_id_of(sandbox)}/process/execute", json=payload, timeout=http_timeout))
         return ProcessResult.from_dict(data)

leap0/_utils/env.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+
+
+_ENV_REF_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}|\$([A-Za-z_][A-Za-z0-9_]*)")
+
+
+def expand_env(value: str, env: Mapping[str, str]) -> str:
+    """Expand ``$NAME`` and ``${NAME}`` placeholders using the provided mapping."""
+    return _ENV_REF_RE.sub(lambda match: env.get(match.group(1) or match.group(2), match.group(0)), value)

leap0/_utils/process.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+
+from .._internal.types import JsonObject
+from .env import expand_env
+
+
+def build_command_payload(*, command: str, cwd: str | None = None, timeout: int | None = None, env: Mapping[str, str] | None = None) -> JsonObject:
+    payload: JsonObject = {"command": expand_env(command, env) if env else command}
+    if cwd is not None:
+        payload["cwd"] = expand_env(cwd, env) if env else cwd
+    if timeout is not None:
+        payload["timeout"] = timeout
+    return payload

leap0/models/process.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from .._schemas.process import ProcessResultDict
 
 @dataclass(slots=True)
@@ -9,41 +9,12 @@ class ProcessResult:
     exit_code: int
     stdout: str
     stderr: str
-    _legacy_result: str | dict[str, str] | None = field(default=None, repr=False)
-
-    @property
-    def result(self) -> str | dict[str, str]:
-        """Backward-compatible alias for legacy result payloads."""
-        if self._legacy_result is not None:
-            return self._legacy_result
-        return self.stdout
 
     @classmethod
     def from_dict(cls, data: ProcessResultDict) -> ProcessResult:
         """Build an instance from a wire-format dictionary."""
-        legacy_result = data.get("result")
-        stdout = data.get("stdout")
-        stderr = data.get("stderr")
-
-        normalized_legacy_result: str | dict[str, str] | None = None
-        if isinstance(legacy_result, dict):
-            normalized_legacy_result = {
-                key: value
-                for key in ("stdout", "stderr")
-                if isinstance((value := legacy_result.get(key)), str)
-            }
-            if stdout is None:
-                stdout = legacy_result.get("stdout", "")
-            if stderr is None:
-                stderr = legacy_result.get("stderr", "")
-        elif legacy_result is not None and stdout is None and stderr is None:
-            normalized_legacy_result = legacy_result if isinstance(legacy_result, str) else None
-            stdout = legacy_result if isinstance(legacy_result, str) else ""
-            stderr = ""
-
         return cls(
             exit_code=int(data.get("exit_code", 0)),
-            stdout=stdout if isinstance(stdout, str) else "",
-            stderr=stderr if isinstance(stderr, str) else "",
-            _legacy_result=normalized_legacy_result,
+            stdout=data.get("stdout") if isinstance(data.get("stdout"), str) else "",
+            stderr=data.get("stderr") if isinstance(data.get("stderr"), str) else "",
         )

tests/_async/test_process.py

@@ -16,3 +16,21 @@ async def run() -> None:
             assert async_mock_transport.request_json.call_args[0][:2] == ("POST", "/v1/sandbox/sbx-1/process/execute")
 
         asyncio.run(run())
+
+    def test_execute_expands_env(self, async_mock_transport):
+        async def run() -> None:
+            async_mock_transport.request_json.return_value = {"exit_code": 0, "stdout": "hello", "stderr": ""}
+
+            await AsyncProcessClient(async_mock_transport).execute(
+                "sbx-1",
+                command="echo $NAME from ${PLACE}",
+                cwd="/workspace/$NAME",
+                env={"NAME": "leap0", "PLACE": "sandbox"},
+            )
+
+            assert async_mock_transport.request_json.call_args.kwargs["json"] == {
+                "command": "echo leap0 from sandbox",
+                "cwd": "/workspace/leap0",
+            }
+
+        asyncio.run(run())

tests/_sync/test_process.py

@@ -11,3 +11,18 @@ def test_execute(self, mock_transport):
         assert result.stdout == "hello"
         assert result.stderr == "warn"
         assert mock_transport.request_json.call_args[0][:2] == ("POST", "/v1/sandbox/sbx-1/process/execute")
+
+    def test_execute_expands_env(self, mock_transport):
+        mock_transport.request_json.return_value = {"exit_code": 0, "stdout": "hello", "stderr": ""}
+
+        ProcessClient(mock_transport).execute(
+            "sbx-1",
+            command="echo $NAME from ${PLACE}",
+            cwd="/workspace/$NAME",
+            env={"NAME": "leap0", "PLACE": "sandbox"},
+        )
+
+        assert mock_transport.request_json.call_args.kwargs["json"] == {
+            "command": "echo leap0 from sandbox",
+            "cwd": "/workspace/leap0",
+        }

tests/models/test_process.py

@@ -9,15 +9,3 @@ def test_from_dict(self):
         assert r.exit_code == 1
         assert r.stdout == "hello"
         assert r.stderr == "error output"
-
-    def test_from_dict_accepts_legacy_result_string(self):
-        r = ProcessResult.from_dict({"exit_code": 0, "result": "hello"})
-        assert r.stdout == "hello"
-        assert r.stderr == ""
-        assert r.result == "hello"
-
-    def test_from_dict_accepts_legacy_result_mapping(self):
-        r = ProcessResult.from_dict({"exit_code": 0, "result": {"stdout": "hello", "stderr": "warn"}})
-        assert r.stdout == "hello"
-        assert r.stderr == "warn"
-        assert r.result == {"stdout": "hello", "stderr": "warn"}

tests/test_import.py

@@ -20,6 +20,7 @@
     AzureRegistryCredentialsDict,
     BasicRegistryCredentials,
     BasicRegistryCredentialsDict,
+    CodeLanguage,
     CreatePtySessionParams,
     CreateSandboxParams,
     CreateSnapshotParams,
@@ -39,6 +40,7 @@
     Sandbox,
     SandboxesClient,
     SnapshotsClient,
+    StreamEventType,
     TemplatesClient,
 )
 
@@ -87,6 +89,8 @@ def test_service_client_imports() -> None:
     assert CreateTemplateParams is not None
     assert RenameTemplateParams is not None
     assert CreatePtySessionParams is not None
+    assert CodeLanguage.PYTHON == "python"
+    assert StreamEventType.STDOUT == "stdout"
     assert DEFAULT_TEMPLATE_NAME == "system/debian:bookworm"
     assert DEFAULT_CODE_INTERPRETER_TEMPLATE_NAME == "system/code-interpreter:v0.1.0"