Pipelines

Create and monitor long-running bioinformatics jobs asynchronously. Pipelines run on managed AWS infrastructure — you submit, poll for status, and download outputs when done. Requires the tools scope.

1. POST /v1/pipelines

2. GET /v1/pipelines/:id

3. status = completed

4. download outputs

Predefined Pipelines

SmartsBio ships 12 ready-to-run multi-step workflows. Pass the pipeline's ID as tool_id in the request body (pipeline_id is accepted as an alias, and is the argument name the SDKs use) and the system handles orchestration, step ordering, and intermediate file handoff automatically. Use the list_pipelines tool to fetch the current list and required inputs at runtime.

# Discover all predefined pipelines and their required inputs
result = client.tools.run(tool_id="list_pipelines", input={})
for p in result["pipelines"]:
    print(f"{p['id']:40s} ~{p['estimated_runtime']}")
    for param in p["required_inputs"]:
        print(f"  {param['name']:30s} {param['description']}")

Pipeline Reference

pipeline_id	Name	Required Inputs	Main Outputs	Est. Runtime
alignment-wes	WES Alignment	fastq_r1, fastq_r2, reference (GRCh38 / GRCh37)	sorted.markdup.bqsr.bam, .bai, fastqc_report.html, multiqc_report.html	2–4 hours
whole-genome-sequencing	Whole Genome Sequencing	fastq_r1, fastq_r2, reference	sorted.markdup.bqsr.bam, .bai, qc_report.html	4–8 hours
rna-seq-analysis	RNA-seq Analysis	fastq_r1, fastq_r2 (optional), gtf, reference	counts_matrix.csv, alignment_summary.txt, multiqc_report.html	1–3 hours
atac-seq	ATAC-seq	fastq_r1, fastq_r2, reference, genome (hg38 / mm10)	peaks.narrowPeak, peaks.bed, fragment_sizes.pdf, multiqc_report.html	1–2 hours
chip-seq	ChIP-seq	fastq_chip, fastq_input, reference, genome	peaks.narrowPeak, motifs/, bigwig/	1–3 hours
gatk-variant-calling	GATK Variant Calling	bam, bai, reference (GRCh38 / GRCh37)	variants.vcf.gz, variants.vcf.gz.tbi, genotyping_summary.txt	1–4 hours
somatic-variant-calling	Somatic Variant Calling	tumor_bam, tumor_bai, normal_bam, normal_bai, reference	somatic.filtered.vcf.gz, somatic.filtered.vcf.gz.tbi, contamination.table	2–5 hours
quality-control	Quality Control	fastq (list of one or more FASTQ files)	multiqc_report.html, per_file_fastqc/	10–30 minutes
protein-binder-design-validated	Protein Binder Design (Validated)	target_pdb or target_sequence, binding_site (optional), n_designs	designs.pdb (top N), affinity_scores.csv, design_report.pdf	30–90 minutes
nanobody-discovery	Nanobody Discovery	antigen_pdb or antigen_sequence, epitope_residues (optional), n_candidates	nanobody_candidates.pdb, binding_scores.csv, screening_report.pdf	1–2 hours
enzyme-engineering	Enzyme Engineering	scaffold_pdb or scaffold_sequence, reaction_smiles, optimization_target	engineered_variants.pdb, activity_predictions.csv, stability_scores.csv	1–3 hours
structure-based-drug-discovery	Structure-Based Drug Discovery	target_pdb or uniprot_id, ligand_library or smiles_list	docking_results.csv, top_poses.sdf, binding_site_report.pdf	30–120 minutes

Run a Predefined Pipeline

Over HTTP, put the pipeline ID in the tool_id body field (pipeline_id also works as an alias). The SDKs take it as pipeline_id / pipelineId. All file references use workspace-relative S3 keys returned by the Files API.

