biolm package

Subpackages

Submodules

biolm.client module

BioLM convenience client and HTTP client re-exports.

class biolm.client.AsyncRateLimiter(max_calls: int, period: float)

Bases: object

limit()
class biolm.client.BioLM(*, entity: str, action: str, type: str | None = None, items: Any | List[Any], params: dict | None = None, api_key: str | None = None, **kwargs)

Bases: object

Universal client for BioLM API.

This is a convenience wrapper that creates a client, makes the request, and returns the result. For long-running operations or when making multiple requests, consider using BioLMApiClient (async) or BioLMApi (sync) directly with proper cleanup via context managers or shutdown().

Args:

entity (str): The entity name (model, database, calculation, etc). action (str): The action to perform (e.g., ‘generate’, ‘encode’, ‘predict’, ‘search’, ‘finetune’). type (str): The type of item (e.g., ‘sequence’, ‘pdb’, ‘fasta_str’). item (Union[Any, List[Any]]): The item(s) to process. params (Optional[dict]): Optional parameters for the action. raise_httpx (bool): Whether to raise HTTPX errors. stop_on_error (bool): Stop on first error if True. output (str): ‘memory’ or ‘disk’. file_path (Optional[str]): Output file path if output=’disk’. api_key (Optional[str]): API key for authentication. compress_requests (bool): Enable gzip compression for POST requests. Default: True. compress_threshold (int): Minimum payload size in bytes to trigger compression. Default: 256.

run() Any

Run the specified action on the entity with the given item(s).

biolm.client.BioLMApi

alias of BlockingBioLMApi

class biolm.client.BioLMApiClient(model_name: str, api_key: str | None = None, base_url: str | None = None, timeout: Timeout = Timeout(connect=10.0, read=1200, write=1200, pool=1200), raise_httpx: bool = False, unwrap_single: bool = True, semaphore: int | Semaphore | None = 16, rate_limit: str | None = None, retry_error_batches: bool = True, compress_requests: bool = True, compress_threshold: int = 256, concurrent_batches: bool = True, http2: bool = True)

Bases: object

async call(func: str, items: List[dict], params: dict | None = None, raw: bool = False)
encode(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)
static extract_max_items(schema: dict) int | None

Extracts the ‘maxItems’ value for the ‘items’ key from the schema. Returns the integer value if found, else None.

generate(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)
async lookup(query: dict | List[dict], *, raw: bool = False, output: str = 'memory', file_path: str | None = None)
predict(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)
async schema(model: str, action: str) dict | None

Fetch the JSON schema for a given model and action, with caching. Returns the schema dict if successful, else None. Uses a module-level cache keyed by (model, action) so it is safe to use from any event loop (sync wrapper, async tests, pytest-xdist workers).

Schema is always fetched from the hosted platform API without credentials. Hub gateways and custom model API hosts may not expose /schema/, and sending API tokens to the public schema endpoint can return 401.

score(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True)
search(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True)
async shutdown()
class biolm.client.CredentialsProvider

Bases: object

static get_auth_headers(api_key: str | None = None) Dict[str, str]
biolm.client.batch_iterable(iterable, batch_size)
biolm.client.is_list_of_lists(items, check_n=10)
biolm.client.parse_rate_limit(rate: str)

biolm.finetune module

Programmatic finetuning client for BioLM.

Wraps the /api/finetune/ endpoints (the SDK-facing companion to the browser console finetuning UI) so callers can launch and track XGBoost and DSM finetuning runs from Python without a session cookie.

Auth follows the same rules as the rest of the SDK: pass api_key= or set BIOLM_TOKEN (see biolm.core.http.CredentialsProvider).

class biolm.finetune.Finetune

Bases: object

Launch and track BioLM finetuning runs.

All methods are classmethods returning plain dicts. *_data arguments accept a list of row dicts ([{"sequence": ..., "label": ...}, ...]) or a raw CSV string; they are sent inline as JSON.

classmethod cancel(run_id: str, **kwargs) dict

Synchronous wrapper for cancel_async().

async classmethod cancel_async(run_id: str, *, api_key: str | None = None, base_url: str | None = None) dict

Cancel an in-flight run.

classmethod dsm_rl(**kwargs) dict

Synchronous wrapper for dsm_rl_async().

async classmethod dsm_rl_async(*, seed_sequences: List[str] | str, oracle_type: str = 'esmc', stability_objective: str = 'thermostability', training_mode: str = 'online', algorithm: str = 'ppo', num_episodes: int = 100, samples_per_episode: int = 64, learning_rate: float = 0.0003, batch_size: int = 8, mask_ratio: float = 0.3, mutation_rate: float = 0.1, run_name: str | None = None, environment_id: int | str | None = None, api_key: str | None = None, base_url: str | None = None) dict

Launch DSM RL protein optimization against an oracle.

classmethod dsm_stage1(**kwargs) dict

Synchronous wrapper for dsm_stage1_async().

