dynamo.runtime | NVIDIA Dynamo Documentation

dynamo.runtime publishes 7 classes and 10 functions. Source: lib/bindings/python/src/dynamo/runtime/__init__.py

Client (class)

A client capable of calling served instances of an endpoint

1 from dynamo.runtime import Client

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

Public methods

instance_ids

1 instance_ids() -> List[int]

Get list of current instance IDs.

Returns

List[int] — A list of currently available instance IDs

source

instances

1 instances() -> List[Instance]

Get a snapshot of the current instances with full transport details.

Like instance_ids(), the result is a snapshot of the watched instance set; pair with wait_for_instances() to block until instances exist.

Returns

List[Instance] — A list of Instance for the currently available instances,
List[Instance] — across all transports (TCP, NATS, …).

source

wait_for_instances

1 wait_for_instances() -> List[int]

Wait for instances to be available for work and return their IDs.

Returns

List[int] — A list of instance IDs that are available for work

source

wait_for_instance_by_runtime_data

1 wait_for_instance_by_runtime_data(key: str, value: str, timeout_s: float | None = None) -> int

Wait for exactly one instance whose MDC runtime_data contains the given string value.

source

random

1 random(request: JsonLike, annotated: bool | None = True, context: Context | None = None) -> AsyncIterator[JsonLike]

Pick a random instance of the endpoint and issue the request

source

round_robin

1 round_robin(request: JsonLike, annotated: bool | None = True, context: Context | None = None) -> AsyncIterator[JsonLike]

Pick the next instance of the endpoint in a round-robin fashion

source

direct

1 direct(request: JsonLike, instance_id: int, annotated: bool | None = True, context: Context | None = None) -> AsyncIterator[JsonLike]

Pick a specific instance of the endpoint

source

generate

1 generate(request: JsonLike, annotated: bool | None = True, context: Context | None = None) -> AsyncIterator[JsonLike]

Generate a response from the endpoint

source

Context (class)

Context wrapper around AsyncEngineContext for Python bindings. Provides tracing and cancellation capabilities for request handling.

1 from dynamo.runtime import Context

1 Context(id: Optional[str] = None, metadata: Optional[Dict[str, str]] = None) -> None

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

Public methods

init

1 __init__(id: Optional[str] = None, metadata: Optional[Dict[str, str]] = None) -> None

Create a new Context instance.

Parameters

Optional[str]

Optional request ID. If None, a default ID will be generated.

metadata

Optional[Dict[str, str]]

Optional propagated metadata map.

source

is_stopped

1 is_stopped() -> bool

Check if the context has been stopped (synchronous).

Returns

bool — True if the context is stopped, False otherwise.

source

is_killed

1 is_killed() -> bool

Check if the context has been killed (synchronous).

Returns

bool — True if the context is killed, False otherwise.

source

stop_generating

1 stop_generating() -> None

Issue a stop generating signal to the context.

source

id

1 id() -> str

Get the context ID.

Returns

str — The context identifier string.

source

detached

1 detached(id: str) -> Context

Create a context with a fresh cancellation controller and request ID while preserving trace parentage and a metadata snapshot.

source

async_killed_or_stopped

1 async_killed_or_stopped() -> asyncio.Future[bool]

Asynchronously wait until the context is killed or stopped.

Returns

asyncio.Future[bool] — True when the context is killed or stopped.

source

notify_first_token

1 notify_first_token() -> None

Fire the first-token signal so the framework can release any deferred engine.abort(). Idempotent; no-op on non-decode requests. Engines normally don’t need this — the framework auto-fires on the first non-empty chunk in the response stream.

source

trace_headers

1 trace_headers() -> Optional[dict[str, str]]

Build W3C trace headers for propagating to downstream inference engines.

Returns

