biolm.core

When to use

• BioLMApi — sync client with schema helpers and manual batching

• BioLMApiClient — async client (await its methods)

• biolm() — legacy one-shot sync wrapper (re-exported from biolm); still common in examples

• Client interfaces — extended examples for batching, disk output, and errors

• Concurrency — async patterns and throughput tuning

Examples

Legacy one-shot call:

from biolm import biolm

result = biolm(entity="esm2-8m", action="encode", type="sequence", items="MSILVTRPSPAGEEL")

Sync HTTP client:

from biolm.core.http import BioLMApi

model = BioLMApi("esm2-8m", raise_httpx=False)
result = model.encode(items=[{"sequence": "MSILV"}, {"sequence": "MDNELE"}])
schema = model.schema("esm2-8m", "encode")

Async HTTP client:

from biolm.core.http import BioLMApiClient
import asyncio

async def main():
    model = BioLMApiClient("esmfold")
    result = await model.predict(items=[{"sequence": "MDNELE"}])
    await model.shutdown()

asyncio.run(main())

API

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.core.http.BioLMApi

alias of BlockingBioLMApi

class biolm.core.http.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)

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)

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.

async shutdown()

See also

• biolm.models — Model (recommended inference interface)

• Batching and Input Flexibility — generators, batch sizes, and manual batching

• Error Handling — stop-on-error, retries, disk output

• Rate Limiting and Throttling — throttling and concurrency

• biolm.core package — full biolm.core module tree