async classmethod dsm_stage1_async(*, train_data: List[Dict[str, Any]] | str, test_data: List[Dict[str, Any]] | str | None = None, valid_data: List[Dict[str, Any]] | str | None = None, sequence_col: str = 'sequence', lr: float = 0.0001, batch_size: int = 8, grad_accum: int = 16, max_steps: int = 50000, max_length: int = 2048, save_every: int = 1000, fp16: bool = False, run_name: str | None = None, environment_id: int | str | None = None, api_key: str | None = None, base_url: str | None = None) dict

Launch DSM Stage 1 (single-chain masked-LM finetune).

classmethod dsm_stage2(**kwargs) dict

Synchronous wrapper for dsm_stage2_async().

async classmethod dsm_stage2_async(*, stage1_checkpoint: str, paired_data: List[Dict[str, Any]] | str, unpaired_data: List[Dict[str, Any]] | str | None = None, heavy_col: str = 'heavy', light_col: str = 'light', use_mixed_training: bool = False, fp16: bool = False, lr: float = 5e-05, batch_size: int = 4, unpaired_batch_size: int = 8, grad_accum: int = 16, max_steps: int = 25000, max_length: int = 300, unpaired_ratio: float = 2.0, run_name: str | None = None, environment_id: int | str | None = None, api_key: str | None = None, base_url: str | None = None) dict

Launch DSM Stage 2 (paired multichain finetune from a Stage 1 checkpoint).

classmethod get_run(run_id: str, **kwargs) dict

Synchronous wrapper for get_run_async().

async classmethod get_run_async(run_id: str, *, api_key: str | None = None, base_url: str | None = None) dict

Fetch sanitized detail (status, results) for a single run.

classmethod list_runs(**kwargs) dict

Synchronous wrapper for list_runs_async().

async classmethod list_runs_async(*, dag: str | None = None, status: str | None = None, page: int = 1, page_size: int = 20, api_key: str | None = None, base_url: str | None = None) dict

List the caller’s finetune runs (paginated).

classmethod progress(run_id: str, **kwargs) dict

Synchronous wrapper for progress_async().

async classmethod progress_async(run_id: str, *, api_key: str | None = None, base_url: str | None = None) dict

Fetch lightweight status + telemetry channel id for a run.

classmethod wait(run_id: str, *, poll_interval: float = 15.0, timeout: float | None = None, api_key: str | None = None, base_url: str | None = None) dict

Block until run_id reaches a terminal state, returning its detail.

classmethod xgboost(**kwargs) dict

Synchronous wrapper for xgboost_async().

async classmethod xgboost_async(*, train_data: List[Dict[str, Any]] | str, embedding_models: List[str], task_type: str = 'classification', target_column: str = 'label', text_column: str = 'sequence', test_data: List[Dict[str, Any]] | str | None = None, validation_data: List[Dict[str, Any]] | str | None = None, n_estimators: int = 100, max_depth: int = 6, learning_rate: float = 0.1, n_splits: int = 5, seed: int = 42, hyperopt: bool = False, hyperopt_n_trials: int | None = None, antibody_mode: bool = False, heavy_column: str = 'heavy', light_column: str = 'light', modality: str = 'protein', run_name: str | None = None, environment_id: int | str | None = None, api_key: str | None = None, base_url: str | None = None) dict

Launch an XGBoost finetune (optionally with Ray Tune hyperopt).

biolm.progress module

Shared Rich progress UI for batch operations. Used by CLI and Model (progress=True).

biolm.progress.rich_progress(total_items: int, description: str = 'Processing...', console: Any | None = None)

Context manager that yields a progress callback for batch operations.

The callback has signature (completed: int, total: int) -> None. Call it after each batch with the number of items completed so far and total items. Uses Rich Progress (spinner, bar, task progress) when Rich is available; otherwise yields a no-op callback.

Args:

total_items: Total number of items to process. description: Description shown in the progress bar. console: Rich Console instance (default: new Console()).

Yields:

A callable (completed: int, total: int) -> None to update progress.

biolm.protocol_runs module

Protocol Submission API client — programmatic run submission, progress tracking, and results retrieval.

This file is a renamed copy of biolmai/protocol_runs.py from py-biolm, kept for backwards compatibility when migrating to the biolm namespace.

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

Bases: object

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(slug: str, version: int | None = None) Dict[str, Any]
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) 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
exception biolm.protocol_runs.ProtocolNotFoundError

Bases: ProtocolRunError

The requested protocol slug/version does not exist or is not accessible.

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

Bases: object

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
download_files(*, output_dir: str | Path = '.', file_type: str = 'csv', overwrite: bool = False) Path

Compatibility alias for download().

progress() Dict[str, Any]
refresh() ProtocolRun
results() Dict[str, Any]
to_dataframe(*, output_dir: str | Path = '.', overwrite: bool = False)

Download CSV zip and return a pandas DataFrame.

wait(timeout: float = 3600.0, show_progress: bool = True) ProtocolRun
exception biolm.protocol_runs.ProtocolRunError

