> For clean Markdown content of this page, append .md to this URL.

# aitune.dynamo.worker

Dynamo worker — serve AITune-tuned models as Dynamo endpoints.

## Module Contents

### Classes

| Name                                                                     | Description                                                              |
| ------------------------------------------------------------------------ | ------------------------------------------------------------------------ |
| [`DynamoWorker`](#aitune-dynamo-worker-DynamoWorker)                     | Base class for AITune Dynamo workers (power-user API).                   |
| [`DynamoWorkerConfig`](#aitune-dynamo-worker-DynamoWorkerConfig)         | Configuration for the high-level :func:`dynamo_worker` entrypoint.       |
| [`_HighLevelDynamoWorker`](#aitune-dynamo-worker-_HighLevelDynamoWorker) | Internal worker built from a DynamoWorkerConfig. Not part of public API. |

### Functions

| Name                                                             | Description                                                              |
| ---------------------------------------------------------------- | ------------------------------------------------------------------------ |
| [`_get_wiring`](#aitune-dynamo-worker-_get_wiring)               | Return (ModelInput, ModelType, request\_class) for *type\_*.             |
| [`_import_dynamo`](#aitune-dynamo-worker-_import_dynamo)         | Import dynamo runtime dependencies, raising a clear error if absent.     |
| [`_pack_response`](#aitune-dynamo-worker-_pack_response)         | Convert a user function's return value to the Dynamo wire-format dict.   |
| [`_run_dynamo_worker`](#aitune-dynamo-worker-_run_dynamo_worker) | Low-level runtime loop. Call `setup()` once, serve until SIGTERM/SIGINT. |
| [`dynamo_worker`](#aitune-dynamo-worker-dynamo_worker)           | Serve a tuned model as a Dynamo worker endpoint.                         |

### Data

[`_VALID_TYPES`](#aitune-dynamo-worker-_VALID_TYPES)

[`logger`](#aitune-dynamo-worker-logger)

### API

```python
class aitune.dynamo.worker.DynamoWorker()
```

Base class for AITune Dynamo workers (power-user API).

Subclass, override :meth:`setup` and :meth:`serve`, then call :meth:`run`.

**`component`** `str = 'backend'`

---

**`endpoint_name`** `str = 'generate'`

---

**`namespace`** `str = 'aitune'`

---

```python
aitune.dynamo.worker.DynamoWorker.on_ready(
    runtime: typing.Any,
    endpoint: typing.Any
) -> None
```

async

Called after the Dynamo endpoint is registered.

Override to call `register_model` or perform post-startup work.

**Parameters:**

**`runtime`** `Any`

The :class:`DistributedRuntime` instance.

---

**`endpoint`** `Any`

The registered Dynamo :class:`Endpoint` object.

---

```python
aitune.dynamo.worker.DynamoWorker.run(
    enable_nats: bool = False
) -> None
```

Start serving. Blocks until SIGTERM/SIGINT.

**Parameters:**

**`enable_nats`** `bool` — default: False

Enable NATS JetStream for KV cache events.

---

```python
aitune.dynamo.worker.DynamoWorker.serve(
    request: typing.Any
) -> collections.abc.AsyncGenerator[typing.Any, None]
```

async

Handle one request. Async generator — yield response chunks.

**Parameters:**

**`request`** `Any`

Incoming request payload.

---

**Raises:**

* `NotImplementedError`: Must be overridden by subclasses.

```python
aitune.dynamo.worker.DynamoWorker.setup() -> None
```

Initialize the model. Called once before serving starts.

**Raises:**

* `NotImplementedError`: Must be overridden by subclasses.

```python
class aitune.dynamo.worker.DynamoWorkerConfig(
    type: typing.Literal['image', 'video', 'embedding'],
    model_path: str,
    mapping: collections.abc.Callable | None = None,
    namespace: str = 'aitune',
    component: str = 'backend',
    endpoint: str = 'generate',
    enable_nats: bool = False,
    model_name: str | None = None
)
```

Dataclass

Configuration for the high-level :func:`dynamo_worker` entrypoint.

**Parameters:**

**`type`** `Literal['image', 'video', 'embedding']`

Modality type. One of `"image"`, `"video"`, `"embedding"`.

---

**`model_path`** `str`

HuggingFace model ID or local path passed to `register_model`.

---

**`mapping`** `Callable | None` — default: None

Optional adapter `fn(DynamoRequest) -> dict`. The dict is
unpacked as `**kwargs` when calling the user function. If `None`
and the user passed a plain callable (not `nn.Module`), the raw
Dynamo request object is passed as the sole positional argument.

---

**`namespace`** `str` — default: 'aitune'

Dynamo service namespace. Default: `"aitune"`.

---

**`component`** `str` — default: 'backend'

Component name within the namespace. Default: `"backend"`.

---

**`endpoint`** `str` — default: 'generate'

Endpoint name within the component. Default: `"generate"`.

---

**`enable_nats`** `bool` — default: False

Enable NATS JetStream for KV cache events. Default: `False`.

---

**`model_name`** `str | None` — default: None

Name advertised to the Dynamo frontend. Defaults to `model_path`.

---

**`component`** `str = 'backend'`

---

**`enable_nats`** `bool = False`

---

**`endpoint`** `str = 'generate'`

---

**`mapping`** `Callable | None = None`

---

**`model_name`** `str | None = None`

---

**`model_path`** `str`

---

**`namespace`** `str = 'aitune'`

---

**`type`** `Literal['image', 'video', 'embedding']`

---

```python
class aitune.dynamo.worker._HighLevelDynamoWorker(
    model_or_fn: torch.nn.Module | collections.abc.Callable,
    config: aitune.dynamo.worker.DynamoWorkerConfig
)
```

**Bases:** [DynamoWorker](#aitune-dynamo-worker-DynamoWorker)

Internal worker built from a DynamoWorkerConfig. Not part of public API.

**`component`** `= config.component`

---

**`endpoint_name`** `= config.endpoint`

---

**`namespace`** `= config.namespace`

---

```python
aitune.dynamo.worker._HighLevelDynamoWorker._version_specific_register_model_kwargs() -> dict
```

From version 1.3.0 dynamo introduces a new WorkerType enum and made it required.

```python
aitune.dynamo.worker._HighLevelDynamoWorker.on_ready(
    runtime: typing.Any,
    endpoint: typing.Any
) -> None
```

async

Register this worker with the Dynamo frontend.

```python
aitune.dynamo.worker._HighLevelDynamoWorker.serve(
    request: typing.Any
) -> collections.abc.AsyncGenerator[typing.Any, None]
```

async

Deserialize request, run user function in executor, pack and yield response.

```python
aitune.dynamo.worker._HighLevelDynamoWorker.setup() -> None
```

No-op: the model is already initialized before dynamo\_worker() is called.

```python
aitune.dynamo.worker._get_wiring(
    type_: str
) -> tuple
```

Return (ModelInput, ModelType, request\_class) for *type\_*.

All imports are deferred because `dynamo` is optional.

**Parameters:**

**`type_`** `str`

Modality type string (`"image"`, `"video"`, `"embedding"`).

---

**Returns:** `tuple`

Tuple of (ModelInput enum value, ModelType enum value, request Pydantic class).

```python
aitune.dynamo.worker._import_dynamo() -> tuple
```

Import dynamo runtime dependencies, raising a clear error if absent.

**Returns:** `tuple`

Tuple of (DistributedRuntime class, dynamo\_worker decorator, uvloop module).

**Raises:**

* `ImportError`: When `ai-dynamo-runtime` or `uvloop` is not installed.

```python
aitune.dynamo.worker._pack_response(
    result: typing.Any,
    config: aitune.dynamo.worker.DynamoWorkerConfig
) -> dict
```

Convert a user function's return value to the Dynamo wire-format dict.

**Parameters:**

**`result`** `Any`

Return value from the user's inference function.

---

**`config`** `DynamoWorkerConfig`

Worker configuration (used for model name and type).

---

**Returns:** `dict`

A dict ready to yield to the Dynamo runtime.

**Raises:**

* `TypeError`: When *result* cannot be converted for *config.type*.

```python
aitune.dynamo.worker._run_dynamo_worker(
    setup: collections.abc.Callable[[], None],
    serve: collections.abc.Callable[[Any], collections.abc.AsyncGenerator[typing.Any, None]],
    namespace: str = 'aitune',
    component: str = 'backend',
    endpoint: str = 'generate',
    enable_nats: bool = False,
    on_ready: collections.abc.Callable[[Any, Any], collections.abc.Coroutine] | None = None
) -> None
```

Low-level runtime loop. Call `setup()` once, serve until SIGTERM/SIGINT.

**Parameters:**

**`setup`** `Callable[[], None]`

Zero-argument initializer called before serving starts.

---

**`serve`** `Callable[[Any], AsyncGenerator[Any, None]]`

Async generator `serve(request) -> AsyncIterable[response]`.

---

**`namespace`** `str` — default: 'aitune'

Dynamo service namespace.

---

**`component`** `str` — default: 'backend'

Component name within the namespace.

---

**`endpoint`** `str` — default: 'generate'

Endpoint name within the component.

---

**`enable_nats`** `bool` — default: False

Enable NATS JetStream for KV cache events.

---

**`on_ready`** `Callable[[Any, Any], Coroutine] | None` — default: None

Optional async callable invoked after endpoint registration.

---

```python
aitune.dynamo.worker.dynamo_worker(
    model_or_fn: torch.nn.Module | collections.abc.Callable,
    config: aitune.dynamo.worker.DynamoWorkerConfig
) -> None
```

Serve a tuned model as a Dynamo worker endpoint.

The minimal path to serving after `ait.tune()` or `ait.load()`:

1. Build a :class:`DynamoWorkerConfig` for your modality.
2. Call this function. It blocks until SIGTERM/SIGINT.

Startup validation happens before any Dynamo runtime is started.
Modality-specific request deserialization, `register_model`, sync-to-async
wrapping, and response packing are handled automatically.

**Parameters:**

**`model_or_fn`** `nn.Module | Callable`

A `torch.nn.Module` (requires `config.mapping`) or any
callable. If callable and `config.mapping` is `None`, the raw
Dynamo request object is passed as the sole argument.

---

**`config`** `DynamoWorkerConfig`

Worker configuration including modality type, model path, and
optional request adapter.

---

**Raises:**

* `ValueError`: If `config.type` is not a supported P0 modality, or if
  a `torch.nn.Module` is passed without a `mapping`.
* `ImportError`: If `ai-dynamo-runtime` is not installed.

```python
aitune.dynamo.worker._VALID_TYPES: frozenset[str] = frozenset({'image', 'video', 'embedding'})
```

```python
aitune.dynamo.worker.logger = getLogger(__name__)
```