# Python API reference (/docs/python/reference)



This page lists the public methods on the `CrowNest` and `AsyncCrowNest`
clients, grouped by resource. Keyword arguments are snake\_case; returned
objects are JSON dictionaries whose field names match the REST API
(camelCase), so a command result exposes `result["exitCode"]`.

List-style methods return the REST envelope, not a bare list. Read records from
`page["data"]`; pagination metadata such as `hasMore` remains on the page
dictionary when the REST endpoint exposes it.

Sync and async clients have full parity: every method below exists on
both, and the async variant takes the same arguments with `await`.
`stream_logs` is a regular iterator on the sync client and an async
iterator on the async client.

## Clients [#clients]

Construct a client directly or as a context manager.

```python
from crownest import CrowNest, AsyncCrowNest, CrowNestApiError

client = CrowNest(
    credential=None,
    api_key=None,
    base_url="https://api.crownest.dev",
    timeout=660,
)
```

* `credential` falls back to `CROWNEST_BEARER_TOKEN`, then
  `CROWNEST_API_KEY` for developer API-key compatibility.
* `api_key` remains accepted as a backwards-compatible alias.
* The sync client supports `with` and `.close()`; the async client
  supports `async with`.
* Failed calls raise `CrowNestApiError` with `.status`, `.code`, and
  `.details`.

## client.sandboxes [#clientsandboxes]

Manages sandbox lifecycle. `create`, `get`, `extend`, and `kill` return a
`SandboxHandle`; `list` returns a page whose `data` entries are
`SandboxHandle` objects.

### create [#create]

```python
client.sandboxes.create(
    *,
    template=None,            # "python-node" by default
    template_version_id=None, # pin an immutable TemplateVersion ("tplv_...")
    ttl_ms=None,              # requested lifetime in ms (plan-limited)
    metadata=None,            # dict[str, str], max 16 keys
    project_id=None,          # "prj_..."
    idempotency_key=None,     # auto-generated when omitted
) -> SandboxHandle
```

Returns a handle whose `expiresAt` reflects the resolved TTL. Raises
`sandbox_ttl_exceeded` if `ttl_ms` exceeds your plan limit, and
`quota_exceeded` at the concurrency limit.

### get [#get]

```python
client.sandboxes.get(sandbox_id) -> SandboxHandle
```

### extend [#extend]

```python
client.sandboxes.extend(
    sandbox_id,
    *,
    ttl_ms,                 # new lifetime in ms (plan-limited)
    idempotency_key=None,   # auto-generated when omitted
) -> SandboxHandle
```

Resets a live sandbox's remaining TTL from now and returns a handle whose
`expiresAt` reflects the refreshed deadline. Raises `invalid_request` when
the requested TTL would not extend the current expiry, `sandbox_destroyed`
when the sandbox is already destroyed, and `sandbox_ttl_exceeded` when
the requested lifetime exceeds your plan limit.

### list [#list]

```python
client.sandboxes.list(*, metadata=None) -> dict
```

Returns live sandboxes; destroyed and failed records are excluded by
default. The response has `data` and `hasMore`; each item in `data` is a
`SandboxHandle`.

### kill [#kill]

```python
client.sandboxes.kill(sandbox_id) -> SandboxHandle
```

Destroys the sandbox and returns the record with status `destroyed`.
Idempotent by state.

### SandboxHandle [#sandboxhandle]

The handle returned by `create` and `get` wraps the sandbox object:

* Attribute and mapping access: `sandbox.id`, `sandbox["status"]`.
* `.to_dict()` returns the plain dictionary.
* `.extend(ttl_ms=...)` refreshes the sandbox TTL.
* `.kill()` destroys the sandbox.
* `.wait_until_ready()` polls until the sandbox can accept work.
* `.commands`, `.files`, `.artifacts`, `.previews` are sub-clients bound
  to the sandbox — the same methods as below without the `sandbox_id`
  argument.

## client.code [#clientcode]

Runs interpreter-style code snippets and manages Code Contexts. SDK
helpers default `language` to `"python"`.

### create\_context [#create_context]

