feat: dynamic endpoint registration

components/src/dynamo/sglang/main.py

@@ -128,6 +128,8 @@ async def register_model():
 
     health_check_payload = SglangHealthCheckPayload(engine).to_dict()
 
+    test_endpoint = component.endpoint("test")
+
     try:
         # Start endpoint immediately and register model concurrently
         # Requests queue until ready_event is set
@@ -139,6 +141,13 @@ async def register_model():
                 health_check_payload=health_check_payload,
             ),
             register_model(),
+            test_endpoint.serve_endpoint(
+                handler.test,
+                graceful_shutdown=True,
+                metrics_labels=metrics_labels,
+                health_check_payload=health_check_payload,
+                http_endpoint_path="/test",
+            ),
         )
     except Exception as e:
         logging.error(f"Failed to serve endpoints: {e}")

components/src/dynamo/sglang/request_handlers/decode_handler.py

@@ -66,6 +66,9 @@ def _build_sampling_params(self, request: dict) -> dict:
 
         return {k: v for k, v in param_mapping.items() if v is not None}
 
+    async def test(self, request: dict):
+        yield {"message": "Hello, world!"}
+
     async def generate(self, request: dict):
         sampling_params = self._build_sampling_params(request)
         input_param = self._get_input_param(request)

lib/bindings/python/rust/lib.rs

