> For clean Markdown content of this page, append .md to this URL. For the complete documentation index, see https://docs.nvidia.com/dynamo/llms.txt. For full content including API reference and SDK examples, see https://docs.nvidia.com/dynamo/llms-full.txt.
# dynamo.common
`dynamo.common` publishes 50 classes and 30 functions. Source: [`components/src/dynamo/common/__init__.py`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/__init__.py)
#### AbstractEmbeddingReceiver (class)
Abstract base class for a receiver of precomputed embeddings from the encode worker.
```python
from dynamo.common.multimodal import AbstractEmbeddingReceiver
```
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L83`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L83)
**Public methods**
receive_embeddings
```python
receive_embeddings(request: TransferRequest) -> tuple[int, torch.Tensor]
```
Abstract method to receive precomputed embeddings for a given request ID.
**Parameters**
**`request`** `TransferRequest`
The TransferRequest object containing information to receive embeddings.
---
**Returns**
* `int` — A tuple containing the tensor ID and the received embeddings as a torch.Tensor.
* `torch.Tensor` — Caller should invoke release\_tensor(tensor\_id) when the tensor is no longer needed to free up resources.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L88)
release_tensor
```python
release_tensor(tensor_id: int) -> None
```
Abstract method to indicate that the tensor associated with the ID is no longer in use. Args: tensor\_id: The ID of the tensor to release.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L104)
#### AbstractEmbeddingSender (class)
Abstract base class for a sender of precomputed embeddings to the downstream worker.
```python
from dynamo.common.multimodal import AbstractEmbeddingSender
```
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L114`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L114)
**Public methods**
send_embeddings
```python
send_embeddings(embeddings: torch.Tensor, stage_embeddings: bool = False) -> tuple[TransferRequest, Awaitable[None]]
```
Abstract method to send precomputed embeddings for a given request ID.
**Parameters**
**`embeddings`** `torch.Tensor`
A torch.Tensor of the embeddings to send.
---
**`stage_embeddings`** `bool`
A boolean indicating whether the embeddings should be staged for the transfer,
---
Returns:
A tuple containing the TransferRequest object and an awaitable that can be awaited to indicate the send is completed.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L119)
#### AiohttpClient (class)
aiohttp-backed concrete client.
```python
from dynamo.common.http import AiohttpClient
```
```python
AiohttpClient(config = None) -> None
```
[`components/src/dynamo/common/http/aiohttp_client.py#L28`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/aiohttp_client.py#L28)
**Public methods**
**init**
```python
__init__(config = None) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/aiohttp_client.py#L31)
close
```python
close() -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/aiohttp_client.py#L130)
#### AsyncEncoderCache (class)
Async wrapper with request coalescing over MultimodalEmbeddingCacheManager.
```python
from dynamo.common.multimodal import AsyncEncoderCache
```
```python
AsyncEncoderCache(cache: MultimodalEmbeddingCacheManager)
```
Provides async get\_or\_compute that deduplicates concurrent requests
for the same key, ensuring only one encoding runs at a time per key.
Thread Safety:
This class is NOT thread-safe. It is designed to run within a single
asyncio event loop. All access must be from the same thread.
[`components/src/dynamo/common/multimodal/async_encoder_cache.py#L45`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/async_encoder_cache.py#L45)
**Public methods**
**init**
```python
__init__(cache: MultimodalEmbeddingCacheManager)
```
Initialize the async encoder cache.
**Parameters**
**`cache`** `MultimodalEmbeddingCacheManager`
Underlying MultimodalEmbeddingCacheManager for storage.
---
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/async_encoder_cache.py#L57)
get
```python
get(key: str) -> Optional[CachedEmbedding]
```
Synchronous get from underlying cache.
**Parameters**
**`key`** `str`
Cache key.
---
**Returns**
* `Optional[CachedEmbedding]` — Cached embedding or None if not found.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/async_encoder_cache.py#L67)
get_or_compute
```python
get_or_compute(key: str, compute_fn: Callable[[], Awaitable[CachedEmbedding]]) -> CachedEmbedding
```
Get from cache or compute with request coalescing.
If the key is in cache, returns immediately.
If another coroutine is already computing this key, waits for that result.
Otherwise, computes and caches the result.
**Parameters**
**`key`** `str`
Cache key (typically content hash).
---
**`compute_fn`** `Callable[[], Awaitable[CachedEmbedding]]`
Async function to compute the embedding if not cached.
---
**Returns**
* `CachedEmbedding` — The cached or computed embedding.
**Raises**
* `Exception` — Re-raises any exception from compute\_fn.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/async_encoder_cache.py#L79)
#### AudioLoader (class)
Async audio loader for multimodal pipelines.
```python
from dynamo.common.multimodal import AudioLoader
```
```python
AudioLoader(http_timeout: float = 30.0, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None) -> None
```
Delegates URL fetching and decoding to vLLM's `MediaConnector` +
`AudioMediaIO` so that the exact same loading logic runs whether the
request arrives via `vllm serve` or through Dynamo. Returns
`(waveform, sample_rate)` tuples at the native sample rate — vLLM's
model-specific `MultiModalDataParser` handles resampling and channel
normalization downstream.
Also supports the NIXL decoded variant for frontend-decoded audio
transferred via RDMA.
[`components/src/dynamo/common/multimodal/audio_loader.py#L74`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/audio_loader.py#L74)
**Public methods**
**init**
```python
__init__(http_timeout: float = 30.0, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/audio_loader.py#L88)
load_audio
```python
load_audio(audio_url: str) -> tuple[np.ndarray, float]
```
Load audio from a URL and return a (waveform, sample\_rate) tuple.
Supports http(s), data: URIs, file:// paths, and bare filesystem paths.
Audio is loaded at the native sample rate — no resampling is performed.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/audio_loader.py#L140)
load_audio_batch
```python
load_audio_batch(audio_mm_items: List[Dict[str, Any]]) -> List[tuple[np.ndarray, float]]
```
Load a batch of audio files from multimodal data items.
Supports two paths:
1. Url variant: Download and decode audio via vLLM's MediaConnector
2. Decoded variant: Read pre-decoded audio via NIXL RDMA
(requires enable\_frontend\_decoding=True)
**Returns**
* `List[tuple[np.ndarray, float]]` — List of (waveform, sample\_rate) tuples.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/audio_loader.py#L192)
#### BaseEngine (class)
Abstract base for all engines — the modality-agnostic lifecycle.
```python
from dynamo.common.backend import BaseEngine
```
`Worker` drives every engine through the same lifecycle regardless of
modality; only the request/response shape of `generate` differs.
That method is therefore declared on the modality-specific subclasses
(`LLMEngine` for token-based inference, `RawEngine` for
raw non-token media generation), not here.
Lifecycle:
1. from\_args(argv) -- parse CLI args, return (engine, WorkerConfig)
2. start() -- start the engine, return EngineConfig metadata.
After start() returns, generate() MUST be ready
to accept calls. Worker begins serving
immediately after start().
3. generate() -- called for each request (concurrent calls expected)
4. abort() -- called when a request is cancelled (optional, default no-op)
5. cleanup() -- called once on shutdown, release all resources
[`components/src/dynamo/common/backend/engine.py#L154`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L154)
**Public methods**
from_args
```python
from_args(cls, argv: list[str] | None = None) -> tuple[BaseEngine, WorkerConfig]
```
Parse CLI args and construct the engine (not yet started).
**Parameters**
**`argv`** `list[str] | None`
Command-line arguments. `None` means `sys.argv[1:]`.
---
**Returns**
* `tuple[BaseEngine, WorkerConfig]` — A `(engine, worker_config)` pair.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L174)
start
```python
start(worker_id: int) -> EngineConfig
```
Start the engine and return registration metadata.
After this returns the engine MUST be ready to accept `generate()`
calls. `Worker` will register the model and begin serving
immediately.
`worker_id` is an opaque, runtime-allocated unique identifier for
this worker. It is stable from `start()` onward for the worker's
lifetime and unique across replicas in the cluster. Engines that
need a per-worker key for cluster-wide bookkeeping should derive it
from this value rather than hashing host/pid or asking operators for a
CLI override. The internal mechanism (discovery instance ID) is not
part of the contract — engines should treat it as opaque.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L189)
abort
```python
abort(context: Context) -> None
```
Abort an in-flight request (optional, default no-op).
Called by Worker when the client disconnects or
the request is cancelled. Override to release engine resources
(KV cache, scheduler slots, etc.).
`context.metadata` in this callback reflects the original
propagated request metadata snapshot. Mutations made to
`context.metadata` during `generate` are not visible here.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L207)
is_quiescent
```python
is_quiescent() -> Optional[bool]
```
Whether in-flight KV transfers are done, so `cleanup` may release GPU memory. The Rust `Worker` polls this on prefill workers between the grace period and `cleanup`:
* `True` — quiescent; exit the drain loop now.
* `False` — busy; poll again next tick.
* `None` — no introspection (default); poll until the drain budget
(`DYN_PREFILL_DRAIN_TIMEOUT_S`) expires. Never frees KV early.
Aggregated/decode workers are never polled. Override only if the engine
can observe transfer completion.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L219)
cleanup
```python
cleanup() -> None
```
Release all engine resources.
`Worker` guarantees:
* `cleanup()` runs after a successful `start()` on shutdown —
the common case.
* `cleanup()` also runs after `start()` raised, on the partial
state the engine may have allocated before failing (inner LLM
handle, sockets, background tasks). Implementations **must**
be null-safe: guard each resource with an `is None` check
so a partially constructed engine can be released without
raising.
* `cleanup()` is **not** called when `start()` was never
invoked (e.g. pre-start shutdown). Engines whose constructors
allocate resources should release them via `__del__` /
context-manager semantics rather than rely on `cleanup()`.
`cleanup()` is never invoked concurrently with `start()` or
another `cleanup()` — `Worker`'s state machine serializes
those transitions. The conformance kit asserts that a second
`cleanup()` call after a successful first is a safe no-op.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L234)
register_prometheus
```python
register_prometheus(metrics: 'EngineMetrics') -> None
```
Bridge a vendor-prefixed Prometheus registry into the runtime's `/metrics` output via `metrics.add_expfmt_callback`. Default no-op. See `dynamo.common.backend.metrics` for helpers. Do not retain `metrics` past return.
Framework-owned lifecycle + per-rank gauges
(`dynamo_component_{cleanup_time_seconds,drain_time_seconds,model_load_time_seconds,total_blocks,gpu_cache_usage_percent,kv_cache_hit_rate}`)
are owned and registered by the framework Rust-side — they do NOT
require the engine to implement this method.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L260)
component_metrics_dp_ranks
```python
component_metrics_dp_ranks() -> list[int]
```
Declare the data-parallel ranks this engine publishes per-rank snapshots for. Empty (default) opts out.
Stable for the engine's lifetime. `Worker` constructs a
`SnapshotPublisher` sized to these ranks and hands it
back via `attach_snapshot_publisher`. The engine then
calls `publisher.publish(rank, snap)` from its stat-logger
thread — event-driven, no polling.
`ComponentSnapshot.kv_cache_hit_rate` is tri-state:
`None` means "no data yet" or "no prefix cache" (gauge
skipped), `0.0` is a legitimate measurement (zero hits).
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L271)
attach_snapshot_publisher
```python
attach_snapshot_publisher(publisher: Any) -> None
```
Framework hands the engine the Rust-owned `SnapshotPublisher` once, after `setup_metrics` constructed it from `component_metrics_dp_ranks`. Stash the reference; call `publisher.publish(rank, snap)` from your stat-logger thereafter.
Only invoked when `component_metrics_dp_ranks` returns
non-empty. Default is no-op so engines that opt out don't need
to override.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L286)
health_check_payload
```python
health_check_payload() -> Optional[dict[str, Any]]
```
Canary payload the runtime sends through `generate` when the endpoint is idle. Return `None` (default) to disable active probing. `Worker` calls this once after `start` and resolves `DYN_HEALTH_CHECK_PAYLOAD` / `--health-check-payload` overrides on top.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L297)
supported_controls
```python
supported_controls() -> set[str]
```
Return the set of engine-control capability keys this engine supports.
Controls are semantic operations on the engine's serving lifecycle.
Engines advertise the keys they implement.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L305)
engine_control
```python
engine_control(control: str, body: dict[str, Any]) -> dict[str, Any]
```
Handle one advertised engine-control request.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L313)
supported_updates
```python
supported_updates() -> set[str]
```
Return the set of engine-update capability keys this engine supports.
Updates are a sibling surface to `supported_controls` for
operations that mutate engine-managed assets rather than the engine's
serving lifecycle. Engines advertise the keys they implement.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L322)
engine_update
```python
engine_update(update: str, body: dict[str, Any]) -> dict[str, Any]
```
Handle one advertised engine-update request.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L331)
on_endpoint_ready
```python
on_endpoint_ready(endpoint) -> None
```
Receive the runtime serving `Endpoint` once, before serving begins.
Default no-op. Engines that publish their own discovery records stash
it for use from `engine_update`. `Worker` calls this exactly
once; a raised exception is fatal to startup.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L338)
#### DiffusionEngine (class)
A `RawEngine` for diffusion-family generation (image/video via VisualGen, DiffGenerator). Names the family only — non-diffusion raw modalities (e.g. TTS audio) subclass `RawEngine` directly. Routing keys off `RawEngine`, so any subclass uses the raw adapter.
```python
from dynamo.common.backend import DiffusionEngine
```
[`components/src/dynamo/common/backend/engine.py#L427`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L427)
#### DisaggregationMode (class)
Disaggregation mode for LLM workers.
```python
from dynamo.common.constants import DisaggregationMode
```
[`components/src/dynamo/common/constants.py#L15`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/constants.py#L15)
#### EmbeddingTransferMode (class)
Embedding transfer mode for LLM workers.
```python
from dynamo.common.constants import EmbeddingTransferMode
```
[`components/src/dynamo/common/constants.py#L24`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/constants.py#L24)
#### EngineConfig (class)
Registration metadata returned by an engine's `start`.
```python
from dynamo.common.backend import EngineConfig
```
```python
EngineConfig(model: str, served_model_name: Optional[str] = None, runtime_data: Optional[dict[str, Any]] = None, llm: Optional[LlmRegistration] = None, model_aliases: list[str] = list()) -> None
```
The neutral fields (`model`, `served_model_name`, `model_aliases`,
`runtime_data`) apply to every modality; token-pipeline metadata lives in
the optional `llm` sub-record, which raw media engines leave `None`.
[`components/src/dynamo/common/backend/engine.py#L135`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L135)
**Public methods**
**init**
```python
__init__(model: str, served_model_name: Optional[str] = None, runtime_data: Optional[dict[str, Any]] = None, llm: Optional[LlmRegistration] = None, model_aliases: list[str] = list()) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py)
#### EngineHealthMonitorConfig (class)
No summary available.
```python
from dynamo.common.engine_monitor import EngineHealthMonitorConfig
```
```python
EngineHealthMonitorConfig(interval: float, check_timeout: float, shutdown_timeout: float) -> None
```
[`components/src/dynamo/common/engine_monitor.py#L40`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/engine_monitor.py#L40)
**Public methods**
from_env
```python
from_env(cls, *, interval: Optional[float] = None, check_timeout: Optional[float] = None, shutdown_timeout: Optional[float] = None) -> 'EngineHealthMonitorConfig'
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/engine_monitor.py#L46)
**init**
```python
__init__(interval: float, check_timeout: float, shutdown_timeout: float) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/engine_monitor.py)
#### ForwardPassMetrics (class)
Per-iteration metrics emitted by InstrumentedScheduler.
```python
from dynamo.common.forward_pass_metrics import ForwardPassMetrics
```
One message is emitted per scheduler iteration (one per forward pass).
An idle heartbeat (all zeros, wall\_time=0) is emitted once when the
engine transitions from active to idle.
[`components/src/dynamo/common/forward_pass_metrics.py#L153`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L153)
#### GenerateChunk (class)
Single chunk yielded by `LLMEngine.generate()`.
```python
from dynamo.common.backend import GenerateChunk
```
Every chunk must include `token_ids` and `index`.
Use `index=0` for single-choice responses. The final chunk must
additionally include `finish_reason`; `completion_usage` is
optional (the OpenAI frontend aggregates it when present, and
matches the Rust `Option` /
`skip_serializing_if = "Option::is_none"` semantics).
Prefill terminals carry `disaggregated_params` for the
PrefillRouter to forward to the decode peer. When the caller
requested logprobs, chunks may also carry `log_probs` and
`top_logprobs` aligned to `token_ids` — see
`dynamo.common.backend.logprobs`.
Encode terminals carry `encoder_result` (an opaque object the
frontend forwards onto the downstream
`PreprocessedRequest.encoder_result`). Construct with
`dynamo.common.backend.multimodal.encoder_terminal_chunk`.
[`components/src/dynamo/common/backend/engine.py#L74`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L74)
#### GenerateRequest (class)
Inbound request dict passed to `LLMEngine.generate()`.
```python
from dynamo.common.backend import GenerateRequest
```
`token_ids` is always present (set by the Rust preprocessor).
The remaining groups are optional — engines should access them
defensively with `.get(key, {})`.
Disaggregated-serving keys (`prefill_result`, `bootstrap_info`)
are set by the frontend's PrefillRouter on decode requests; engines
read them via `dynamo.common.backend.disagg` helpers.
Multimodal keys (`multi_modal_data`, `mm_processor_kwargs`,
`mm_routing_info`) are populated by the frontend preprocessor when
the request carries media. `encoder_result` is set by the
frontend when forwarding a request from an Encode worker
to a downstream Prefill/Aggregated peer; engines read it via
`dynamo.common.backend.multimodal.require_encoder_result`. All
four are object-shaped (`dict`) by contract.
`model` carries the requested model name (set by the Rust
preprocessor). Engines that support dynamic LoRA read it to route a
request to a loaded adapter.
[`components/src/dynamo/common/backend/engine.py#L34`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L34)
#### HttpClient (class)
Backend-neutral HTTP client.
```python
from dynamo.common.http import HttpClient
```
```python
HttpClient(config: Optional[HttpConfigBase] = None) -> None
```
Subclasses own a backend-specific session/client singleton on the
instance. Callers reach the public surface via `fetch_bytes`
and the unified exception classes above; the concrete backend type
is invisible past instantiation.
[`components/src/dynamo/common/http/base.py#L50`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L50)
**Public methods**
**init**
```python
__init__(config: Optional[HttpConfigBase] = None) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L59)
fetch_bytes
```python
fetch_bytes(url: str, timeout: float, *, policy: Optional[UrlValidationPolicy] = None) -> bytes
```
Fetch `url` and return the response body.
Single-shot: no retries. Raises one of the unified exception
classes above; callers never see native httpx/aiohttp classes.
`policy=None`: use the backend's built-in redirect handling.
`policy` set: follow redirects manually and revalidate each
hop against the policy via `url_validator.validate_url`.
This is the SSRF-safe path; raises `UrlValidationError`
if any hop fails or the chain exceeds `_MAX_REDIRECTS`.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L63)
close
```python
close() -> None
```
Close the backend session/client. Idempotent.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L135)
#### HttpConnectionError (class)
Network-layer failure: DNS, refused, reset, half-close.
```python
from dynamo.common.http import HttpConnectionError
```
[`components/src/dynamo/common/http/base.py#L36`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L36)
#### HttpError (class)
Base class for all HTTP fetch failures.
```python
from dynamo.common.http import HttpError
```
[`components/src/dynamo/common/http/base.py#L28`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L28)
#### HttpStatusError (class)
Server responded with a non-2xx status.
```python
from dynamo.common.http import HttpStatusError
```
```python
HttpStatusError(status: int, message: str, url: str) -> None
```
[`components/src/dynamo/common/http/base.py#L40`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L40)
**Public methods**
**init**
```python
__init__(status: int, message: str, url: str) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L43)
#### HttpTimeoutError (class)
Timeout during connect / read / pool-wait.
```python
from dynamo.common.http import HttpTimeoutError
```
[`components/src/dynamo/common/http/base.py#L32`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/base.py#L32)
#### HttpxClient (class)
httpx-backed concrete client.
```python
from dynamo.common.http import HttpxClient
```
```python
HttpxClient(config = None) -> None
```
[`components/src/dynamo/common/http/httpx_client.py#L36`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/httpx_client.py#L36)
**Public methods**
**init**
```python
__init__(config = None) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/httpx_client.py#L39)
close
```python
close() -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/httpx_client.py#L153)
#### ImageLoader (class)
No summary available.
```python
from dynamo.common.multimodal import ImageLoader
```
```python
ImageLoader(cache_size: int = CACHE_SIZE_MAXIMUM, http_timeout: float = 30.0, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None)
```
[`components/src/dynamo/common/multimodal/image_loader.py#L59`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/image_loader.py#L59)
**Public methods**
**init**
```python
__init__(cache_size: int = CACHE_SIZE_MAXIMUM, http_timeout: float = 30.0, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None)
```
Initialize the ImageLoader with caching, HTTP settings, and optional NIXL config for receiving frontend decoding.
**Parameters**
**`cache_size`** `int`
Maximum number of images to store in the in-memory LRU cache.
Defaults to CACHE\_SIZE\_MAXIMUM.
---
**`http_timeout`** `float`
Timeout in seconds for HTTP requests when fetching remote images.
Defaults to 30.0 seconds.
---
**`enable_frontend_decoding`** `bool`
If True, enables NIXL RDMA for transferring
decoded images directly from frontend memory, bypassing standard
network transport. Defaults to False.
---
**`url_policy`** `UrlValidationPolicy | None`
Policy for validating URLs. Defaults to UrlValidationPolicy.from\_env().
---
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/image_loader.py#L62)
load_image
```python
load_image(image_url: str) -> Image.Image
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/image_loader.py#L186)
load_image_batch
```python
load_image_batch(image_mm_items: List[Dict[str, Any]], *, preserve_uuid_slots: bool = False) -> list[Any]
```
Load a batch of images from multimodal data items.
Supports three paths:
1. Url variant: Download and decode image from URL (default)
2. Decoded variant: Read pre-decoded image via NIXL RDMA (requires enable\_frontend\_decoding=True)
3. UuidOnly variant: Preserve an aligned empty slot for backend cache lookup
when preserve\_uuid\_slots=True
**Parameters**
**`image_mm_items`** `List[Dict[str, Any]]`
List of multimodal data items for images
---
**`preserve_uuid_slots`** `bool`
Allow UUID-only items and preserve their positions
as None. This is enabled only by backends that resolve such slots.
---
**Returns**
* `list[Any]` — Loaded images, with None for UUID-only cache slots
**Raises**
* `HttpStatusError` — If any image fails with an HTTP status error (e.g. 415 Unsupported Media Type); the status is preserved so the frontend returns the correct client-error code instead of 500.
* `UrlValidationError` — If a media URL is rejected by the SSRF policy; preserved as a ValueError so the frontend returns a 4xx, not 500.
* `Exception` — If any image fails to load for any other reason
* `ValueError` — If enable\_frontend\_decoding=True but nixl\_connector is None
* `ValueError` — If a UUID-only slot is received without opting in
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/image_loader.py#L263)
#### LLMEngine (class)
Abstract base for token-based inference engines.
```python
from dynamo.common.backend import LLMEngine
```
The token pipeline: the Rust preprocessor tokenizes the prompt and sets
`token_ids` on the request; `generate` yields token chunks that
the Rust postprocessor detokenizes. Registered with
`ModelInput.Tokens` and served through the token request adapter.
[`components/src/dynamo/common/backend/engine.py#L347`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L347)
**Public methods**
generate
```python
generate(request: GenerateRequest, context: Context) -> AsyncGenerator[GenerateChunk, None]
```
Yield streaming response chunks for a single request.
Called concurrently for multiple in-flight requests.
Each chunk: `{"token_ids": [...], "index": 0}`
Final chunk must include: `{"token_ids": [...], "index": 0,
"finish_reason": "...", "completion_usage": {...}}`
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L356)
kv_event_sources
```python
kv_event_sources() -> list[KvEventSource]
```
KV event sources, one per data-parallel rank. Default opts out of KV-aware routing. `Worker` calls once after `start`.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L371)
logits_processor_spec
```python
logits_processor_spec() -> 'LogitsProcessorSpec | None'
```
Return backend-neutral logits-processor activation data.
The default opts out. An engine that overrides this method resolves
and caches the specification during startup, then passes it to
`logits_processors_for_request` from `generate`. The
engine integration remains responsible for realizing each entry into
its inference library's processor type.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L376)
#### LlmRegistration (class)
Token-pipeline registration metadata (KV cache, data-parallel layout, disaggregation bootstrap). Set by `LLMEngine`s; `RawEngine`s leave `EngineConfig.llm` `None`. A `None` field isn't advertised (the router falls back to its defaults).
```python
from dynamo.common.backend import LlmRegistration
```
```python
LlmRegistration(context_length: Optional[int] = None, kv_cache_block_size: Optional[int] = None, total_kv_blocks: Optional[int] = None, max_num_seqs: Optional[int] = None, max_num_batched_tokens: Optional[int] = None, data_parallel_size: Optional[int] = None, data_parallel_start_rank: Optional[int] = None, bootstrap_host: Optional[str] = None, bootstrap_port: Optional[int] = None) -> None
```
[`components/src/dynamo/common/backend/engine.py#L109`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L109)
**Public methods**
**init**
```python
__init__(context_length: Optional[int] = None, kv_cache_block_size: Optional[int] = None, total_kv_blocks: Optional[int] = None, max_num_seqs: Optional[int] = None, max_num_batched_tokens: Optional[int] = None, data_parallel_size: Optional[int] = None, data_parallel_start_rank: Optional[int] = None, bootstrap_host: Optional[str] = None, bootstrap_port: Optional[int] = None) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py)
#### LoRAInfo (class)
Metadata for a loaded LoRA adapter.
```python
from dynamo.common.lora import LoRAInfo
```
```python
LoRAInfo(id: int, path: str) -> None
```
[`components/src/dynamo/common/lora/manager.py#L123`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L123)
**Public methods**
**init**
```python
__init__(id: int, path: str) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py)
#### LoRAManager (class)
Minimal Python wrapper around Rust core with extension points.
```python
from dynamo.common.lora import LoRAManager
```
```python
LoRAManager(cache_path: Optional[Path] = None)
```
The manager uses the Rust-based LoRADownloader for local, S3, and Hugging Face
sources, and allows registering custom Python sources for other protocols.
[`components/src/dynamo/common/lora/manager.py#L36`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L36)
**Public methods**
**init**
```python
__init__(cache_path: Optional[Path] = None)
```
Initialize LoRA manager.
**Parameters**
**`cache_path`** `Optional[Path]`
Optional custom cache path. If not provided, uses DYN\_LORA\_PATH env var.
---
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L44)
register_custom_source
```python
register_custom_source(scheme: str, source: LoRASourceProtocol) -> None
```
Register a custom Python source for a URI scheme.
**Parameters**
**`scheme`** `str`
URI scheme without "://" (e.g., "hf" for hf:// URIs)
---
**`source`** `LoRASourceProtocol`
LoRA source implementing LoRASourceProtocol
---
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L58)
download_lora
```python
download_lora(lora_uri: str) -> Dict[str, Any]
```
Download LoRA if needed, return local path.
The source is inferred from the URI scheme:
* file:// -> Local filesystem (Rust)
* s3:// -> S3 (Rust)
* hf:// -> Hugging Face Hub (Rust)
* Custom schemes -> Registered Python sources
**Parameters**
**`lora_uri`** `str`
Source URI (file://, s3://, hf://, or custom scheme)
---
**Returns**
* `Dict[str, Any]` — Dictionary with: - status: "success" or "error" - local\_path: Local path to LoRA (if successful) - message: Error message (if error)
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L68)
is_cached
```python
is_cached(lora_uri: str) -> bool
```
Check if LoRA is already cached locally.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L114)
#### LoRASourceProtocol (class)
Protocol for custom Python LoRA sources. Users can implement this to add custom sources.
```python
from dynamo.common.lora import LoRASourceProtocol
```
[`components/src/dynamo/common/lora/manager.py#L21`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L21)
**Public methods**
download
```python
download(lora_uri: str, dest_path: Path) -> Path
```
Download LoRA to dest\_path, return actual path
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L27)
exists
```python
exists(lora_uri: str) -> bool
```
Check if LoRA exists in this source
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L31)
#### LocalEmbeddingReceiver (class)
Receiver that reads embeddings from a local file path provided in the serialized request.
```python
from dynamo.common.multimodal import LocalEmbeddingReceiver
```
```python
LocalEmbeddingReceiver()
```
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L203`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L203)
**Public methods**
**init**
```python
__init__()
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L208)
receive_embeddings
```python
receive_embeddings(request: TransferRequest) -> tuple[int, torch.Tensor]
```
Receive precomputed embeddings for a given request ID.
**Parameters**
**`request`** `TransferRequest`
The TransferRequest object containing information to receive embeddings for.
---
**Returns**
* `int` — A tuple containing the tensor ID and the received embeddings as a torch.Tensor.
* `torch.Tensor` — Caller should invoke release\_tensor(tensor\_id) when the tensor is no longer needed to free up resources.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L213)
release_tensor
```python
release_tensor(tensor_id: int) -> None
```
Indicate that the tensor associated with the ID is no longer in use.
**Parameters**
**`tensor_id`** `int`
The ID of the tensor to release.
---
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L235)
#### LocalEmbeddingSender (class)
Sender that saves embeddings to a local file and sends the file path as the serialized request.
```python
from dynamo.common.multimodal import LocalEmbeddingSender
```
```python
LocalEmbeddingSender()
```
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L136`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L136)
**Public methods**
**init**
```python
__init__()
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L141)
save_embeddings_to_file
```python
save_embeddings_to_file(embedding_key: str, embeddings: torch.Tensor) -> str
```
Save the embeddings to a local file and return the file path.
**Parameters**
**`embedding_key`** `str`
A unique key for the embeddings.
---
**`embeddings`** `torch.Tensor`
A torch.Tensor of the embeddings to save.
---
Returns:
The file path where the embeddings are saved.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L145)
send_embeddings
```python
send_embeddings(embeddings: torch.Tensor, stage_embeddings: bool = False) -> tuple[TransferRequest, Awaitable[None]]
```
Send precomputed embeddings for a given request ID.
**Parameters**
**`embeddings`** `torch.Tensor`
A torch.Tensor of the embeddings to send.
---
**`stage_embeddings`** `bool`
A boolean indicating whether the embeddings should be staged for the transfer,
---
Returns:
A tuple containing the TransferRequest object and an awaitable that can be awaited to indicate the send is completed.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L168)
#### MetadataUploader (class)
No summary available.
```python
from dynamo.common.metadata_upload import MetadataUploader
```
```python
MetadataUploader(url: str) -> None
```
[`components/src/dynamo/common/metadata_upload.py#L141`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/metadata_upload.py#L141)
**Public methods**
```python
from_settings(cls, settings: dict[str, Any] | None) -> MetadataUploader | None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/metadata_upload.py#L153)
```python
from_backend_request(cls, request: dict[str, Any]) -> MetadataUploader | None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/metadata_upload.py#L165)
```python
upload_choice(choice_index: int, metadata: dict[str, Any]) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/metadata_upload.py#L169)
```python
__init__(url: str) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/metadata_upload.py)
#### MultimodalEmbeddingCacheManager (class)
LRU cache for encoder embeddings.
```python
from dynamo.common.memory import MultimodalEmbeddingCacheManager
```
```python
MultimodalEmbeddingCacheManager(capacity_bytes: int)
```
Stores tensors keyed by content hash with automatic eviction
when capacity is exceeded.
Thread Safety:
This class is NOT thread-safe. It is designed to run within a single
thread (e.g., an asyncio event loop). All access must be from the same
thread to avoid race conditions. This is intentional to keep the
implementation simple and avoid locking overhead.
[`components/src/dynamo/common/memory/multimodal_embedding_cache_manager.py#L43`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/memory/multimodal_embedding_cache_manager.py#L43)
**Public methods**
**init**
```python
__init__(capacity_bytes: int)
```
Initialize the encoder cache.
**Parameters**
**`capacity_bytes`** `int`
Maximum cache capacity in bytes.
---
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/memory/multimodal_embedding_cache_manager.py#L57)
get
```python
get(key: str) -> Optional[CachedEmbedding]
```
Get a cached embedding from the cache.
If found, the entry is moved to the end (most recently used).
**Parameters**
**`key`** `str`
Cache key (typically content hash).
---
**Returns**
* `Optional[CachedEmbedding]` — The cached embedding, or None if not found.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/memory/multimodal_embedding_cache_manager.py#L95)
keys
```python
keys() -> list[str]
```
Return the current cache keys in LRU order.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/memory/multimodal_embedding_cache_manager.py#L116)
set
```python
set(key: str, entry: CachedEmbedding) -> bool
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/memory/multimodal_embedding_cache_manager.py#L120)
set_with_delta
```python
set_with_delta(key: str, entry: CachedEmbedding) -> CacheMutation
```
Store a cached embedding in the cache.
If the key already exists, the old value is replaced.
If adding the entry would exceed capacity, LRU entries are evicted.
If the tensor itself is larger than capacity, it is not stored.
**Parameters**
**`key`** `str`
Cache key (typically content hash).
---
**`entry`** `CachedEmbedding`
CachedEmbedding to cache.
---
**Returns**
* `CacheMutation` — CacheMutation describing whether the entry was stored plus the
* `CacheMutation` — authoritative add/remove delta caused by this mutation.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/memory/multimodal_embedding_cache_manager.py#L123)
#### NixlReadEmbeddingReceiver (class)
NIXL READ based embedding transfer receiver.
```python
from dynamo.common.multimodal import NixlReadEmbeddingReceiver
```
```python
NixlReadEmbeddingReceiver(embedding_hidden_size: int = 8 * 1024, max_item_mm_token: int = 1024, max_items: int = 1024) -> None
```
Uses nixl\_connect.Connector which now natively provides a shared singleton
Connection (NIXL agent) and reference-counted Remote agent lifecycle.
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L852`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L852)
**Public methods**
**init**
```python
__init__(embedding_hidden_size: int = 8 * 1024, max_item_mm_token: int = 1024, max_items: int = 1024) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L859)
receive_embeddings
```python
receive_embeddings(request: TransferRequest) -> tuple[int, torch.Tensor]
```
Receive precomputed embeddings for a given request ID.
**Parameters**
**`request`** `TransferRequest`
The TransferRequest object containing information to receive embeddings for.
---
**Returns**
* `int` — A tuple containing the tensor ID and the received embeddings as a torch.Tensor.
* `torch.Tensor` — Caller should invoke release\_tensor(tensor\_id) when the tensor is no longer needed to free up resources.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L883)
release_tensor
```python
release_tensor(tensor_id: int) -> None
```
Indicate that the tensor associated with the ID is no longer in use.
**Parameters**
**`tensor_id`** `int`
The ID of the tensor to release.
---
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L942)
#### NixlReadEmbeddingSender (class)
NIXL READ based embedding transfer sender.
```python
from dynamo.common.multimodal import NixlReadEmbeddingSender
```
```python
NixlReadEmbeddingSender()
```
Uses nixl\_connect.Connector which now natively provides a shared singleton
Connection (NIXL agent) and reference-counted Remote agent lifecycle.
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L795`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L795)
**Public methods**
**init**
```python
__init__()
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L802)
send_embeddings
```python
send_embeddings(embeddings: torch.Tensor, stage_embeddings: bool = False) -> tuple[TransferRequest, Awaitable[None]]
```
Send precomputed embeddings.
**Parameters**
**`embeddings`** `torch.Tensor`
A torch.Tensor of the embeddings to send.
---
**`stage_embeddings`** `bool`
A boolean indicating whether the embeddings should be staged for the transfer,
---
Returns:
A tuple containing the TransferRequest object and an awaitable that can be awaited to indicate the send is completed.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L806)
#### NixlWriteEmbeddingReceiver (class)
Counter part of 'NixlWriteEmbeddingSender', see 'NixlWriteEmbeddingSender' for details. The receiver manages a ring buffer for sender to write the embeddings into, and respond to the sender's transfer request with the buffer information for the WRITE transfer.
```python
from dynamo.common.multimodal import NixlWriteEmbeddingReceiver
```
```python
NixlWriteEmbeddingReceiver(buffer_size = 2 * 8 * 1024 * 1024 * 256 * 2)
```
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L635`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L635)
**Public methods**
**init**
```python
__init__(buffer_size = 2 * 8 * 1024 * 1024 * 256 * 2)
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L642)
receive_embeddings
```python
receive_embeddings(request: TransferRequest, receive_timeout = 60) -> tuple[int, torch.Tensor]
```
Receive precomputed embeddings for a given request ID.
**Parameters**
**`request`** `TransferRequest`
The TransferRequest object containing information to receive embeddings for.
---
**`receive_timeout`**
Maximum time to wait for the transfer to complete before raising a TimeoutError.
---
**Returns**
* `int` — A tuple containing the tensor ID and the received embeddings as a torch.Tensor.
* `torch.Tensor` — Caller should invoke release\_tensor(tensor\_id) when the tensor is no longer needed to free up resources.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L666)
release_tensor
```python
release_tensor(tensor_id: int) -> None
```
Indicate that the tensor associated with the ID is no longer in use.
**Parameters**
**`tensor_id`** `int`
The ID of the tensor to release.
---
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L784)
#### NixlWriteEmbeddingSender (class)
NIXL WRITE-based implementation of the embedding sender interface.
```python
from dynamo.common.multimodal import NixlWriteEmbeddingSender
```
```python
NixlWriteEmbeddingSender()
```
Designed for scenarios where the sender transmits dynamically allocated
tensors. Because these tensors allocation is external to the sender,
NIXL memory registration will perform on each send request. The receiver
will manage a pre-allocated buffer, so its NIXL metadata is consistent once
initialized. In such acenarios, let sender initiate the WRITE operations requires
minimal metadata exchange.
Protocol:
1. Record the receiver NIXL metadata, this is done:
* Implicitly through the first transfer request as fallback if the metadata
hasn't been recorded.
* \[REMOVED] Explicitly through add\_agent() API before calling send\_embeddings().
The receiver provides get\_agent\_metadata() API to return its NIXL metadata.
This complicates the implementation and add extra responsiblity on the caller side,
will revisit the necessity if metadata exchange overhead is significant.
2. The sender prepares the embeddings and produces a TransferRequest
containing sender contact and tensor metadata (shape, dtype, size, etc).
3. The receiver responds with (optional) receiver contact, target tensor
metadata (buffer address, device, etc) and done signal through NIXL notification.
4. The sender performs a NIXL WRITE to push the data into the
receiver's buffer.
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L376`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L376)
**Public methods**
**init**
```python
__init__()
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L402)
send_embeddings
```python
send_embeddings(embeddings: torch.Tensor, stage_embeddings: bool = False) -> tuple[TransferRequest, asyncio.Future]
```
Send precomputed embeddings.
**Parameters**
**`embeddings`** `torch.Tensor`
A torch.Tensor of the embeddings to send.
---
**`stage_embeddings`** `bool`
A boolean indicating whether the embeddings should be staged for the transfer,
---
Returns:
A tuple containing the TransferRequest object and an awaitable that can be awaited to indicate the send is completed.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L587)
#### NvCreateVideoRequest (class)
Request for video generation (/v1/videos endpoint).
```python
from dynamo.common.protocols import NvCreateVideoRequest
```
Matches Rust NvCreateVideoRequest in lib/llm/src/protocols/openai/videos.rs.
[`components/src/dynamo/common/protocols/video_protocol.py#L50`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/protocols/video_protocol.py#L50)
#### NvVideosResponse (class)
Response structure for video generation.
```python
from dynamo.common.protocols import NvVideosResponse
```
Matches Rust NvVideosResponse in lib/llm/src/protocols/openai/videos.rs.
[`components/src/dynamo/common/protocols/video_protocol.py#L107`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/protocols/video_protocol.py#L107)
#### OnceLock (class)
No summary available.
```python
from dynamo.common.lora import OnceLock
```
```python
OnceLock() -> None
```
[`components/src/dynamo/common/lora/once.py#L15`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/once.py#L15)
**Public methods**
**init**
```python
__init__() -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/once.py#L16)
get_or_init
```python
get_or_init(initializer: Callable[[], T]) -> T
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/once.py#L20)
get
```python
get() -> T | None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/once.py#L31)
#### QueuedRequestMetrics (class)
Metrics for requests waiting in the queue (not scheduled this iteration).
```python
from dynamo.common.forward_pass_metrics import QueuedRequestMetrics
```
All token counts here are raw totals -- prefix cache effects are unknown
until a request is actually scheduled.
[`components/src/dynamo/common/forward_pass_metrics.py#L122`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L122)
#### RLAdminValidationError (class)
Validation error whose message can be returned directly to RL clients.
```python
from dynamo.common.rl import RLAdminValidationError
```
[`components/src/dynamo/common/rl/admin.py#L21`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L21)
#### RLRouteRegistry (class)
Registry for worker RL admin route descriptors.
```python
from dynamo.common.rl import RLRouteRegistry
```
```python
RLRouteRegistry(runtime: Any, *, logger_: logging.Logger | None = None) -> None
```
[`components/src/dynamo/common/rl/admin.py#L88`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L88)
**Public methods**
**init**
```python
__init__(runtime: Any, *, logger_: logging.Logger | None = None) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L91)
add_route
```python
add_route(name: str, handler: RLRouteHandler) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L101)
add_routes
```python
add_routes(routes: Mapping[str, RLRouteHandler]) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L104)
describe
```python
describe() -> dict[str, Any]
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L108)
dispatch
```python
dispatch(request: Mapping[str, Any] | None = None) -> dict[str, Any]
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L122)
dispatch_stream
```python
dispatch_stream(request: Mapping[str, Any] | None = None) -> AsyncIterator[dict[str, Any]]
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L149)
#### RawEngine (class)
Engines for raw, non-token generation (image, video, audio).
```python
from dynamo.common.backend import RawEngine
```
Named for the *contract*, not a use case: unlike `LLMEngine` there
is no token pipeline — the frontend forwards the OpenAI-shaped request as a
JSON object and `generate` yields the response object(s) directly.
Registered with `ModelInput.Text` and served through the raw request
adapter (no tokenization or KV cache). The `dict` contract is
modality-neutral, so a new media modality is a new engine, not a new
framework path; one engine may serve several modalities. Yield one
(terminal) object, or intermediate progress objects ending with a terminal
one. Subclasses like `DiffusionEngine` add no contract.
[`components/src/dynamo/common/backend/engine.py#L397`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L397)
**Public methods**
generate
```python
generate(request: RawRequest, context: Context) -> AsyncGenerator[RawResponseChunk, None]
```
Yield response object(s) for a single raw-media request.
`request` is the raw OpenAI-shaped request body (see
`RawRequest`); yield the response body object(s) (see
`RawResponseChunk`). For non-streaming modalities yield exactly
one (terminal) object; for streaming modalities yield intermediate
progress objects ending with the terminal one.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/engine.py#L411)
#### ScheduledRequestMetrics (class)
Metrics for requests scheduled in this iteration
```python
from dynamo.common.forward_pass_metrics import ScheduledRequestMetrics
```
[`components/src/dynamo/common/forward_pass_metrics.py#L84`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L84)
#### TokenBudget (class)
Advertise which token-overflow requests the frontend may reject early.
```python
from dynamo.common.token_budget import TokenBudget
```
```python
TokenBudget(combined_limit: int, reject_prompt_overflow: bool, reject_total_overflow: bool) -> None
```
A false flag delegates that overflow dimension to the backend. The backend
remains responsible for any clamping, truncation, or rejection after that.
[`components/src/dynamo/common/token_budget.py#L15`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/token_budget.py#L15)
**Public methods**
**init**
```python
__init__(combined_limit: int, reject_prompt_overflow: bool, reject_total_overflow: bool) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/token_budget.py)
#### TransferRequest (class)
Data class for transfer requests containing necessary information for embedding transfer.
```python
from dynamo.common.multimodal import TransferRequest
```
[`components/src/dynamo/common/multimodal/embedding_transfer.py#L73`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/embedding_transfer.py#L73)
#### UnsupportedFpmVersionError (class)
Raised when a ForwardPassMetrics message has an unrecognised version.
```python
from dynamo.common.forward_pass_metrics import UnsupportedFpmVersionError
```
[`components/src/dynamo/common/forward_pass_metrics.py#L200`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L200)
#### VideoData (class)
Video data in response.
```python
from dynamo.common.protocols import VideoData
```
Matches Rust VideoData in lib/llm/src/protocols/openai/videos.rs.
[`components/src/dynamo/common/protocols/video_protocol.py#L91`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/protocols/video_protocol.py#L91)
#### VideoLoader (class)
No summary available.
```python
from dynamo.common.multimodal import VideoLoader
```
```python
VideoLoader(http_timeout: float = 60.0, num_frames: int = NUM_FRAMES_DEFAULT, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None) -> None
```
[`components/src/dynamo/common/multimodal/video_loader.py#L89`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/video_loader.py#L89)
**Public methods**
**init**
```python
__init__(http_timeout: float = 60.0, num_frames: int = NUM_FRAMES_DEFAULT, enable_frontend_decoding: bool = False, url_policy: UrlValidationPolicy | None = None) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/video_loader.py#L92)
load_video
```python
load_video(video_url: str) -> tuple[np.ndarray, Dict[str, Any]]
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/video_loader.py#L190)
load_video_batch
```python
load_video_batch(video_mm_items: List[Dict[str, Any]]) -> List[tuple[np.ndarray, Dict[str, Any]]]
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/multimodal/video_loader.py#L232)
#### WelfordAccumulator (class)
Welford's online algorithm for count / sum / population-variance.
```python
from dynamo.common.forward_pass_metrics import WelfordAccumulator
```
```python
WelfordAccumulator() -> None
```
Numerically stable single-pass computation -- avoids catastrophic
cancellation that sum-of-squares can suffer with large values.
Usage
```
acc = WelfordAccumulator()
for v in values:
acc.add(v)
print(acc.n, acc.s, acc.variance())
```
[`components/src/dynamo/common/forward_pass_metrics.py#L48`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L48)
**Public methods**
**init**
```python
__init__() -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L64)
add
```python
add(v: int) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L70)
variance
```python
variance() -> float
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L78)
#### Worker (class)
Drive the Rust `Worker` for a single engine instance.
```python
from dynamo.common.backend import Worker
```
```python
Worker(engine: BaseEngine, config: WorkerConfig)
```
Accepts any `BaseEngine` — an `LLMEngine` (token pipeline)
or a `DiffusionEngine` (raw media pipeline). The request adapter is
selected from the engine kind (`raw=isinstance(engine, RawEngine)`);
`WorkerConfig.model_input` is validated against that kind.
[`components/src/dynamo/common/backend/worker.py#L217`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/worker.py#L217)
**Public methods**
**init**
```python
__init__(engine: BaseEngine, config: WorkerConfig)
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/worker.py#L225)
run
```python
run() -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/worker.py#L229)
#### WorkerConfig (class)
No summary available.
```python
from dynamo.common.backend import WorkerConfig
```
```python
WorkerConfig(namespace: str, component: str = 'backend', endpoint: str = 'generate', model_name: str = '', served_model_name: Optional[str] = None, model_input: ModelInput = (lambda: ModelInput.Tokens)(), endpoint_types: str = 'chat,completions', discovery_backend: str = 'etcd', request_plane: str = 'tcp', event_plane: Optional[str] = None, use_kv_events: bool = False, custom_jinja_template: Optional[str] = None, tool_call_parser: Optional[str] = None, reasoning_parser: Optional[str] = None, exclude_tools_when_tool_choice_none: bool = True, enable_local_indexer: bool = True, enable_kv_routing: bool = True, metrics_labels: list[tuple[str, str]] = list(), disaggregation_mode: DisaggregationMode = DisaggregationMode.AGGREGATED, health_check_payload: Optional[dict] = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto', route_to_encoder: bool = False, media_decoder: Optional[MediaDecoder] = None, media_fetcher: Optional[MediaFetcher] = None, kv_state_endpoint: Optional[str] = None, default_thinking_mode: Optional[str] = None) -> None
```
[`components/src/dynamo/common/backend/worker.py#L101`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/worker.py#L101)
**Public methods**
from_runtime_config
```python
from_runtime_config(cls, runtime_cfg, model_name: str, served_model_name: Optional[str] = None, model_input: Optional[ModelInput] = None, **overrides) -> 'WorkerConfig'
```
Build from any object that carries DynamoRuntimeConfig fields.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/worker.py#L148)
**init**
```python
__init__(namespace: str, component: str = 'backend', endpoint: str = 'generate', model_name: str = '', served_model_name: Optional[str] = None, model_input: ModelInput = (lambda: ModelInput.Tokens)(), endpoint_types: str = 'chat,completions', discovery_backend: str = 'etcd', request_plane: str = 'tcp', event_plane: Optional[str] = None, use_kv_events: bool = False, custom_jinja_template: Optional[str] = None, tool_call_parser: Optional[str] = None, reasoning_parser: Optional[str] = None, exclude_tools_when_tool_choice_none: bool = True, enable_local_indexer: bool = True, enable_kv_routing: bool = True, metrics_labels: list[tuple[str, str]] = list(), disaggregation_mode: DisaggregationMode = DisaggregationMode.AGGREGATED, health_check_payload: Optional[dict] = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto', route_to_encoder: bool = False, media_decoder: Optional[MediaDecoder] = None, media_fetcher: Optional[MediaFetcher] = None, kv_state_endpoint: Optional[str] = None, default_thinking_mode: Optional[str] = None) -> None
```
No summary available.
[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/backend/worker.py)
#### add\_config\_dump\_args (function)
Add arguments to the parser to dump the config to a file.
```python
from dynamo.common.config_dump import add_config_dump_args
```
```python
add_config_dump_args(parser: argparse.ArgumentParser) -> None
```
**Parameters**
**`parser`** `argparse.ArgumentParser`
The parser to add the arguments to
---
[`components/src/dynamo/common/config_dump/config_dumper.py#L159`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/config_dump/config_dumper.py#L159)
#### close\_http\_client (function)
Close the active singleton. Idempotent. Safe across resets.
```python
from dynamo.common.http import close_http_client
```
```python
close_http_client() -> None
```
Clears the resolved client so a fresh env-var reading happens on
the next call (primarily useful in tests that vary
`DYN_HTTP_BACKEND`).
[`components/src/dynamo/common/http/__init__.py#L87`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/__init__.py#L87)
#### decode (function)
Decode a ForwardPassMetrics message, returning None for unknown versions.
```python
from dynamo.common.forward_pass_metrics import decode
```
```python
decode(data: bytes) -> ForwardPassMetrics | None
```
Returns None (and logs a warning) if the message cannot be decoded or
carries a version this code does not understand, so callers can simply
skip unsupported messages without crashing.
[`components/src/dynamo/common/forward_pass_metrics.py#L204`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L204)
#### dump\_config (function)
Dump the configuration to a file or stdout.
```python
from dynamo.common.config_dump import dump_config
```
```python
dump_config(dump_config_to: Optional[str], config: Any) -> None
```
If dump\_config\_to is not provided, the config will be logged to stdout at VERBOSE level.
**Parameters**
**`dump_config_to`** `Optional[str]`
Optional path to dump the config to. If None, logs to stdout.
---
**`config`** `Any`
The configuration object to dump (must be JSON-serializable).
---
[`components/src/dynamo/common/config_dump/config_dumper.py#L72`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/config_dump/config_dumper.py#L72)
#### encode (function)
No summary available.
```python
from dynamo.common.forward_pass_metrics import encode
```
```python
encode(metrics: ForwardPassMetrics) -> bytes
```
[`components/src/dynamo/common/forward_pass_metrics.py#L196`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/forward_pass_metrics.py#L196)
#### env\_bool (function)
Parse a boolean environment variable using Dynamo's common true values.
```python
from dynamo.common.rl import env_bool
```
```python
env_bool(name: str, default: bool = False) -> bool
```
[`components/src/dynamo/common/rl/admin.py#L25`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L25)
#### fetch\_bytes (function)
Singleton-backed convenience wrapper over `HttpClient.fetch_bytes`.
```python
from dynamo.common.http import fetch_bytes
```
```python
fetch_bytes(url, timeout, *, policy = None) -> bytes
```
[`components/src/dynamo/common/http/__init__.py#L82`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/__init__.py#L82)
#### fetch\_model (function)
No summary available.
```python
from dynamo.common.model_fetch import fetch_model
```
```python
fetch_model(remote_name: str, ignore_weights: bool = False) -> str
```
[`components/src/dynamo/common/model_fetch.py#L87`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/model_fetch.py#L87)
#### fetch\_model\_in\_subprocess (function)
Fetch a model in a short-lived process before snapshotting.
```python
from dynamo.common.model_fetch import fetch_model_in_subprocess
```
```python
fetch_model_in_subprocess(remote_name: str, ignore_weights: bool = False) -> str
```
[`components/src/dynamo/common/model_fetch.py#L44`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/model_fetch.py#L44)
#### first\_endpoint\_response (function)
Return the first response from an async-generator endpoint handler.
```python
from dynamo.common.rl import first_endpoint_response
```
```python
first_endpoint_response(endpoint_handler: EndpointGenerator, body: dict[str, Any]) -> dict[str, Any]
```
The generator is explicitly closed before returning so handlers that hold
resources across their yield (e.g. load\_lora/unload\_lora holding a per-LoRA
lock) release them promptly rather than waiting for garbage collection.
[`components/src/dynamo/common/rl/admin.py#L33`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L33)
#### get\_config\_dump (function)
Collect comprehensive config information about a backend instance.
```python
from dynamo.common.config_dump import get_config_dump
```
```python
get_config_dump(config: Any, extra_info: Optional[Dict[str, Any]] = None) -> str
```
**Parameters**
**`config`** `Any`
Any JSON-serializable object containing the backend configuration.
---
**`extra_info`** `Optional[Dict[str, Any]]`
Optional dict of additional information to include in the dump.
---
**Returns**
* `str` — JSON string containing comprehensive information.
Returns error information if collection fails, ensuring some diagnostic data is always available.
[`components/src/dynamo/common/config_dump/config_dumper.py#L108`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/config_dump/config_dumper.py#L108)
#### get\_default\_client (function)
Return the process-wide singleton client, instantiating on first call.
```python
from dynamo.common.http import get_default_client
```
```python
get_default_client() -> HttpClient
```
[`components/src/dynamo/common/http/__init__.py#L73`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/http/__init__.py#L73)
#### get\_environment\_vars (function)
Get relevant environment variables based on prefixes.
```python
from dynamo.common.config_dump import get_environment_vars
```
```python
get_environment_vars(prefixes: Optional[List[str]] = None, include_sensitive: bool = False, additional_vars: Optional[Set[str]] = None) -> Dict[str, str]
```
**Parameters**
**`prefixes`** `Optional[List[str]]`
List of environment variable prefixes to capture.
If None, uses DEFAULT\_ENV\_PREFIXES.
---
**`include_sensitive`** `bool`
If False, redacts values of potentially sensitive variables.
Default is False for security.
---
**`additional_vars`** `Optional[Set[str]]`
Set of specific variable names to include regardless of prefix.
---
**Returns**
* `Dict[str, str]` — Dictionary of environment variable names to values.
* `Dict[str, str]` — Sensitive values are replaced with "\" unless include\_sensitive is True.
**Examples**
```python
>>> get_environment_vars() # Uses default prefixes
>>> get_environment_vars(prefixes=["MY_APP_"]) # Custom prefixes only
>>> get_environment_vars(additional_vars={"PATH", "HOME"}) # Include specific vars
```
[`components/src/dynamo/common/config_dump/environment.py#L51`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/config_dump/environment.py#L51)
#### get\_fs (function)
Initialize fsspec filesystem for the given URL.
```python
from dynamo.common.storage import get_fs
```
```python
get_fs(fs_url: str) -> DirFileSystem
```
**Parameters**
**`fs_url`** `str`
The URL of the filesystem to initialize. e.g. s3://bucket, gs\://bucket, file:///local/path
---
**Returns**
* `DirFileSystem` — The initialized DirFileSystem wrapper for the filesystem.
* `DirFileSystem` — fs.fs.protocol to get the protocol of the filesystem
* `DirFileSystem` — fs.path to get the bucket or root path
* `DirFileSystem` — path to the object in the filesystem - f"\{fs.fs.protocol}://\{fs.path}/\{path}"
[`components/src/dynamo/common/storage.py#L43`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/storage.py#L43)
#### get\_gpu\_info (function)
Get GPU information if available.
```python
from dynamo.common.config_dump import get_gpu_info
```
```python
get_gpu_info() -> Optional[Dict[str, Any]]
```
**Returns**
* `Optional[Dict[str, Any]]` — Dictionary containing GPU details if available, None otherwise.
* `Optional[Dict[str, Any]]` — Attempts to use nvidia-smi via subprocess with XML output format.
This is a best-effort function and returns None if GPU info cannot be obtained.
[`components/src/dynamo/common/config_dump/system_info.py#L98`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/config_dump/system_info.py#L98)
#### get\_lora\_manager (function)
Return the LoRAManager singleton, or None when DYN\_LORA\_ENABLED is unset.
```python
from dynamo.common.lora import get_lora_manager
```
```python
get_lora_manager() -> Optional[LoRAManager]
```
Initializes on first call. Initialization errors propagate to the caller —
OnceLock does not cache failures, so subsequent calls will retry.
[`components/src/dynamo/common/lora/manager.py#L145`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/lora/manager.py#L145)
#### get\_media\_url (function)
Build a public URL for a file stored in the media filesystem.
```python
from dynamo.common.storage import get_media_url
```
```python
get_media_url(fs: DirFileSystem, storage_path: str, base_url: Optional[str] = None) -> str
```
**Parameters**
**`fs`** `DirFileSystem`
The DirFileSystem returned by `get_fs()`.
---
**`storage_path`** `str`
Relative path within the filesystem (e.g. "videos/req-id.mp4").
---
**`base_url`** `Optional[str]`
Optional CDN / proxy base URL. When set, the returned URL is
`{base_url}/{storage_path}`. When *None*, the URL is constructed
from the filesystem's protocol and root path.
---
**Returns**
* `str` — Public URL string for the uploaded file.
**Examples**
```python
>>> fs = get_fs("memory://media")
>>> get_media_url(fs, "videos/request.mp4", "https://cdn.example.com/media")
'https://cdn.example.com/media/videos/request.mp4'
>>> get_media_url(fs, "videos/request.mp4")
'memory:///media/videos/request.mp4'
```
[`components/src/dynamo/common/storage.py#L73`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/storage.py#L73)
#### get\_native\_offloading\_capacity\_tokens (function)
Read native offloading capacity from a worker's runtime metadata.
```python
from dynamo.common.native_offloading import get_native_offloading_capacity_tokens
```
```python
get_native_offloading_capacity_tokens(runtime_data: object) -> int | None
```
[`components/src/dynamo/common/native_offloading.py#L28`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/native_offloading.py#L28)
#### get\_runtime\_info (function)
Get Python runtime information.
```python
from dynamo.common.config_dump import get_runtime_info
```
```python
get_runtime_info() -> Dict[str, Any]
```
**Returns**
* `Dict[str, Any]` — Dictionary containing Python version, executable path, and command-line arguments.
Gracefully handles errors by returning partial information.
[`components/src/dynamo/common/config_dump/system_info.py#L60`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/config_dump/system_info.py#L60)
#### get\_system\_info (function)
Get comprehensive system information.
```python
from dynamo.common.config_dump import get_system_info
```
```python
get_system_info() -> Dict[str, Any]
```
**Returns**
* `Dict[str, Any]` — Dictionary containing platform, architecture, processor, hostname,
* `Dict[str, Any]` — and operating system details.
Gracefully handles errors by returning partial information.
[`components/src/dynamo/common/config_dump/system_info.py#L13`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/config_dump/system_info.py#L13)
#### main (function)
No summary available.
```python
from dynamo.common.recv_forward_pass_metrics import main
```
```python
main() -> None
```
[`components/src/dynamo/common/recv_forward_pass_metrics.py#L89`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/recv_forward_pass_metrics.py#L89)
#### native\_offloading\_capacity (function)
Build runtime metadata from an authoritative backend token capacity.
```python
from dynamo.common.native_offloading import native_offloading_capacity
```
```python
native_offloading_capacity(total_tokens: object) -> dict[str, int] | None
```
[`components/src/dynamo/common/native_offloading.py#L14`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/native_offloading.py#L14)
#### publish\_token\_budget (function)
Publish an engine's token-overflow contract to the Dynamo frontend.
```python
from dynamo.common.token_budget import publish_token_budget
```
```python
publish_token_budget(runtime_config: Any, token_budget: TokenBudget) -> None
```
[`components/src/dynamo/common/token_budget.py#L32`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/token_budget.py#L32)
#### register\_encoder (function)
Decorator to register custom encoders for specific types.
```python
from dynamo.common.config_dump import register_encoder
```
```python
register_encoder(type_class: type) -> Any
```
Usage:
@register\_encoder(MyClass)
def encode\_my\_class(obj: MyClass):
return \{"field": obj.field}
[`components/src/dynamo/common/config_dump/config_dumper.py#L209`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/config_dump/config_dumper.py#L209)
#### register\_model\_taint\_route (function)
Register POST /engine/update/model\_taints on the system status server.
```python
from dynamo.common.model_taints import register_model_taint_route
```
```python
register_model_taint_route(runtime: DistributedRuntime, endpoint: Endpoint) -> None
```
[`components/src/dynamo/common/model_taints.py#L17`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/model_taints.py#L17)
#### register\_rl\_routes (function)
Register worker system routes and optionally expose route descriptors.
```python
from dynamo.common.rl import register_rl_routes
```
```python
register_rl_routes(runtime: Any, registry: RLRouteRegistry, routes: Mapping[str, RLRouteHandler], *, enable_dispatch: bool) -> None
```
[`components/src/dynamo/common/rl/admin.py#L155`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L155)
#### require\_lora\_load\_request (function)
Validate the shared URI-based LoRA load request shape.
```python
from dynamo.common.rl import require_lora_load_request
```
```python
require_lora_load_request(request: Mapping[str, Any] | None) -> tuple[str, str]
```
[`components/src/dynamo/common/rl/admin.py#L54`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L54)
#### require\_lora\_unload\_request (function)
Validate the shared LoRA unload request shape.
```python
from dynamo.common.rl import require_lora_unload_request
```
```python
require_lora_unload_request(request: Mapping[str, Any] | None) -> str
```
[`components/src/dynamo/common/rl/admin.py#L76`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/rl/admin.py#L76)
#### run (function)
No summary available.
```python
from dynamo.common.recv_forward_pass_metrics import run
```
```python
run(args: argparse.Namespace) -> None
```
[`components/src/dynamo/common/recv_forward_pass_metrics.py#L136`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/recv_forward_pass_metrics.py#L136)
#### upload\_to\_fs (function)
Upload bytes to the media filesystem and return the public URL.
```python
from dynamo.common.storage import upload_to_fs
```
```python
upload_to_fs(fs: DirFileSystem, storage_path: str, data: bytes, base_url: Optional[str] = None) -> str
```
This is the canonical helper for all backends (vLLM, SGLang, TRT-LLM)
to store generated images/videos and produce a response URL.
**Parameters**
**`fs`** `DirFileSystem`
The DirFileSystem returned by `get_fs()`.
---
**`storage_path`** `str`
Relative path within the filesystem (e.g. "images/req-id/file.png").
---
**`data`** `bytes`
Raw bytes to upload.
---
**`base_url`** `Optional[str]`
Optional CDN / proxy base URL for URL rewriting.
---
**Returns**
* `str` — Public URL string for the uploaded file.
[`components/src/dynamo/common/storage.py#L104`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/common/storage.py#L104)