biolm.protocols

Package layout

Legacy import paths biolm.protocol_runs and biolm.protocol_runtime remain as compatibility shims.

When to use which

Local (SDK + pipeline)

• biolm protocol validate — YAML/schema checks

• biolm protocol run-local PROTOCOL.yaml --input key=value — compile and run locally

• Protocol.execute / run_local_protocol()

• Requires pip install "biolm-sdk[pipeline]"

• Supported features: see Local Protocol Profile v1

Hosted (platform API)

• biolm protocol run SLUG -i inputs.json [--wait] — submit registered protocol

• biolm.run_protocol() — submit by slug and block until results

• ProtocolClient — submit, poll, download, cancel

• Full protocol feature set (gather, foreach, task-output expressions, etc.)

Schema reference

Field-level protocol schema (inputs, tasks, execution, outputs, full JSON Schema): Protocol Schema Reference.

Examples

Validate a protocol file:

biolm protocol validate my-protocol.yaml

Submit inputs to the registered protocol slug:

biolm protocol run my-protocol-slug -i inputs.json --wait

The CLI submits JSON inputs, not the local YAML file. Use protocol list to discover registered slugs and status, wait, cancel, results, or download to manage a run by ID.

Validate from Python:

from biolm.protocols import Protocol

result = Protocol.validate("my-protocol.yaml")
if not result.is_valid:
    for err in result.errors:
        print(err.path, err.message)

Run locally from Python:

from biolm.protocols import Protocol

protocol = Protocol("my-protocol.yaml")
result = protocol.execute(inputs={"sequence": "MKLLIV"})
print(result.records)
print(result.selected_records)  # when protocol defines outputs[]

Run locally from the CLI:

pip install "biolm-sdk[pipeline]"
biolm protocol run-local my-protocol.yaml --input sequence=MKLLIV --json

Run on the platform from Python:

from biolm import run_protocol

results = run_protocol("my-protocol-slug", inputs={"sequences": ["MKTAYIAKQRQ"]})

Programmatic hosted runs

For progress tracking, cancellation, and result download, use ProtocolClient directly:

from biolm.protocols import ProtocolClient

client = ProtocolClient()
run = client.submit("my-protocol-slug", inputs={"sequences": ["MKTAYIAKQRQ"]})
run.wait()
print(run.results())

API

class biolm.protocols.Protocol(yaml_path: str)

Load and validate BioLM protocol YAML files.

Args:

yaml_path: Path to a protocol YAML file. The file is loaded and

validated on construction; invalid YAML raises ValueError.

Use validate() as a classmethod to validate without instantiating.

execute(inputs: Dict[str, Any] | None = None)

Execute protocol locally with given inputs.

Args:

inputs: Input values for the protocol (optional, uses defaults from protocol if not provided).

Returns:

biolm.protocols.runtime.LocalRunResult with dataframe, records, and metadata.

Requires:

biolm[pipeline] optional dependencies.

classmethod validate(yaml_path: str) → ProtocolValidationResult

Validate a protocol YAML file.

biolm.run_protocol(slug: str, inputs: dict, *, run_name: str | None = None, api_key: str | None = None, base_url: str | None = None, timeout: float = 3600.0, show_progress: bool = True, poll_interval: float = 5.0) → dict

Submit a BioLM protocol run and block until results are ready.

biolm.protocols.run_local_protocol(protocol: dict, inputs: dict[str, Any] | None = None, *, output_dir: str | Path | None = None, verbose: bool = False, **kwargs: Any) → LocalRunResult

Run a protocol dict locally (main public API).

class biolm.protocols.ProtocolClient(api_key: str | None = None, base_url: str | None = None)

Submit and monitor BioLM protocol runs from Python.

Wraps the /api/protocols/ REST endpoints. Use submit() to start a run, run_and_wait() for a blocking workflow, or get_run() to poll an existing run ID.

get_run(run_id: str) → ProtocolRun

Reconnect to an existing run by ID.

list(search: str | None = None, page: int = 1, page_size: int = 20) → Dict[str, Any]

run_and_wait(slug: str, inputs: Dict[str, Any], run_name: str | None = None, timeout: float = 3600.0, show_progress: bool = True, poll_interval: float = 5.0) → Dict[str, Any]

submit(slug: str, inputs: Dict[str, Any], version: int | None = None, run_name: str | None = None, environment_id: int | None = None, files: Dict[str, Any] | None = None) → ProtocolRun

class biolm.protocols.ProtocolRun(data: Dict[str, Any], client: ProtocolClient)

A submitted protocol run returned by ProtocolClient.submit().

cancel() → Dict[str, Any]

Cancel this run (idempotent; may error if already terminal).

download(output_dir: str | Path = '.', file_type: str = 'csv', overwrite: bool = False) → Path

results() → Dict[str, Any]

wait(timeout: float = 3600.0, show_progress: bool = True, poll_interval: float = 5.0) → ProtocolRun

Block until this run reaches a terminal state.

Waiting is websocket-first: it connects to the run’s telemetry channel for live progress. If the websockets dependency is missing, the connection fails, or the stream closes while the run is still nonterminal, it transparently falls back to REST polling of the run detail endpoint until the run succeeds, fails, or is cancelled.

Parameters:

• timeout – Total deadline in seconds shared across the websocket and REST polling phases. The deadline is not reset after a websocket failure.

• show_progress – Print status changes to stdout.

• poll_interval – Seconds between REST polls during fallback. Each sleep is clamped to the remaining deadline and must be greater than zero.

Returns:

self for chaining.

Raises:

• ProtocolRunError – If the run fails or is cancelled.

• TimeoutError – If the total deadline elapses before completion.

• ValueError – If poll_interval is not greater than zero.

class biolm.protocols.runtime.LocalRunResult(dataframe: ~pandas.core.frame.DataFrame, records: list[dict[str, typing.Any]], plan: ~biolm.protocols.runtime.spec.ExecutionPlan, pipeline: ~biolm.pipeline.data.DataPipeline | None = None, run_id: str | None = None, output_selections: list[biolm.protocols.outputs.OutputSelection] = <factory>, selected_records: list[dict[str, typing.Any]] = <factory>)

Result of a local protocol run.

output_selections: list[biolm.protocols.outputs.OutputSelection]

records: list[dict[str, Any]]

selected_records: list[dict[str, Any]]

to_seqframe(*, molecule_type: str | None = None, path: str | Path | None = None)

Materialize results as a SeqFrame (requires seqframe extra).

See also

• Protocol Workflows — authoring, local, and hosted guides

• Local Protocol Profile v1 — local execution profile (v1)