@@ -644,14 +644,15 @@ impl Component {
 
 #[pymethods]
 impl Endpoint {
-    #[pyo3(signature = (generator, graceful_shutdown = true, metrics_labels = None, health_check_payload = None))]
+    #[pyo3(signature = (generator, graceful_shutdown = true, metrics_labels = None, health_check_payload = None, http_endpoint_path = None))]
     fn serve_endpoint<'p>(
         &self,
         py: Python<'p>,
         generator: PyObject,
         graceful_shutdown: Option<bool>,
         metrics_labels: Option<Vec<(String, String)>>,
         health_check_payload: Option<&Bound<'p, PyDict>>,
+        http_endpoint_path: Option<&str>,
     ) -> PyResult<Bound<'p, PyAny>> {
         let engine = Arc::new(engine::PythonAsyncEngine::new(
             generator,
@@ -689,6 +690,10 @@ impl Endpoint {
             builder = builder.health_check_payload(payload);
         }
 
+        if let Some(http_endpoint_path) = http_endpoint_path {
+            builder = builder.http_endpoint_path(http_endpoint_path);
+        }
+
         let graceful_shutdown = graceful_shutdown.unwrap_or(true);
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
             builder

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

@@ -115,7 +115,7 @@ class Endpoint:
 
     ...
 
-    async def serve_endpoint(self, handler: RequestHandler, graceful_shutdown: bool = True, metrics_labels: Optional[List[Tuple[str, str]]] = None, health_check_payload: Optional[Dict[str, Any]] = None) -> None:
+    async def serve_endpoint(self, handler: RequestHandler, graceful_shutdown: bool = True, metrics_labels: Optional[List[Tuple[str, str]]] = None, health_check_payload: Optional[Dict[str, Any]] = None, http_endpoint_path: Optional[str] = None) -> None:
         """
         Serve an endpoint discoverable by all connected clients at
         `{{ namespace }}/components/{{ component_name }}/endpoints/{{ endpoint_name }}`

lib/llm/src/http/service.rs

@@ -21,6 +21,7 @@
 mod openai;
 
 pub mod disconnect;
+pub mod dynamic_endpoint;
 pub mod error;
 pub mod health;
 pub mod metrics;

lib/llm/src/http/service/dynamic_endpoint.rs

@@ -0,0 +1,183 @@
+// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use super::{RouteDoc, service_v2};
+use crate::types::Annotated;
+use axum::{
+    Json, Router,
+    http::{Method, StatusCode},
+    response::IntoResponse,
+    routing::post,
+};
+use dynamo_runtime::instances::list_all_instances;
+use dynamo_runtime::{DistributedRuntime, Runtime, component::Client};
+use dynamo_runtime::{pipeline::PushRouter, stream::StreamExt};
+use std::sync::Arc;
+
+pub const DYNAMIC_ENDPOINT_PATH: &str = "dynamic_endpoint";
+
+pub fn dynamic_endpoint_router(
+    state: Arc<service_v2::State>,
+    path: Option<String>,
+) -> (Vec<RouteDoc>, Router) {
+    let wildcard_path = "/{*path}";
+    let path = path.unwrap_or_else(|| wildcard_path.to_string());
+
+    let docs: Vec<RouteDoc> = vec![RouteDoc::new(Method::POST, &path)];
+
+    let router = Router::new()
+        .route(&path, post(dynamic_endpoint_handler))
+        .with_state(state);
+
+    (docs, router)
+}
+
+async fn dynamic_endpoint_handler(
+    axum::extract::State(state): axum::extract::State<Arc<service_v2::State>>,
+    axum::extract::Path(path): axum::extract::Path<String>,
+) -> impl IntoResponse {
+    let Some(etcd_client) = state.etcd_client() else {
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(serde_json::json!({
+                "message": "Failed to get etcd client"
+            })),
+        );
+    };
+
+    let instances = match list_all_instances(etcd_client).await {
+        Ok(instances) => instances,
+        Err(_) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({
+                    "message": "Failed to get instances"
+                })),
+            );
+        }
+    };
+
+    let dynamic_endpoints = instances
+        .iter()
+        .filter_map(|instance| instance.http_endpoint_path.clone())
+        .collect::<Vec<String>>();
+
+    let fmt_path = format!("/{}", &path);
+    if !dynamic_endpoints.contains(&fmt_path) {
+        return (
+            StatusCode::NOT_FOUND,
+            Json(serde_json::json!({
+                "message": "Dynamic endpoint not found"
+            })),
+        );
+    }
+
+    let rt = match Runtime::from_current() {
+        Ok(rt) => rt,
+        Err(_) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({
+                    "message": "Failed to get runtime"
+                })),
+            );
+        }
+    };
+    let drt = match DistributedRuntime::from_settings(rt).await {
+        Ok(drt) => drt,
+        Err(_) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({
+                    "message": "Failed to get distributed runtime"
+                })),
+            );
+        }
+    };
+
+    // grab all instances that expose this endpoint
+    let target_instances = instances
+        .iter()
+        .filter(|instance| instance.http_endpoint_path == Some(fmt_path.clone()))
+        .collect::<Vec<_>>();
+
+    // use pushrouter .direct to forward the request to the filtered instances sequentially
+    let mut target_clients: Vec<Client> = Vec::new();
+    for instance in target_instances {
+        let ns = match drt.namespace(instance.namespace.clone()) {
+            Ok(ns) => ns,
+            Err(e) => {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(serde_json::json!({
+                        "message": "Failed to get namespace"
+                    })),
+                );
+            }
+        };
+        let c = match ns.component(instance.component.clone()) {
+            Ok(c) => c,
+            Err(_) => {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(serde_json::json!({
+                        "message": "Failed to get component"
+                    })),
+                );
+            }
+        };
+        let ep = c.endpoint(path.clone());
+        let c = match ep.client().await {
+            Ok(c) => c,
+            Err(_) => {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(serde_json::json!({
+                        "message": "Failed to get client"
+                    })),
+                );
+            }
+        };
+        target_clients.push(c.clone());
+    }
+
+    let mut all_responses = Vec::new();
+    for client in target_clients {
+        let router = match PushRouter::<(), Annotated<serde_json::Value>>::from_client(
+            client,
+            Default::default(),
+        )
+        .await
+        {
+            Ok(router) => router,
+            Err(_) => {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(serde_json::json!({
+                        "message": "Failed to get router"
+                    })),
+                );
+            }
+        };
+        let mut stream = match router.round_robin(().into()).await {
+            Ok(s) => s,
+            Err(_) => {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(serde_json::json!({"message": "Failed to route"})),
+                );
+            }
+        };
+
+        while let Some(resp) = stream.next().await {
+            all_responses.push(resp);
+        }
+    }
+
+    return (
+        StatusCode::OK,
+        Json(serde_json::json!({
+            "responses": all_responses
+        })),
+    );
+}

lib/llm/src/http/service/service_v2.rs

@@ -325,6 +325,7 @@ impl HttpServiceConfigBuilder {
             super::openai::list_models_router(state.clone(), var(HTTP_SVC_MODELS_PATH_ENV).ok()),
             super::health::health_check_router(state.clone(), var(HTTP_SVC_HEALTH_PATH_ENV).ok()),
             super::health::live_check_router(state.clone(), var(HTTP_SVC_LIVE_PATH_ENV).ok()),
+            super::dynamic_endpoint::dynamic_endpoint_router(state.clone(), None),
         ];
 
         let endpoint_routes =

lib/runtime/src/component.rs

@@ -101,6 +101,8 @@ pub struct Instance {
     pub namespace: String,
     pub instance_id: i64,
     pub transport: TransportType,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub http_endpoint_path: Option<String>,
 }
 
 impl Instance {
@@ -463,7 +465,7 @@ impl Endpoint {
         .expect("Endpoint name and component name should be valid")
     }
 
-    /// The fully path of an instance in etcd
+    /// The full path of an instance in etcd
     pub fn etcd_path_with_lease_id(&self, lease_id: i64) -> String {
         format!("{INSTANCE_ROOT_PATH}/{}", self.unique_path(lease_id))
     }

lib/runtime/src/component/endpoint.rs

@@ -44,6 +44,10 @@ pub struct EndpointConfig {
     #[educe(Debug(ignore))]
     #[builder(default, setter(into, strip_option))]
     health_check_payload: Option<serde_json::Value>,
+
+    /// Expose this endpoint over HTTP at this path
+    #[builder(default, setter(into, strip_option))]
+    http_endpoint_path: Option<String>,
 }
 
 impl EndpointConfigBuilder {
@@ -67,6 +71,7 @@ impl EndpointConfigBuilder {
             metrics_labels,
             graceful_shutdown,
             health_check_payload,
+            http_endpoint_path,
         ) = self.build_internal()?.dissolve();
         let lease = lease.or(endpoint.drt().primary_lease());
         let lease_id = lease.as_ref().map(|l| l.id()).unwrap_or(0);
@@ -128,6 +133,7 @@ impl EndpointConfigBuilder {
         let subject = endpoint.subject_to(lease_id);
         let etcd_path = endpoint.etcd_path_with_lease_id(lease_id);
         let etcd_client = endpoint.component.drt.etcd_client.clone();
+        let http_endpoint_path = http_endpoint_path.clone();
 
         // Register health check target in SystemHealth if provided
         if let Some(health_check_payload) = &health_check_payload {
@@ -137,6 +143,7 @@ impl EndpointConfigBuilder {
                 namespace: namespace_name.clone(),
                 instance_id: lease_id,
                 transport: TransportType::NatsTcp(subject.clone()),
+                http_endpoint_path: http_endpoint_path.clone(),
             };
             tracing::debug!(subject = %subject, "Registering endpoint health check target");
             let guard = system_health.lock().unwrap();
@@ -230,6 +237,7 @@ impl EndpointConfigBuilder {
             namespace: namespace_name,
             instance_id: lease_id,
             transport: TransportType::NatsTcp(subject),
+            http_endpoint_path: http_endpoint_path,
         };
 
         let info = serde_json::to_vec_pretty(&info)?;