Bases: Exception

A protocol run failed, was cancelled, or the API returned an error.

biolm.protocols module

Protocol schema validation and execution for BioLM.

class biolm.protocols.Protocol(yaml_path: str)

Bases: object

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 with given inputs.

Args:

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

Returns:

Protocol execution results.

Note:

Protocol execution is not yet implemented. This is a placeholder.

classmethod fetch_by_id(protocol_id: str, api_key: str | None = None, base_url: str | None = None) dict

Fetch a protocol from the platform by ID (synchronous wrapper).

Args:

protocol_id: Protocol ID to fetch api_key: Optional API key for authentication base_url: Optional base URL (defaults to BIOLMAI_BASE_API_URL)

Returns:

Protocol data as dict (same structure as YAML)

Raises:

FileNotFoundError: If protocol not found (404) PermissionError: If not authenticated (401) ValueError: If API request fails

async classmethod fetch_by_id_async(protocol_id: str, api_key: str | None = None, base_url: str | None = None) dict

Fetch a protocol from the platform by ID.

Args:

protocol_id: Protocol ID to fetch api_key: Optional API key for authentication base_url: Optional base URL (defaults to BIOLMAI_BASE_API_URL)

Returns:

Protocol data as dict (same structure as YAML)

Raises:

FileNotFoundError: If protocol not found (404) PermissionError: If not authenticated (401) ValueError: If API request fails

classmethod init(output_path: str, example: str | None = None, force: bool = False) str

Initialize a new protocol YAML file.

Args:

output_path: Path where the protocol file should be created example: Optional example template name to use force: If True, overwrite existing file

Returns:

Path to the created file

Raises:

FileExistsError: If file exists and force=False ValueError: If example name is invalid FileNotFoundError: If example file doesn’t exist

static render_report(protocol_data: dict, source: str = 'file', console=None) None

Render a formatted report of the protocol using Rich.

Args:

protocol_data: Protocol data dictionary source: Source description (e.g., “file”, “platform”) console: Optional Rich Console instance (creates one if not provided)

Raises:

ValueError: If protocol_data is invalid or missing required fields

classmethod validate(yaml_path: str) ProtocolValidationResult

Validate a protocol YAML file.

Args:

yaml_path: Path to protocol YAML file.

Returns:

ProtocolValidationResult with validation results.

