dynamo.frontend | NVIDIA Dynamo Documentation

dynamo.frontend publishes 12 classes and 31 functions. Source: components/src/dynamo/frontend/__init__.py

EngineFactory (class)

No summary available.

from dynamo.frontend.vllm_processor import EngineFactory

EngineFactory(config: FrontendConfig, flags: Namespace)

components/src/dynamo/frontend/vllm_processor.py#L931

Public methods

init

__init__(config: FrontendConfig, flags: Namespace)

No summary available.

source

chat_engine_factory

chat_engine_factory(instance_id: ModelCardInstanceId, mdc: ModelDeploymentCard, routed_engine: RoutedEngine) -> PythonAsyncEngine

Called by Rust when a model is discovered.

source

FrontendArgGroup (class)

Frontend configuration parameters.

from dynamo.frontend.frontend_args import FrontendArgGroup

components/src/dynamo/frontend/frontend_args.py#L171

Public methods

add_arguments

add_arguments(parser) -> None

No summary available.

source

FrontendConfig (class)

Configuration for the Dynamo frontend.

from dynamo.frontend.frontend_args import FrontendConfig

components/src/dynamo/frontend/frontend_args.py#L53

Public methods

validate

validate() -> None

No summary available.

source

PreprocessError (class)

Raised by preprocess workers for user-facing errors (e.g., n!=1).

from dynamo.frontend.utils import PreprocessError

PreprocessError(message: str)

Carries a plain message because the worker→main-process boundary pickles the exception; the main process re-raises a Dynamo-typed exception so PyO3 can route it through the proper backend-error path.

components/src/dynamo/frontend/utils.py#L83

Public methods

init

__init__(message: str)

No summary available.

source

PreprocessResult (class)

No summary available.

from dynamo.frontend.prepost import PreprocessResult

PreprocessResult(request_for_sampling: ChatCompletionRequest, tool_parser: ToolParser | None, chat_template_kwargs: dict[str, Any], engine_prompt: dict[str, Any], prompt_token_ids: list[int], guided_decoding: dict[str, Any] | None = None) -> None

components/src/dynamo/frontend/prepost.py#L45

Public methods

init

__init__(request_for_sampling: ChatCompletionRequest, tool_parser: ToolParser | None, chat_template_kwargs: dict[str, Any], engine_prompt: dict[str, Any], prompt_token_ids: list[int], guided_decoding: dict[str, Any] | None = None) -> None

No summary available.

source

SglangEngineFactory (class)

No summary available.

from dynamo.frontend.sglang_processor import SglangEngineFactory

SglangEngineFactory(config: FrontendConfig, debug_perf: bool = False, tool_call_parser_name: str | None = None, reasoning_parser_name: str | None = None, chat_template: str | None = None)

components/src/dynamo/frontend/sglang_processor.py#L784

Public methods

init

__init__(config: FrontendConfig, debug_perf: bool = False, tool_call_parser_name: str | None = None, reasoning_parser_name: str | None = None, chat_template: str | None = None)

No summary available.

source

chat_engine_factory

chat_engine_factory(instance_id: ModelCardInstanceId, mdc: ModelDeploymentCard, routed_engine: RoutedEngine) -> PythonAsyncEngine

Called by Rust when a model is discovered.

source

SglangPreprocessResult (class)

Result of SGLang preprocessing.

from dynamo.frontend.sglang_prepost import SglangPreprocessResult

SglangPreprocessResult(prompt_token_ids: list[int], tool_call_parser: ToolCallParserType | None, reasoning_parser: ReasoningParser | None, guided_decoding: dict[str, Any] | None, request: dict[str, Any], force_reasoning: bool = False) -> None

components/src/dynamo/frontend/sglang_prepost.py#L43

Public methods

init

__init__(prompt_token_ids: list[int], tool_call_parser: ToolCallParserType | None, reasoning_parser: ReasoningParser | None, guided_decoding: dict[str, Any] | None, request: dict[str, Any], force_reasoning: bool = False) -> None

No summary available.

source

SglangPreprocessWorkerResult (class)

Picklable return value from the SGLang preprocess worker.

from dynamo.frontend.sglang_processor import SglangPreprocessWorkerResult

SglangPreprocessWorkerResult(prompt_token_ids: list[int], dynamo_preproc: dict[str, Any], request: dict[str, Any], force_reasoning: bool = False, effective_reasoning_parser_name: str | None = None) -> None