# 1. Upload inputs
fastq_r1 = client.files.upload("sample_R1.fastq.gz", workspace_id=ws_id, path="input/")
fastq_r2 = client.files.upload("sample_R2.fastq.gz", workspace_id=ws_id, path="input/")

# 2. Start the predefined WES alignment pipeline
pipeline = client.pipelines.create(
    pipeline_id="alignment-wes",      # ← predefined pipeline ID
    workspace_id=ws_id,
    input={
        "fastq_r1": fastq_r1["key"],
        "fastq_r2": fastq_r2["key"],
        "reference": "GRCh38",
        "output_path": "results/wes/",
    },
)
print(f"Started {pipeline['id']}  status={pipeline['status']}")

# 3. Wait and download
pipeline = client.pipelines.wait(
    pipeline["id"],
    workspace_id=ws_id,
    poll_interval=30,
    on_progress=lambda p: print(f"  {p['progress_pct']}%"),
)
for key in pipeline["output_paths"]:
    client.files.download(key, workspace_id=ws_id, dest="./output/")

Example — Protein Binder Design

# Upload target structure
target = client.files.upload("target.pdb", workspace_id=ws_id, path="input/")

# Start protein binder design pipeline
pipeline = client.pipelines.create(
    pipeline_id="protein-binder-design-validated",
    workspace_id=ws_id,
    input={
        "target_pdb": target["key"],
        "binding_site": "A:45-60,A:102-115",  # optional chain:residue ranges
        "n_designs": 10,
        "output_path": "results/binders/",
    },
)

# Wait for completion (~30-90 min)
pipeline = client.pipelines.wait(
    pipeline["id"],
    workspace_id=ws_id,
    poll_interval=60,
    on_progress=lambda p: print(f"  Step: {p.get('current_step', '?')}  {p['progress_pct']}%"),
)

# Download top designs and affinity scores
for key in pipeline["output_paths"]:
    client.files.download(key, workspace_id=ws_id, dest="./binders/")

Custom Tool Pipelines

The same tool_id field also accepts any single pipeline-capable tool. See Available Tools for the full list — tools marked Pipeline or Both support this mode.

Saved & Custom Pipeline Specs

A pipeline spec is a JSON file that declares a pipeline's inputs and steps. Save any finished run as a spec with POST /v1/pipelines/:id/save-spec, then re-run it on different files by passing spec_file (a *.pipeline.json in your workspace) or spec (the spec object inline) instead of tool_id. See Pipeline Specs for the format and a worked example. The SDKs don't wrap these options yet — use HTTP or the CLI.

# Re-run a saved spec on a new sample
curl -X POST https://api.smarts.bio/v1/pipelines \
  -H "Authorization: Bearer sk_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "spec_file": "pipelines/specs/sort-and-index.pipeline.json",
    "workspace_id": "ws_abc123",
    "input": {
      "inputBam": "input/sample2.bam",
      "sampleName": "sample2"
    }
  }'

The Pipeline object

Field	Type	Description
id	string	Unique pipeline ID (e.g. `pipe_abc123`).
tool_id	string	Tool that was run — or `pipeline:{id}` for predefined pipelines.
workspace_id	string	Workspace the pipeline belongs to.
status	string	`queued` → `running` → `completed` \| `failed` \| `cancelled`.
progress_pct	integer	Completion percentage 0–100. Updated during execution.
current_step	string \| null	Human-readable name of the step currently executing (predefined pipelines only).
input	object	The exact input parameters submitted.
output_paths	string[]	Paths of output files relative to workspace root (populated when `completed`).
logs_path	string	Path to the execution log file, relative to workspace root.
error	string \| null	Error message if `failed`, otherwise null.
created_at	string	ISO 8601 creation timestamp.
started_at	string \| null	When execution began.
completed_at	string \| null	When execution finished (success or failure).

POST/v1/pipelinestools scope

