Add self-validating sandbox example E2E tests

.github/workflows/examples-e2e.yml

@@ -49,6 +49,8 @@ jobs:
     env:
       KOYEB_API_TOKEN: ${{ secrets.KOYEB_API_TOKEN }}
       KOYEB_API_HOST: ${{ vars.KOYEB_API_HOST || 'https://app.koyeb.com' }}
+      KOYEB_PROJECT_ID: ${{ vars.KOYEB_PROJECT_ID }}
+      KOYEB_REGION: ${{ vars.KOYEB_REGION || 'na' }}
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
@@ -64,6 +66,8 @@ jobs:
     env:
       KOYEB_API_TOKEN: ${{ secrets.KOYEB_API_TOKEN }}
       KOYEB_API_HOST: ${{ vars.KOYEB_API_HOST || 'https://app.koyeb.com' }}
+      KOYEB_PROJECT_ID: ${{ vars.KOYEB_PROJECT_ID }}
+      KOYEB_REGION: ${{ vars.KOYEB_REGION || 'na' }}
       KOYEB_SNAPSHOT_BENCHMARK_INSTANCE_TYPE: micro
     steps:
       - uses: actions/checkout@v4

examples/05_environment_variables.py

@@ -28,7 +28,11 @@ def main():
     try:
         # Create a secret
         secret_response = secrets_api.create_secret(
-            secret=CreateSecret(name=secret_name, value=secret_value)
+            secret=CreateSecret(
+                name=secret_name,
+                value=secret_value,
+                project_id=os.getenv("KOYEB_PROJECT_ID") or None,
+            )
         )
         secret_id = secret_response.secret.id
         print(f"Created secret: {secret_name}")

examples/05_environment_variables_async.py

@@ -29,7 +29,11 @@ async def main():
     try:
         # Create a secret
         secret_response = secrets_api.create_secret(
-            secret=CreateSecret(name=secret_name, value=secret_value)
+            secret=CreateSecret(
+                name=secret_name,
+                value=secret_value,
+                project_id=os.getenv("KOYEB_PROJECT_ID") or None,
+            )
         )
         secret_id = secret_response.secret.id
         print(f"Created secret: {secret_name}")

examples/20_config_files.py

@@ -35,7 +35,11 @@ def main():
     try:
         # Create a secret
         secret_response = secrets_api.create_secret(
-            secret=CreateSecret(name=secret_name, value=secret_value)
+            secret=CreateSecret(
+                name=secret_name,
+                value=secret_value,
+                project_id=os.getenv("KOYEB_PROJECT_ID") or None,
+            )
         )
         secret = secret_response.secret
         secret_id = secret.id

examples/20_config_files_async.py

@@ -36,7 +36,11 @@ async def main():
     try:
         # Create a secret
         secret_response = secrets_api.create_secret(
-            secret=CreateSecret(name=secret_name, value=secret_value)
+            secret=CreateSecret(
+                name=secret_name,
+                value=secret_value,
+                project_id=os.getenv("KOYEB_PROJECT_ID") or None,
+            )
         )
         secret = secret_response.secret
         secret_id = secret.id

examples/README.md

@@ -7,6 +7,8 @@ A collection of examples demonstrating the Koyeb Sandbox SDK capabilities.
 ```bash
 # Set your API token
 export KOYEB_API_TOKEN=your_api_token_here
+export KOYEB_PROJECT_ID=your_project_id # Optional
+export KOYEB_REGION=na                  # Optional
 
 # Optional: API host override (used by the config-file examples)
 export KOYEB_API_HOST=https://app.koyeb.com
@@ -31,9 +33,9 @@ uv run python examples/00_run_all_async.py
 ```
 
 The GitHub Actions workflow needs the `KOYEB_API_TOKEN` repository secret.
-It also accepts the `KOYEB_API_HOST` repository variable. The API token
-selects the Koyeb organization; resources land in that organization's
-default project.
+It also accepts `KOYEB_API_HOST`, `KOYEB_PROJECT_ID`, and `KOYEB_REGION` repository variables.
+The API token selects the Koyeb organization.
+`KOYEB_PROJECT_ID` selects the project inside that organization.
 
 ## Examples
 

