> ## Documentation Index
> Fetch the complete documentation index at: https://docs.bioptimus.com/llms.txt
> Use this file to discover all available pages before exploring further.

# bioptimus.models.clients.http

HTTP client for direct API endpoints.

Sends JSON payloads to a model server via plain HTTP POST using `requests` (synchronous) and `aiohttp` (asynchronous).

***

#### is\_transient

```python theme={null}
def is_transient(exc: BaseException) -> bool
```

Returns whether an exception is a transient failure worth retrying.

Transient failures include connection resets, timeouts, and HTTP responses carrying a retryable server-side status code (e.g. 503 from momentary GPU back-pressure). Deterministic failures such as HTTP 400 (malformed request) are not retried.

<ParamField body="exc">
  The exception raised while dispatching a request.
</ParamField>

**Returns**:

`True` if the failure is transient and should be retried.

***

#### tile\_retry\_logger

```python theme={null}
def tile_retry_logger(
    logger: logging.Logger,
    tile_index: int,
    *,
    max_retries: int = DEFAULT_MAX_RETRIES
) -> Callable[[int, BaseException, float], None]
```

Builds an `on_retry` callback that logs a per-tile dispatch retry.

Returns the callback expected by [`retry_sync`](/sdk-reference/models/clients/http#retry_sync) / [`retry_async`](/sdk-reference/models/clients/http#retry_async), emitting a structured `tile_dispatch_retry` warning. Shared by every per-tile dispatch path so the retry log event and fields stay identical.

<ParamField body="logger">
  The logger to emit on.
</ParamField>

<ParamField body="tile_index">
  Index of the tile being retried, recorded in the event.
</ParamField>

<ParamField body="max_retries">
  Retry budget, used to report the total attempt count.
</ParamField>

**Returns**:

A callback taking `(attempt, exc, delay)` that logs the retry.

***

#### retry\_sync

```python theme={null}
def retry_sync(
        attempt: Callable[[], T],
        *,
        max_retries: int = DEFAULT_MAX_RETRIES,
        base_delay: float = DEFAULT_RETRY_BASE_DELAY,
        is_retryable: Callable[[BaseException], bool] = is_transient,
        sleep: Callable[[float], None] = time.sleep,
        on_retry: Callable[[int, BaseException, float], None] | None = None
) -> T
```

Runs *attempt*, retrying transient failures with exponential backoff.

Synchronous twin of [`retry_async`](/sdk-reference/models/clients/http#retry_async) for threaded (non-async) callers, such as the `TorchModel` thread-pool dispatch. Shares the same backoff policy, defaults, and `on_retry` contract so both paths behave identically.

<ParamField body="attempt">
  Zero-argument callable that performs one attempt.
</ParamField>

<ParamField body="max_retries">
  Number of retries after the initial attempt (so the total number of attempts is `max_retries + 1`).
</ParamField>

<ParamField body="base_delay">
  Seed for the exponential backoff (`base_delay * 2**n`).
</ParamField>

<ParamField body="is_retryable">
  Predicate deciding whether a raised exception is transient and worth retrying.
</ParamField>

<ParamField body="sleep">
  Blocking sleep function; injectable so tests can skip real waits.
</ParamField>

<ParamField body="on_retry">
  Optional callback invoked before each retry with the zero-based attempt number, the exception, and the delay.
</ParamField>

**Returns**:

The result of the first successful *attempt*.

**Raises**:

* `BaseException` - Re-raises the last exception when the retry budget is exhausted or the exception is not retryable.

***

#### retry\_async

```python theme={null}
async def retry_async(
        attempt: Callable[[], Awaitable[T]],
        *,
        max_retries: int = DEFAULT_MAX_RETRIES,
        base_delay: float = DEFAULT_RETRY_BASE_DELAY,
        is_retryable: Callable[[BaseException], bool] = is_transient,
        sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
        on_retry: Callable[[int, BaseException, float], None] | None = None
) -> T
```

Runs *attempt*, retrying transient failures with exponential backoff.

The retry mechanism is transport-agnostic: the work to perform, the logging, and (optionally) the sleep function are all supplied by the caller, keeping this function reusable and easy to test.

<ParamField body="attempt">
  Zero-argument coroutine factory that performs one attempt.
</ParamField>

<ParamField body="max_retries">
  Number of retries after the initial attempt (so the total number of attempts is `max_retries + 1`).
</ParamField>

<ParamField body="base_delay">
  Seed for the exponential backoff (`base_delay * 2**n`).
</ParamField>

<ParamField body="is_retryable">
  Predicate deciding whether a raised exception is transient and worth retrying.
</ParamField>

<ParamField body="sleep">
  Async sleep function; injectable so tests can skip real waits.
</ParamField>

<ParamField body="on_retry">
  Optional callback invoked before each retry with the zero-based attempt number, the exception, and the delay.
</ParamField>

**Returns**:

The result of the first successful *attempt*.

**Raises**:

* `BaseException` - Re-raises the last exception when the retry budget is exhausted or the exception is not retryable.

## HTTPClient

```python theme={null}
class HTTPClient(base_url: str, endpoints: Dict[str, str], timeout: float = 30.0)
```

Sends JSON payloads over HTTP.

Fully configured at construction — `predict` / `embed` only need the serialized body.

<ParamField body="base_url">
  Root URL of the model server, e.g. `"http://localhost:8080"`.
</ParamField>

<ParamField body="endpoints">
  Mapping of endpoint type to path. At least one key must be present; see the example below.
</ParamField>

<ParamField body="timeout">
  HTTP request timeout in seconds.
</ParamField>

**Example**:

```python theme={null}
client = HTTPClient(
    "http://localhost:8080",
    endpoints={
        "prediction": "/api/predict/m-optimus",
        "embedding": "/api/embed/m-optimus",
    },
)
pred_json = client.predict(request_json)
emb_json = client.embed(request_json)
```

***

#### close

```python theme={null}
def close() -> None
```

Closes the underlying HTTP session and releases connections.

***

#### predict

```python theme={null}
@log_failures
def predict(body: str) -> str
```

POST to the prediction endpoint.

<ParamField body="body">
  Serialized JSON request payload.
</ParamField>

**Returns**:

Response body as a JSON string.

***

#### embed

```python theme={null}
@log_failures
def embed(body: str) -> str
```

POST to the embedding endpoint.

<ParamField body="body">
  Serialized JSON request payload.
</ParamField>

**Returns**:

Response body as a JSON string.

***

#### predict\_with\_embedding

```python theme={null}
@log_failures
def predict_with_embedding(body: str) -> str
```

POST to the combined prediction+embedding endpoint.

Returns both the gene prediction and the tile embedding from a single forward pass. Only models that expose both prediction and embedding configure this endpoint; others raise `NotImplementedError`.

<ParamField body="body">
  Serialized JSON request payload.
</ParamField>

**Returns**:

Response body as a JSON string, with `embedding` populated
alongside `output`.

***

#### metadata

```python theme={null}
@log_failures
def metadata() -> str
```

GET the metadata endpoint.

**Returns**:

Response body as a JSON string.

***

#### predict\_async

```python theme={null}
@log_failures
async def predict_async(body: str, session: Any = None) -> str
```

POST to the prediction endpoint asynchronously.

<ParamField body="body">
  Serialized JSON request payload.
</ParamField>

<ParamField body="session">
  An `aiohttp.ClientSession` for connection pooling.
</ParamField>

**Returns**:

Response body as a JSON string.

***

#### embed\_async

```python theme={null}
@log_failures
async def embed_async(body: str, session: Any = None) -> str
```

POST to the embedding endpoint asynchronously.

<ParamField body="body">
  Serialized JSON request payload.
</ParamField>

<ParamField body="session">
  An `aiohttp.ClientSession` for connection pooling.
</ParamField>

**Returns**:

Response body as a JSON string.

***

#### predict\_with\_embedding\_async

```python theme={null}
@log_failures
async def predict_with_embedding_async(body: str, session: Any = None) -> str
```

POST to the combined prediction+embedding endpoint asynchronously.

<ParamField body="body">
  Serialized JSON request payload.
</ParamField>

<ParamField body="session">
  An `aiohttp.ClientSession` for connection pooling.
</ParamField>

**Returns**:

Response body as a JSON string, with `embedding` populated
alongside `output`.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.