Python SDK

computalot is the Python SDK for Computalot.

Install


python3 -m pip install --user --break-system-packages \
  https://computalot.com/docs/downloads/computalot-0.2.1-py3-none-any.whl
export PATH="$HOME/.local/bin:$PATH"

Setup


export COMPUTALOT_CONTROLLER_URL="https://computalot.com"
export COMPUTALOT_API_TOKEN="flk_..."

Quickstart: first authenticated probe


from computalot import ComputalotClient
 
client = ComputalotClient(
    controller_url="https://computalot.com",
    token="flk_...",
)
 
docs = client.docs_index()
jobs = client.list_jobs(limit=5)
 
print(docs["status"])
print(len(jobs.get("jobs", [])))

When your project is ready, use the submit examples below.

Submitting jobs


# Basic
job = client.submit_structured(
    runner_command=["python", "evaluate.py"],
    payload={"dataset": "smoke"},
    project="my-project",
)
 
# With fan-out
job = client.submit_structured(
    runner_command=["python", "evaluate.py"],
    payload={"models": ["gpt-4", "claude"], "config": {}},
    fan_out={"by": "models"},
    merge_strategy="keyed",
    project="my-project",
)
 
# With GPU requirements
job = client.submit_job({
    "type": "structured_runner",
    "runner_command": ["python", "train.py"],
    "payload": {"epochs": 100},
    "project": "my-project",
    "requirements": {"profile": "gpu", "gpu_count": 1},
    "checkpointing": {"enabled": True, "resume_from_latest": True},
})
 
# With dependencies (DAG)
eval_job = client.submit_structured(
    runner_command=["python", "evaluate.py"],
    depends_on=[train_job["id"]],
    project="my-project",
)

Inspecting jobs


job = client.get_job(job_id)
tasks = client.job_tasks(job_id)
events = client.job_events(job_id)
metrics = client.job_metrics(job_id)
 
latest_progress = tasks["tasks"][0].get("latest_progress")
latest_checkpoint = tasks["tasks"][0].get("checkpoint")
resume_state = tasks["tasks"][0].get("resume_state")

For long checkpointed jobs, latest_checkpoint can include durable publication fields such as artifact_id, artifact_source, publish_status, and published_at. When the latest checkpoint was published as an artifact, retries rewrite _resume.checkpoint.path to the downloaded local checkpoint path.

Artifacts


# Relay upload (up to 2 GiB)
meta = client.upload_artifact("/path/to/file.parquet", filename="file.parquet")
 
# Register an object that already exists at an external URL
external = client.register_artifact(
    url="https://storage.example/large-file.parquet",
    sha256="abc123...",
    filename="large-file.parquet",
)
 
client.download_artifact(meta["id"], "/path/to/output.parquet")
artifacts = client.list_artifacts()
client.delete_artifact(meta["id"])

download_artifact() prefers the signed object-store URL from the artifact metadata when one is available. When the server supports HTTP Range, the download resumes from <dest>.partial. Before the final rename, the SDK validates the completed file against the artifact size and SHA-256 metadata.

Direct and multipart upload helpers return a local 410. Use upload_artifact() or register_artifact() instead. Computalot deduplicates retained local and R2 bytes by content, per account, against the default 100 GiB quota. list_artifacts() includes the authoritative quota limit, used, and remaining bytes. delete_artifact() returns 409 artifact_in_use only while a non-terminal job references the artifact. Terminal references do not block deletion.

Waiting and reading results


final = client.wait_for_job(job["id"])
results = client.get_results(job["id"])

CLI


export COMPUTALOT_CONTROLLER_URL="https://computalot.com"
export COMPUTALOT_API_TOKEN="flk_..."
 
computalot docs --llm
computalot jobs --limit 5
computalot job <job_id>

When a project is ready, the submit helpers are:


computalot submit --project my-project python -c "print('done')"
computalot run --project my-project python evaluate.py
computalot run --project my-project --gpu python train.py