Add Prefect worker example

README.md

@@ -63,6 +63,7 @@ Give an agent a sandboxed place to run the code it writes.
 | [Covalent](examples/covalent-tenki/) | Each workflow task in its own microVM |
 | [AgentBox](examples/agentbox-tenki/) | Coding-agent boxes as Firecracker microVMs |
 | [ComputeSDK](examples/computesdk-tenki/) | Tenki as a provider for the unified sandbox interface |
+| [Prefect](examples/prefect-tenki/) | Each flow run in its own microVM via a Prefect 3.x worker |
 
 ### Agent platforms
 

examples/prefect-tenki/README.md

@@ -0,0 +1,71 @@
+# Prefect flow runs on Tenki
+
+Run [Prefect 3.x](https://docs.prefect.io) flow runs where **each flow run executes in its own disposable [Tenki](https://tenki.cloud) microVM** — a fresh Linux VM with real root, created when the run is scheduled and destroyed the moment it finishes. In Prefect 3.x, a *worker* polls a *work pool* and provisions infrastructure per flow run; [`prefect-tenki`](https://github.com/luxorlabs/tenki-prefect) adds a worker type `tenki` that makes that infrastructure a Tenki sandbox.
+
+> **Not on PyPI yet.** `prefect-tenki` is pre-alpha (`0.1.0.dev0`) and installs from git — [`requirements.txt`](requirements.txt) pins a known-good commit.
+
+## Setup (Python 3.10+)
+
+```bash
+uv venv                                 # or: python3.11 -m venv .venv
+uv pip install -r requirements.txt      # prefect-tenki (from git) + prefect + tenki SDK
+export TENKI_API_KEY=tk_...             # from the Tenki dashboard
+```
+
+## Walkthrough
+
+**1. Create a work pool.** Installing `prefect-tenki` registers the worker type via a `prefect.collections` entry point, so the CLI discovers it:
+
+```bash
+prefect work-pool create --type tenki my-tenki-pool
+```
+
+**2. Save credentials as a block.** The API key lives in a Prefect `TenkiCredentials` block, used only by the worker process to call the Tenki API — it is **never injected into the guest VM's environment**:
+
+```python
+from prefect_tenki import TenkiCredentials
+
+TenkiCredentials(api_key="tk_...").save("tenki-prod")
+```
+
+**3. Set job variables** on the pool (UI or `prefect work-pool update`): `workspace_id`, sizing (`cpu_cores` default 2, `memory_mb` default 4096, optional `disk_size_gb`), and `image` *or* `snapshot_id`.
+
+> **The image must be able to run Prefect.** Inside the sandbox the worker runs Prefect's prepared `prefect flow-run execute` command — the guest pulls your flow code itself. So the image/snapshot needs **Prefect + your flow's dependencies preinstalled**, and `allow_outbound: true` (the default) so it can **reach the Prefect API and your code source**. That also means the Prefect API must be reachable *from the microVM*: use Prefect Cloud or a server on a real address — `http://localhost:4200` only exists on your laptop.
+
+**4. Start a worker** (any machine with the venv + credentials — your laptop works):
+
+```bash
+prefect worker start --pool my-tenki-pool
+```
+
+**5. Deploy and run:**
+
+```bash
+python -c "
+from prefect import flow
+
+@flow(log_prints=True)
+def hello(): print('hello from a Tenki microVM')
+
+hello.from_source('https://github.com/you/your-repo', entrypoint='flows.py:hello') \
+    .deploy(name='hello-tenki', work_pool_name='my-tenki-pool')
+"
+prefect deployment run 'hello/hello-tenki'
+```
+
+Each run now gets its own microVM: the worker creates a sandbox (`create_timeout_seconds`, default 180 s), streams stdout/stderr into the flow-run logs, and closes it when the command exits.
+
+## Lifecycle guarantees
+
+- **Credential isolation** — the API key goes only into the worker's SDK client; the guest env gets Prefect's own variables, never your Tenki credentials.
+- **Cancellation-shielded cleanup** — sandbox create/close run inside shielded scopes, so a cancelled or crashing run can't leak a VM mid-request; teardown always runs.
+- **`kill_infrastructure`** — cancelling a flow run in Prefect terminates the sandbox by ID (the sandbox ID is the run's infrastructure PID).
+- **Server-side backstop** — `max_duration_seconds` (default 3600) self-destructs the sandbox even if the worker dies. No leaked billing.
+
+## Verify (fast smoke check)
+
+```bash
+node verify.mjs      # or: .venv/bin/python verify.py
+```
+
+CI can't run a real flow: the guest must reach the Prefect API, and a localhost `prefect server` is unreachable from inside the microVM. So [`verify.py`](verify.py) does what the upstream repo's own live smoke test does — checks the worker type registers, then drives **`TenkiWorker.run()` directly** with a stand-in command: create a live microVM → run it → assert exit 0 → teardown. That exercises everything Tenki-specific in the worker; the Prefect-side orchestration is covered by Prefect itself and the plugin's CI in [luxorlabs/tenki-prefect](https://github.com/luxorlabs/tenki-prefect).

examples/prefect-tenki/requirements.txt

@@ -0,0 +1 @@
+prefect-tenki @ git+https://github.com/luxorlabs/tenki-prefect.git@1778874be00ddd725137e4aa1fd23e5d8d744f76

examples/prefect-tenki/verify.mjs

@@ -0,0 +1,11 @@
+/**
+ * Shim so the cookbook's Node verify harness can run this Python example.
+ * Runs verify.py with the example's local venv if present, else `python3`.
+ * Setup (Python 3.10+): `uv venv && uv pip install -r requirements.txt` (see README).
+ */
+import { spawnSync } from "node:child_process";
+import { existsSync } from "node:fs";
+
+const py = existsSync(".venv/bin/python") ? ".venv/bin/python" : process.env.PYTHON || "python3";
+const r = spawnSync(py, ["verify.py"], { stdio: "inherit" });
+process.exit(r.status ?? 1);

examples/prefect-tenki/verify.py

@@ -0,0 +1,91 @@
+"""
+Smoke-verify the Prefect example without a Prefect server:
+  1. `prefect-tenki` imports and registers the "tenki" worker type, and
+  2. `TenkiWorker.run()` drives the real per-flow-run infrastructure lifecycle
+     against live Tenki — create microVM → run command → teardown — exit 0.
+
+We deliberately do NOT execute a full flow run: the guest must reach the
+Prefect API, and a localhost `prefect server` is unreachable from inside the
+microVM. Driving run() directly is the upstream repo's own smoke pattern
+(tests/live_worker_smoke.py) and exercises everything Tenki-specific.
+
+Token/workspace from env (CI) or ~/.config/tenki/config.yaml (local `tenki login`).
+"""
+import logging
+import os
+import sys
+import tempfile
+
+# Keep ~/.prefect untouched — prefect writes its profile/db under PREFECT_HOME.
+os.environ.setdefault("PREFECT_HOME", tempfile.mkdtemp(prefix="prefect-tenki-verify-"))
+
+import anyio
+from tenki import AsyncClient
+
+from prefect_tenki import TenkiCredentials, TenkiWorker, TenkiWorkerJobConfiguration
+
+assert TenkiWorker.type == "tenki", TenkiWorker.type  # the work-pool type name
+
+
+def cfg(key):
+    try:
+        with open(os.path.expanduser("~/.config/tenki/config.yaml")) as f:
+            for line in f:
+                if line.startswith(key + ":"):
+                    return line.split(":", 1)[1].strip()
+    except Exception:
+        pass
+    return ""
+
+
+token = os.environ.get("TENKI_AUTH_TOKEN") or os.environ.get("TENKI_API_KEY") or cfg("auth_token")
+if not token:
+    print("No token. Set TENKI_AUTH_TOKEN, or run `tenki login`.")
+    sys.exit(1)
+
+# SDK auth gap: a bare `tenki login` session token is sent as Bearer (rejected);
+# prefix it `cookie:` so it goes as a cookie. A `tk_` API key works as-is.
+if not token.startswith(("tk_", "ory_st_", "cookie:")):
+    token = f"cookie:{token}"
+
+
+async def main():
+    credentials = TenkiCredentials(api_key=token)
+
+    workspace_id = os.environ.get("TENKI_WORKSPACE_ID") or cfg("current_workspace_id")
+    if not workspace_id:
+        async with AsyncClient(**credentials.get_client_options()) as client:
+            workspaces = list((await client.who_am_i()).workspaces)
+        if len(workspaces) != 1:
+            raise RuntimeError(f"{len(workspaces)} workspaces accessible; set TENKI_WORKSPACE_ID.")
+        workspace_id = workspaces[0].id
+
+    # Same shape a `prefect worker start` builds per flow run, minus the
+    # Prefect-prepared `prefect flow-run execute` command.
+    configuration = TenkiWorkerJobConfiguration(
+        name="prefect-tenki-verify",
+        command="set -eu; test \"$(python3 -c 'print(6*7)')\" = \"42\"; echo ok",
+        credentials=credentials,
+        workspace_id=workspace_id,
+        cpu_cores=1,
+        memory_mb=1024,
+        allow_inbound=False,
+        allow_outbound=False,
+        max_duration_seconds=120,
+        create_timeout_seconds=180,
+        command_timeout_seconds=60,
+        stream_output=True,
+    )
+    worker = TenkiWorker(work_pool_name="prefect-tenki-verify")
+    result = await worker.run(object(), configuration)
+    if result.status_code != 0:
+        raise AssertionError(f"exit {result.status_code} (sandbox {result.identifier})")
+    print("✓ prefect-tenki: TenkiWorker.run() → live microVM → flow command exit 0 → teardown")
+
+
+logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+try:
+    anyio.run(main)
+except Exception as e:  # noqa
+    print(f"✗ {type(e).__name__}: {e}")
+    sys.exit(1)