Add command and entrypoint to sandbox

examples/19_entrypoint_and_command.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""Create sandboxes with custom entrypoint and command"""
+
+import os
+import sys
+import random
+import string
+
+from koyeb import Sandbox
+
+
+def main():
+    api_token = os.getenv("KOYEB_API_TOKEN")
+    if not api_token:
+        print("Error: KOYEB_API_TOKEN not set")
+        return 1
+
+    suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+
+    # Example 1: Custom command with args
+    print("=== Example 1: Custom command ===")
+    sandbox = None
+    try:
+        sandbox = Sandbox.create(
+            image="ubuntu",
+            name=f"custom-command-{suffix}",
+            command="/bin/sh",
+            args=["-c", "touch /tmp/command-was-here && sleep infinity"],
+            api_token=api_token,
+            env={"LOG_LEVEL": "DEBUG"} 
+        )
+        result = sandbox.exec("cat /tmp/command-was-here && echo 'File exists'")
+        print(f"  {result.stdout.strip()}")
+        assert result.exit_code == 0, "Expected /tmp/command-was-here to exist"
+        print("  OK: custom command created the file")
+    except Exception as e:
+        print(f"  Error: {e}")
+        return 1
+    finally:
+        if sandbox:
+            sandbox.delete()
+
+    # Example 2: Custom entrypoint with command
+    # Use a custom entrypoint that runs python, proving the entrypoint was used.
+    print("=== Example 2: Custom entrypoint ===")
+    sandbox = None
+    try:
+        sandbox = Sandbox.create(
+            image="python:3.12-slim",
+            name=f"custom-entrypoint-{suffix}",
+            entrypoint=["python3", "-c"],
+            command="import os; os.makedirs('/tmp', exist_ok=True); open('/tmp/started-by-python', 'w').write('yes'); import time; time.sleep(999999)",
+            api_token=api_token,
+        )
+        # The file was created by python3 (the entrypoint), not bash
+        result = sandbox.exec("cat /tmp/started-by-python")
+        content = result.stdout.strip()
+        print(f"  Marker content: {content}")
+        assert content == "yes", f"Expected 'yes', got '{content}'"
+        print("  OK: python3 entrypoint created the marker file")
+    except Exception as e:
+        print(f"  Error: {e}")
+        return 1
+    finally:
+        if sandbox:
+            sandbox.delete()
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

koyeb/sandbox/sandbox.py

@@ -127,6 +127,9 @@ def create(
         app_id: Optional[str] = None,
         enable_mesh: bool = None,
         poll_interval: float = DEFAULT_POLL_INTERVAL,
+        entrypoint: Optional[List[str]] = None,
+        command: Optional[str] = None,
+        args: Optional[List[str]] = None,
     ) -> Sandbox:
         """
             Create a new sandbox instance.
