> 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 5 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)

#### MockEngineArgs (class)

No summary available.

```python
from dynamo.mocker import MockEngineArgs
```

```python
MockEngineArgs(engine_type: str = 'vllm', num_gpu_blocks: Optional[int] = None, block_size: int = 0, max_num_seqs: Optional[int] = 256, max_num_batched_tokens: Optional[int] = 8192, enable_prefix_caching: bool = True, enable_chunked_prefill: bool = True, speedup_ratio: float = 1.0, decode_speedup_ratio: float = 1.0, dp_size: int = 1, startup_time: Optional[float] = None, worker_type: str = 'aggregated', planner_profile_data: Optional[str | os.PathLike[str]] = None, aic_backend: Optional[str] = None, aic_system: Optional[str] = None, aic_backend_version: Optional[str] = None, aic_tp_size: Optional[int] = None, aic_model_path: Optional[str] = None, aic_moe_tp_size: Optional[int] = None, aic_moe_ep_size: Optional[int] = None, aic_attention_dp_size: Optional[int] = None, aic_nextn: Optional[int] = None, aic_nextn_accept_rates: Optional[str] = None, aic_mtp_seed: int = 42, aic_gemm_dtype: Optional[str] = None, aic_moe_dtype: Optional[str] = None, aic_fmha_dtype: Optional[str] = None, aic_kv_cache_dtype: Optional[str] = None, aic_comm_dtype: Optional[str] = None, gpu_memory_utilization: Optional[float] = None, mem_fraction_static: Optional[float] = None, free_gpu_memory_fraction: Optional[float] = None, enable_local_indexer: bool = False, bootstrap_port: Optional[int] = None, handoff_session_timeout_ms: int = 300000, kv_bytes_per_token: Optional[int] = None, kv_transfer_bandwidth: Optional[float] = None, kv_transfer_timing_mode: str = 'full_prompt', reasoning: Optional[ReasoningConfig] = None, response_replay_trace_path: Optional[str | os.PathLike[str]] = None, zmq_kv_events_port: Optional[int] = None, zmq_replay_port: Optional[int] = None, preemption_mode: str = 'lifo', router_queue_policy: Optional[str] = None, sglang: Optional[SglangArgs] = None, trtllm: Optional[TrtllmArgs] = None, num_g2_blocks: Optional[int] = None, num_g3_blocks: Optional[int] = None, offload_batch_size: Optional[int] = None, bandwidth_g1_to_g2_gbps: Optional[float] = None, bandwidth_g2_to_g1_gbps: Optional[float] = None, bandwidth_g2_to_g3_gbps: Optional[float] = None, bandwidth_g3_to_g2_gbps: Optional[float] = None, enable_g4_storage: bool = False, bandwidth_g2_to_g4_gbps: Optional[float] = None, bandwidth_g4_to_g2_gbps: Optional[float] = None, max_model_len: Optional[int] = None, g1_backend: Optional[str] = None) -> None
```

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

**Public methods**

<h4 id="api-dynamo-core-mockengineargs-init">
  **init**
</h4>

```python
__init__(engine_type: str = 'vllm', num_gpu_blocks: Optional[int] = None, block_size: int = 0, max_num_seqs: Optional[int] = 256, max_num_batched_tokens: Optional[int] = 8192, enable_prefix_caching: bool = True, enable_chunked_prefill: bool = True, speedup_ratio: float = 1.0, decode_speedup_ratio: float = 1.0, dp_size: int = 1, startup_time: Optional[float] = None, worker_type: str = 'aggregated', planner_profile_data: Optional[str | os.PathLike[str]] = None, aic_backend: Optional[str] = None, aic_system: Optional[str] = None, aic_backend_version: Optional[str] = None, aic_tp_size: Optional[int] = None, aic_model_path: Optional[str] = None, aic_moe_tp_size: Optional[int] = None, aic_moe_ep_size: Optional[int] = None, aic_attention_dp_size: Optional[int] = None, aic_nextn: Optional[int] = None, aic_nextn_accept_rates: Optional[str] = None, aic_mtp_seed: int = 42, aic_gemm_dtype: Optional[str] = None, aic_moe_dtype: Optional[str] = None, aic_fmha_dtype: Optional[str] = None, aic_kv_cache_dtype: Optional[str] = None, aic_comm_dtype: Optional[str] = None, gpu_memory_utilization: Optional[float] = None, mem_fraction_static: Optional[float] = None, free_gpu_memory_fraction: Optional[float] = None, enable_local_indexer: bool = False, bootstrap_port: Optional[int] = None, handoff_session_timeout_ms: int = 300000, kv_bytes_per_token: Optional[int] = None, kv_transfer_bandwidth: Optional[float] = None, kv_transfer_timing_mode: str = 'full_prompt', reasoning: Optional[ReasoningConfig] = None, response_replay_trace_path: Optional[str | os.PathLike[str]] = None, zmq_kv_events_port: Optional[int] = None, zmq_replay_port: Optional[int] = None, preemption_mode: str = 'lifo', router_queue_policy: Optional[str] = None, sglang: Optional[SglangArgs] = None, trtllm: Optional[TrtllmArgs] = None, num_g2_blocks: Optional[int] = None, num_g3_blocks: Optional[int] = None, offload_batch_size: Optional[int] = None, bandwidth_g1_to_g2_gbps: Optional[float] = None, bandwidth_g2_to_g1_gbps: Optional[float] = None, bandwidth_g2_to_g3_gbps: Optional[float] = None, bandwidth_g3_to_g2_gbps: Optional[float] = None, enable_g4_storage: bool = False, bandwidth_g2_to_g4_gbps: Optional[float] = None, bandwidth_g4_to_g2_gbps: Optional[float] = None, max_model_len: Optional[int] = None, g1_backend: Optional[str] = None) -> None
```