Optional[dict[str, str]] — {"traceparent": "00-<trace_id>-<span_id>-<flags>"} when this
Optional[dict[str, str]] — request carries trace context, None otherwise. Also emits tracestate,
Optional[dict[str, str]] — x-request-id, request-id when upstream propagated them.
Optional[dict[str, str]] — Forward unchanged to the inference engine’s trace_headers kwarg.

source

current_span

1 current_span() -> SpanProxy

Handle on the framework’s engine.generate span. Use it to set_attribute / add_event / set_status on the parent span. Returns a silent no-op proxy when no parent was plumbed in (test contexts) or the OTel bridge isn’t installed.

Engines normally reach this through dynamo.common.backend.telemetry.current_span(context).

source

start_span

1 start_span(name: str, attrs: Optional[dict[str, Any]] = None) -> SpanProxy

Open a child span under engine.generate with a dynamic name. The returned SpanProxy is a context manager — the span ends on __exit__ / close() / drop.

Engines normally reach this through dynamo.common.backend.telemetry.start_span(context, name).

source

DistributedRuntime (class)

The runtime object for dynamo applications

1 from dynamo.runtime import DistributedRuntime

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

Public methods

endpoint

1 endpoint(path: str) -> Endpoint

Get an endpoint directly by path.

Parameters

path

str

Endpoint path in format ‘namespace.component.endpoint’ or ‘dyn://namespace.component.endpoint’

Returns

Endpoint — The requested endpoint

Raises

ValueError — If path format is invalid (not 3 parts separated by dots)
Exception — If namespace or component creation fails

