Docs · reference
Python package
The denv Python client and its Harbor, Inspect, and OpenEnv adapters, method by method.
pip install "denv[all]" # or denv[harbor], denv[inspect], denv[openenv]
export DENV_SERVER=https://api.denv.ai DENV_API_KEY=dak_…
Package denv, version 0.1.0, Python ≥ 3.11. Dependencies: httpx, pydantic, pyyaml. Optional extras pull harbor, inspect-ai, openenv-core. The package registers the entry point inspect_ai = denv._inspect_registry so Inspect finds the sandbox without an import.
denv.Client
from denv import Client
c = Client(server=None, api_key=None, timeout=900.0)
| Argument | Default | Meaning |
|---|---|---|
server |
DENV_SERVER, else http://localhost:8080 |
Control plane base URL. |
api_key |
DENV_API_KEY |
Required; RuntimeError("DENV_API_KEY is not set") otherwise. |
timeout |
900 s | Default HTTP timeout. |
All methods are synchronous. Any HTTP error raises RuntimeError("denv <METHOD> <path>: <status> <message>").
Sandboxes and versions
| Method | Returns | Behaviour |
|---|---|---|
sandbox(*, image=None, from_version=None, cmd=None, env=None, ports=None, egress="deny", volumes=None, name=None, project=None, timeout_sec=600) |
Sandbox |
POST /v1/sandboxes. One of image or from_version is required (ValueError). Returns when healthy. |
destroy(env_id) |
None |
DELETE /v1/envs/{id}. |
current_version(env_id) |
str |
The environment's CurrentVersionID. |
snapshot(env_id, timeout_sec=300) |
str |
POST …/snapshot, waits for the deployment, returns the new version id. |
fork(env_id, name=None, timeout_sec=600, live=True) |
Sandbox |
With live=True (default) snapshots first, then creates a sandbox from that version; with live=False forks the current version. |
verify(env_id, resume=False, timeout_sec=900) |
Verdict |
POST …/verify, waits, reads the verdict phase's evidence. Raises RuntimeError if the deployment failed. |
Inside an environment
| Method | Returns | Behaviour |
|---|---|---|
exec(env_id, service, cmd: list[str], timeout_sec=120) |
ExecResult |
POST …/exec. |
put_tar(env_id, service, path, tar_bytes) |
None |
PUT …/files; the archive is extracted into path. |
get_tar(env_id, service, path) |
bytes |
GET …/files; a tar of path. |
put_file(env_id, service, source, target_path) |
None |
Uploads one local file to the file path target_path. |
put_bytes(env_id, service, target_path, data) |
None |
Writes data to the file path target_path. |
put_dir(env_id, service, source_dir, target_dir) |
None |
Uploads a directory's contents into target_dir. |
get_file(env_id, service, source_path, target) |
None |
Downloads one file to the local path target. |
get_bytes(env_id, service, source_path) |
bytes |
The first regular file of the archive; FileNotFoundError if none. |
get_dir(env_id, service, source_dir, target_dir) |
None |
Extracts the directory's contents under target_dir (docker cp src/. semantics). |
Dataclasses
@dataclass
class Sandbox:
id: str; name: str; status: str
endpoints: dict[str, str] # service → "127.0.0.1:port" (only with egress allow)
raw: dict # the environment object as returned
@dataclass
class ExecResult:
exit_code: int; stdout: str; stderr: str
@dataclass
class Verdict:
deployment_id: str
reward: float | None # None when the deployment failed
refused: list[str] # "<path>: <reason>"
missing: list[str]
source: str # reward.txt | stdout | refused | error
status: str # succeeded | failed | rolled_back
evidence: dict # the verdict phase's evidence as returned
shell_command
from denv.client import shell_command
argv = shell_command(script, *, cwd=None, env=None, user=None) -> list[str]
Builds one argv that runs script through bash -c when the image has bash and sh -c otherwise, with env exported first and cwd applied; a non-root user wraps it in su -s /bin/sh <user> -c. Used by the adapters.
Example
from denv import Client
c = Client()
sbx = c.sandbox(image="python:3.12-slim", cmd=["sleep", "infinity"])
c.put_bytes(sbx.id, "main", "/work/a.txt", b"alpha")
print(c.exec(sbx.id, "main", ["cat", "/work/a.txt"]).stdout) # alpha
child = c.fork(sbx.id) # carries /work
c.destroy(child.id); c.destroy(sbx.id)
Harbor adapter
harbor run --dataset <ds> --agent <agent> -e denv.harbor:DenvEnvironment
denv.harbor.DenvEnvironment subclasses Harbor's BaseEnvironment. type() returns "denv"; capabilities reports disable_internet=True, docker_compose=True (not mounted, so Harbor downloads logs).
| Task input | What the adapter does |
|---|---|
environment/docker-compose.yaml (or .yml) present |
The file is deployed as environment harbor-<session> of a project named sandboxes (created if missing). The agent runs in service main. |
[environment] docker_image set |
sandbox(image=…, cmd=["sleep","infinity"], volumes=["/app","/logs", workdir?]). |
Only a Dockerfile |
Refused at _validate_definition: denv: this task has only a Dockerfile. Set [environment] docker_image in task.toml (or add a compose file) until denv ships host-side builds. |
allow_internet = true |
egress: allow; otherwise deny. |
After start, /logs/agent /logs/verifier /logs/artifacts /tests /solution /app are created in the service. exec(command, cwd, env, timeout_sec, user) runs through shell_command with Harbor's merged env and resolved user (default timeout 600 s). upload_file/upload_dir/download_file/download_dir map to the client's put_file/put_dir/get_file/get_dir. stop(delete=True) destroys the environment; delete=False leaves it.
Not mapped: cpus, memory_mb, storage_mb, gpus from task.toml (the sidecar defaults apply); fork reuse across -k attempts; a separate-verifier mode. Compose tasks always use service main as the agent service.
Inspect adapter
Task(..., sandbox="denv") # image from DENV_SANDBOX_IMAGE, default python:3.12-slim
Task(..., sandbox=("denv", "denv.yaml")) # a config file
Task(..., sandbox=("denv", DenvSandboxConfig(image="…"))) # inline
Task(..., sandbox=("denv", "ver_01…")) # fork of a version
Inspect treats a string config as a file path, so an image name must go in a file or a DenvSandboxConfig; a bare non-file string that does not start with ver_ raises ValueError. A denv.yaml or denv.yml next to the task is picked up automatically (config_files()).
denv.yaml keys (DenvSandboxConfig):
| Key | Default | Meaning |
|---|---|---|
image |
— | One container from this image, command sleep infinity. |
from_version |
— | Or: fork this version. |
egress |
deny |
deny or allow. |
service |
main |
Service exec runs in. |
volumes |
["/work"] |
Paths that persist across snapshot and fork (image form). |
Behaviour: task_init constructs a Client so a missing key fails before any model call; sample_init creates one sandbox per sample and returns {"default": DenvSandbox}; sample_cleanup destroys it; exec(cmd, input, cwd, env, user, timeout) joins the argv with shlex.quote, writes input to /tmp/.denv_stdin and redirects it, runs through shell_command (default timeout 300 s) and returns Inspect's ExecResult; write_file/read_file use put_bytes/get_bytes; connection() returns the command denv exec --service <svc> <env> -- sh; default_concurrency() is 8. task_cleanup and cli_cleanup are no-ops.
OpenEnv adapter
from denv.openenv import DenvProvider
env = MyEnv.from_docker_image("org/my-env:v1", provider=DenvProvider())
DenvProvider(client=None, egress="deny") implements OpenEnv's ContainerProvider:
| Method | Behaviour |
|---|---|
start_container(image, port=None, env_vars=None, **kwargs) |
sandbox(image, cmd=kwargs.get("cmd"), env, ports=[port or 8000], egress). Sets and returns base_url = "http://<endpoint>". Raises RuntimeError when no endpoint was published, which is the case with egress="deny" because a denied network is internal; pass egress="allow". |
wait_for_ready(base_url, timeout_s=30) |
Polls GET <base_url>/health until the status is below 500. |
fork() |
client.fork(env_id) (snapshot then fork) and returns a new DenvProvider bound to the child, base_url set if the child published an endpoint. |
stop_container() / close() |
Destroy the environment. |
Not yet available: from_denv_version, denv export --openenv.
Conformance suite
make e2e-frameworks # all checks
./scripts/e2e-frameworks.sh verify harbor # a subset
scripts/e2e-frameworks.sh starts a control plane and a sidecar on the local machine (Docker and python/.venv required) and runs python/tests/conformance.py. Checks: client (exec, files, fork carries /work, fork is isolated), verify (correct answer scores 1, a symlink scores 0 and is refused, a FIFO scores 0, the agent stays frozen without resume), inspect (an eval with mockllm/model, accuracy 1.0, plugin found via the entry point), harbor (the oracle agent on a fixture task, reward 1), openenv (python -m http.server as the image, ready, stopped). The script exits non-zero with the sidecar and control-plane logs on failure.