No summary available.

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

<h4 id="api-dynamo-core-mockengineargs-from-json">
  from_json
</h4>

```python
from_json(config_json: str) -> MockEngineArgs
```

No summary available.

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

<h4 id="api-dynamo-core-mockengineargs-copy">
  copy
</h4>

```python
copy() -> MockEngineArgs
```

No summary available.

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

<h4 id="api-dynamo-core-mockengineargs-is-prefill">
  is_prefill
</h4>

```python
is_prefill() -> bool
```

No summary available.

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

<h4 id="api-dynamo-core-mockengineargs-is-decode">
  is_decode
</h4>

```python
is_decode() -> bool
```

No summary available.

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

<h4 id="api-dynamo-core-mockengineargs-with-overrides">
  with_overrides
</h4>

```python
with_overrides(bootstrap_port: Optional[int] = None, zmq_kv_events_port: Optional[int] = None, zmq_replay_port: Optional[int] = None, kv_bytes_per_token: Optional[int] = None, num_gpu_blocks: Optional[int] = None, aic_backend: Optional[str] = None, aic_system: Optional[str] = None, aic_backend_version: Optional[str] = None, aic_tp_size: Optional[int] = None, aic_model_path: Optional[str] = None, aic_moe_tp_size: Optional[int] = None, aic_moe_ep_size: Optional[int] = None, aic_attention_dp_size: Optional[int] = None, aic_nextn: Optional[int] = None, aic_nextn_accept_rates: Optional[str] = None, aic_mtp_seed: Optional[int] = None, aic_gemm_dtype: Optional[str] = None, aic_moe_dtype: Optional[str] = None, aic_fmha_dtype: Optional[str] = None, aic_kv_cache_dtype: Optional[str] = None, aic_comm_dtype: Optional[str] = None, gpu_memory_utilization: Optional[float] = None, mem_fraction_static: Optional[float] = None, free_gpu_memory_fraction: Optional[float] = None, enable_prefix_caching: Optional[bool] = None, worker_type: Optional[str] = None) -> MockEngineArgs
```

No summary available.

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

#### 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#L51`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L51)

**Public methods**

<h4 id="api-dynamo-mocker-args-profiledataresult-init">
  **init**
</h4>

```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#L54)

#### 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#L1875`](https://github.com/ai-dynamo/dynamo/blob/main/lib/bindings/python/src/dynamo/_core.pyi#L1875)

**Public methods**

<h4 id="api-dynamo-core-reasoningconfig-init">
  **init**
</h4>

```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#L1876)

#### SglangArgs (class)

No summary available.

```python
from dynamo.mocker import SglangArgs
```

```python
SglangArgs(schedule_policy: Optional[str] = None, page_size: Optional[int] = None, max_prefill_tokens: Optional[int] = None, chunked_prefill_size: Optional[int] = None, clip_max_new_tokens: Optional[int] = None, schedule_conservativeness: Optional[float] = None) -> None
```

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

**Public methods**

<h4 id="api-dynamo-core-sglangargs-init">
  **init**
</h4>

```python
__init__(schedule_policy: Optional[str] = None, page_size: Optional[int] = None, max_prefill_tokens: Optional[int] = None, chunked_prefill_size: Optional[int] = None, clip_max_new_tokens: Optional[int] = None, schedule_conservativeness: Optional[float] = None) -> None
```

No summary available.

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

#### TrtllmArgs (class)

No summary available.

```python
from dynamo.mocker import TrtllmArgs
```

```python
TrtllmArgs(capacity_scheduler_policy: Optional[str] = None) -> None
```

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

**Public methods**

<h4 id="api-dynamo-core-trtllmargs-init">
  **init**
</h4>

```python
__init__(capacity_scheduler_policy: Optional[str] = None) -> None
```

No summary available.

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

#### 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: MockEngineArgs, *, 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, aic_mtp_seed: int | None = None) -> MockEngineArgs
```

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

#### 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) -> MockEngineArgs
```

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

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

No summary available.

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

```python
build_runtime_config(engine_args: MockEngineArgs) -> tuple[int, ModelRuntimeConfig]
```

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

#### 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#L104`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L104)

#### 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#L34`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L34)

#### 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#L126`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L126)

#### 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) -> MockEngineArgs
```

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

#### main (function)

No summary available.

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

```python
main()
```

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

#### non\_negative\_float (function)

No summary available.

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

```python
non_negative_float(value: str) -> float
```

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

#### non\_negative\_int (function)

No summary available.

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

```python
non_negative_int(value: str) -> int
```

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

#### 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#L175`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L175)

#### 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#L168`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L168)

#### positive\_int (function)

No summary available.

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

```python
positive_int(value: str) -> int
```

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

#### prefetch\_model (function)

Pre-fetch model from HuggingFace to avoid rate limiting with many workers.

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

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

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

#### 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#L69`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L69)

#### 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)
```

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

#### 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#L122`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/args.py#L122)

#### 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#L63`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/mocker/main.py#L63)