Python API

dynamo.mocker

Mock engine used to run Dynamo workflows without a real GPU backend.
View as Markdown

dynamo.mocker publishes 5 classes and 16 functions. Source: components/src/dynamo/mocker/__init__.py

No summary available.

from dynamo.mocker import MockEngineArgs
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, max_model_len: Optional[int] = None) -> None

lib/bindings/python/src/dynamo/_core.pyi#L1956

Public methods

init

__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, max_model_len: Optional[int] = None) -> None

No summary available.

source

from_json

from_json(config_json: str) -> MockEngineArgs

No summary available.

source

copy

copy() -> MockEngineArgs

No summary available.

source

is_prefill

is_prefill() -> bool

No summary available.

source

is_decode

is_decode() -> bool

No summary available.

source

with_overrides

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

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

from dynamo.mocker.args import ProfileDataResult
ProfileDataResult(npz_path: Path | None, tmpdir: tempfile.TemporaryDirectory | None)

components/src/dynamo/mocker/args.py#L35

Public methods

init

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

No summary available.

source

No summary available.

from dynamo.mocker import ReasoningConfig
ReasoningConfig(start_thinking_token_id: int, end_thinking_token_id: int, thinking_ratio: float) -> None

lib/bindings/python/src/dynamo/_core.pyi#L1928

Public methods

init

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

No summary available.

source

No summary available.

from dynamo.mocker import SglangArgs
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#L1937

Public methods

init

__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

No summary available.

from dynamo.mocker import TrtllmArgs
TrtllmArgs(capacity_scheduler_policy: Optional[str] = None) -> None

lib/bindings/python/src/dynamo/_core.pyi#L1949

Public methods

init

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

No summary available.

source

No summary available.

from dynamo.mocker.config import apply_worker_engine_args_overrides
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#L322

No summary available.

from dynamo.mocker.config import build_mocker_engine_args
build_mocker_engine_args(args: argparse.Namespace) -> MockEngineArgs

components/src/dynamo/mocker/config.py#L216

No summary available.

from dynamo.mocker.config import build_runtime_config
build_runtime_config(engine_args: MockEngineArgs) -> tuple[int, ModelRuntimeConfig]

components/src/dynamo/mocker/config.py#L340

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

from dynamo.mocker.main import compute_stagger_delay
compute_stagger_delay(num_workers: int, stagger_delay: float) -> float

components/src/dynamo/mocker/main.py#L122

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

from dynamo.mocker.main import graceful_shutdown
graceful_shutdown(runtimes: list)

components/src/dynamo/mocker/main.py#L34

Launch mocker worker(s) with isolated DistributedRuntime instances.

from dynamo.mocker.main import launch_workers
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#L144

No summary available.

from dynamo.mocker.config import load_mocker_engine_args
load_mocker_engine_args(args: argparse.Namespace) -> MockEngineArgs

components/src/dynamo/mocker/config.py#L310

No summary available.

from dynamo.mocker.main import main
main()

components/src/dynamo/mocker/main.py#L286

Parse command-line arguments for the Dynamo mocker engine.

from dynamo.mocker.args import parse_args
parse_args(argv: list[str] | None = None) -> argparse.Namespace

Returns

  • argparse.Namespace — argparse.Namespace: Parsed command-line arguments.

components/src/dynamo/mocker/args.py#L159

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

from dynamo.mocker.args import parse_bootstrap_ports
parse_bootstrap_ports(ports_str: str | None) -> list[int]

components/src/dynamo/mocker/args.py#L152

No summary available.

from dynamo.mocker.args import positive_int
positive_int(value: str) -> int

components/src/dynamo/mocker/args.py#L25

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

from dynamo.mocker.main import prefetch_model
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#L45

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

from dynamo.mocker.args import resolve_planner_profile_data
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#L53

No summary available.

from dynamo.mocker import run_mocker_trace_replay
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)

components/src/dynamo/mocker/__init__.py#L42

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

from dynamo.mocker.args import validate_worker_type_args
validate_worker_type_args(args: argparse.Namespace) -> None

components/src/dynamo/mocker/args.py#L106

Main worker function that launches mocker instances.

from dynamo.mocker.main import worker
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#L77