koyeb/sandbox/control_plane.py

@@ -109,8 +109,13 @@ class SyncControlPlane:
     def __init__(self, clients: Any):
         self._clients = clients
 
-    def create_app(self, payload: Dict[str, Any]) -> str:
-        reply = self._clients.apps.create_app(app=CreateApp(**payload))
+    def create_app(
+        self, payload: Dict[str, Any], project_id: Optional[str] = None
+    ) -> str:
+        kwargs: Dict[str, Any] = {"app": CreateApp(**payload)}
+        if project_id:
+            kwargs["_headers"] = {"x-koyeb-project-id": project_id}
+        reply = self._clients.apps.create_app(**kwargs)
         return reply.app.id
 
     def delete_app(self, app_id: str) -> None:
@@ -121,8 +126,13 @@ def get_app(self, app_id: str) -> AppInfo:
         domains = [d.name for d in getattr(app, "domains", None) or []]
         return AppInfo(id=app.id, name=app.name, domains=domains)
 
-    def create_service(self, payload: Dict[str, Any]) -> str:
-        reply = self._clients.services.create_service(service=CreateService(**payload))
+    def create_service(
+        self, payload: Dict[str, Any], project_id: Optional[str] = None
+    ) -> str:
+        kwargs: Dict[str, Any] = {"service": CreateService(**payload)}
+        if project_id:
+            kwargs["_headers"] = {"x-koyeb-project-id": project_id}
+        reply = self._clients.services.create_service(**kwargs)
         return reply.service.id
 
     def get_service(self, service_id: str) -> ServiceInfo:
@@ -196,8 +206,13 @@ class AsyncControlPlane:
     def __init__(self, clients: Any):
         self._clients = clients
 
-    async def create_app(self, payload: Dict[str, Any]) -> str:
-        reply = await self._clients.apps.create_app(app=AsyncCreateApp(**payload))
+    async def create_app(
+        self, payload: Dict[str, Any], project_id: Optional[str] = None
+    ) -> str:
+        kwargs: Dict[str, Any] = {"app": AsyncCreateApp(**payload)}
+        if project_id:
+            kwargs["_headers"] = {"x-koyeb-project-id": project_id}
+        reply = await self._clients.apps.create_app(**kwargs)
         return reply.app.id
 
     async def delete_app(self, app_id: str) -> None:
@@ -208,10 +223,13 @@ async def get_app(self, app_id: str) -> AppInfo:
         domains = [d.name for d in getattr(app, "domains", None) or []]
         return AppInfo(id=app.id, name=app.name, domains=domains)
 
-    async def create_service(self, payload: Dict[str, Any]) -> str:
-        reply = await self._clients.services.create_service(
-            service=AsyncCreateService(**payload)
-        )
+    async def create_service(
+        self, payload: Dict[str, Any], project_id: Optional[str] = None
+    ) -> str:
+        kwargs: Dict[str, Any] = {"service": AsyncCreateService(**payload)}
+        if project_id:
+            kwargs["_headers"] = {"x-koyeb-project-id": project_id}
+        reply = await self._clients.services.create_service(**kwargs)
         return reply.service.id
 
     async def get_service(self, service_id: str) -> ServiceInfo:

koyeb/sandbox/sandbox.py

