> This page is for version dev.
> For other versions, use one of these documentation indexes:
> - Latest (v1.5.1) (default): https://docs.nvidia.com/dynamo/latest/llms.txt
> - dev: https://docs.nvidia.com/dynamo/dev/llms.txt
> - v1.5.1: https://docs.nvidia.com/dynamo/v1.5.1/llms.txt
> - v1.5.0: https://docs.nvidia.com/dynamo/v1.5.0/llms.txt
> - v1.4.2: https://docs.nvidia.com/dynamo/v1.4.2/llms.txt
> - v1.4.1: https://docs.nvidia.com/dynamo/v1.4.1/llms.txt
> - v1.4.0: https://docs.nvidia.com/dynamo/v1.4.0/llms.txt
> - v1.3.0: https://docs.nvidia.com/dynamo/v1.3.0/llms.txt
> - v1.2.1: https://docs.nvidia.com/dynamo/v1.2.1/llms.txt
> - v1.2.0: https://docs.nvidia.com/dynamo/v1.2.0/llms.txt
> - v1.1.1: https://docs.nvidia.com/dynamo/v1.1.1/llms.txt
> - v1.1.0: https://docs.nvidia.com/dynamo/v1.1.0/llms.txt
> - v1.0.2: https://docs.nvidia.com/dynamo/v1.0.2/llms.txt
> - v1.0.1: https://docs.nvidia.com/dynamo/v1.0.1/llms.txt
> - v1.0.0: https://docs.nvidia.com/dynamo/v1.0.0/llms.txt
> - v0.9.1: https://docs.nvidia.com/dynamo/v-0-9-1/llms.txt
> - v0.9.0: https://docs.nvidia.com/dynamo/v-0-9-0/llms.txt
> - v0.8.1: https://docs.nvidia.com/dynamo/v-0-8-1/llms.txt
> - v0.8.0: https://docs.nvidia.com/dynamo/v-0-8-0/llms.txt
> - v0.7.1: https://docs.nvidia.com/dynamo/v-0-7-1/llms.txt
> - v0.7.0: https://docs.nvidia.com/dynamo/v-0-7-0/llms.txt

> 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.mocker

`dynamo.mocker` publishes 2 classes and 18 functions. Source: [`components/src/dynamo/mocker/__init__.py`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/__init__.py)

#### ProfileDataResult (class)

Result of processing --planner-profile-data argument. Cleans up tmpdir on deletion.

```python
from dynamo.mocker.args import ProfileDataResult
```

```python
ProfileDataResult(npz_path: Path | None, tmpdir: tempfile.TemporaryDirectory | None)
```

[`components/src/dynamo/mocker/args.py#L27`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L27)

**Public methods**

#### **init**

```python
__init__(npz_path: Path | None, tmpdir: tempfile.TemporaryDirectory | None)
```

No summary available.

[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L30)

#### ReasoningConfig (class)

No summary available.

```python
from dynamo.mocker import ReasoningConfig
```

```python
ReasoningConfig(start_thinking_token_id: int, end_thinking_token_id: int, thinking_ratio: float) -> None
```