```python
client.code.create_context(
    sandbox_id,
    *,
    language="python",
    cwd=None,
    timeout_ms=None,
    idempotency_key=None,
) -> dict
```

Returns a Code Context dictionary with `id`, `sandboxId`, `language`, `cwd`,
and `createdAt`.

### list\_contexts [#list_contexts]

```python
client.code.list_contexts(sandbox_id) -> dict
```

Returns active Code Contexts for the sandbox. `sandbox.code.list_contexts()`
is the sandbox-bound equivalent. The response has `data` and `hasMore`.

### get\_context [#get_context]

```python
client.code.get_context(sandbox_id, context_id) -> dict
```

Returns one active Code Context. `sandbox.code.get_context(context_id)` is
the sandbox-bound equivalent.

### delete\_context [#delete_context]

```python
client.code.delete_context(sandbox_id, context_id) -> dict
```

### run [#run]

```python
client.code.run(
    sandbox_id,
    code,
    *,
    language="python",
    context_id=None,
    cwd=None,
    timeout_ms=None,
    artifact_policy=None,
    idempotency_key=None,
) -> dict
```

### run\_stream [#run_stream]

```python
client.code.run_stream(sandbox_id, code, *, language="python", ...)
# sync: Iterator[dict]; async: AsyncIterator[dict]
```

## client.commands [#clientcommands]

Runs and tracks processes inside a sandbox. A non-zero exit code doesn't
raise; check the returned dictionary's `status` and `exitCode`.

### run [#run-1]

```python
client.commands.run(
    sandbox_id,
    command,
    *,
    cwd=None,             # working directory, default /workspace
    env=None,             # dict[str, str]; CROWNEST* keys rejected
    timeout_ms=None,      # default 60000, max 600000
    collect=None,         # list of {"path": str, "name": str?}
    collect_on=None,      # "success" (default) | "always"
    idempotency_key=None, # auto-generated when omitted
) -> dict
```

Waits for the command to exit and returns the full command record
including `exitCode`, `stdout`, and `stderr`.

### start [#start]

```python
client.commands.start(
    sandbox_id,
    command,
    *,
    cwd=None,
    env=None,
    timeout_ms=None,
    idempotency_key=None,
) -> dict
```

Returns immediately with status `queued` or `starting`. `start` doesn't
accept `collect` or `collect_on`.

### get [#get-1]

```python
client.commands.get(command_id) -> dict
```

### cancel [#cancel]

```python
client.commands.cancel(command_id, *, mode=None) -> dict
```

`mode` is `"graceful"` (default, `SIGTERM`) or `"force"` (`SIGKILL`).

### logs [#logs]

```python
client.commands.logs(command_id, *, after_seq=None, limit=None) -> dict
```

Returns stored log chunks
(`{"commandId", "createdAt", "data", "seq", "stream"}`). When `limit` is
omitted the server applies its own bound. The response has `data` and
`hasMore`.

### stream\_logs [#stream_logs]

```python
client.commands.stream_logs(command_id, *, after_seq=None, reconnect=True)
# sync: Iterator[dict]; async: AsyncIterator[dict]
```

Streams events live and blocks until the terminal event. Events are
dictionaries with `type` of `log`, `heartbeat`, `terminal`, or `error`:

```python
cmd = client.commands.start(sandbox.id, "pytest -q")
for event in client.commands.stream_logs(cmd["id"]):
    if event["type"] == "log":
        print(event["data"], end="")
    elif event["type"] == "terminal":
        print("exit code:", event["command"]["exitCode"])
```

## client.files [#clientfiles]

Manages files inside `/workspace`. Direct reads and writes are capped at
256 KB; use `download_url` for larger files. Paths outside the workspace
raise `path_outside_workspace`.

### read [#read]

```python
client.files.read(sandbox_id, path, *, encoding=None) -> str
```

`encoding` is `"utf8"` (default) or `"base64"`.

### write [#write]

```python
client.files.write(
    sandbox_id,
    path,
    content,
    *,
    create_parents=None,  # default False
    encoding=None,        # "utf8" (default) | "base64"
    overwrite=None,       # default True
) -> dict
```

Returns the file's stat record.

