CrowNest
Python SDK

Python API reference

Every public client and method in the crownest Python package, with parameters and return shapes.

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

Construct a client directly or as a context manager.

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

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

create

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

client.sandboxes.get(sandbox_id) -> SandboxHandle

extend

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

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

client.sandboxes.kill(sandbox_id) -> SandboxHandle

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

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

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

create_context

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

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

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

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

run

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

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

client.commands

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

run

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

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

client.commands.get(command_id) -> dict

cancel

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

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

logs

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

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:

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

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

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

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

write

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

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

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

Returns an envelope dictionary with data.

stat

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

mkdir

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

parents defaults to False.

move

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

overwrite defaults to False.

download_url

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

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

create

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

client.artifacts.get(artifact_id) -> dict

list

client.artifacts.list(sandbox_id) -> dict

Returns a page dictionary with data and hasMore.

download

client.artifacts.download(artifact_id) -> bytes

download_url

client.artifacts.download_url(artifact_id) -> dict

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

delete

client.artifacts.delete(artifact_id) -> dict

Idempotent by state; returns the artifact with deletedAt set.

client.previews

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

create

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

client.previews.get(preview_id) -> dict

list

client.previews.list(sandbox_id) -> dict

Returns a page dictionary with data and hasMore.

revoke

client.previews.revoke(preview_id) -> dict

Idempotent; returns the preview with revokedAt set.

client.projects

create

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

Requires an org-wide key with project:create.

list

client.projects.list() -> dict

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

client.workspace_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

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

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

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

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

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

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

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

Starts extraction and command execution.

get

client.workspace_runs.get(workspace_run_id) -> dict

list

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

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

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

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

evidence

client.workspace_runs.evidence(workspace_run_id) -> dict

Returns the durable Evidence Bundle after the run is terminal.

client.api_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.

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

Next steps

Python SDK

Install the crownest package, construct sync and async clients, and run your first sandbox from Python.

CLI

Install the crownest CLI, log in, configure credentials, and understand exit codes and output formats.

On this page

Clientsclient.sandboxescreategetextendlistkillSandboxHandleclient.codecreate_contextlist_contextsget_contextdelete_contextrunrun_streamclient.commandsrunstartgetcancellogsstream_logsclient.filesreadwritedeleteliststatmkdirmovedownload_urlclient.artifactscreategetlistdownloaddownload_urldeleteclient.previewscreategetlistrevokeclient.projectscreatelistclient.workspace_runsrun_archivecreateupload_archivecreate_archive_transferupload_archive_to_transferfinalize_archivestartgetlistlist_eventsstream_eventscancelevidenceclient.api_keysNext steps