Create and queue a pipeline run. Returns immediately with the pipeline ID and status: "queued". Supply exactly one of tool_id (predefined pipeline or pipeline-capable tool), spec_file (saved spec) or spec (inline spec) — sending none or more than one returns 400 validation_error. Each organization runs at most 5 compute tasks at once; further steps wait in queued and start as slots free up — they never fail because of the limit.

Body field	Type	Description
tool_id	string	ID of a predefined pipeline (e.g. `alignment-wes`) or a pipeline-capable tool (e.g. `gatk_toolkit`). `pipeline_id` is accepted as an alias. Use this, `spec_file` or `spec`.
spec_file	string	Workspace-relative path of a saved spec (e.g. `pipelines/specs/sort-and-index.pipeline.json`). The run is named after the file. See Pipeline Specs.
spec	object	An inline spec object — for a spec that lives outside the workspace, e.g. committed to your own repo.
workspace_id *	string	Workspace for input/output files.
input *	object	Pipeline or tool-specific input parameters (file keys, settings, etc.). For a spec, keyed by its declared input names; pass `{}` if it has none.

# Predefined pipeline
pipeline = client.pipelines.create(
    pipeline_id="gatk-variant-calling",
    workspace_id=ws_id,
    input={
        "bam": bam["key"],
        "bai": bai["key"],
        "reference": "GRCh38",
        "output_path": "variants/",
    },
)

# Custom tool pipeline
pipeline = client.pipelines.create(
    tool_id="bwa_toolkit",
    workspace_id=ws_id,
    input={"fastq": fastq["key"], "reference": "GRCh38"},
)

print(f"Pipeline {pipeline['id']} is {pipeline['status']}")

GET/v1/pipelines/:idtools scope

Get the current status of a pipeline. Poll until status is completed, failed, or cancelled. Recommended interval: 15–30 seconds.

Parameter	Type	Description
id *	string (path)	Pipeline ID.
workspace_id *	string (query)	Workspace the pipeline belongs to.

pipeline = client.pipelines.get("pipe_abc123", workspace_id=ws_id)
print(f"Step: {pipeline.get('current_step', '?')}")
print(f"Status: {pipeline['status']}  {pipeline['progress_pct']}%")

GET/v1/pipelinestools scope

List pipelines in a workspace, optionally filtered by status.

Parameter	Type	Description
workspace_id *	string	Workspace to list pipelines from.
status	string	Filter: `queued`, `running`, `completed`, `failed`.
limit	integer	Max results (default 20, max 100).

pipelines = client.pipelines.list(workspace_id=ws_id)
for p in pipelines:
    print(f"  {p['id']}  {p['tool_id']:<35}  {p['status']}  {p['progress_pct']}%")

running = client.pipelines.list(workspace_id=ws_id, status="running")

DELETE/v1/pipelines/:idtools scope

Cancel a queued or running pipeline. Has no effect on completed or already-cancelled pipelines.

result = client.pipelines.cancel("pipe_abc123", workspace_id=ws_id)
print(result["message"])  # "Pipeline cancelled."

POST/v1/pipelines/:id/save-spectools scope

Save a finished pipeline run as a reusable pipeline spec at pipelines/specs/<name>.pipeline.json in the run's workspace. The files the run used become declared inputs, so the spec can be re-run on different data with POST /v1/pipelines + spec_file. Failed runs can be saved too (a warning asks you to review the result). Returns 201 with the saved path and any warnings.

Parameter	Type	Description
id *	string (path)	The run ID returned by `POST /v1/pipelines`.
workspace_id *	string (body)	Workspace the run belongs to.
name	string (body)	Name for the saved pipeline; slugified into the file name. Defaults to the source pipeline's name.
description	string (body)	Description stored in the spec. Defaults to `Saved from run <run folder>`.
overwrite	boolean (body)	Replace an existing spec with the same name. Default `false`.
parameterize_sample_name	boolean (body)	For ad-hoc runs, turn a shared output-filename prefix into a `sampleName` input. Default `true`.