@@ -161,6 +164,8 @@ def create(
                 app_id: If provided, create the sandbox service in an existing app instead of creating a new one.
                 enable_mesh: Enable or disable mesh for this sandbox. Disabled by default
                 poll_interval: Time between health checks in seconds when wait_ready is True (default: 0.5)
+                entrypoint: Override the default entrypoint of the Docker image (e.g., ["/bin/sh", "-c"])
+                command: Override the default command of the Docker image (e.g., "python app.py")
 
         Returns:
                 Sandbox: A new Sandbox instance
@@ -206,6 +211,9 @@ def create(
             app_id=app_id,
             enable_mesh=enable_mesh,
             poll_interval=poll_interval,
+            entrypoint=entrypoint,
+            command=command,
+            args=args,
         )
 
         if wait_ready:
@@ -241,6 +249,9 @@ def _create_sync(
         app_id: Optional[str] = None,
         enable_mesh: bool = None,
         poll_interval: float = DEFAULT_POLL_INTERVAL,
+        entrypoint: Optional[List[str]] = None,
+        command: Optional[str] = None,
+        args: Optional[List[str]] = None,
     ) -> Sandbox:
         """
         Synchronous creation method that returns creation parameters.
@@ -272,7 +283,8 @@ def _create_sync(
 
         env_vars = build_env_vars(env)
         docker_source = create_docker_source(
-            image, [], privileged=privileged, image_registry_secret=registry_secret
+            image, privileged=privileged, image_registry_secret=registry_secret,
+            entrypoint=entrypoint, command=command, args=args,
         )
 
         deployment_definition = create_deployment_definition(
@@ -1095,6 +1107,9 @@ async def create(
         app_id: Optional[str] = None,
         enable_mesh: bool = False,
         poll_interval: float = DEFAULT_POLL_INTERVAL,
+        entrypoint: Optional[List[str]] = None,
+        command: Optional[str] = None,
+        args: Optional[List[str]] = None,
     ) -> AsyncSandbox:
         """
             Create a new sandbox instance with async support.
@@ -1131,6 +1146,8 @@ async def create(
                 app_id: If provided, create the sandbox service in an existing app instead of creating a new one.
                 enable_mesh: Enable or disable mesh for this sandbox. Disabled by default
                 poll_interval: Time between health checks in seconds when wait_ready is True (default: 0.5)
+                entrypoint: Override the default entrypoint of the Docker image (e.g., ["/bin/sh", "-c"])
+                command: Override the default command of the Docker image (e.g., "python app.py")
 
         Returns:
                 AsyncSandbox: A new AsyncSandbox instance
@@ -1169,6 +1186,9 @@ async def create(
                 app_id=app_id,
                 enable_mesh=enable_mesh,
                 poll_interval=poll_interval,
+                entrypoint=entrypoint,
+                command=command,
+                args=args,
             ),
         )
 

koyeb/sandbox/test_utils.py

@@ -0,0 +1,64 @@
+import unittest
+
+from koyeb.sandbox.utils import create_docker_source
+
+
+class TestCreateDockerSource(unittest.TestCase):
+    """Tests for create_docker_source entrypoint, command, and args support."""
+
+    def test_default_no_entrypoint_no_command(self):
+        ds = create_docker_source("myimage")
+        self.assertIsNone(ds.command)
+        self.assertIsNone(ds.args)
+        self.assertIsNone(ds.entrypoint)
+
+    def test_command_and_args(self):
+        ds = create_docker_source("myimage", command="python", args=["-u", "app.py"])
+        self.assertEqual(ds.command, "python")
+        self.assertEqual(ds.args, ["-u", "app.py"])
+        self.assertIsNone(ds.entrypoint)
+
+    def test_command_only(self):
+        ds = create_docker_source("myimage", command="python app.py")
+        self.assertEqual(ds.command, "python app.py")
+        self.assertIsNone(ds.args)
+        self.assertIsNone(ds.entrypoint)
+
+    def test_entrypoint_only(self):
+        ds = create_docker_source("myimage", entrypoint=["/bin/sh", "-c"])
+        self.assertIsNone(ds.command)
+        self.assertIsNone(ds.args)
+        self.assertEqual(ds.entrypoint, ["/bin/sh", "-c"])
+
+    def test_entrypoint_and_command(self):
+        ds = create_docker_source(
+            "myimage", entrypoint=["/bin/sh", "-c"], command="python app.py"
+        )
+        self.assertEqual(ds.command, "python app.py")
+        self.assertEqual(ds.entrypoint, ["/bin/sh", "-c"])
+        self.assertIsNone(ds.args)
+
+    def test_entrypoint_command_and_args(self):
+        ds = create_docker_source(
+            "myimage", entrypoint=["/bin/sh", "-c"], command="python", args=["app.py"]
+        )
+        self.assertEqual(ds.command, "python")
+        self.assertEqual(ds.entrypoint, ["/bin/sh", "-c"])
+        self.assertEqual(ds.args, ["app.py"])
+
+    def test_privileged_and_registry_secret_still_work(self):
+        ds = create_docker_source(
+            "myimage",
+            privileged=True,
+            image_registry_secret="my-secret",
+            entrypoint=["/entrypoint.sh"],
+            command="serve",
+        )
+        self.assertTrue(ds.privileged)
+        self.assertEqual(ds.image_registry_secret, "my-secret")
+        self.assertEqual(ds.entrypoint, ["/entrypoint.sh"])
+        self.assertEqual(ds.command, "serve")
+
+
+if __name__ == "__main__":
+    unittest.main()

koyeb/sandbox/utils.py

@@ -144,29 +144,34 @@ def build_env_vars(env: Optional[Dict[str, str]]) -> List[DeploymentEnv]:
 
 def create_docker_source(
     image: str,
-    command_args: List[str],
     privileged: Optional[bool] = None,
     image_registry_secret: Optional[str] = None,
+    entrypoint: Optional[List[str]] = None,
+    command: Optional[str] = None,
+    args: Optional[List[str]] = None,
 ) -> DockerSource:
     """
     Create Docker source configuration.
 
     Args:
         image: Docker image name
-        command_args: Command and arguments to run (optional, empty list means use image default)
         privileged: If True, run the container in privileged mode (default: None/False)
         image_registry_secret: Name of the secret containing registry credentials
             for pulling private images
+        entrypoint: Override the default entrypoint of the Docker image
+        command: Override the default command of the Docker image
+        args: Arguments to pass to the command
 
     Returns:
         DockerSource object
     """
     return DockerSource(
         image=image,
-        command=command_args[0] if command_args else None,
-        args=list(command_args[1:]) if len(command_args) > 1 else None,
+        command=command,
+        args=args,
         privileged=privileged,
         image_registry_secret=image_registry_secret,
+        entrypoint=entrypoint,
     )