### delete [#delete]

```python
client.files.delete(sandbox_id, path) -> None
```

Deletes a file or empty directory. A non-empty directory raises
`directory_not_empty`; there's no recursive delete.

### list [#list-1]

```python
client.files.list(sandbox_id, path="/workspace") -> dict
```

Returns an envelope dictionary with `data`.

### stat [#stat]

```python
client.files.stat(sandbox_id, path) -> dict
```

### mkdir [#mkdir]

```python
client.files.mkdir(sandbox_id, path, *, parents=None) -> dict
```

`parents` defaults to `False`.

### move [#move]

```python
client.files.move(sandbox_id, from_path, to_path, *, overwrite=None) -> dict
```

`overwrite` defaults to `False`.

### download\_url [#download_url]

```python
client.files.download_url(sandbox_id, path) -> dict
```

Returns `{"url", "method": "GET", "authMode": "api_key"}` — a
short-lived signed URL for files of any size.

## client.artifacts [#clientartifacts]

Exports workspace files to durable storage and manages them afterward.
Artifacts outlive their sandbox.

### create [#create-1]

```python
client.artifacts.create(
    sandbox_id,
    *,
    path,                  # source path inside /workspace
    name=None,             # display name, derived from path when omitted
    idempotency_key=None,  # auto-generated when omitted
) -> dict
```

### get [#get-2]

```python
client.artifacts.get(artifact_id) -> dict
```

### list [#list-2]

```python
client.artifacts.list(sandbox_id) -> dict
```

Returns a page dictionary with `data` and `hasMore`.

### download [#download]

```python
client.artifacts.download(artifact_id) -> bytes
```

### download\_url [#download_url-1]

```python
client.artifacts.download_url(artifact_id) -> dict
```

Returns `{"url", "method", "headers", "authMode"}` for a short-lived
signed download.

### delete [#delete-1]

```python
client.artifacts.delete(artifact_id) -> dict
```

Idempotent by state; returns the artifact with `deletedAt` set.

## client.previews [#clientpreviews]

Exposes authenticated HTTP services from a sandbox at URLs like
`https://p-a1b2c3.crownest.dev`.

### create [#create-2]

```python
client.previews.create(sandbox_id, *, port, auth_mode=None) -> dict
```

`port` is the in-sandbox port (1–65535). `auth_mode` accepts
`"authenticated"` or `"token"` and defaults to `"authenticated"`. The
response is `{"preview": {...}, "previewToken": "pvt_..."}` for newly
created token-mode previews; the token is returned only once. One active
preview exists per sandbox and port; creating the same one again returns
the existing preview envelope without recovering the token.

### get [#get-3]

```python
client.previews.get(preview_id) -> dict
```

### list [#list-3]

```python
client.previews.list(sandbox_id) -> dict
```

Returns a page dictionary with `data` and `hasMore`.

### revoke [#revoke]

```python
client.previews.revoke(preview_id) -> dict
```

Idempotent; returns the preview with `revokedAt` set.

## client.projects [#clientprojects]

### create [#create-3]

```python
client.projects.create(name="Agent Workspace") -> dict
```

Requires an org-wide key with `project:create`.

### list [#list-4]

```python
client.projects.list() -> dict
```

Returns the projects your API key can access, each with `id`, `orgId`,
`name`, and `createdAt` in `data`.

## client.workspace\_runs [#clientworkspace_runs]

Runs a command inside an uploaded archive using the `python-node` template.
Use Workspace Runs when you have a `.tar.gz` or `.tgz` archive and want
CrowNest to execute a command inside a fresh sandbox.

### run\_archive [#run_archive]

```python
client.workspace_runs.run_archive(
    content,                # bytes
    *,
    command,
    sha256,
    size_bytes,
    project_id=None,
    template=None,          # "python-node" by default
    template_version_id=None,
    sandbox_id=None,
    keep_sandbox=None,
    timeout_ms=None,
    metadata=None,
    source_metadata=None,
    artifacts=None,         # list of {"path": str, "name": str?}
    idempotency_key=None,
) -> dict
```