endpoint = runtime.endpoint(“demo.backend.generate”) endpoint = runtime.endpoint(“dyn://demo.backend.generate”)

source

shutdown

1 shutdown() -> None

Shutdown the runtime by triggering the cancellation token

source

set_health_status

1 set_health_status(ready: bool) -> None

Explicitly set the system-level health status (Ready / NotReady).

source

register_engine_route

1 register_engine_route(route_name: str, callback: Callable[[dict], Awaitable[dict]]) -> None

Parameters

route_name

str

The route path (e.g., “control/start_profile” creates /engine/control/start_profile)

callback

Callable[[dict], Awaitable[dict]]

Async function with signature: async def(body: dict) -> dict

async def start_profile(body: dict) -> dict: await engine.start_profile(**body) return {“status”: “ok”, “message”: “Profiling started”}

runtime.register_engine_route(“control/start_profile”, start_profile)

The callback receives the JSON request body as a dict and should return a dict that will be serialized as the JSON response.

For GET requests or empty bodies, an empty dict {} is passed.

source

Endpoint (class)

An Endpoint is a single API endpoint

1 from dynamo.runtime import Endpoint

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

Public methods

serve_endpoint

1 serve_endpoint(handler: RequestHandler, graceful_shutdown: bool = True, metrics_labels: Optional[List[Tuple[str, str]]] = None, health_check_payload: Optional[Dict[str, Any]] = None) -> None

Serve an endpoint discoverable by all connected clients at {{ namespace }}/components/{{ component_name }}/endpoints/{{ endpoint_name }}

Parameters

handler

RequestHandler

The request handler function

graceful_shutdown

bool

Whether to wait for inflight requests to complete during shutdown (default: True)

metrics_labels

Optional[List[Tuple[str, str]]]

Optional list of metrics labels to add to the metrics

health_check_payload

Optional[Dict[str, Any]]

Optional dict containing the health check request payload that will be used to verify endpoint health

source

serve_bidirectional_endpoint

1 serve_bidirectional_endpoint(handler: Callable[..., AsyncIterator[JsonLike]], graceful_shutdown: bool = True, metrics_labels: Optional[List[Tuple[str, str]]] = None) -> None

Serve a bidirectional (streaming-input, streaming-output) endpoint.

The handler is an async generator function — async def generate(request_stream) or async def generate(request_stream, context) — so calling it returns an async iterator of response frames directly (it is not awaited). request_stream is a PyAsyncRequestStream yielding inbound frames as JSON-like Python objects; the generator yields response frames as JSON-like Python objects.

Request-stream end (when __anext__ raises StopAsyncIteration) is not a cancellation signal: the caller has merely stopped sending input. The engine must keep yielding response chunks until it chooses to return or observes context.is_stopped().

Parameters

handler

Callable[..., AsyncIterator[JsonLike]]

The async generator factory described above

graceful_shutdown

bool

Whether to wait for inflight requests to complete during shutdown (default: True)

metrics_labels

Optional[List[Tuple[str, str]]]

Optional list of metrics labels to add to the metrics

source

client

1 client(router_mode: Optional[RouterMode] = None) -> Client

Create a Client capable of calling served instances of this endpoint.

By default this uses round-robin routing when router_mode is not provided.

source

connection_id

1 connection_id() -> int

Opaque unique ID for this worker. May change over worker lifetime.

source

unregister_endpoint_instance

1 unregister_endpoint_instance() -> None

Unregister this endpoint instance from discovery.

This removes the endpoint from the instances bucket, preventing the router from sending requests to this worker. Use this when a worker is sleeping and should not receive any requests.

source

register_endpoint_instance

1 register_endpoint_instance() -> None

Re-register this endpoint instance to discovery.

This adds the endpoint back to the instances bucket, allowing the router to send requests to this worker again. Use this when a worker wakes up and should start receiving requests.

source

LogHandler (class)

Custom logging handler that sends log messages to the Rust env_logger

1 from dynamo.runtime.logging import LogHandler

lib/bindings/python/src/dynamo/runtime/logging.py#L28

Public methods

emit

1 emit(record: logging.LogRecord) -> None

Emit a log record

source

PyAsyncRequestStream (class)

Python-visible inbound iterator handed to bidirectional engine handlers as the first positional argument. Yields request frames as JSON-like Python objects.

1 from dynamo.runtime import PyAsyncRequestStream

Request-stream end is not a cancellation signal: when this iterator raises StopAsyncIteration, the caller has merely stopped sending input. The engine should keep yielding response chunks until it chooses to return or observes context.is_stopped().

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

VllmColorFormatter (class)

Formatter that matches Rust tracing’s compact colored output style.

1 from dynamo.runtime.logging import VllmColorFormatter

Used for vLLM logs routed through a StreamHandler (bypassing the Rust bridge) so that VLLM_LOGGING_LEVEL is respected independently of DYN_LOG while still producing visually consistent colored output.

lib/bindings/python/src/dynamo/runtime/logging.py#L65

Public methods

format

1 format(record: logging.LogRecord) -> str

No summary available.

source

configure_dynamo_logging (function)

A single place to configure logging for Dynamo.

1 from dynamo.runtime.logging import configure_dynamo_logging

1 configure_dynamo_logging(service_name: str | None = None, worker_id: int | None = None) -> None

lib/bindings/python/src/dynamo/runtime/logging.py#L140

configure_logger (function)

Called once to configure the Python logger to use the LogHandler

1 from dynamo.runtime.logging import configure_logger

1 configure_logger(service_name: str | None, worker_id: int | None, level: int = logging.INFO) -> None

lib/bindings/python/src/dynamo/runtime/logging.py#L106

configure_sglang_logging (function)

SGLang allows us to create a custom logging config file

1 from dynamo.runtime.logging import configure_sglang_logging

1 configure_sglang_logging(dyn_level: int) -> None

lib/bindings/python/src/dynamo/runtime/logging.py#L218

configure_vllm_logging (function)

Configure vLLM logging for the main process and subprocesses.

1 from dynamo.runtime.logging import configure_vllm_logging

1 configure_vllm_logging(dyn_level: int) -> None

Main process: replaces vLLM’s StreamHandler with a new StreamHandler that uses VllmColorFormatter and writes directly to stderr. This bypasses the Rust LogHandler bridge so that VLLM_LOGGING_LEVEL is respected independently of DYN_LOG (the Rust bridge filters based on DYN_LOG).

Subprocesses (EngineCore, workers): use vLLM’s DEFAULT_LOGGING_CONFIG (StreamHandler to stderr) since the Rust runtime is not initialized there. Setting VLLM_CONFIGURE_LOGGING=1 without VLLM_LOGGING_CONFIG_PATH causes vLLM to use its built-in default config in spawned subprocesses.

The dyn_level param is kept for signature compatibility but does not control the vLLM logger level. Use VLLM_LOGGING_LEVEL env var instead.

lib/bindings/python/src/dynamo/runtime/logging.py#L255

construct_formatter_prefix (function)

No summary available.

1 from dynamo.runtime.logging import construct_formatter_prefix

1 construct_formatter_prefix(service_name: str | None, worker_id: int | None) -> str

lib/bindings/python/src/dynamo/runtime/logging.py#L129

dynamo_endpoint (function)

Decorator that can parse a request payload into a Pydantic model before the endpoint runs.

1 from dynamo.runtime import dynamo_endpoint

1 dynamo_endpoint(request_model: Union[Type[BaseModel], Type[Any]], response_model: Type[BaseModel]) -> Callable

Parsing applies only when request_model is a BaseModel subclass and the wrapper receives one or two positional arguments — (request) or (self, request). With three or more positional arguments, or when the payload arrives by keyword, it is forwarded untouched. A str payload is parsed with parse_raw and a dict with parse_obj; any other type, including an already-constructed request_model instance, is rejected. response_model is reserved for future validation; yielded items pass through unchanged today.

Parameters

request_model

Union[Type[BaseModel], Type[Any]]

Request class used to parse str or dict payloads. Pass a non-BaseModel type, as examples/custom_backend does with str, to skip parsing entirely.

response_model

Type[BaseModel]

Expected response class. Currently accepted but not enforced.

Raises

ValueError — On the first __anext__() of the returned generator, not when the decorated function is called, because the wrapper is itself an async generator. Raised when the payload fails validation or is neither str nor dict.

Examples

1 >>> from pydantic import BaseModel
2 >>> from dynamo.runtime import dynamo_endpoint
3 >>>
4 >>> class Request(BaseModel):
5 ...     data: str
6 >>> class Response(BaseModel):
7 ...     char: str
8 >>>
9 >>> @dynamo_endpoint(Request, Response)
10 ... async def generate(request):
11 ...     for char in request.data:
12 ...         yield char

lib/bindings/python/src/dynamo/runtime/__init__.py#L54

dynamo_worker (function)

Decorator that creates a DistributedRuntime and passes it to the worker function.

1 from dynamo.runtime import dynamo_worker

1 dynamo_worker(enable_nats: Optional[bool] = None)

Parameters

enable_nats

Optional[bool]

Deprecated. NATS enablement is now determined automatically from the event-plane configuration. This parameter is accepted for backwards compatibility but will be removed in a future release.

lib/bindings/python/src/dynamo/runtime/__init__.py#L21

get_bool_env_var (function)

No summary available.

1 from dynamo.runtime.logging import get_bool_env_var

1 get_bool_env_var(name: str, default: str = 'false') -> bool

lib/bindings/python/src/dynamo/runtime/logging.py#L317

log_level_mapping (function)

The DYN_LOG variable is set using “debug” or “trace” or “info. This function maps those to the appropriate logging level and defaults to INFO if the variable is not set or a bad value.

1 from dynamo.runtime.logging import log_level_mapping

1 log_level_mapping(level: str) -> int

lib/bindings/python/src/dynamo/runtime/logging.py#L174

python_log_level_mapping (function)

Return the lowest Python level enabled by a Rust-style DYN_LOG filter.

1 from dynamo.runtime.logging import python_log_level_mapping

1 python_log_level_mapping(filters: str) -> int

lib/bindings/python/src/dynamo/runtime/logging.py#L196