@@ -252,6 +252,7 @@ def create(
         delete_after_delay: int = 0,
         delete_after_inactivity_delay: int = 0,
         app_id: Optional[str] = None,
+        project_id: Optional[str] = None,
         enable_mesh: Optional[bool] = None,
         poll_interval: float = DEFAULT_POLL_INTERVAL,
         entrypoint: Optional[List[str]] = None,
@@ -298,6 +299,7 @@ def create(
                 delete_after_inactivity_delay: If >0, automatically delete the sandbox if service sleeps due to inactivity
                     after this many seconds.
                 app_id: If provided, create the sandbox service in an existing app instead of creating a new one.
+                project_id: Project for new sandbox apps and services. Defaults to KOYEB_PROJECT_ID.
                 enable_mesh: Mesh tri-state: None (default) = auto, True = enabled, False = disabled
                 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"])
@@ -352,6 +354,9 @@ def create(
             if not api_token:
                 raise MissingApiTokenError()
 
+        if project_id is None:
+            project_id = os.getenv("KOYEB_PROJECT_ID") or None
+
         snapshot_id, snapshot_type = _resolve_snapshot_reference(
             snapshot, api_token, host
         )
@@ -371,6 +376,7 @@ def create(
             deep_sleep_value=_experimental_deep_sleep_value,
             delete_after_delay=delete_after_delay,
             delete_after_inactivity_delay=delete_after_inactivity_delay,
+            project_id=project_id,
             enable_mesh=enable_mesh,
             entrypoint=entrypoint,
             command=command,
@@ -442,7 +448,7 @@ def _create_sync(
 
         created_app = False
         if app_id is None:
-            app_id = cp.create_app(spec.app_payload())
+            app_id = cp.create_app(spec.app_payload(), project_id=spec.project_id)
             created_app = True
 
         def _delete_created_app() -> None:
@@ -457,7 +463,9 @@ def _delete_created_app() -> None:
                 )
 
         try:
-            service_id = cp.create_service(spec.create_service_payload(app_id))
+            service_id = cp.create_service(
+                spec.create_service_payload(app_id), project_id=spec.project_id
+            )
         except ApiException as e:
             _delete_created_app()
             raise SandboxError(f"Failed to create sandbox '{spec.name}': {e}") from e
@@ -1660,6 +1668,7 @@ async def create(
         delete_after_delay: int = 0,
         delete_after_inactivity_delay: int = 0,
         app_id: Optional[str] = None,
+        project_id: Optional[str] = None,
         enable_mesh: Optional[bool] = None,
         poll_interval: float = DEFAULT_POLL_INTERVAL,
         entrypoint: Optional[List[str]] = None,
@@ -1708,6 +1717,7 @@ async def create(
                 delete_after_inactivity_delay: If >0, automatically delete the sandbox if service sleeps due to inactivity
                     after this many seconds.
                 app_id: If provided, create the sandbox service in an existing app instead of creating a new one.
+                project_id: Project for new sandbox apps and services. Defaults to KOYEB_PROJECT_ID.
                 enable_mesh: Mesh tri-state: None (default) = auto, True = enabled, False = disabled
                 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"])
@@ -1736,6 +1746,9 @@ async def create(
             if not api_token:
                 raise MissingApiTokenError()
 
+        if project_id is None:
+            project_id = os.getenv("KOYEB_PROJECT_ID") or None
+
         snapshot_id, snapshot_type = _resolve_snapshot_reference(
             snapshot, api_token, host
         )
@@ -1755,6 +1768,7 @@ async def create(
             deep_sleep_value=_experimental_deep_sleep_value,
             delete_after_delay=delete_after_delay,
             delete_after_inactivity_delay=delete_after_inactivity_delay,
+            project_id=project_id,
             enable_mesh=enable_mesh,
             entrypoint=entrypoint,
             command=command,
@@ -1773,7 +1787,9 @@ async def create(
         # Use provided app_id or create a new app
         created_app = False
         if app_id is None:
-            app_id = await cp.create_app(spec.app_payload())
+            app_id = await cp.create_app(
+                spec.app_payload(), project_id=spec.project_id
+            )
             created_app = True
 
         async def _delete_created_app() -> None:
@@ -1788,7 +1804,9 @@ async def _delete_created_app() -> None:
                 )
 
         try:
-            service_id = await cp.create_service(spec.create_service_payload(app_id))
+            service_id = await cp.create_service(
+                spec.create_service_payload(app_id), project_id=spec.project_id
+            )
         except AsyncApiException as e:
             await _delete_created_app()
             raise SandboxError(f"Failed to create sandbox '{name}': {e}") from e

koyeb/sandbox/snapshot.py

@@ -353,6 +353,8 @@ def spawn(
             create_params["api_token"] = self.api_token
         if self.host:
             create_params["host"] = self.host
+        if self.project_id:
+            create_params["project_id"] = self.project_id
         if self.sandbox_secret:
             create_params["sandbox_secret"] = self.sandbox_secret
 

koyeb/sandbox/spec.py

@@ -381,6 +381,7 @@ class SandboxSpec:
     deep_sleep_value: int = 3900
     delete_after_delay: int = 0
     delete_after_inactivity_delay: int = 0
+    project_id: Optional[str] = None
     enable_mesh: Optional[bool] = None
     entrypoint: Optional[List[str]] = None
     command: Optional[str] = None

koyeb/sandbox/test_egress_policy.py

@@ -45,6 +45,24 @@ def test_allowlist_sends_deny_all_with_destinations(self, mock_get_clients):
             [d.cidr for d in egress.allow_list], ["1.2.3.4/32", "10.0.0.0/8"]
         )
 
+    @patch("koyeb.sandbox.sandbox.get_api_clients")
+    def test_project_id_scopes_created_app_and_service(self, mock_get_clients):
+        clients = MagicMock()
+        clients.apps.create_app.return_value.app.id = "mock-app-id"
+        mock_get_clients.return_value = clients
+
+        Sandbox.create(
+            name="t",
+            api_token="tok",
+            project_id="project-id",
+            wait_ready=False,
+        )
+
+        app_headers = clients.apps.create_app.call_args.kwargs["_headers"]
+        service_headers = clients.services.create_service.call_args.kwargs["_headers"]
+        self.assertEqual(app_headers["x-koyeb-project-id"], "project-id")
+        self.assertEqual(service_headers["x-koyeb-project-id"], "project-id")
+
     @patch("koyeb.sandbox.sandbox.get_api_clients")
     def test_mutually_exclusive_fails_before_any_api_call(self, mock_get_clients):
         with self.assertRaises(EgressPolicyError):
@@ -73,6 +91,28 @@ def test_async_create_forwards_egress_kwargs(self, mock_get_clients):
         egress = service.definition.network_policy.egress
         self.assertEqual(egress.mode, EgressPolicyMode.EGRESS_POLICY_MODE_DENY_ALL)
 
+    @patch("koyeb.sandbox.sandbox.get_async_api_clients")
+    def test_async_project_id_scopes_created_app_and_service(self, mock_get_clients):
+        clients = MagicMock()
+        clients.apps.create_app = AsyncMock()
+        clients.apps.create_app.return_value.app.id = "mock-app-id"
+        clients.services.create_service = AsyncMock()
+        mock_get_clients.return_value = clients
+
+        asyncio.run(
+            AsyncSandbox.create(
+                name="t",
+                api_token="tok",
+                project_id="project-id",
+                wait_ready=False,
+            )
+        )
+
+        app_headers = clients.apps.create_app.call_args.kwargs["_headers"]
+        service_headers = clients.services.create_service.call_args.kwargs["_headers"]
+        self.assertEqual(app_headers["x-koyeb-project-id"], "project-id")
+        self.assertEqual(service_headers["x-koyeb-project-id"], "project-id")
+
     @patch("koyeb.sandbox.sandbox.get_async_api_clients")
     def test_async_mutually_exclusive_fails_before_any_api_call(self, mock_get_clients):
         with self.assertRaises(EgressPolicyError):

koyeb/sandbox/test_snapshot.py

@@ -0,0 +1,24 @@
+from datetime import datetime
+from unittest.mock import patch
+
+from koyeb.sandbox.sandbox import Sandbox
+from koyeb.sandbox.snapshot import Snapshot, SnapshotStatus, SnapshotType
+
+
+@patch.object(Sandbox, "create")
+def test_spawn_preserves_snapshot_project(mock_create):
+    snapshot = Snapshot(
+        id="snapshot-id",
+        name="snapshot",
+        service_id="service-id",
+        snapshot_type=SnapshotType.FILESYSTEM,
+        status=SnapshotStatus.AVAILABLE,
+        created_at=datetime.now(),
+        project_id="project-id",
+        api_token="token",
+        sandbox_secret="secret",
+    )
+
+    snapshot.spawn(name="restored")
+
+    assert mock_create.call_args.kwargs["project_id"] == "project-id"