components/src/dynamo/frontend/sglang_processor.py#L176

Public methods

init

__init__(prompt_token_ids: list[int], dynamo_preproc: dict[str, Any], request: dict[str, Any], force_reasoning: bool = False, effective_reasoning_parser_name: str | None = None) -> None

No summary available.

source

SglangProcessor (class)

No summary available.

from dynamo.frontend.sglang_processor import SglangProcessor

SglangProcessor(tokenizer, routed_engine: RoutedEngine, tool_call_parser_name: str | None, reasoning_parser_name: str | None, eos_token_ids: list[int] | None, debug_perf: bool = False, preprocess_pool: ProcessPoolExecutor | None = None, preprocess_workers: int = 0, stream_interval: int = 1, default_thinking_mode: str | None = None)

components/src/dynamo/frontend/sglang_processor.py#L355

Public methods

init

__init__(tokenizer, routed_engine: RoutedEngine, tool_call_parser_name: str | None, reasoning_parser_name: str | None, eos_token_ids: list[int] | None, debug_perf: bool = False, preprocess_pool: ProcessPoolExecutor | None = None, preprocess_workers: int = 0, stream_interval: int = 1, default_thinking_mode: str | None = None)

No summary available.

source

generator

generator(request: dict[str, Any], context: Any | None = None) -> AsyncGenerator[dict[str, Any], None]

Main entry point: preprocess, route, post-process a chat request.

source

SglangStreamingPostProcessor (class)

Streaming post-processor using SGLang parsers and HF tokenizer detokenization.

from dynamo.frontend.sglang_prepost import SglangStreamingPostProcessor

SglangStreamingPostProcessor(*, tokenizer, tool_call_parser: ToolCallParserType | None, reasoning_parser: ReasoningParser | None, history_tool_calls_count: int = 0, sglang_tools: list[SglangTool] | None = None, tool_call_parser_name: str | None = None, eos_token_ids: list[int] | None = None, prompt_token_ids: list[int] | None = None) -> None

Handles:

Incremental detokenization across tokenizer-safe boundaries
Reasoning content extraction via SGLang ReasoningParser
Tool call parsing via SGLang FunctionCallParser or JsonArrayParser

components/src/dynamo/frontend/sglang_prepost.py#L953

Public methods

init

__init__(*, tokenizer, tool_call_parser: ToolCallParserType | None, reasoning_parser: ReasoningParser | None, history_tool_calls_count: int = 0, sglang_tools: list[SglangTool] | None = None, tool_call_parser_name: str | None = None, eos_token_ids: list[int] | None = None, prompt_token_ids: list[int] | None = None) -> None

No summary available.

source

process_output

process_output(engine_response: dict[str, Any]) -> dict[str, Any] | None

Process a single engine response chunk into an OpenAI SSE choice dict.

Parameters

engine_response

dict[str, Any]

Dict with token_ids and optional finish_reason.

Returns

dict[str, Any] | None — OpenAI choice dict or None if nothing to emit yet.

source

StreamingPostProcessor (class)

No summary available.

from dynamo.frontend.prepost import StreamingPostProcessor

StreamingPostProcessor(*, tokenizer: TokenizerLike, request_for_sampling: ChatCompletionRequest, sampling_params: SamplingParams, prompt_token_ids: Sequence[int], tool_parser: ToolParser | None, reasoning_parser_class: type[ReasoningParser] | None, chat_template_kwargs: dict[str, Any], stream_response: bool = True) -> None

components/src/dynamo/frontend/prepost.py#L597

Public methods

init

__init__(*, tokenizer: TokenizerLike, request_for_sampling: ChatCompletionRequest, sampling_params: SamplingParams, prompt_token_ids: Sequence[int], tool_parser: ToolParser | None, reasoning_parser_class: type[ReasoningParser] | None, chat_template_kwargs: dict[str, Any], stream_response: bool = True) -> None

No summary available.

source

process_output

process_output(output: Any) -> dict[str, Any] | None

No summary available.

source

VllmProcessor (class)

No summary available.

from dynamo.frontend.vllm_processor import VllmProcessor