class biolm.protocols.ProtocolValidationResult(is_valid: bool, errors: ~typing.List[~biolm.protocols.ValidationError] = <factory>, warnings: ~typing.List[str] = <factory>, statistics: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Result of protocol validation.

add_error(message: str, path: str = '', error_type: str = 'unknown')

Add an error to the result.

add_warning(message: str)

Add a warning to the result.

errors: List[ValidationError]
is_valid: bool
statistics: Dict[str, Any]
warnings: List[str]
class biolm.protocols.ValidationError(message: str, path: str = '', error_type: str = 'unknown')

Bases: object

Represents a single validation error.

error_type: str = 'unknown'
message: str
path: str = ''

biolm.volumes module

Volume management for BioLM (Python SDK not yet implemented; use CLI/hub).

class biolm.volumes.Volume(name: str | None = None, api_key: str | None = None)

Bases: object

Volume management interface (not yet implemented).

Use hub CLI and storage workflows today. This class defines the intended Python API: list, create, and get.

Args:

name: Volume name. If None, uses the default volume. api_key: Optional API token; defaults to BIOLM_TOKEN.

create(name: str, **kwargs) Dict[str, Any]

Create a new volume.

Args:

name: Name of the volume to create. **kwargs: Additional volume parameters.

Returns:

Created volume information.

Note:

Volume creation is not yet implemented. This is a placeholder.

delete(name: str | None = None) bool

Delete a volume.

Args:

name: Volume name. If None, uses the volume name from initialization.

Returns:

True if deletion was successful.

Note:

Volume deletion is not yet implemented. This is a placeholder.

get(name: str | None = None) Dict[str, Any]

Get volume information.

Args:

name: Volume name. If None, uses the volume name from initialization.

Returns:

Volume information dictionary.

Note:

Volume retrieval is not yet implemented. This is a placeholder.

list() List[Dict[str, Any]]

List all available volumes.

Returns:

List of volume dictionaries.

Note:

Volume listing is not yet implemented. This is a placeholder.

biolm.workspaces module

Workspace management for BioLM (Python SDK not yet implemented; use CLI).

class biolm.workspaces.Workspace(name: str | None = None, api_key: str | None = None)

Bases: object

Workspace management interface (not yet implemented).

Use biolm workspace CLI commands today. This class defines the intended Python API: list, create, and get.

Args:

name: Workspace name. If None, uses the default workspace. api_key: Optional API token; defaults to BIOLM_TOKEN.

create(name: str, **kwargs) Dict[str, Any]

Create a new workspace.

Args:

name: Name of the workspace to create. **kwargs: Additional workspace parameters.

Returns:

Created workspace information.

Note:

Workspace creation is not yet implemented. This is a placeholder.

delete(name: str | None = None) bool

Delete a workspace.

Args:

name: Workspace name. If None, uses the workspace name from initialization.

Returns:

True if deletion was successful.

Note:

Workspace deletion is not yet implemented. This is a placeholder.

get(name: str | None = None) Dict[str, Any]

Get workspace information.

Args:

name: Workspace name. If None, uses the workspace name from initialization.

Returns:

Workspace information dictionary.

Note:

Workspace retrieval is not yet implemented. This is a placeholder.

list() List[Dict[str, Any]]

List all available workspaces.

Returns:

List of workspace dictionaries.

Note:

Workspace listing is not yet implemented. This is a placeholder.

Module contents

Top-level package for BioLM.

class biolm.BioLM(*, entity: str, action: str, type: str | None = None, items: Any | List[Any], params: dict | None = None, api_key: str | None = None, **kwargs)

Bases: object

Universal client for BioLM API.

This is a convenience wrapper that creates a client, makes the request, and returns the result. For long-running operations or when making multiple requests, consider using BioLMApiClient (async) or BioLMApi (sync) directly with proper cleanup via context managers or shutdown().

Args:

entity (str): The entity name (model, database, calculation, etc). action (str): The action to perform (e.g., ‘generate’, ‘encode’, ‘predict’, ‘search’, ‘finetune’). type (str): The type of item (e.g., ‘sequence’, ‘pdb’, ‘fasta_str’). item (Union[Any, List[Any]]): The item(s) to process. params (Optional[dict]): Optional parameters for the action. raise_httpx (bool): Whether to raise HTTPX errors. stop_on_error (bool): Stop on first error if True. output (str): ‘memory’ or ‘disk’. file_path (Optional[str]): Output file path if output=’disk’. api_key (Optional[str]): API key for authentication. compress_requests (bool): Enable gzip compression for POST requests. Default: True. compress_threshold (int): Minimum payload size in bytes to trigger compression. Default: 256.

run() Any

Run the specified action on the entity with the given item(s).

biolm.BioLMApi

alias of BlockingBioLMApi

class biolm.BioLMApiClient(model_name: str, api_key: str | None = None, base_url: str | None = None, timeout: Timeout = Timeout(connect=10.0, read=1200, write=1200, pool=1200), raise_httpx: bool = False, unwrap_single: bool = True, semaphore: int | Semaphore | None = 16, rate_limit: str | None = None, retry_error_batches: bool = True, compress_requests: bool = True, compress_threshold: int = 256, concurrent_batches: bool = True, http2: bool = True)

Bases: object

async call(func: str, items: List[dict], params: dict | None = None, raw: bool = False)
encode(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)
static extract_max_items(schema: dict) int | None

Extracts the ‘maxItems’ value for the ‘items’ key from the schema. Returns the integer value if found, else None.

generate(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)
async lookup(query: dict | List[dict], *, raw: bool = False, output: str = 'memory', file_path: str | None = None)
predict(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)
async schema(model: str, action: str) dict | None

Fetch the JSON schema for a given model and action, with caching. Returns the schema dict if successful, else None. Uses a module-level cache keyed by (model, action) so it is safe to use from any event loop (sync wrapper, async tests, pytest-xdist workers).

Schema is always fetched from the hosted platform API without credentials. Hub gateways and custom model API hosts may not expose /schema/, and sending API tokens to the public schema endpoint can return 401.

score(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True)
search(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True)
async shutdown()
class biolm.Finetune

Bases: object

Launch and track BioLM finetuning runs.

All methods are classmethods returning plain dicts. *_data arguments accept a list of row dicts ([{"sequence": ..., "label": ...}, ...]) or a raw CSV string; they are sent inline as JSON.

classmethod cancel(run_id: str, **kwargs) dict

Synchronous wrapper for cancel_async().

async classmethod cancel_async(run_id: str, *, api_key: str | None = None, base_url: str | None = None) dict

Cancel an in-flight run.

classmethod dsm_rl(**kwargs) dict

Synchronous wrapper for dsm_rl_async().

async classmethod dsm_rl_async(*, seed_sequences: List[str] | str, oracle_type: str = 'esmc', stability_objective: str = 'thermostability', training_mode: str = 'online', algorithm: str = 'ppo', num_episodes: int = 100, samples_per_episode: int = 64, learning_rate: float = 0.0003, batch_size: int = 8, mask_ratio: float = 0.3, mutation_rate: float = 0.1, run_name: str | None = None, environment_id: int | str | None = None, api_key: str | None = None, base_url: str | None = None) dict

Launch DSM RL protein optimization against an oracle.

classmethod dsm_stage1(**kwargs) dict

Synchronous wrapper for dsm_stage1_async().

async classmethod dsm_stage1_async(*, train_data: List[Dict[str, Any]] | str, test_data: List[Dict[str, Any]] | str | None = None, valid_data: List[Dict[str, Any]] | str | None = None, sequence_col: str = 'sequence', lr: float = 0.0001, batch_size: int = 8, grad_accum: int = 16, max_steps: int = 50000, max_length: int = 2048, save_every: int = 1000, fp16: bool = False, run_name: str | None = None, environment_id: int | str | None = None, api_key: str | None = None, base_url: str | None = None) dict

Launch DSM Stage 1 (single-chain masked-LM finetune).

classmethod dsm_stage2(**kwargs) dict

Synchronous wrapper for dsm_stage2_async().

async classmethod dsm_stage2_async(*, stage1_checkpoint: str, paired_data: List[Dict[str, Any]] | str, unpaired_data: List[Dict[str, Any]] | str | None = None, heavy_col: str = 'heavy', light_col: str = 'light', use_mixed_training: bool = False, fp16: bool = False, lr: float = 5e-05, batch_size: int = 4, unpaired_batch_size: int = 8, grad_accum: int = 16, max_steps: int = 25000, max_length: int = 300, unpaired_ratio: float = 2.0, run_name: str | None = None, environment_id: int | str | None = None, api_key: str | None = None, base_url: str | None = None) dict

Launch DSM Stage 2 (paired multichain finetune from a Stage 1 checkpoint).

classmethod get_run(run_id: str, **kwargs) dict

Synchronous wrapper for get_run_async().

async classmethod get_run_async(run_id: str, *, api_key: str | None = None, base_url: str | None = None) dict

Fetch sanitized detail (status, results) for a single run.

classmethod list_runs(**kwargs) dict

Synchronous wrapper for list_runs_async().

async classmethod list_runs_async(*, dag: str | None = None, status: str | None = None, page: int = 1, page_size: int = 20, api_key: str | None = None, base_url: str | None = None) dict

List the caller’s finetune runs (paginated).

classmethod progress(run_id: str, **kwargs) dict

Synchronous wrapper for progress_async().

async classmethod progress_async(run_id: str, *, api_key: str | None = None, base_url: str | None = None) dict

Fetch lightweight status + telemetry channel id for a run.

classmethod wait(run_id: str, *, poll_interval: float = 15.0, timeout: float | None = None, api_key: str | None = None, base_url: str | None = None) dict

Block until run_id reaches a terminal state, returning its detail.

classmethod xgboost(**kwargs) dict

Synchronous wrapper for xgboost_async().

async classmethod xgboost_async(*, train_data: List[Dict[str, Any]] | str, embedding_models: List[str], task_type: str = 'classification', target_column: str = 'label', text_column: str = 'sequence', test_data: List[Dict[str, Any]] | str | None = None, validation_data: List[Dict[str, Any]] | str | None = None, n_estimators: int = 100, max_depth: int = 6, learning_rate: float = 0.1, n_splits: int = 5, seed: int = 42, hyperopt: bool = False, hyperopt_n_trials: int | None = None, antibody_mode: bool = False, heavy_column: str = 'heavy', light_column: str = 'light', modality: str = 'protein', run_name: str | None = None, environment_id: int | str | None = None, api_key: str | None = None, base_url: str | None = None) dict

Launch an XGBoost finetune (optionally with Ray Tune hyperopt).

class biolm.Model(name: str, api_key: str | None = None, **kwargs)

Bases: object

User-friendly model interface for BioLM API.

Args:

name (str): The model name (e.g., ‘esm2-8m’, ‘esmfold’). api_key (Optional[str]): API key for authentication. **kwargs: Additional arguments passed to BioLMApi.

encode(items: Any | List[Any], type: str | None = None, params: dict | None = None, progress: bool = True, progress_callback: Callable[[int, int], None] | None = None, **kwargs)

Encode using the model.

Args:

items: Single item or list of items to encode. type: Type of item (e.g., ‘sequence’). Required if items are not dicts. params: Optional parameters for the encoding. progress: If True (default), show Rich progress bar (multiple items only). progress_callback: Optional (completed, total) callback; overrides progress=True if provided. **kwargs: Additional arguments (stop_on_error, output, file_path, etc.).

Returns:

Encoding results.

generate(items: Any | List[Any], type: str | None = None, params: dict | None = None, progress: bool = True, progress_callback: Callable[[int, int], None] | None = None, **kwargs)

Generate using the model.

Args:

items: Single item or list of items to generate from. type: Type of item (e.g., ‘context’, ‘pdb’). Required if items are not dicts. params: Optional parameters for the generation. progress: If True (default), show Rich progress bar (multiple items only). progress_callback: Optional (completed, total) callback; overrides progress=True if provided. **kwargs: Additional arguments (stop_on_error, output, file_path, etc.).

Returns:

Generation results.

get_example(action: str | None = None, format: str = 'python', **kwargs) str

Get SDK usage example for this model.

Args:

action: Action name (encode, predict, generate, lookup). If None, generates for first available action. format: Output format (‘python’, ‘markdown’, ‘rst’, ‘json’). **kwargs: Additional arguments passed to ExampleGenerator (base_url).

Returns:

Formatted example string.

get_examples(format: str = 'python', **kwargs) str

Get SDK usage examples for all supported actions of this model.

Args:

format: Output format (‘python’, ‘markdown’, ‘rst’, ‘json’). **kwargs: Additional arguments passed to ExampleGenerator (base_url).

Returns:

Formatted examples string with all actions.

lookup(query: dict | List[dict], **kwargs)

Lookup using the model.

Args:

query: Query dict or list of query dicts. **kwargs: Additional arguments (raw, output, file_path).

Returns:

Lookup results.

predict(items: Any | List[Any], type: str | None = None, params: dict | None = None, progress: bool = True, progress_callback: Callable[[int, int], None] | None = None, **kwargs)

Predict using the model.

Args:

items: Single item or list of items to predict. type: Type of item (e.g., ‘sequence’, ‘pdb’). Required if items are not dicts. params: Optional parameters for the prediction. progress: If True (default), show Rich progress bar (multiple items only). progress_callback: Optional (completed, total) callback; overrides progress=True if provided. **kwargs: Additional arguments (stop_on_error, output, file_path, etc.).

Returns:

Prediction results.

class biolm.Protocol(yaml_path: str)

Bases: object

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 with given inputs.

Args:

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

Returns:

Protocol execution results.

Note:

Protocol execution is not yet implemented. This is a placeholder.

classmethod fetch_by_id(protocol_id: str, api_key: str | None = None, base_url: str | None = None) dict

Fetch a protocol from the platform by ID (synchronous wrapper).

Args:

protocol_id: Protocol ID to fetch api_key: Optional API key for authentication base_url: Optional base URL (defaults to BIOLMAI_BASE_API_URL)

Returns:

Protocol data as dict (same structure as YAML)

Raises:

FileNotFoundError: If protocol not found (404) PermissionError: If not authenticated (401) ValueError: If API request fails

async classmethod fetch_by_id_async(protocol_id: str, api_key: str | None = None, base_url: str | None = None) dict

Fetch a protocol from the platform by ID.

Args:

protocol_id: Protocol ID to fetch api_key: Optional API key for authentication base_url: Optional base URL (defaults to BIOLMAI_BASE_API_URL)

Returns:

Protocol data as dict (same structure as YAML)

Raises:

FileNotFoundError: If protocol not found (404) PermissionError: If not authenticated (401) ValueError: If API request fails

classmethod init(output_path: str, example: str | None = None, force: bool = False) str

Initialize a new protocol YAML file.

Args:

output_path: Path where the protocol file should be created example: Optional example template name to use force: If True, overwrite existing file

Returns:

Path to the created file

Raises:

FileExistsError: If file exists and force=False ValueError: If example name is invalid FileNotFoundError: If example file doesn’t exist

static render_report(protocol_data: dict, source: str = 'file', console=None) None

Render a formatted report of the protocol using Rich.

Args:

protocol_data: Protocol data dictionary source: Source description (e.g., “file”, “platform”) console: Optional Rich Console instance (creates one if not provided)

Raises:

ValueError: If protocol_data is invalid or missing required fields

classmethod validate(yaml_path: str) ProtocolValidationResult

Validate a protocol YAML file.

Args:

yaml_path: Path to protocol YAML file.

Returns:

ProtocolValidationResult with validation results.

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

Bases: object

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(slug: str, version: int | None = None) Dict[str, Any]
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) 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
exception biolm.ProtocolNotFoundError