Status	Meaning
201	Saved. The body carries `path`, a summary of the spec, and `warnings`.
400	Missing `workspace_id`, or the run can't be turned into a re-runnable spec (e.g. an agent-driven workflow).
404	No pipeline run with that ID that you started — the same response whether it doesn't exist or isn't yours.
409	A spec with that name already exists and `overwrite` is not set. Choose another `name` or send `overwrite: true`.

curl -X POST https://api.smarts.bio/v1/pipelines/pipe_abc123/save-spec \
  -H "Authorization: Bearer sk_live_..." \
  -H "Content-Type: application/json" \
  -d '{ "workspace_id": "ws_abc123", "name": "sort-and-index" }'
# → 201
# {
#   "status": "success",
#   "path": "pipelines/specs/sort-and-index.pipeline.json",
#   "s3Key": "organizations/.../files/pipelines/specs/sort-and-index.pipeline.json",
#   "spec": {
#     "id": "sort-and-index-3f9a1c",
#     "name": "sort-and-index",
#     "inputs": [
#       { "name": "inputBam", "type": "file", "description": "..." },
#       { "name": "sampleName", "type": "string", "description": "...", "default": "sample" }
#     ],
#     "steps": 2
#   },
#   "warnings": []
# }
#
# → 409 if the name is taken:
# { "error": "A pipeline named \"sort-and-index\" already exists at pipelines/specs/sort-and-index.pipeline.json.",
#   "path": "pipelines/specs/sort-and-index.pipeline.json",
#   "hint": "Choose a different name, or retry with overwrite: true." }

POST/v1/pipelines/validatetools scope

Check a pipeline spec without running it. Applies the same checks a run does, so a spec that validates here will start. Returns { valid, errors, warnings } — every problem at once, not just the first. Warnings never block a run.

Body field	Type	Description
workspace_id *	string	Workspace to resolve `spec_file` against.
spec_file	string	Workspace-relative path of a saved spec. Use this or `spec`.
spec	object	An inline spec object. Use this or `spec_file`.

A spec_file that isn't valid JSON comes back as valid: false; one that doesn't exist returns 404.

curl -X POST https://api.smarts.bio/v1/pipelines/validate \
  -H "Authorization: Bearer sk_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "workspace_id": "ws_abc123",
    "spec_file": "pipelines/specs/sort-and-index.pipeline.json"
  }'
# {
#   "valid": false,
#   "errors": [
#     "Step \"index\", parameter \"inputFile\": ${sort.outputFile} reads from \"sort\", which is not a dependency of \"index\". Add \"sort\" to its dependsOn, or the value will be unavailable when the step runs."
#   ],
#   "warnings": []
# }

Polling with `.wait()`

Both SDKs include a convenience .wait() method that polls until the pipeline finishes and raises on failure.

pipeline = client.pipelines.create(
    pipeline_id="rna-seq-analysis",
    workspace_id=ws_id,
    input={"fastq_r1": r1["key"], "fastq_r2": r2["key"],
           "gtf": gtf["key"], "reference": "GRCh38"},
)

pipeline = client.pipelines.wait(
    pipeline["id"],
    workspace_id=ws_id,
    poll_interval=30,           # seconds
    timeout=14400,              # 4 hours
    on_progress=lambda p: print(f"  [{p.get('current_step','?')}]  {p['progress_pct']}%"),
)

if pipeline["status"] == "completed":
    for key in pipeline["output_paths"]:
        client.files.download(key, workspace_id=ws_id, dest="./output/")
else:
    print(f"Failed: {pipeline['error']}")

Accessing logs

The logs_path field is populated as soon as a pipeline starts running.

pipeline = client.pipelines.get(pipeline_id, workspace_id=ws_id)
if pipeline.get("logs_path"):
    client.files.download(pipeline["logs_path"], workspace_id=ws_id, dest="./logs/")

Available Tools →Pipeline Specs →Pipeline Examples →Files API →