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", [])))

Once you have a ready project, 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-running checkpointed jobs, latest_checkpoint can include durable publication fields like artifact_id, artifact_source, publish_status, and published_at. On retry, _resume.checkpoint.path is rewritten to the downloaded local checkpoint path when the latest checkpoint was published as an artifact.

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 artifact metadata when available, resumes from <dest>.partial with HTTP Range when supported, and verifies the completed file against artifact size and SHA-256 metadata before the final rename.

Direct and multipart upload helpers return a local 410; use upload_artifact() or register_artifact() instead. Retained local/R2 bytes are content-deduplicated per account against the default 100 GiB quota. list_artifacts() includes authoritative quota limit/used/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>

Once 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