Bases: ProtocolRunError

The requested protocol slug/version does not exist or is not accessible.

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

Bases: object

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
download_files(*, output_dir: str | Path = '.', file_type: str = 'csv', overwrite: bool = False) Path

Compatibility alias for download().

progress() Dict[str, Any]
refresh() ProtocolRun
results() Dict[str, Any]
to_dataframe(*, output_dir: str | Path = '.', overwrite: bool = False)

Download CSV zip and return a pandas DataFrame.

wait(timeout: float = 3600.0, show_progress: bool = True) ProtocolRun
exception biolm.ProtocolRunError

Bases: Exception

A protocol run failed, was cancelled, or the API returned an error.

class biolm.Volume(name: str | None = None, api_key: str | None = None)

Bases: object

Volume management interface (not yet implemented).

Use hub CLI and storage workflows today. This class defines the intended Python API: list, create, and get.

Args:

name: Volume name. If None, uses the default volume. api_key: Optional API token; defaults to BIOLM_TOKEN.

create(name: str, **kwargs) Dict[str, Any]

Create a new volume.

Args:

name: Name of the volume to create. **kwargs: Additional volume parameters.

Returns:

Created volume information.

Note:

Volume creation is not yet implemented. This is a placeholder.

delete(name: str | None = None) bool