Creates the run, creates a staged archive transfer, uploads the archive body,
finalizes the archive, and starts execution. It returns the started run.

### create [#create-4]

```python
client.workspace_runs.create(
    *,
    command,
    project_id=None,
    template=None,          # "python-node" by default
    template_version_id=None,
    sandbox_id=None,
    keep_sandbox=None,
    timeout_ms=None,
    metadata=None,
    source_metadata=None,
    artifacts=None,         # list of {"path": str, "name": str?}
    idempotency_key=None,
) -> dict
```

Returns a run record with `id` and `status="awaiting_archive"`.

### upload\_archive [#upload_archive]

```python
client.workspace_runs.upload_archive(
    workspace_run_id,
    *,
    body,                   # bytes
    sha256,
    size_bytes,
    idempotency_key=None,
) -> dict
```

Uploads a small archive directly through the API. Capped at 8 MiB.

### create\_archive\_transfer [#create_archive_transfer]

```python
client.workspace_runs.create_archive_transfer(
    workspace_run_id,
    *,
    sha256,
    size_bytes,
    idempotency_key=None,
) -> dict
```

Creates a staged upload target with `uploadUrl`, `method`, `headers`,
`expiresAt`, and `maxSizeBytes`.

### upload\_archive\_to\_transfer [#upload_archive_to_transfer]

```python
client.workspace_runs.upload_archive_to_transfer(
    transfer,               # dict from create_archive_transfer
    *,
    body,                   # bytes
) -> None
```

Uploads bytes to the staged transfer target. External upload URLs are sent
without CrowNest API authentication.

### finalize\_archive [#finalize_archive]

```python
client.workspace_runs.finalize_archive(
    workspace_run_id,
    *,
    upload_id,
    sha256,
    size_bytes,
    idempotency_key=None,
) -> dict
```

Verifies and attaches the staged upload to the run.

### start [#start-1]

```python
client.workspace_runs.start(workspace_run_id, *, idempotency_key=None) -> dict
```

Starts extraction and command execution.

### get [#get-4]

```python
client.workspace_runs.get(workspace_run_id) -> dict
```

### list [#list-5]

```python
client.workspace_runs.list(
    *,
    limit=None,
    metadata=None,
    project_id=None,
    status=None,
) -> dict
```

`status` accepts: `"awaiting_archive"`, `"archive_uploaded"`, `"starting"`,
`"extracting"`, `"running"`, `"collecting"`, `"succeeded"`, `"failed"`,
`"canceled"`. The response has `data` and `hasMore`.

### list\_events [#list_events]

```python
client.workspace_runs.list_events(
    workspace_run_id,
    *,
    after_seq=None,
    limit=None,
) -> dict
```

Returns a bounded replay window without opening an SSE stream. The response
has `data`, `hasMore`, and `nextSeq`.

### stream\_events [#stream_events]

```python
client.workspace_runs.stream_events(
    workspace_run_id,
    *,
    after_seq=None,
    reconnect=True,
)
# sync: Iterator[dict]; async: AsyncIterator[dict]
```

Streams `status`, `archive_progress`, `stdout`, `stderr`,
`artifact_collected`, `artifact_error`, `heartbeat`, `terminal`, and `error`
events.

### cancel [#cancel-1]

```python
client.workspace_runs.cancel(
    workspace_run_id,
    *,
    idempotency_key=None,
) -> dict
```

### evidence [#evidence]

```python
client.workspace_runs.evidence(workspace_run_id) -> dict
```

Returns the durable Evidence Bundle after the run is terminal.

## client.api\_keys [#clientapi_keys]

Trusted automation can inspect and revoke API keys when granted
`api_key:read` and `api_key:revoke`. API key creation stays in the
dashboard.

```python
client.api_keys.list() -> dict
client.api_keys.get("key_...") -> dict
client.api_keys.revoke("key_...") -> dict
```

## Next steps [#next-steps]

* [Python SDK overview](/docs/python) — install, clients, and a worked
  example.
* [API error reference](/docs/api/errors) — codes raised as
  `CrowNestApiError`.
* [Sandboxes concept](/docs/concepts/sandboxes) — lifecycle and TTL.