[`lib/bindings/python/src/dynamo/_core.pyi#L1957`](https://github.com/ai-dynamo/dynamo/blob/main/lib/bindings/python/src/dynamo/_core.pyi#L1957)

**Public methods**

#### **init**

```python
__init__(start_thinking_token_id: int, end_thinking_token_id: int, thinking_ratio: float) -> None
```

No summary available.

[source](https://github.com/ai-dynamo/dynamo/blob/main/lib/bindings/python/src/dynamo/_core.pyi#L1958)

#### apply\_worker\_engine\_args\_overrides (function)

No summary available.

```python
from dynamo.mocker.config import apply_worker_engine_args_overrides
```

```python
apply_worker_engine_args_overrides(engine_args: Mapping[str, Any], *, kv_bytes_per_token: int | None = None, bootstrap_port: int | None = None, zmq_kv_events_port: int | None = None, zmq_replay_port: int | None = None, ais_mtp_seed: int | None = None) -> dict
```

[`components/src/dynamo/mocker/config.py#L191`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/config.py#L191)

#### build\_mocker\_engine\_args (function)

No summary available.

```python
from dynamo.mocker.config import build_mocker_engine_args
```

```python
build_mocker_engine_args(args: argparse.Namespace) -> dict
```

[`components/src/dynamo/mocker/config.py#L59`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/config.py#L59)

#### build\_runtime\_config (function)

No summary available.

```python
from dynamo.mocker.config import build_runtime_config
```

```python
build_runtime_config(engine_args: Mapping[str, Any]) -> tuple[int, ModelRuntimeConfig]
```

[`components/src/dynamo/mocker/config.py#L217`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/config.py#L217)

#### compute\_stagger\_delay (function)

Compute the stagger delay based on worker count to give the frontend time to process registrations. Returns the delay in seconds between worker launches.

```python
from dynamo.mocker.main import compute_stagger_delay
```

```python
compute_stagger_delay(num_workers: int, stagger_delay: float) -> float
```

[`components/src/dynamo/mocker/main.py#L124`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L124)

#### engine\_limit (function)

No summary available.

```python
from dynamo.mocker.config import engine_limit
```

```python
engine_limit(config: Mapping[str, Any], name: str) -> int | None
```

[`components/src/dynamo/mocker/config.py#L32`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/config.py#L32)

#### graceful\_shutdown (function)

Shutdown dynamo distributed runtime instances. The endpoints will be immediately invalidated so no new requests will be accepted.

```python
from dynamo.mocker.main import graceful_shutdown
```

```python
graceful_shutdown(runtimes: list)
```

[`components/src/dynamo/mocker/main.py#L35`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L35)

#### launch\_workers (function)

Launch mocker worker(s) with isolated DistributedRuntime instances.

```python
from dynamo.mocker.main import launch_workers
```

```python
launch_workers(args: argparse.Namespace, base_engine_args)
```

Each worker gets its own DistributedRuntime, which means:

* Separate etcd/NATS connections
* Separate Component instances (no shared overhead)
* Independent service registration and stats scraping
* But still sharing the same tokio runtime (efficient)

[`components/src/dynamo/mocker/main.py#L146`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L146)

#### load\_mocker\_engine\_args (function)

No summary available.

```python
from dynamo.mocker.config import load_mocker_engine_args
```

```python
load_mocker_engine_args(args: argparse.Namespace) -> dict
```

[`components/src/dynamo/mocker/config.py#L185`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/config.py#L185)

#### main (function)

No summary available.

```python
from dynamo.mocker.main import main
```

```python
main()
```

[`components/src/dynamo/mocker/main.py#L312`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L312)

#### normalize\_mocker\_config (function)

No summary available.

```python
from dynamo.mocker.config import normalize_mocker_config
```

```python
normalize_mocker_config(config: Mapping[str, Any] | str | None = None) -> dict
```

[`components/src/dynamo/mocker/config.py#L19`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/config.py#L19)

#### parse\_args (function)

Parse command-line arguments for the Dynamo mocker engine.

```python
from dynamo.mocker.args import parse_args
```

```python
parse_args(argv: list[str] | None = None) -> argparse.Namespace
```

**Returns**

* `argparse.Namespace` — argparse.Namespace: Parsed command-line arguments.

[`components/src/dynamo/mocker/args.py#L151`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L151)

#### parse\_bootstrap\_ports (function)

Parse comma-separated bootstrap ports string into list of integers.

```python
from dynamo.mocker.args import parse_bootstrap_ports
```

```python
parse_bootstrap_ports(ports_str: str | None) -> list[int]
```

[`components/src/dynamo/mocker/args.py#L144`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L144)

#### performance\_config (function)

No summary available.

```python
from dynamo.mocker.config import performance_config
```

```python
performance_config(config: Mapping[str, Any]) -> dict | None
```

[`components/src/dynamo/mocker/config.py#L25`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/config.py#L25)

#### prefetch\_model (function)

Resolve `model_path` to a local directory, fetching config/tokenizer if needed.

```python
from dynamo.mocker.main import prefetch_model
```

```python
prefetch_model(model_path: str) -> str
```

`fetch_model` returns the cached snapshot directory without contacting the
hub when config.json and the tokenizer files are already present, and downloads
only those files otherwise. Resolving once here means neither the workers nor
the KV-bytes estimate below resolve a hub ID over the network. Returns the
original path on failure so callers degrade to their previous behavior.

[`components/src/dynamo/mocker/main.py#L46`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L46)

#### resolve\_planner\_profile\_data (function)

Resolve --planner-profile-data to an NPZ file path.

```python
from dynamo.mocker.args import resolve_planner_profile_data
```

```python
resolve_planner_profile_data(planner_profile_data: Path | None) -> ProfileDataResult
```

Handles backward compatibility by accepting either:

1. A mocker-format NPZ file (returned as-is)
2. A profiler-style results directory (converted to mocker-format NPZ)

**Parameters**

**`planner_profile_data`** `Path | None`

Path from --planner-profile-data argument.

---

**Returns**

* `ProfileDataResult` — ProfileDataResult with npz\_path and optional tmpdir for cleanup.

**Raises**

* `FileNotFoundError` — If path doesn't contain valid profile data in any supported format.

[`components/src/dynamo/mocker/args.py#L45`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L45)

#### run\_mocker\_trace\_replay (function)

No summary available.

```python
from dynamo.mocker import run_mocker_trace_replay
```

```python
run_mocker_trace_replay(trace_files, extra_engine_args = None, router_config = None, num_workers = 1, replay_concurrency = None, router_mode = 'round_robin', arrival_speedup_ratio = 1.0, trace_block_size = None, trace_format = 'mooncake', trace_shared_prefix_ratio = 0.0, trace_num_prefix_groups = 0, model_name = None, sla_ttft_ms = None, sla_itl_ms = None, sla_e2e_ms = None, capture_per_request = False, agentic_lanes = None, execution_model = None)
```

[`components/src/dynamo/mocker/__init__.py#L36`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/__init__.py#L36)

#### validate\_worker\_type\_args (function)

Resolve disaggregation mode from --disaggregation-mode or legacy boolean flags. Raises ValueError if validation fails.

```python
from dynamo.mocker.args import validate_worker_type_args
```

```python
validate_worker_type_args(args: argparse.Namespace) -> None
```

[`components/src/dynamo/mocker/args.py#L98`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L98)

#### worker (function)

Main worker function that launches mocker instances.

```python
from dynamo.mocker.main import worker
```

```python
worker()
```

Each mocker gets its own DistributedRuntime instance for true isolation,
while still sharing the same event loop and tokio runtime.

[`components/src/dynamo/mocker/main.py#L78`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L78)