Delete a volume.

Args:

name: Volume name. If None, uses the volume name from initialization.

Returns:

True if deletion was successful.

Note:

Volume deletion is not yet implemented. This is a placeholder.

get(name: str | None = None) Dict[str, Any]

Get volume information.

Args:

name: Volume name. If None, uses the volume name from initialization.

Returns:

Volume information dictionary.

Note:

Volume retrieval is not yet implemented. This is a placeholder.

list() List[Dict[str, Any]]

List all available volumes.

Returns:

List of volume dictionaries.

Note:

Volume listing is not yet implemented. This is a placeholder.

class biolm.Workspace(name: str | None = None, api_key: str | None = None)

Bases: object

Workspace management interface (not yet implemented).

Use biolm workspace CLI commands today. This class defines the intended Python API: list, create, and get.

Args:

name: Workspace name. If None, uses the default workspace. api_key: Optional API token; defaults to BIOLM_TOKEN.

create(name: str, **kwargs) Dict[str, Any]

Create a new workspace.

Args:

name: Name of the workspace to create. **kwargs: Additional workspace parameters.

Returns:

Created workspace information.

Note:

Workspace creation is not yet implemented. This is a placeholder.

delete(name: str | None = None) bool

Delete a workspace.

Args:

name: Workspace name. If None, uses the workspace name from initialization.

