dynamo.runtime

A client capable of calling served instances of an endpoint

from dynamo.runtime import Client

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

Public methods

instance_ids

instance_ids() -> List[int]

Get list of current instance IDs.

Returns

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

source

instances

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

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

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

wait_for_instances_by_runtime_data

wait_for_instances_by_runtime_data(key: str, value: str, min_count: int, timeout_s: float | None = None) -> List[int]

Wait until at least min_count instances have MDC runtime_data containing the given string value, and return their sorted instance IDs.

source

random

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

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

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

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

Generate a response from the endpoint

source

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

from dynamo.runtime import Context

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

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

Public methods

init

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

Create a new Context instance.

Parameters

source

is_stopped

is_stopped() -> bool

Check if the context has been stopped (synchronous).

Returns

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

source

is_killed

is_killed() -> bool

Check if the context has been killed (synchronous).

Returns

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

source

stop_generating

stop_generating() -> None

Issue a stop generating signal to the context.

source

id

id() -> str

Get the context ID.

Returns

• str — The context identifier string.

source

detached

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

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

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

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

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

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

The runtime object for dynamo applications

from dynamo.runtime import DistributedRuntime

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

Public methods

endpoint

endpoint(path: str) -> Endpoint

Get an endpoint directly by path.

Parameters

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

source

shutdown

shutdown() -> None

Shutdown the runtime by triggering the cancellation token

source

set_health_status

set_health_status(ready: bool) -> None

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

source

register_engine_route

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

Register an async callback for /engine/{route_name} on the system status server.

Parameters

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

An Endpoint is a single API endpoint

from dynamo.runtime import Endpoint

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

Public methods

first_token_source

first_token_source(worker_type: WorkerType) -> Optional[FirstTokenSource]

No summary available.

source

serve_endpoint

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

source

serve_bidirectional_endpoint

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

source

client

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

connection_id() -> int

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

source

unregister_endpoint_instance

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

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

Custom logging handler that sends log messages to the Rust env_logger

from dynamo.runtime.logging import LogHandler

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

Public methods

emit

emit(record: logging.LogRecord) -> None

Emit a log record

source

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

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#L242

Return exactly one response from a round-robin Dynamo endpoint call.

from dynamo.runtime import UnaryClient

UnaryClient(client: _RoundRobinClient) -> None

lib/bindings/python/src/dynamo/runtime/_unary.py#L39

Public methods

init

__init__(client: _RoundRobinClient) -> None

No summary available.

source

complete

complete(request: Any, *, context: Context | None = None) -> Any

Return one response, linking the call to context when provided.

source

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

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

format(record: logging.LogRecord) -> str

No summary available.

source

A single place to configure logging for Dynamo.

from dynamo.runtime.logging import configure_dynamo_logging

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

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

Called once to configure the Python logger to use the LogHandler

from dynamo.runtime.logging import configure_logger

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

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

SGLang allows us to create a custom logging config file

from dynamo.runtime.logging import configure_sglang_logging

configure_sglang_logging(dyn_level: int) -> None

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

Configure vLLM logging for the main process and subprocesses.

from dynamo.runtime.logging import configure_vllm_logging

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

No summary available.

from dynamo.runtime.logging import construct_formatter_prefix

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

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

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

from dynamo.runtime import dynamo_endpoint

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

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

>>> from pydantic import BaseModel
>>> from dynamo.runtime import dynamo_endpoint
>>>
>>> class Request(BaseModel):
...     data: str
>>> class Response(BaseModel):
...     char: str
>>>
>>> @dynamo_endpoint(Request, Response)
... async def generate(request):
...     for char in request.data:
...         yield char

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

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

from dynamo.runtime import dynamo_worker

dynamo_worker(enable_nats: Optional[bool] = None)

Parameters

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

No summary available.

from dynamo.runtime.logging import get_bool_env_var

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

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

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.

from dynamo.runtime.logging import log_level_mapping

log_level_mapping(level: str) -> int

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

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

from dynamo.runtime.logging import python_log_level_mapping

python_log_level_mapping(filters: str) -> int

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

Serve a unary handler that accepts Dynamo’s per-request context.

from dynamo.runtime import serve_unary_endpoint

serve_unary_endpoint(endpoint: _Endpoint, handler: _UnaryHandler) -> None

lib/bindings/python/src/dynamo/runtime/_unary.py#L56