VllmProcessor(tokenizer: TokenizerLike, input_processor: InputProcessor, output_processor: OutputProcessor, tool_parser_class: type[ToolParser] | None, reasoning_parser_class: type[ReasoningParser] | None, routed_engine: RoutedEngine, block_size: int = 16, enable_auto_tool_choice: bool = False, default_chat_template_kwargs: dict[str, Any] | None = None, default_thinking_mode: str | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto')

components/src/dynamo/frontend/vllm_processor.py#L268

Public methods

init

__init__(tokenizer: TokenizerLike, input_processor: InputProcessor, output_processor: OutputProcessor, tool_parser_class: type[ToolParser] | None, reasoning_parser_class: type[ReasoningParser] | None, routed_engine: RoutedEngine, block_size: int = 16, enable_auto_tool_choice: bool = False, default_chat_template_kwargs: dict[str, Any] | None = None, default_thinking_mode: str | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto')

No summary available.

source

generator

generator(request: dict[str, Any], context: Any | None = None) -> AsyncGenerator[dict[str, Any], None]

Run a single request through the engine. Does pre and post processing on this machine, delegates model inference to a backend using the router.

source

apply_default_thinking_mode_to_template_kwargs (function)

Merge deployment thinking default unless the request already controls it.

from dynamo.frontend.thinking import apply_default_thinking_mode_to_template_kwargs

apply_default_thinking_mode_to_template_kwargs(chat_template_kwargs: dict[str, Any], default_thinking_mode: str | None, *, request_has_root_thinking: bool = False) -> dict[str, Any]

components/src/dynamo/frontend/thinking.py#L33

async_main (function)

Main async entry point for the Dynamo frontend.

from dynamo.frontend.main import async_main

async_main()

Initializes the distributed runtime, configures routing, and starts the HTTP server or interactive mode based on command-line arguments.

components/src/dynamo/frontend/main.py#L322

build_response_format_guided_decoding (function)

Build Dynamo guided decoding from OpenAI chat response_format.

from dynamo.frontend.sglang_prepost import build_response_format_guided_decoding

build_response_format_guided_decoding(request: dict[str, Any]) -> dict[str, Any] | None

components/src/dynamo/frontend/sglang_prepost.py#L635

build_tool_call_guided_decoding (function)

Build tool-call guidance through vLLM’s configured tool parser.

from dynamo.frontend.prepost import build_tool_call_guided_decoding

build_tool_call_guided_decoding(request: ChatCompletionRequest, tool_parser: ToolParser | None, *, parser_guided_decoding: dict[str, Any] | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto') -> dict[str, Any] | None

components/src/dynamo/frontend/prepost.py#L187

build_tool_call_guided_decoding (function)

Build native-SGLang-like tool call constraints for guided decoding.

from dynamo.frontend.sglang_prepost import build_tool_call_guided_decoding

build_tool_call_guided_decoding(request: dict[str, Any], *, tool_call_parser_name: str | None, sglang_tools: list[SglangTool] | None) -> dict[str, Any] | None

components/src/dynamo/frontend/sglang_prepost.py#L558

convert_tools (function)

Convert OpenAI tool dicts to SGLang Tool objects.

from dynamo.frontend.sglang_prepost import convert_tools

convert_tools(tools: list[dict[str, Any]] | None) -> list[SglangTool] | None

components/src/dynamo/frontend/sglang_prepost.py#L192

create_parsers (function)

Create tool call and reasoning parsers for a request.

from dynamo.frontend.sglang_prepost import create_parsers

create_parsers(request: dict[str, Any], *, tool_call_parser_name: str | None, reasoning_parser_name: str | None, sglang_tools: list[SglangTool] | None = None, force_reasoning: bool = False) -> tuple[ToolCallParserType | None, ReasoningParser | None]

Shared by both the single-process preprocessing path and the pool path (which must recreate non-picklable parsers in the main process).

If sglang_tools is provided, reuses them; otherwise converts from the request’s tools field.

For tool_choice="required" or a named function, uses JsonArrayParser (matching native SGLang) since guided decoding constrains the output to a JSON array. Otherwise uses the model-specific FunctionCallParser.

components/src/dynamo/frontend/sglang_prepost.py#L266

detect_force_reasoning_from_template (function)

Return True if the chat template auto-opens a reasoning block.

from dynamo.frontend.sglang_prepost import detect_force_reasoning_from_template

detect_force_reasoning_from_template(chat_template: str | None) -> bool

Intended to be called once at processor startup with tokenizer.chat_template and cached on the processor.

components/src/dynamo/frontend/sglang_prepost.py#L79

extract_mm_urls (function)

Extract media and vLLM image processor-cache UUIDs from chat messages.

from dynamo.frontend.utils import extract_mm_urls

extract_mm_urls(messages: list[dict[str, Any]]) -> tuple[dict[str, list[dict[str, str]]] | None, dict[str, list[str | None]] | None]

URL-backed parts become Url variants. Image parts with no URL and an opaque uuid become UuidOnly variants for vLLM’s multimodal processor cache. Cache UUIDs on audio and video are rejected. Image UUID lists preserve slot order

({"image_url": [{"Url": "https://..."}, {"UuidOnly": "image-1"}]},
 {"image_url": ["image-1", "image-1"]})

The UUID map is None when no user UUID is present. A media content part with neither a URL nor UUID is rejected instead of being silently dropped.

components/src/dynamo/frontend/utils.py#L100

graceful_shutdown (function)

Handle graceful shutdown of the distributed runtime.

from dynamo.frontend.main import graceful_shutdown

graceful_shutdown(runtime: DistributedRuntime) -> None

Parameters

runtime

DistributedRuntime

The DistributedRuntime instance to shut down.

components/src/dynamo/frontend/main.py#L475

handle_engine_error (function)

Classify an invalid engine response and return an OpenAI-style error dict.

from dynamo.frontend.utils import handle_engine_error

handle_engine_error(engine_response: Any, request_id: str, logger: logging.Logger) -> dict[str, Any]

Called when engine_response is None or missing ‘token_ids’.

components/src/dynamo/frontend/utils.py#L184

load_frontend_route_extensions (function)

Load trusted frontend route extensions.

from dynamo.frontend.main import load_frontend_route_extensions

load_frontend_route_extensions(extension_names: list[str]) -> list[FrontendRoute]

Each value is either a name registered under the dynamo.frontend.routes entry-point group (preferred) or a direct module:function path.

components/src/dynamo/frontend/main.py#L217

main (function)

Entry point for the Dynamo frontend CLI.

from dynamo.frontend.main import main

main() -> None

components/src/dynamo/frontend/main.py#L484

make_backend_error (function)

Build an OpenAI-style error dict, guarding against None/missing message.

from dynamo.frontend.utils import make_backend_error

make_backend_error(engine_response: dict[str, Any]) -> dict[str, Any]

components/src/dynamo/frontend/utils.py#L162

make_internal_error (function)

Build an OpenAI-style internal error dict with request-specific fallback.

from dynamo.frontend.utils import make_internal_error

make_internal_error(request_id: str, detail: str | None = None) -> dict[str, Any]

components/src/dynamo/frontend/utils.py#L173

map_finish_reason (function)

No summary available.

from dynamo.frontend.vllm_processor import map_finish_reason

map_finish_reason(raw_reason: str | None) -> FinishReason | None

components/src/dynamo/frontend/vllm_processor.py#L64

nvext_extra_field_requested (function)

Return whether a request opted into a response nvext field.

from dynamo.frontend.utils import nvext_extra_field_requested

nvext_extra_field_requested(request: dict[str, Any], field: str) -> bool

components/src/dynamo/frontend/utils.py#L69

parse_args (function)

Parse command-line arguments for the Dynamo frontend.

from dynamo.frontend.main import parse_args

parse_args() -> tuple[FrontendConfig, Optional[Namespace], Optional[Namespace]]

Returns

tuple[FrontendConfig, Optional[Namespace], Optional[Namespace]] — Tuple of (FrontendConfig, vllm_flags, sglang_flags).

components/src/dynamo/frontend/main.py#L245

preprocess_chat_request (function)

No summary available.

from dynamo.frontend.prepost import preprocess_chat_request

preprocess_chat_request(request: dict[str, Any] | ChatCompletionRequest, *, tokenizer: TokenizerLike, renderer: _Renderer, tool_parser_class: type[ToolParser] | None, exclude_tools_when_tool_choice_none: bool = True, enable_auto_tool_choice: bool = False, default_chat_template_kwargs: dict[str, Any] | None = None, default_thinking_mode: str | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto') -> PreprocessResult

components/src/dynamo/frontend/prepost.py#L491

preprocess_chat_request (function)

Preprocess a chat request using SGLang tokenizer and parser APIs.

from dynamo.frontend.sglang_prepost import preprocess_chat_request

preprocess_chat_request(request: dict[str, Any], *, tokenizer, tool_call_parser_name: str | None, reasoning_parser_name: str | None, exclude_tools_when_tool_choice_none: bool = True, template_force_reasoning: bool = False, default_thinking_mode: str | None = None) -> SglangPreprocessResult

template_force_reasoning is the static per-server flag derived from the chat template (see detect_force_reasoning_from_template); the effective per-request value combines it with the configured parser and request-level thinking controls.

Synchronous — suitable for both main-process and worker-process execution.

components/src/dynamo/frontend/sglang_prepost.py#L720

random_call_id (function)

Generate a random tool call ID in OpenAI format.

from dynamo.frontend.utils import random_call_id

random_call_id() -> str

components/src/dynamo/frontend/utils.py#L64

random_uuid (function)

Generate a random 16-character hex UUID.

from dynamo.frontend.utils import random_uuid

random_uuid() -> str

components/src/dynamo/frontend/utils.py#L59

read_jinja_chat_template (function)

Read a Jinja chat template using backend-specific file semantics.

from dynamo.frontend.utils import read_jinja_chat_template

read_jinja_chat_template(template_path: str, *, backend: ChatProcessorBackend) -> str

components/src/dynamo/frontend/utils.py#L18

resolve_chat_template (function)

Return a chat template stored beside the model, or None.

from dynamo.frontend.utils import resolve_chat_template

resolve_chat_template(source_path: str, *, backend: ChatProcessorBackend = 'vllm') -> str | None

Covers models (e.g. Qwen3-Omni) whose template lives in chat_template.json or chat_template.jinja rather than tokenizer_config.json, which the HF tokenizer does not merge. The backend selects native .jinja file semantics.

components/src/dynamo/frontend/utils.py#L33

resolve_request_force_reasoning (function)

Resolve the effective force_reasoning flag for a single request.

from dynamo.frontend.sglang_prepost import resolve_request_force_reasoning

resolve_request_force_reasoning(request: dict[str, Any], reasoning_parser_name: str | None, template_default: bool) -> bool

Mirrors sglang.srt.entrypoints.openai.serving_chat._get_reasoning_from_request combined with template_manager.force_reasoning:

opt-out families (glm45/qwen3/kimi_k2/…): on by default, chat_template_kwargs.enable_thinking=False (or thinking=False for kimi_k2) disables it.
MiniMax-M3 defaults to adaptive, but SGLang still enables the reasoning parser unless chat_template_kwargs.thinking_mode is explicitly "disabled".
Mistral is enabled only when reasoning_effort is present and not "none".
opt-in families (deepseek-v3/gemma4): off by default, enabled by chat_template_kwargs.{thinking,enable_thinking}=True.
anything else: follow the statically-detected template default.

components/src/dynamo/frontend/sglang_prepost.py#L120

runtime_default_thinking_mode (function)

Read deployment-level default thinking mode from model runtime metadata.

from dynamo.frontend.thinking import runtime_default_thinking_mode

runtime_default_thinking_mode(runtime_config: dict[str, Any] | None) -> str | None

components/src/dynamo/frontend/thinking.py#L20

setup_engine_factory (function)

When using vllm pre and post processor, create the EngineFactory that creates the engines that run requests.

from dynamo.frontend.main import setup_engine_factory

setup_engine_factory(config: FrontendConfig, vllm_flags: Namespace) -> EngineFactory

components/src/dynamo/frontend/main.py#L100

setup_sglang_engine_factory (function)

When using sglang pre and post processor, create the SglangEngineFactory that creates the engines that run requests.

from dynamo.frontend.main import setup_sglang_engine_factory

setup_sglang_engine_factory(config: FrontendConfig, sglang_flags: Optional[Namespace] = None)

components/src/dynamo/frontend/main.py#L113

validate_model_name (function)

Validate that model-name is a non-empty string.

from dynamo.frontend.frontend_args import validate_model_name

validate_model_name(value: str) -> str

components/src/dynamo/frontend/frontend_args.py#L35

validate_model_path (function)

Validate that model-path is a valid directory on disk.

from dynamo.frontend.frontend_args import validate_model_path

validate_model_path(value: str) -> str

components/src/dynamo/frontend/frontend_args.py#L44

worker_warmup (function)

Dummy task to ensure a ProcessPoolExecutor worker is fully initialized.

from dynamo.frontend.utils import worker_warmup

worker_warmup() -> bool

components/src/dynamo/frontend/utils.py#L78