Returns:

True if deletion was successful.

Note:

Workspace deletion is not yet implemented. This is a placeholder.

get(name: str | None = None) Dict[str, Any]

Get workspace information.

Args:

name: Workspace name. If None, uses the workspace name from initialization.

Returns:

Workspace information dictionary.

Note:

Workspace retrieval is not yet implemented. This is a placeholder.

list() List[Dict[str, Any]]

List all available workspaces.

Returns:

List of workspace dictionaries.

Note:

Workspace listing is not yet implemented. This is a placeholder.

biolm.biolm(*, entity: str, action: str, type: str | None = None, items: Any | List[Any], params: dict | None = None, api_key: str | None = None, **kwargs) Any

Call a BioLM model in one step (sync, blocking).

Wraps biolm.client.BioLM and returns the API result. Single-item calls return a dict; batch calls return a list.

Args:

entity: Model slug (e.g. "esm2-8m", "esmfold"). action: Model action (e.g. "encode", "predict", "generate"). type: Item type when items are plain strings (e.g. "sequence"). items: One item, a list of items, or a generator of items. params: Optional action-specific parameters. api_key: Optional API token; defaults to BIOLM_TOKEN. **kwargs: Passed through to biolm.client.BioLM.

Returns:

A single result dict or a list of result dicts (one per input item).

biolm.encode(model_name: str, items: Any | List[Any], type: str | None = None, params: dict | None = None, **kwargs)

Quick encoding using a model.

biolm.generate(model_name: str, items: Any | List[Any], type: str | None = None, params: dict | None = None, **kwargs)

Quick generation using a model.

biolm.get_example(model_name: str, action: str | None = None, format: str = 'python', api_key: str | None = None, base_url: str | None = None) str

Generate SDK usage example for a model (synchronous).

Args:

model_name: Name of the model. action: Action name (optional, will try to detect). format: Output format (‘python’, ‘markdown’, ‘rst’, ‘json’). api_key: Optional API key. base_url: Optional base URL.

Returns:

Formatted example string.

biolm.list_models(api_key: str | None = None, base_url: str | None = None) List[Dict[str, Any]]

List all available models from community-api-models endpoint (synchronous).

Args:

api_key: Optional API key. base_url: Optional base URL.

Returns:

List of model dictionaries.

biolm.load_csv(file_path: str | Path | IO, sequence_key: str | None = None) List[Dict[str, Any]]

Load data from a CSV file.

Parses a CSV file and returns a list of dictionaries suitable for use with BioLM API requests. Each row becomes a dictionary with column headers as keys. All values are kept as strings (no type inference).

Args:

file_path: Path to CSV file (str, Path) or file-like object. sequence_key: Optional key name to validate exists in CSV; if provided, raises ValueError if column is missing.

Returns:

List of dictionaries, one per row.

Raises:

FileNotFoundError: If file path doesn’t exist. ValueError: If file is empty or sequence_key column is missing.

Example:
default
>>> items = load_csv("data.csv", sequence_key="sequence")
>>> items[0]
{'sequence': 'ACDEFGHIKLMNPQRSTVWY', 'id': 'seq1', 'score': '0.95'}
biolm.load_fasta(file_path: str | Path | IO) List[Dict[str, Any]]

Load sequences from a FASTA file.

Parses a FASTA file and returns a list of dictionaries suitable for use with BioLM API requests. Each dictionary contains: - “sequence”: The sequence string - “id”: Sequence identifier from header (if available) - “metadata”: Additional metadata from header (if available)

Supports multi-line sequences (wrapped sequences).

Args:

file_path: Path to FASTA file (str, Path) or file-like object

Returns:

List of dictionaries, each containing sequence data

Raises:

FileNotFoundError: If file path doesn’t exist ValueError: If file is empty or malformed

Example:
default
>>> items = load_fasta("sequences.fasta")
>>> items[0]
{'sequence': 'ACDEFGHIKLMNPQRSTVWY', 'id': 'seq1', 'metadata': {}}
biolm.load_json(file_path: str | Path | IO) List[Dict[str, Any]]

Load data from a JSON file or JSONL (newline-delimited JSON) file.

Parses a JSON file and returns a list of dictionaries suitable for use with BioLM API requests. Supports: - Single JSON object: Returns list with one item - JSON array: Returns list of items - JSONL format (newline-delimited): Returns list of items, one per line

Args:

file_path: Path to JSON/JSONL file (str, Path), file-like object, or “-” for stdin

Returns:

List of dictionaries, each containing data suitable for API requests

Raises:

FileNotFoundError: If file path doesn’t exist ValueError: If file is empty or malformed json.JSONDecodeError: If JSON is invalid

Example:
default
>>> items = load_json("data.json")
>>> items[0]
{'sequence': 'ACDEFGHIKLMNPQRSTVWY', 'id': 'seq1'}
default
>>> items = load_json("data.jsonl")  # JSONL format
>>> len(items)
3
biolm.load_pdb(file_path: str | Path | IO) List[Dict[str, Any]]

Load PDB structure(s) from a PDB file.

Reads a PDB file and returns a list of dictionaries suitable for use with BioLM API requests. For single-model PDBs, returns one item. For multi-model PDBs (with MODEL/ENDMDL records), returns one item per model.

Args:

file_path: Path to PDB file (str, Path) or file-like object

Returns:

List of dictionaries, each containing “pdb” key with PDB content. Single-model files return [{“pdb”: “…”}]. Multi-model files return [{“pdb”: “…”}, {“pdb”: “…”}, …].

Raises:

FileNotFoundError: If file path doesn’t exist ValueError: If file is empty

Example:
default
>>> items = load_pdb("structure.pdb")
>>> items[0]
{'pdb': 'ATOM      1  N   MET A   1      20.154  16.967  19.502...'}
biolm.predict(model_name: str, items: Any | List[Any], type: str | None = None, params: dict | None = None, **kwargs)

Quick prediction using a model.

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) dict

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

biolm.to_csv(data: List[Dict[str, Any]], file_path: str | Path | IO, fieldnames: List[str] | None = None) None

Write data to a CSV file.

Converts a list of dictionaries (API response format) to CSV format.

Args:

data: List of dictionaries to write. file_path: Output file path (str, Path) or file-like object. fieldnames: Optional list of column names; if not provided, inferred from the first item’s keys (missing keys filled with empty strings).

Raises:

ValueError: If data is empty.

Example:
default
>>> data = [{"sequence": "ACDEFGHIKLMNPQRSTVWY", "score": 0.95}]
>>> to_csv(data, "output.csv")
biolm.to_fasta(data: List[Dict[str, Any]], file_path: str | Path | IO, sequence_key: str = 'sequence') None

Write sequences to a FASTA file.

Converts a list of dictionaries (API response format) to FASTA format. Each dictionary should contain a sequence field (default: “sequence”).

Args:

data: List of dictionaries containing sequence data file_path: Output file path (str, Path) or file-like object sequence_key: Key to use for sequence data (default: “sequence”)

Raises:

ValueError: If sequence_key is missing from any item KeyError: If required keys are missing

Example:
default
>>> data = [{"sequence": "ACDEFGHIKLMNPQRSTVWY", "id": "seq1"}]
>>> to_fasta(data, "output.fasta")
biolm.to_json(data: List[Dict[str, Any]], file_path: str | Path | IO, indent: int = 2, jsonl: bool = False) None

Write data to a JSON file or JSONL (newline-delimited JSON) file.

Converts a list of dictionaries (API response format) to JSON format.

Args:

data: List of dictionaries to write. file_path: Output file path (str, Path), file-like object, or “-” for stdout. indent: Indentation level for JSON (default: 2). Use None for compact JSON. jsonl: If True, write as JSONL (one JSON object per line); if False, write as JSON array. Default: False.

Raises:

ValueError: If data is empty.

Example:
default
>>> data = [{"sequence": "ACDEFGHIKLMNPQRSTVWY", "score": 0.95}]
>>> to_json(data, "output.json")
>>> to_json(data, "output.jsonl", jsonl=True)
biolm.to_pdb(data: List[Dict[str, Any]], file_path: str | Path | IO, pdb_key: str = 'pdb') None

Write PDB structure(s) to a PDB file.

Converts a list of dictionaries (API response format) to PDB format. Each dictionary should contain a PDB content field (default: “pdb”). If multiple items are provided, they are concatenated.

Args:

data: List of dictionaries containing PDB data file_path: Output file path (str, Path) or file-like object pdb_key: Key to use for PDB content (default: “pdb”)

Raises:

ValueError: If data is empty or pdb_key is missing from any item

Example:
default
>>> data = [{"pdb": "ATOM      1  N   MET A   1..."}]
>>> to_pdb(data, "output.pdb")

We speak the language of bio-AI

© 2022 - 2026 BioLM. All Rights Reserved.