[model-gateway] move python to binding folder (#13295)

sgl-router/bindings/python/.coveragerc

@@ -0,0 +1,6 @@
+[run]
+source = sglang_router
+omit = */mini_lb.py
+
+[report]
+fail_under = 80

sgl-router/bindings/python/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "sglang-router-python"
+version = "0.2.2"
+edition = "2021"
+
+[lib]
+name = "sglang_router_rs"
+crate-type = ["cdylib"]
+
+[dependencies]
+pyo3 = { version = "0.27.1", features = ["extension-module", "abi3-py38"] }
+tokio = { version = "1.42.0", features = ["full"] }
+
+[dependencies.sglang-router]
+path = "../.."
+default-features = true
+package = "sglang-router"
+
+[features]
+default = ["pyo3/extension-module"]
+vendored-openssl = ["sglang-router/vendored-openssl"]
+
+[profile.ci]
+inherits = "release"
+opt-level = 2       # Lighter optimization (still fast runtime, much faster compile)
+lto = "thin"        # Thin LTO - good balance
+codegen-units = 16  # More parallelization for faster builds
+strip = true

sgl-router/bindings/python/MANIFEST.in

@@ -0,0 +1,9 @@
+# Must include:
+include Cargo.toml              # Python bindings Cargo configuration
+include ../../Cargo.toml        # Main Rust project configuration
+include ../../build.rs          # Build script for protobuf generation
+include ../../LICENSE
+recursive-include src *.rs      # Python bindings wrapper
+recursive-include ../../src *.rs  # Main Rust source files
+recursive-include ../../src/proto *.proto  # Protobuf definitions
+recursive-include sglang_router *.py  # Python source files

sgl-router/bindings/python/README.md

@@ -0,0 +1,71 @@
+# SGLang Model Gateway Python Bindings
+
+This directory contains the Python bindings for the SGLang Router, built using [maturin](https://github.com/PyO3/maturin) and [PyO3](https://github.com/PyO3/pyo3).
+
+## Directory Structure
+
+```
+bindings/python/
+├── src/                    # Rust source code for Python bindings
+│   └── lib.rs              # PyO3 bindings implementation
+├── sglang_router/          # Python source code
+│   ├── __init__.py
+│   ├── version.py
+│   ├── launch_server.py
+│   ├── launch_router.py
+│   ├── router.py
+│   ├── router_args.py
+│   └── mini_lb.py
+├── Cargo.toml              # Rust package configuration for bindings
+├── pyproject.toml          # Python package configuration
+├── setup.py                # Setup configuration
+├── MANIFEST.in             # Package manifest
+├── .coveragerc             # Test coverage configuration
+└── README.md               # This file
+```
+
+## Building
+
+### Development Build
+
+```bash
+# Install maturin
+pip install maturin
+
+# Build and install in development mode
+cd sgl-router/bindings/python
+maturin develop --features vendored-openssl
+```
+
+### Production Build
+
+```bash
+# Build wheel
+cd sgl-router/bindings/python
+maturin build --release --out dist --features vendored-openssl
+
+# Install the built wheel
+pip install dist/sglang_router-*.whl
+```
+
+## Testing
+
+```bash
+# Run Python tests
+cd sgl-router
+pytest py_test/
+```
+
+## Configuration
+
+- **pyproject.toml**: Defines package metadata, dependencies, and build configuration
+- **python-source**: Set to "." to indicate Python source is in the same directory as pyproject.toml
+- **module-name**: `sglang_router.sglang_router_rs` - the Rust extension module name
+
+## Notes
+
+- The Rust bindings source code is located in `src/lib.rs`
+- The bindings have their own `Cargo.toml` in this directory
+- The main sglang-router library is located in `../../` and is used as a dependency
+- The package includes both Python code and Rust extensions built with PyO3
+- PyO3 types are prefixed with `Py` in Rust but exposed to Python without the prefix using the `name` attribute

sgl-router/bindings/python/pyproject.toml

@@ -0,0 +1,46 @@
+[build-system]
+requires = ["maturin>=1.0,<2.0"]
+build-backend = "maturin"
+
+[project]
+name = "sglang-router"
+version = "0.2.3"
+description = "High-performance Rust-based load balancer for SGLang with multiple routing algorithms and prefill-decode disaggregation support"
+authors = [
+    {name = "Byron Hsu", email = "byronhsu1230@gmail.com"},
+    {name = "Chang Su", email = "mckvtl@gmail.com"},
+    {name = "Keyang Ru", email = "rukeyang@gmail.com"}
+]
+requires-python = ">=3.8"
+readme = "README.md"
+license = { text = "Apache-2.0" }
+classifiers = [
+    "Programming Language :: Python :: Implementation :: CPython",
+    "Programming Language :: Rust",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
+]
+
+dependencies = [
+    "setproctitle",
+    "aiohttp",
+    "orjson",
+    "uvicorn",
+    "fastapi",
+]
+
+[project.optional-dependencies]
+dev = [
+    "requests>=2.25.0",
+]
+
+
+[tool.maturin]
+python-source = "."
+module-name = "sglang_router.sglang_router_rs"

sgl-router/bindings/python/setup.py

@@ -0,0 +1,22 @@
+import os
+
+from setuptools import setup
+
+no_rust = os.environ.get("SGLANG_ROUTER_BUILD_NO_RUST") == "1"
+
+rust_extensions = []
+if not no_rust:
+    from setuptools_rust import Binding, RustExtension
+
+    rust_extensions.append(
+        RustExtension(
+            target="sglang_router_rs",
+            path="Cargo.toml",
+            binding=Binding.PyO3,
+        )
+    )
+
+setup(
+    rust_extensions=rust_extensions,
+    zip_safe=False,
+)

sgl-router/bindings/python/sglang_router/__init__.py

@@ -0,0 +1,3 @@
+from sglang_router.version import __version__
+
+__all__ = ["__version__"]

sgl-router/bindings/python/sglang_router/launch_router.py

@@ -0,0 +1,113 @@
+import argparse
+import logging
+import sys
+from typing import List, Optional
+
+import setproctitle
+from sglang_router.mini_lb import MiniLoadBalancer
+from sglang_router.router_args import RouterArgs
+
+logger = logging.getLogger("router")
+
+try:
+    from sglang_router.router import Router
+except ImportError:
+    Router = None
+    logger.warning(
+        "Rust Router is not installed, only python MiniLB (debugging only) is available"
+    )
+
+
+def launch_router(args: argparse.Namespace) -> Optional[Router]:
+    """
+    Launch the SGLang router with the configuration from parsed arguments.
+
+    Args:
+        args: Namespace object containing router configuration
+            Can be either raw argparse.Namespace or converted RouterArgs
+
+    Returns:
+        Router instance if successful, None if failed
+    """
+    setproctitle.setproctitle("sglang::router")
+    try:
+        # Convert to RouterArgs if needed
+        if not isinstance(args, RouterArgs):
+            router_args = RouterArgs.from_cli_args(args)
+        else:
+            router_args = args
+
+        if router_args.mini_lb:
+            mini_lb = MiniLoadBalancer(router_args)
+            mini_lb.start()
+        else:
+            # TODO: support tracing for router(Rust).
+            del router_args.enable_trace
+            del router_args.otlp_traces_endpoint
+
+            if Router is None:
+                raise RuntimeError("Rust Router is not installed")
+            router_args._validate_router_args()
+            router = Router.from_args(router_args)
+            router.start()
+
+    except Exception as e:
+        logger.error(f"Error starting router: {e}")
+        raise e
+
+
+class CustomHelpFormatter(
+    argparse.RawDescriptionHelpFormatter, argparse.ArgumentDefaultsHelpFormatter
+):
+    """Custom formatter that preserves both description formatting and shows defaults"""
+
+    pass
+
+
+def parse_router_args(args: List[str]) -> RouterArgs:
+    """Parse command line arguments and return RouterArgs instance."""
+    parser = argparse.ArgumentParser(
+        description="""SGLang Router - High-performance request distribution across worker nodes
+
+Usage:
+This launcher enables starting a router with individual worker instances. It is useful for
+multi-node setups or when you want to start workers and router separately.
+
+Examples:
+  # Regular mode
+  python -m sglang_router.launch_router --worker-urls http://worker1:8000 http://worker2:8000
+
+  # PD disaggregated mode with same policy for both
+  python -m sglang_router.launch_router --pd-disaggregation \\
+    --prefill http://prefill1:8000 9000 --prefill http://prefill2:8000 \\
+    --decode http://decode1:8001 --decode http://decode2:8001 \\
+    --policy cache_aware
+
+  # PD mode with optional bootstrap ports
+  python -m sglang_router.launch_router --pd-disaggregation \\
+    --prefill http://prefill1:8000 9000 \\    # With bootstrap port
+    --prefill http://prefill2:8000 none \\    # Explicitly no bootstrap port
+    --prefill http://prefill3:8000 \\         # Defaults to no bootstrap port
+    --decode http://decode1:8001 --decode http://decode2:8001
+
+  # PD mode with different policies for prefill and decode
+  python -m sglang_router.launch_router --pd-disaggregation \\
+    --prefill http://prefill1:8000 --prefill http://prefill2:8000 \\
+    --decode http://decode1:8001 --decode http://decode2:8001 \\
+    --prefill-policy cache_aware --decode-policy power_of_two
+
+    """,
+        formatter_class=CustomHelpFormatter,
+    )
+
+    RouterArgs.add_cli_args(parser, use_router_prefix=False)
+    return RouterArgs.from_cli_args(parser.parse_args(args), use_router_prefix=False)
+
+
+def main() -> None:
+    router_args = parse_router_args(sys.argv[1:])
+    launch_router(router_args)
+
+
+if __name__ == "__main__":
+    main()

sgl-router/bindings/python/sglang_router/launch_server.py

@@ -0,0 +1,213 @@
+import argparse
+import asyncio
+import copy
+import logging
+import multiprocessing as mp
+import os
+import random
+import signal
+import sys
+import time
+from typing import List
+
+import requests
+from setproctitle import setproctitle
+from sglang_router.launch_router import RouterArgs, launch_router
+
+from sglang.srt.server_args import ServerArgs
+from sglang.srt.utils import is_port_available
+
+
+def setup_logger():
+    logger = logging.getLogger("router")
+    logger.setLevel(logging.INFO)
+
+    formatter = logging.Formatter(
+        "[Router (Python)] %(asctime)s - %(levelname)s - %(message)s - %(filename)s:%(lineno)d",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    handler = logging.StreamHandler()
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+
+    return logger
+
+
+logger = setup_logger()
+
+
+# Create new process group
+def run_server(server_args, dp_rank):
+    """
+    Note:
+
+    1. Without os.setpgrp(), all processes share the same PGID. When you press Ctrl+C, the terminal sends SIGINT to all processes in the group simultaneously.
+    This can cause leaf processes to terminate first, which messes up the cleaning order and produces orphaned processes.
+
+    Terminal (PGID=100)
+    └── Main Python Process (PGID=100)
+        └── Server Process 1 (PGID=100)
+            └── Scheduler 1
+            └── Detokenizer 1
+        └── Server Process 2 (PGID=100)
+            └── Scheduler 2
+            └── Detokenizer 2
+
+    2. With os.setpgrp(), the main Python process and its children are in a separate group. Now:
+
+    Terminal (PGID=100)
+    └── Main Python Process (PGID=200)
+        └── Server Process 1 (PGID=300)
+            └── Scheduler 1
+            └── Detokenizer 1
+        └── Server Process 2 (PGID=400)
+            └── Scheduler 2
+            └── Detokenizer 2
+    """
+    # create new process group
+    os.setpgrp()
+
+    setproctitle("sglang::server")
+    # Set SGLANG_DP_RANK environment variable
+    os.environ["SGLANG_DP_RANK"] = str(dp_rank)
+
+    # Launch server in appropriate mode (HTTP or gRPC)
+    if server_args.grpc_mode:
+        from sglang.srt.entrypoints.grpc_server import serve_grpc
+
+        asyncio.run(serve_grpc(server_args))
+    else:
+        from sglang.srt.entrypoints.http_server import launch_server
+
+        launch_server(server_args)
+
+
+def launch_server_process(
+    server_args: ServerArgs, worker_port: int, dp_id: int
+) -> mp.Process:
+    """Launch a single server process with the given args and port."""
+    server_args = copy.deepcopy(server_args)
+    server_args.port = worker_port
+    server_args.base_gpu_id = dp_id * server_args.tp_size
+    server_args.dp_size = 1
+
+    proc = mp.Process(target=run_server, args=(server_args, dp_id))
+    proc.start()
+    return proc
+
+
+def wait_for_server_health(host: str, port: int, timeout: int = 300) -> bool:
+    """Wait for server to be healthy by checking /health endpoint."""
+    start_time = time.perf_counter()
+    url = f"http://{host}:{port}/health"
+
+    while time.perf_counter() - start_time < timeout:
+        try:
+            response = requests.get(url, timeout=5)
+            if response.status_code == 200:
+                return True
+        except requests.exceptions.RequestException:
+            pass
+        time.sleep(1)
+    return False
+
+
+def find_available_ports(base_port: int, count: int) -> List[int]:
+    """Find consecutive available ports starting from base_port."""
+    available_ports = []
+    current_port = base_port
+
+    while len(available_ports) < count:
+        if is_port_available(current_port):
+            available_ports.append(current_port)
+        current_port += random.randint(100, 1000)
+
+    return available_ports
+
+
+def cleanup_processes(processes: List[mp.Process]):
+    for process in processes:
+        logger.info(f"Terminating process group {process.pid}")
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+        except ProcessLookupError:
+            # Process group may already be terminated
+            pass
+
+    # Wait for processes to terminate
+    for process in processes:
+        process.join(timeout=5)
+        if process.is_alive():
+            logger.warning(
+                f"Process {process.pid} did not terminate gracefully, forcing kill"
+            )
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+
+    logger.info("All process groups terminated")
+
+
+def main():
+    # CUDA runtime isn't fork-safe, which can lead to subtle bugs or crashes
+    mp.set_start_method("spawn")
+
+    parser = argparse.ArgumentParser(
+        description="Launch SGLang router and server processes"
+    )
+
+    ServerArgs.add_cli_args(parser)
+    RouterArgs.add_cli_args(parser, use_router_prefix=True, exclude_host_port=True)
+    parser.add_argument(
+        "--router-dp-worker-base-port",
+        type=int,
+        default=31000,
+        help="Base port number for data parallel workers",
+    )
+    # No extra retry/CB flags here; RouterArgs.add_cli_args already defines them with router- prefix
+
+    args = parser.parse_args()
+    server_args = ServerArgs.from_cli_args(args)
+    router_args = RouterArgs.from_cli_args(args, use_router_prefix=True)
+
+    # Find available ports for workers
+    worker_ports = find_available_ports(
+        args.router_dp_worker_base_port, server_args.dp_size
+    )
+
+    # Start server processes
+    server_processes = []
+
+    for i, worker_port in enumerate(worker_ports):
+        logger.info(f"Launching DP server process {i} on port {worker_port}")
+        proc = launch_server_process(server_args, worker_port, i)
+        server_processes.append(proc)
+
+    signal.signal(signal.SIGINT, lambda sig, frame: cleanup_processes(server_processes))
+    signal.signal(
+        signal.SIGTERM, lambda sig, frame: cleanup_processes(server_processes)
+    )
+    signal.signal(
+        signal.SIGQUIT, lambda sig, frame: cleanup_processes(server_processes)
+    )
+
+    # Update router args with worker URLs
+    # Use grpc:// protocol if server is in gRPC mode, otherwise http://
+    protocol = "grpc" if server_args.grpc_mode else "http"
+    router_args.worker_urls = [
+        f"{protocol}://{server_args.host}:{port}" for port in worker_ports
+    ]
+
+    # Start the router
+    try:
+        launch_router(router_args)
+    except Exception as e:
+        logger.error(f"Failed to start router: {e}")
+        cleanup_processes(server_processes)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

sgl-router/bindings/python/sglang_router/mini_lb.py

@@ -0,0 +1,488 @@
+"""
+Minimal HTTP load balancer for prefill and decode servers for testing.
+"""
+
+import asyncio
+import ipaddress
+import logging
+import random
+import urllib
+from http import HTTPStatus
+from itertools import chain
+from typing import Optional
+
+import aiohttp
+import orjson
+import uvicorn
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import ORJSONResponse, Response, StreamingResponse
+from sglang_router.router_args import RouterArgs
+
+try:
+    from sglang.srt.tracing.trace import (
+        process_tracing_init,
+        trace_get_remote_propagate_context,
+        trace_req_finish,
+        trace_req_start,
+        trace_set_thread_info,
+        trace_slice_end,
+        trace_slice_start,
+    )
+
+    trace_package_imported = True
+except ImportError:
+    trace_package_imported = False
+
+logger = logging.getLogger(__name__)
+
+AIOHTTP_STREAM_READ_CHUNK_SIZE = (
+    1024 * 64
+)  # 64KB, to prevent aiohttp's "Chunk too big" error
+
+
+def maybe_wrap_ipv6_address(address: str) -> str:
+    try:
+        ipaddress.IPv6Address(address)
+        return f"[{address}]"
+    except ValueError:
+        return address
+
+
+class MiniLoadBalancer:
+    def __init__(
+        self,
+        router_args: RouterArgs,
+    ):
+        self._validate_router_args(router_args)
+
+        self.host = router_args.host
+        self.port = router_args.port
+        self.timeout = router_args.request_timeout_secs
+        self.prefill_urls = [url[0] for url in router_args.prefill_urls]
+        self.prefill_bootstrap_ports = [url[1] for url in router_args.prefill_urls]
+        self.decode_urls = router_args.decode_urls
+        self.otlp_traces_endpoint = router_args.otlp_traces_endpoint
+        self.enable_trace = router_args.enable_trace
+        if self.enable_trace and not trace_package_imported:
+            logger.warning(
+                "Tracing is not supported in this environment. Please install sglang."
+            )
+            self.enable_trace = False
+
+    def _validate_router_args(self, router_args: RouterArgs):
+        logger.warning(
+            "\x1b[33mMiniLB is only for debugging purposes, it only supports random policy!\033[0m"
+        )
+
+        # NOTE: too many arguments unsupported, just validate some important ones
+        if router_args.policy != "random":
+            logger.warning("[MiniLB] Overriding policy to random")
+            router_args.policy = "random"
+
+        if not router_args.pd_disaggregation:
+            raise ValueError("MiniLB only supports PD disaggregation mode")
+
+        if len(router_args.prefill_urls) == 0 or len(router_args.decode_urls) == 0:
+            raise ValueError(
+                "MiniLB requires at least one prefill and one decode server"
+            )
+
+    def start(self):
+        global lb
+        lb = self
+        if self.enable_trace:
+            process_tracing_init(self.otlp_traces_endpoint, "sglang")
+            trace_set_thread_info("Mini lb")
+        uvicorn.run(app, host=self.host, port=self.port)
+
+    def select_pair(self):
+        assert len(self.prefill_urls) > 0, "No prefill servers available"
+        assert len(self.decode_urls) > 0, "No decode servers available"
+        pidx = random.randint(0, len(self.prefill_urls) - 1)
+        didx = random.randint(0, len(self.decode_urls) - 1)
+        return (
+            self.prefill_urls[pidx],
+            self.prefill_bootstrap_ports[pidx],
+            self.decode_urls[didx],
+        )
+
+    async def generate(
+        self, modified_request, prefill_server, decode_server, endpoint
+    ) -> ORJSONResponse:
+        assert endpoint[0] != "/", f"Endpoint should not start with '/': {endpoint}"
+
+        async with aiohttp.ClientSession(
+            timeout=aiohttp.ClientTimeout(
+                total=self.timeout
+            )  # Add timeout for request reliability
+        ) as session:
+            headers = {}
+            bootstrap_room_list = []
+            if self.enable_trace:
+                bootstrap_room_list = (
+                    modified_request["bootstrap_room"]
+                    if isinstance(modified_request["bootstrap_room"], list)
+                    else [modified_request["bootstrap_room"]]
+                )
+                trace_context = trace_get_remote_propagate_context(bootstrap_room_list)
+                headers = {"trace_context": trace_context}
+
+            tasks = [
+                session.post(
+                    f"{prefill_server}/{endpoint}",
+                    json=modified_request,
+                    headers=headers,
+                ),
+                session.post(
+                    f"{decode_server}/{endpoint}",
+                    json=modified_request,
+                    headers=headers,
+                ),
+            ]
+
+            for bootstrap_room in bootstrap_room_list:
+                trace_slice_end("mini_lb_launch", bootstrap_room, auto_next_anon=True)
+
+            # Wait for both responses to complete. Prefill should end first.
+            prefill_response, decode_response = await asyncio.gather(*tasks)
+
+            if "return_logprob" in modified_request:
+
+                prefill_json = await prefill_response.json()
+                ret_json = await decode_response.json()
+
+                # merge `meta_info.input_token_logprobs` from prefill to decode
+                if "meta_info" in ret_json:
+                    if "input_token_logprobs" in ret_json["meta_info"]:
+                        ret_json["meta_info"]["input_token_logprobs"] = (
+                            prefill_json["meta_info"]["input_token_logprobs"]
+                            + ret_json["meta_info"]["input_token_logprobs"]
+                        )
+            else:
+                ret_json = await decode_response.json()
+
+            for bootstrap_room in bootstrap_room_list:
+                trace_slice_end(
+                    "wait_PD_finish",
+                    bootstrap_room,
+                    thread_finish_flag=True,
+                )
+                trace_req_finish(bootstrap_room)
+
+            return ORJSONResponse(
+                content=ret_json,
+                status_code=decode_response.status,
+            )
+
+    async def generate_stream(
+        self, modified_request, prefill_server, decode_server, endpoint="generate"
+    ):
+        assert endpoint[0] != "/", f"Endpoint should not start with '/': {endpoint}"
+
+        async def stream_results():
+            async with aiohttp.ClientSession(
+                timeout=aiohttp.ClientTimeout(
+                    total=self.timeout
+                )  # Add timeout for request reliability
+            ) as session:
+                # Create the tasks for both prefill and decode requests
+                headers = {}
+                bootstrap_room_list = []
+                if self.enable_trace:
+                    bootstrap_room_list = (
+                        modified_request["bootstrap_room"]
+                        if isinstance(modified_request["bootstrap_room"], list)
+                        else [modified_request["bootstrap_room"]]
+                    )
+                    trace_context = trace_get_remote_propagate_context(
+                        bootstrap_room_list
+                    )
+                    headers = {"trace_context": trace_context}
+
+                tasks = [
+                    session.post(
+                        f"{prefill_server}/{endpoint}",
+                        json=modified_request,
+                        headers=headers,
+                    ),
+                    session.post(
+                        f"{decode_server}/{endpoint}",
+                        json=modified_request,
+                        headers=headers,
+                    ),
+                ]
+
+                for bootstrap_room in bootstrap_room_list:
+                    trace_slice_end(
+                        "mini_lb_launch", bootstrap_room, auto_next_anon=True
+                    )
+                # Wait for both responses to complete. Since this is streaming, they return immediately.
+                prefill_response, decode_response = await asyncio.gather(*tasks)
+
+                if modified_request.get("return_logprob", False):
+                    prefill_chunks = []
+                    async for chunk in prefill_response.content:
+                        prefill_chunks.append(chunk)
+
+                    first_prefill_chunk = (
+                        prefill_chunks[0].decode("utf-8")[5:].strip("\n")
+                    )
+                    first_prefill_chunk_json = orjson.loads(first_prefill_chunk)
+
+                    async for chunk in decode_response.content:
+                        # Note: This is inefficient
+                        # merge prefill input_token_logprobs, output_token_logprobs to decode
+                        decoded_chunk = chunk.decode("utf-8")
+                        if (
+                            decoded_chunk
+                            and decoded_chunk.startswith("data:")
+                            and "[DONE]" not in decoded_chunk
+                        ):
+                            ret_json = orjson.loads(decoded_chunk[5:].strip("\n"))
+                            ret_json["meta_info"]["input_token_logprobs"] = (
+                                first_prefill_chunk_json["meta_info"][
+                                    "input_token_logprobs"
+                                ]
+                                + ret_json["meta_info"]["input_token_logprobs"]
+                            )
+
+                            yield b"data: " + orjson.dumps(ret_json) + b"\n\n"
+                        else:
+                            yield chunk
+                else:
+                    async for chunk in decode_response.content.iter_chunked(
+                        AIOHTTP_STREAM_READ_CHUNK_SIZE
+                    ):
+                        yield chunk
+
+            for bootstrap_room in bootstrap_room_list:
+                trace_slice_end(
+                    "wait_PD_finish",
+                    bootstrap_room,
+                    thread_finish_flag=True,
+                )
+                trace_req_finish(bootstrap_room)
+
+        return StreamingResponse(
+            stream_results(),
+            media_type="text/event-stream",
+        )
+
+
+app = FastAPI()
+lb: Optional[MiniLoadBalancer] = None
+
+
+@app.get("/health")
+async def health_check():
+    return Response(status_code=200)
+
+
+@app.get("/health_generate")
+async def health_generate():
+    async with aiohttp.ClientSession() as session:
+        # Create the tasks
+        tasks = []
+        for server in chain(lb.prefill_urls, lb.decode_urls):
+            tasks.append(session.get(f"{server}/health_generate"))
+        for i, response in enumerate(asyncio.as_completed(tasks)):
+            await response
+    return Response(status_code=200)
+
+
+@app.post("/flush_cache")
+async def flush_cache():
+    async with aiohttp.ClientSession() as session:
+        # Create the tasks
+        tasks = []
+        for server in chain(lb.prefill_urls, lb.decode_urls):
+            tasks.append(session.post(f"{server}/flush_cache"))
+        for i, response in enumerate(asyncio.as_completed(tasks)):
+            await response
+    return Response(status_code=200)
+
+
+@app.get("/get_server_info")
+async def get_server_info():
+    prefill_infos = []
+    decode_infos = []
+    all_internal_states = []
+
+    async with aiohttp.ClientSession() as session:
+        for server in lb.prefill_urls:
+            server_info = await session.get(f"{server}/get_server_info")
+            prefill_infos.append(await server_info.json())
+        for server in lb.decode_urls:
+            server_info = await session.get(f"{server}/get_server_info")
+            info_json = await server_info.json()
+            decode_infos.append(info_json)
+            # Extract internal_states from decode servers
+            if "internal_states" in info_json:
+                all_internal_states.extend(info_json["internal_states"])
+
+    # Return format expected by bench_one_batch_server.py
+    if all_internal_states:
+        return {
+            "internal_states": all_internal_states,
+            "prefill": prefill_infos,
+            "decode": decode_infos,
+        }
+    else:
+        # Fallback with dummy data if no internal states found
+        return {
+            "internal_states": [
+                {
+                    "last_gen_throughput": 0.0,
+                    "avg_spec_accept_length": None,
+                }
+            ],
+            "prefill": prefill_infos,
+            "decode": decode_infos,
+        }
+
+
+@app.get("/get_model_info")
+async def get_model_info():
+    if not lb or not lb.prefill_urls:
+        raise HTTPException(
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE,
+            detail="There is no server registered",
+        )
+
+    target_server_url = lb.prefill_urls[0]
+    endpoint_url = f"{target_server_url}/get_model_info"
+
+    async with aiohttp.ClientSession() as session:
+        try:
+            async with session.get(endpoint_url) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    raise HTTPException(
+                        status_code=HTTPStatus.BAD_GATEWAY,
+                        detail=(
+                            f"Failed to get model info from {target_server_url}"
+                            f"Status: {response.status}, Response: {error_text}"
+                        ),
+                    )
+
+                model_info_json = await response.json()
+                return ORJSONResponse(content=model_info_json)
+
+        except aiohttp.ClientError as e:
+            raise HTTPException(
+                status_code=HTTPStatus.SERVICE_UNAVAILABLE,
+                detail=f"Failed to get model info from backend",
+            )
+
+
+@app.post("/generate")
+async def handle_generate_request(request_data: dict):
+    prefill_server, bootstrap_port, decode_server = lb.select_pair()
+
+    # Parse and transform prefill_server for bootstrap data
+    parsed_url = urllib.parse.urlparse(prefill_server)
+    hostname = maybe_wrap_ipv6_address(parsed_url.hostname)
+    modified_request = request_data.copy()
+
+    batch_size = _get_request_batch_size(modified_request)
+    if batch_size is not None:
+        modified_request.update(
+            {
+                "bootstrap_host": [hostname] * batch_size,
+                "bootstrap_port": [bootstrap_port] * batch_size,
+                "bootstrap_room": [
+                    _generate_bootstrap_room() for _ in range(batch_size)
+                ],
+            }
+        )
+    else:
+        modified_request.update(
+            {
+                "bootstrap_host": hostname,
+                "bootstrap_port": bootstrap_port,
+                "bootstrap_room": _generate_bootstrap_room(),
+            }
+        )
+
+    if request_data.get("stream", False):
+        return await lb.generate_stream(
+            modified_request, prefill_server, decode_server, "generate"
+        )
+    else:
+        return await lb.generate(
+            modified_request, prefill_server, decode_server, "generate"
+        )
+
+
+async def _forward_to_backend(request_data: dict, endpoint_name: str):
+    prefill_server, bootstrap_port, decode_server = lb.select_pair()
+
+    # Parse and transform prefill_server for bootstrap data
+    parsed_url = urllib.parse.urlparse(prefill_server)
+    hostname = maybe_wrap_ipv6_address(parsed_url.hostname)
+    modified_request = request_data.copy()
+    modified_request.update(
+        {
+            "bootstrap_host": hostname,
+            "bootstrap_port": bootstrap_port,
+            "bootstrap_room": _generate_bootstrap_room(),
+        }
+    )
+
+    if request_data.get("stream", False):
+        return await lb.generate_stream(
+            modified_request,
+            prefill_server,
+            decode_server,
+            endpoint=endpoint_name,
+        )
+    else:
+        return await lb.generate(
+            modified_request,
+            prefill_server,
+            decode_server,
+            endpoint=endpoint_name,
+        )
+
+
+@app.post("/v1/chat/completions")
+async def handle_chat_completion_request(request_data: dict):
+    return await _forward_to_backend(request_data, "v1/chat/completions")
+
+
+@app.post("/v1/completions")
+async def handle_completion_request(request_data: dict):
+    return await _forward_to_backend(request_data, "v1/completions")
+
+
+def _generate_bootstrap_room():
+    bootstrap_room = random.randint(0, 2**63 - 1)
+    if lb.enable_trace:
+        trace_req_start(bootstrap_room, bootstrap_room, role="router")
+        trace_slice_start("mini_lb_launch", bootstrap_room)
+    return bootstrap_room
+
+
+# We may utilize `GenerateReqInput`'s logic later
+def _get_request_batch_size(request):
+    if (text := request.get("text")) is not None:
+        return None if isinstance(text, str) else len(text)
+    if (input_ids := request.get("input_ids")) is not None:
+        return None if isinstance(input_ids[0], int) else len(input_ids)
+    return None
+
+
+@app.get("/v1/models")
+async def get_models():
+    prefill_server = lb.prefill_urls[0]  # Get the first prefill server
+    async with aiohttp.ClientSession() as session:
+        try:
+            response = await session.get(f"{prefill_server}/v1/models")
+            if response.status != 200:
+                raise HTTPException(
+                    status_code=response.status,
+                    detail=f"Prefill server error: Status {response.status}",
+                )
+            return ORJSONResponse(content=await response.json())
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))

sgl-router/bindings/python/sglang_router/router.py

@@ -0,0 +1,224 @@
+from typing import Optional
+
+from sglang_router.router_args import RouterArgs
+from sglang_router.sglang_router_rs import (
+    BackendType,
+    HistoryBackendType,
+    PolicyType,
+    PyOracleConfig,
+    PyPostgresConfig,
+)
+from sglang_router.sglang_router_rs import Router as _Router
+
+
+def policy_from_str(policy_str: Optional[str]) -> PolicyType:
+    """Convert policy string to PolicyType enum."""
+    if policy_str is None:
+        return None
+    policy_map = {
+        "random": PolicyType.Random,
+        "round_robin": PolicyType.RoundRobin,
+        "cache_aware": PolicyType.CacheAware,
+        "power_of_two": PolicyType.PowerOfTwo,
+        "bucket": PolicyType.Bucket,
+    }
+    return policy_map[policy_str]
+
+
+def backend_from_str(backend_str: Optional[str]) -> BackendType:
+    """Convert backend string to BackendType enum."""
+    if isinstance(backend_str, BackendType):
+        return backend_str
+    if backend_str is None:
+        return BackendType.Sglang
+    backend_map = {"sglang": BackendType.Sglang, "openai": BackendType.Openai}
+    backend_lower = backend_str.lower()
+    if backend_lower not in backend_map:
+        raise ValueError(
+            f"Unknown backend: {backend_str}. Valid options: {', '.join(backend_map.keys())}"
+        )
+    return backend_map[backend_lower]
+
+
+def history_backend_from_str(backend_str: Optional[str]) -> HistoryBackendType:
+    """Convert history backend string to HistoryBackendType enum."""
+    if isinstance(backend_str, HistoryBackendType):
+        return backend_str
+    if backend_str is None:
+        return HistoryBackendType.Memory
+    backend_lower = backend_str.lower()
+    if backend_lower == "memory":
+        return HistoryBackendType.Memory
+    elif backend_lower == "none":
+        # Use getattr to access 'None' which is a Python keyword
+        return getattr(HistoryBackendType, "None")
+    elif backend_lower == "oracle":
+        return HistoryBackendType.Oracle
+    elif backend_lower == "postgres":
+        return HistoryBackendType.Postgres
+    else:
+        raise ValueError(f"Unknown history backend: {backend_str}")
+
+
+class Router:
+    """
+    A high-performance router for distributing requests across worker nodes.
+
+    Args:
+        worker_urls: List of URLs for worker nodes that will handle requests. Each URL should include
+            the protocol, host, and port (e.g., ['http://worker1:8000', 'http://worker2:8000'])
+        policy: Load balancing policy to use. Options:
+            - PolicyType.Random: Randomly select workers
+            - PolicyType.RoundRobin: Distribute requests in round-robin fashion
+            - PolicyType.CacheAware: Distribute requests based on cache state and load balance
+            - PolicyType.PowerOfTwo: Select best of two random workers based on load (PD mode only)
+        host: Host address to bind the router server. Supports IPv4, IPv6 (e.g., ::, ::1), or 0.0.0.0 for all interfaces. Default: '0.0.0.0'
+        port: Port number to bind the router server. Default: 3001
+        worker_startup_timeout_secs: Timeout in seconds for worker startup. Default: 300
+        worker_startup_check_interval: Interval in seconds between checks for worker initialization. Default: 10
+        cache_threshold: Cache threshold (0.0-1.0) for cache-aware routing. Routes to cached worker
+            if the match rate exceeds threshold, otherwise routes to the worker with the smallest
+            tree. Default: 0.5
+        balance_abs_threshold: Load balancing is triggered when (max_load - min_load) > abs_threshold
+            AND max_load > min_load * rel_threshold. Otherwise, use cache aware. Default: 32
+        balance_rel_threshold: Load balancing is triggered when (max_load - min_load) > abs_threshold
+            AND max_load > min_load * rel_threshold. Otherwise, use cache aware. Default: 1.0001
+        eviction_interval_secs: Interval in seconds between cache eviction operations in cache-aware
+            routing. Default: 60
+        max_payload_size: Maximum payload size in bytes. Default: 256MB
+        max_tree_size: Maximum size of the approximation tree for cache-aware routing. Default: 2^24
+        dp_aware: Enable data parallelism aware schedule. Default: False
+        enable_igw: Enable IGW (Inference-Gateway) mode for multi-model support. When enabled,
+            the router can manage multiple models simultaneously with per-model load balancing
+            policies. Default: False
+        api_key: The api key used for the authorization with the worker.
+            Useful when the dp aware scheduling strategy is enabled.
+            Default: None
+        log_dir: Directory to store log files. If None, logs are only output to console. Default: None
+        log_level: Logging level. Options: 'debug', 'info', 'warn', 'error'.
+        service_discovery: Enable Kubernetes service discovery. When enabled, the router will
+            automatically discover worker pods based on the selector. Default: False
+        selector: Dictionary mapping of label keys to values for Kubernetes pod selection.
+            Example: {"app": "sglang-worker"}. Default: {}
+        service_discovery_port: Port to use for service discovery. The router will generate
+            worker URLs using this port. Default: 80
+        service_discovery_namespace: Kubernetes namespace to watch for pods. If not provided,
+            watches pods across all namespaces (requires cluster-wide permissions). Default: None
+        prefill_selector: Dictionary mapping of label keys to values for Kubernetes pod selection
+            for prefill servers (PD mode only). Default: {}
+        decode_selector: Dictionary mapping of label keys to values for Kubernetes pod selection
+            for decode servers (PD mode only). Default: {}
+        prometheus_port: Port to expose Prometheus metrics. Default: None
+        prometheus_host: Host address to bind the Prometheus metrics server. Default: None
+        pd_disaggregation: Enable PD (Prefill-Decode) disaggregated mode. Default: False
+        prefill_urls: List of (url, bootstrap_port) tuples for prefill servers (PD mode only)
+        decode_urls: List of URLs for decode servers (PD mode only)
+        prefill_policy: Specific load balancing policy for prefill nodes (PD mode only).
+            If not specified, uses the main policy. Default: None
+        decode_policy: Specific load balancing policy for decode nodes (PD mode only).
+            If not specified, uses the main policy. Default: None
+        request_id_headers: List of HTTP headers to check for request IDs. If not specified,
+            uses common defaults: ['x-request-id', 'x-correlation-id', 'x-trace-id', 'request-id'].
+            Example: ['x-my-request-id', 'x-custom-trace-id']. Default: None
+        bootstrap_port_annotation: Kubernetes annotation name for bootstrap port (PD mode).
+            Default: 'sglang.ai/bootstrap-port'
+        request_timeout_secs: Request timeout in seconds. Default: 600
+        max_concurrent_requests: Maximum number of concurrent requests allowed for rate limiting. Default: 256
+        queue_size: Queue size for pending requests when max concurrent limit reached (0 = no queue, return 429 immediately). Default: 100
+        queue_timeout_secs: Maximum time (in seconds) a request can wait in queue before timing out. Default: 60
+        rate_limit_tokens_per_second: Token bucket refill rate (tokens per second). If not set, defaults to max_concurrent_requests. Default: None
+        cors_allowed_origins: List of allowed origins for CORS. Empty list allows all origins. Default: []
+        health_failure_threshold: Number of consecutive health check failures before marking worker unhealthy. Default: 3
+        health_success_threshold: Number of consecutive health check successes before marking worker healthy. Default: 2
+        health_check_timeout_secs: Timeout in seconds for health check requests. Default: 5
+        health_check_interval_secs: Interval in seconds between runtime health checks. Default: 60
+        health_check_endpoint: Health check endpoint path. Default: '/health'
+        model_path: Model path for loading tokenizer (HuggingFace model ID or local path). Default: None
+        tokenizer_path: Explicit tokenizer path (overrides model_path tokenizer if provided). Default: None
+    """
+
+    def __init__(self, router: _Router):
+        self._router = router
+
+    @staticmethod
+    def from_args(args: RouterArgs) -> "Router":
+        """Create a router from a RouterArgs instance."""
+
+        args_dict = vars(args)
+        # Convert RouterArgs to _Router parameters
+        args_dict["worker_urls"] = (
+            []
+            if args_dict["service_discovery"] or args_dict["pd_disaggregation"]
+            else args_dict["worker_urls"]
+        )
+        args_dict["policy"] = policy_from_str(args_dict["policy"])
+        args_dict["prefill_urls"] = (
+            args_dict["prefill_urls"] if args_dict["pd_disaggregation"] else None
+        )
+        args_dict["decode_urls"] = (
+            args_dict["decode_urls"] if args_dict["pd_disaggregation"] else None
+        )
+        args_dict["prefill_policy"] = policy_from_str(args_dict["prefill_policy"])
+        args_dict["decode_policy"] = policy_from_str(args_dict["decode_policy"])
+
+        # Convert backend
+        args_dict["backend"] = backend_from_str(args_dict.get("backend"))
+
+        # Convert history_backend to enum first
+        history_backend_raw = args_dict.get("history_backend", "memory")
+        history_backend = history_backend_from_str(history_backend_raw)
+
+        # Convert Oracle config if needed
+        oracle_config = None
+        if history_backend == HistoryBackendType.Oracle:
+            # Prioritize TNS alias over connect descriptor
+            tns_alias = args_dict.get("oracle_tns_alias")
+            connect_descriptor = args_dict.get("oracle_connect_descriptor")
+
+            # Use TNS alias if provided, otherwise use connect descriptor
+            final_descriptor = tns_alias if tns_alias else connect_descriptor
+
+            oracle_config = PyOracleConfig(
+                password=args_dict.get("oracle_password"),
+                username=args_dict.get("oracle_username"),
+                connect_descriptor=final_descriptor,
+                wallet_path=args_dict.get("oracle_wallet_path"),
+                pool_min=args_dict.get("oracle_pool_min", 1),
+                pool_max=args_dict.get("oracle_pool_max", 16),
+                pool_timeout_secs=args_dict.get("oracle_pool_timeout_secs", 30),
+            )
+        args_dict["oracle_config"] = oracle_config
+        args_dict["history_backend"] = history_backend
+
+        # Convert Postgres config if needed
+        postgres_config = None
+        if history_backend == HistoryBackendType.Postgres:
+            postgres_config = PyPostgresConfig(
+                db_url=args_dict.get("postgres_db_url"),
+                pool_max=args_dict.get("postgres_pool_max", 16),
+            )
+        args_dict["postgres_config"] = postgres_config
+
+        # Remove fields that shouldn't be passed to Rust Router constructor
+        fields_to_remove = [
+            "mini_lb",
+            "oracle_wallet_path",
+            "oracle_tns_alias",
+            "oracle_connect_descriptor",
+            "oracle_username",
+            "oracle_password",
+            "oracle_pool_min",
+            "oracle_pool_max",
+            "oracle_pool_timeout_secs",
+        ]
+        for field in fields_to_remove:
+            args_dict.pop(field, None)
+
+        return Router(_Router(**args_dict))
+
+    def start(self) -> None:
+        """Start the router server.
+
+        This method blocks until the server is shut down.
+        """
+        self._router.start()

sgl-router/bindings/python/sglang_router/router_args.py

@@ -0,0 +1,763 @@
+import argparse
+import dataclasses
+import logging
+import os
+from typing import Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclasses.dataclass
+class RouterArgs:
+    # Worker configuration
+    worker_urls: List[str] = dataclasses.field(default_factory=list)
+    host: str = "0.0.0.0"
+    port: int = 30000
+
+    # PD-specific configuration
+    mini_lb: bool = False
+    pd_disaggregation: bool = False  # Enable PD disaggregated mode
+    prefill_urls: List[tuple] = dataclasses.field(
+        default_factory=list
+    )  # List of (url, bootstrap_port)
+    decode_urls: List[str] = dataclasses.field(default_factory=list)
+
+    # Routing policy
+    policy: str = "cache_aware"
+    prefill_policy: Optional[str] = None  # Specific policy for prefill nodes in PD mode
+    decode_policy: Optional[str] = None  # Specific policy for decode nodes in PD mode
+    worker_startup_timeout_secs: int = 600
+    worker_startup_check_interval: int = 30
+    cache_threshold: float = 0.3
+    balance_abs_threshold: int = 64
+    balance_rel_threshold: float = 1.5
+    eviction_interval_secs: int = 120
+    max_tree_size: int = 2**26
+    max_payload_size: int = 512 * 1024 * 1024  # 512MB default for large batches
+    bucket_adjust_interval_secs: int = 5
+    dp_aware: bool = False
+    enable_igw: bool = False  # Enable IGW (Inter-Gateway) mode for multi-model support
+    api_key: Optional[str] = None
+    log_dir: Optional[str] = None
+    log_level: Optional[str] = None
+    # Service discovery configuration
+    service_discovery: bool = False
+    selector: Dict[str, str] = dataclasses.field(default_factory=dict)
+    service_discovery_port: int = 80
+    service_discovery_namespace: Optional[str] = None
+    # PD service discovery configuration
+    prefill_selector: Dict[str, str] = dataclasses.field(default_factory=dict)
+    decode_selector: Dict[str, str] = dataclasses.field(default_factory=dict)
+    bootstrap_port_annotation: str = "sglang.ai/bootstrap-port"
+    # Prometheus configuration
+    prometheus_port: Optional[int] = None
+    prometheus_host: Optional[str] = None
+    # Request ID headers configuration
+    request_id_headers: Optional[List[str]] = None
+    # Request timeout in seconds
+    request_timeout_secs: int = 1800
+    # Max concurrent requests for rate limiting (-1 to disable)
+    max_concurrent_requests: int = -1
+    # Queue size for pending requests when max concurrent limit reached
+    queue_size: int = 100
+    # Maximum time (in seconds) a request can wait in queue before timing out
+    queue_timeout_secs: int = 60
+    # Token bucket refill rate (tokens per second). If not set, defaults to max_concurrent_requests
+    rate_limit_tokens_per_second: Optional[int] = None
+    # CORS allowed origins
+    cors_allowed_origins: List[str] = dataclasses.field(default_factory=list)
+    # Retry configuration
+    retry_max_retries: int = 5
+    retry_initial_backoff_ms: int = 50
+    retry_max_backoff_ms: int = 30_000
+    retry_backoff_multiplier: float = 1.5
+    retry_jitter_factor: float = 0.2
+    disable_retries: bool = False
+    # Health check configuration
+    health_failure_threshold: int = 3
+    health_success_threshold: int = 2
+    health_check_timeout_secs: int = 5
+    health_check_interval_secs: int = 60
+    health_check_endpoint: str = "/health"
+    # Circuit breaker configuration
+    cb_failure_threshold: int = 10
+    cb_success_threshold: int = 3
+    cb_timeout_duration_secs: int = 60
+    cb_window_duration_secs: int = 120
+    disable_circuit_breaker: bool = False
+    model_path: Optional[str] = None
+    tokenizer_path: Optional[str] = None
+    chat_template: Optional[str] = None
+    # Tokenizer cache configuration
+    tokenizer_cache_enable_l0: bool = False
+    tokenizer_cache_l0_max_entries: int = 10000
+    tokenizer_cache_enable_l1: bool = False
+    tokenizer_cache_l1_max_memory: int = 50 * 1024 * 1024  # 50MB
+    reasoning_parser: Optional[str] = None
+    tool_call_parser: Optional[str] = None
+    # MCP server configuration
+    mcp_config_path: Optional[str] = None
+    # Backend selection
+    backend: str = "sglang"
+    # History backend configuration
+    history_backend: str = "memory"
+    oracle_wallet_path: Optional[str] = None
+    oracle_tns_alias: Optional[str] = None
+    oracle_connect_descriptor: Optional[str] = None
+    oracle_username: Optional[str] = None
+    oracle_password: Optional[str] = None
+    oracle_pool_min: int = 1
+    oracle_pool_max: int = 16
+    oracle_pool_timeout_secs: int = 30
+    # mTLS configuration for worker communication
+    client_cert_path: Optional[str] = None
+    client_key_path: Optional[str] = None
+    ca_cert_paths: List[str] = dataclasses.field(default_factory=list)
+    # Trace
+    enable_trace: bool = False
+    otlp_traces_endpoint: str = "localhost:4317"
+
+    @staticmethod
+    def add_cli_args(
+        parser: argparse.ArgumentParser,
+        use_router_prefix: bool = False,
+        exclude_host_port: bool = False,
+    ):
+        """
+        Add router-specific arguments to an argument parser.
+
+        Args:
+            parser: The argument parser to add arguments to
+            use_router_prefix: If True, prefix all arguments with 'router-' to avoid conflicts
+            exclude_host_port: If True, don't add host and port arguments (used when inheriting from server)
+        """
+        prefix = "router-" if use_router_prefix else ""
+
+        # Worker configuration
+        if not exclude_host_port:
+            parser.add_argument(
+                "--host",
+                type=str,
+                default=RouterArgs.host,
+                help="Host address to bind the router server. Supports IPv4, IPv6 (e.g., ::, ::1), or 0.0.0.0 for all interfaces",
+            )
+            parser.add_argument(
+                "--port",
+                type=int,
+                default=RouterArgs.port,
+                help="Port number to bind the router server",
+            )
+
+        parser.add_argument(
+            "--worker-urls",
+            type=str,
+            nargs="*",
+            default=[],
+            help="List of worker URLs. Supports IPv4 and IPv6 addresses (use brackets for IPv6, e.g., http://[::1]:8000 http://192.168.1.1:8000)",
+        )
+
+        # Routing policy configuration
+        parser.add_argument(
+            f"--{prefix}policy",
+            type=str,
+            default=RouterArgs.policy,
+            choices=["random", "round_robin", "cache_aware", "power_of_two"],
+            help="Load balancing policy to use. In PD mode, this is used for both prefill and decode unless overridden",
+        )
+        parser.add_argument(
+            f"--{prefix}prefill-policy",
+            type=str,
+            default=None,
+            choices=["random", "round_robin", "cache_aware", "power_of_two", "bucket"],
+            help="Specific policy for prefill nodes in PD mode. If not specified, uses the main policy",
+        )
+        parser.add_argument(
+            f"--{prefix}decode-policy",
+            type=str,
+            default=None,
+            choices=["random", "round_robin", "cache_aware", "power_of_two"],
+            help="Specific policy for decode nodes in PD mode. If not specified, uses the main policy",
+        )
+
+        # PD-specific arguments
+        parser.add_argument(
+            f"--{prefix}mini-lb",
+            action="store_true",
+            help="Enable MiniLB",
+        )
+        parser.add_argument(
+            f"--{prefix}pd-disaggregation",
+            action="store_true",
+            help="Enable PD (Prefill-Decode) disaggregated mode",
+        )
+        parser.add_argument(
+            f"--{prefix}prefill",
+            nargs="+",
+            action="append",
+            help="Prefill server URL and optional bootstrap port. Can be specified multiple times. "
+            "Format: --prefill URL [BOOTSTRAP_PORT]. "
+            "BOOTSTRAP_PORT can be a port number, 'none', or omitted (defaults to none).",
+        )
+        parser.add_argument(
+            f"--{prefix}decode",
+            nargs=1,
+            action="append",
+            metavar=("URL",),
+            help="Decode server URL. Can be specified multiple times.",
+        )
+        parser.add_argument(
+            f"--{prefix}worker-startup-timeout-secs",
+            type=int,
+            default=RouterArgs.worker_startup_timeout_secs,
+            help="Timeout in seconds for worker startup",
+        )
+        parser.add_argument(
+            f"--{prefix}worker-startup-check-interval",
+            type=int,
+            default=RouterArgs.worker_startup_check_interval,
+            help="Interval in seconds between checks for worker startup",
+        )
+        parser.add_argument(
+            f"--{prefix}cache-threshold",
+            type=float,
+            default=RouterArgs.cache_threshold,
+            help="Cache threshold (0.0-1.0) for cache-aware routing",
+        )
+        parser.add_argument(
+            f"--{prefix}balance-abs-threshold",
+            type=int,
+            default=RouterArgs.balance_abs_threshold,
+            help="Load balancing is triggered when (max_load - min_load) > abs_threshold AND max_load > min_load * rel_threshold. Otherwise, use cache aware",
+        )
+        parser.add_argument(
+            f"--{prefix}balance-rel-threshold",
+            type=float,
+            default=RouterArgs.balance_rel_threshold,
+            help="Load balancing is triggered when (max_load - min_load) > abs_threshold AND max_load > min_load * rel_threshold. Otherwise, use cache aware",
+        )
+        parser.add_argument(
+            f"--{prefix}bucket-adjust-interval-secs",
+            type=int,
+            default=RouterArgs.bucket_adjust_interval_secs,
+            help="Interval in seconds between bucket boundary adjustment operations",
+        )
+        parser.add_argument(
+            f"--{prefix}eviction-interval-secs",
+            type=int,
+            default=RouterArgs.eviction_interval_secs,
+            help="Interval in seconds between cache eviction operations",
+        )
+        parser.add_argument(
+            f"--{prefix}max-tree-size",
+            type=int,
+            default=RouterArgs.max_tree_size,
+            help="Maximum size of the approximation tree for cache-aware routing",
+        )
+        parser.add_argument(
+            f"--{prefix}max-payload-size",
+            type=int,
+            default=RouterArgs.max_payload_size,
+            help="Maximum payload size in bytes",
+        )
+        parser.add_argument(
+            f"--{prefix}dp-aware",
+            action="store_true",
+            help="Enable data parallelism aware schedule",
+        )
+        parser.add_argument(
+            f"--{prefix}enable-igw",
+            action="store_true",
+            help="Enable IGW (Inference-Gateway) mode for multi-model support",
+        )
+        parser.add_argument(
+            f"--{prefix}api-key",
+            type=str,
+            default=None,
+            help="The api key used for the authorization with the worker.  Useful when the dp aware scheduling strategy is enaled.",
+        )
+        parser.add_argument(
+            f"--{prefix}log-dir",
+            type=str,
+            default=None,
+            help="Directory to store log files. If not specified, logs are only output to console.",
+        )
+        parser.add_argument(
+            f"--{prefix}log-level",
+            type=str,
+            default="info",
+            choices=["debug", "info", "warn", "error"],
+            help="Set the logging level. If not specified, defaults to INFO.",
+        )
+        parser.add_argument(
+            f"--{prefix}service-discovery",
+            action="store_true",
+            help="Enable Kubernetes service discovery",
+        )
+        parser.add_argument(
+            f"--{prefix}selector",
+            type=str,
+            nargs="+",
+            default={},
+            help="Label selector for Kubernetes service discovery (format: key1=value1 key2=value2)",
+        )
+        parser.add_argument(
+            f"--{prefix}service-discovery-port",
+            type=int,
+            default=RouterArgs.service_discovery_port,
+            help="Port to use for discovered worker pods",
+        )
+        parser.add_argument(
+            f"--{prefix}service-discovery-namespace",
+            type=str,
+            help="Kubernetes namespace to watch for pods. If not provided, watches all namespaces (requires cluster-wide permissions)",
+        )
+        parser.add_argument(
+            f"--{prefix}prefill-selector",
+            type=str,
+            nargs="+",
+            default={},
+            help="Label selector for prefill server pods in PD mode (format: key1=value1 key2=value2)",
+        )
+        parser.add_argument(
+            f"--{prefix}decode-selector",
+            type=str,
+            nargs="+",
+            default={},
+            help="Label selector for decode server pods in PD mode (format: key1=value1 key2=value2)",
+        )
+        # Prometheus configuration
+        parser.add_argument(
+            f"--{prefix}prometheus-port",
+            type=int,
+            default=29000,
+            help="Port to expose Prometheus metrics. If not specified, Prometheus metrics are disabled",
+        )
+        parser.add_argument(
+            f"--{prefix}prometheus-host",
+            type=str,
+            default="0.0.0.0",
+            help="Host address to bind the Prometheus metrics server. Supports IPv4, IPv6 (e.g., ::, ::1), or 0.0.0.0 for all interfaces",
+        )
+        parser.add_argument(
+            f"--{prefix}request-id-headers",
+            type=str,
+            nargs="*",
+            help="Custom HTTP headers to check for request IDs (e.g., x-request-id x-trace-id). If not specified, uses common defaults.",
+        )
+        parser.add_argument(
+            f"--{prefix}request-timeout-secs",
+            type=int,
+            default=RouterArgs.request_timeout_secs,
+            help="Request timeout in seconds",
+        )
+        # Retry configuration
+        parser.add_argument(
+            f"--{prefix}retry-max-retries",
+            type=int,
+            default=RouterArgs.retry_max_retries,
+        )
+        parser.add_argument(
+            f"--{prefix}retry-initial-backoff-ms",
+            type=int,
+            default=RouterArgs.retry_initial_backoff_ms,
+        )
+        parser.add_argument(
+            f"--{prefix}retry-max-backoff-ms",
+            type=int,
+            default=RouterArgs.retry_max_backoff_ms,
+        )
+        parser.add_argument(
+            f"--{prefix}retry-backoff-multiplier",
+            type=float,
+            default=RouterArgs.retry_backoff_multiplier,
+        )
+        parser.add_argument(
+            f"--{prefix}retry-jitter-factor",
+            type=float,
+            default=RouterArgs.retry_jitter_factor,
+        )
+        parser.add_argument(
+            f"--{prefix}disable-retries",
+            action="store_true",
+            help="Disable retries (equivalent to setting retry_max_retries=1)",
+        )
+        # Circuit breaker configuration
+        parser.add_argument(
+            f"--{prefix}cb-failure-threshold",
+            type=int,
+            default=RouterArgs.cb_failure_threshold,
+        )
+        parser.add_argument(
+            f"--{prefix}cb-success-threshold",
+            type=int,
+            default=RouterArgs.cb_success_threshold,
+        )
+        parser.add_argument(
+            f"--{prefix}cb-timeout-duration-secs",
+            type=int,
+            default=RouterArgs.cb_timeout_duration_secs,
+        )
+        parser.add_argument(
+            f"--{prefix}cb-window-duration-secs",
+            type=int,
+            default=RouterArgs.cb_window_duration_secs,
+        )
+        parser.add_argument(
+            f"--{prefix}disable-circuit-breaker",
+            action="store_true",
+            help="Disable circuit breaker (equivalent to setting cb_failure_threshold to u32::MAX)",
+        )
+        # Health check configuration
+        parser.add_argument(
+            f"--{prefix}health-failure-threshold",
+            type=int,
+            default=RouterArgs.health_failure_threshold,
+            help="Number of consecutive health check failures before marking worker unhealthy",
+        )
+        parser.add_argument(
+            f"--{prefix}health-success-threshold",
+            type=int,
+            default=RouterArgs.health_success_threshold,
+            help="Number of consecutive health check successes before marking worker healthy",
+        )
+        parser.add_argument(
+            f"--{prefix}health-check-timeout-secs",
+            type=int,
+            default=RouterArgs.health_check_timeout_secs,
+            help="Timeout in seconds for health check requests",
+        )
+        parser.add_argument(
+            f"--{prefix}health-check-interval-secs",
+            type=int,
+            default=RouterArgs.health_check_interval_secs,
+            help="Interval in seconds between runtime health checks",
+        )
+        parser.add_argument(
+            f"--{prefix}health-check-endpoint",
+            type=str,
+            default=RouterArgs.health_check_endpoint,
+            help="Health check endpoint path",
+        )
+        parser.add_argument(
+            f"--{prefix}max-concurrent-requests",
+            type=int,
+            default=RouterArgs.max_concurrent_requests,
+            help="Maximum number of concurrent requests allowed (for rate limiting). Set to -1 to disable rate limiting.",
+        )
+        parser.add_argument(
+            f"--{prefix}queue-size",
+            type=int,
+            default=RouterArgs.queue_size,
+            help="Queue size for pending requests when max concurrent limit reached (0 = no queue, return 429 immediately)",
+        )
+        parser.add_argument(
+            f"--{prefix}queue-timeout-secs",
+            type=int,
+            default=RouterArgs.queue_timeout_secs,
+            help="Maximum time (in seconds) a request can wait in queue before timing out",
+        )
+        parser.add_argument(
+            f"--{prefix}rate-limit-tokens-per-second",
+            type=int,
+            default=RouterArgs.rate_limit_tokens_per_second,
+            help="Token bucket refill rate (tokens per second). If not set, defaults to max_concurrent_requests",
+        )
+        parser.add_argument(
+            f"--{prefix}cors-allowed-origins",
+            type=str,
+            nargs="*",
+            default=[],
+            help="CORS allowed origins (e.g., http://localhost:3000 https://example.com)",
+        )
+        # Tokenizer configuration
+        parser.add_argument(
+            f"--{prefix}model-path",
+            type=str,
+            default=None,
+            help="Model path for loading tokenizer (HuggingFace model ID or local path)",
+        )
+        parser.add_argument(
+            f"--{prefix}tokenizer-path",
+            type=str,
+            default=None,
+            help="Explicit tokenizer path (overrides model_path tokenizer if provided)",
+        )
+        parser.add_argument(
+            f"--{prefix}chat-template",
+            type=str,
+            default=None,
+            help="Chat template path (optional)",
+        )
+        parser.add_argument(
+            f"--{prefix}tokenizer-cache-enable-l0",
+            action="store_true",
+            default=RouterArgs.tokenizer_cache_enable_l0,
+            help="Enable L0 (whole-string exact match) tokenizer cache (default: False)",
+        )
+        parser.add_argument(
+            f"--{prefix}tokenizer-cache-l0-max-entries",
+            type=int,
+            default=RouterArgs.tokenizer_cache_l0_max_entries,
+            help="Maximum number of entries in L0 tokenizer cache (default: 10000)",
+        )
+        parser.add_argument(
+            f"--{prefix}tokenizer-cache-enable-l1",
+            action="store_true",
+            default=RouterArgs.tokenizer_cache_enable_l1,
+            help="Enable L1 (prefix matching) tokenizer cache (default: False)",
+        )
+        parser.add_argument(
+            f"--{prefix}tokenizer-cache-l1-max-memory",
+            type=int,
+            default=RouterArgs.tokenizer_cache_l1_max_memory,
+            help="Maximum memory for L1 tokenizer cache in bytes (default: 50MB)",
+        )
+        parser.add_argument(
+            f"--{prefix}reasoning-parser",
+            type=str,
+            default=None,
+            help="Specify the parser for reasoning models (e.g., deepseek-r1, qwen3)",
+        )
+        parser.add_argument(
+            f"--{prefix}tool-call-parser",
+            type=str,
+            default=None,
+            help="Specify the parser for handling tool-call interactions",
+        )
+        # MCP server configuration
+        parser.add_argument(
+            f"--{prefix}mcp-config-path",
+            type=str,
+            default=None,
+            help="Path to MCP (Model Context Protocol) server configuration file",
+        )
+        # Backend selection
+        parser.add_argument(
+            f"--{prefix}backend",
+            type=str,
+            default=RouterArgs.backend,
+            choices=["sglang", "openai"],
+            help="Backend runtime to use (default: sglang)",
+        )
+        # History backend configuration
+        parser.add_argument(
+            f"--{prefix}history-backend",
+            type=str,
+            default=RouterArgs.history_backend,
+            choices=["memory", "none", "oracle", "postgres"],
+            help="History storage backend for conversations and responses (default: memory)",
+        )
+        # Oracle configuration
+        parser.add_argument(
+            f"--{prefix}oracle-wallet-path",
+            type=str,
+            default=os.getenv("ATP_WALLET_PATH"),
+            help="Path to Oracle ATP wallet directory (env: ATP_WALLET_PATH)",
+        )
+        parser.add_argument(
+            f"--{prefix}oracle-tns-alias",
+            type=str,
+            default=os.getenv("ATP_TNS_ALIAS"),
+            help="Oracle TNS alias from tnsnames.ora (env: ATP_TNS_ALIAS).",
+        )
+        parser.add_argument(
+            f"--{prefix}oracle-connect-descriptor",
+            type=str,
+            default=os.getenv("ATP_DSN"),
+            help="Oracle connection descriptor/DSN (full connection string) (env: ATP_DSN)",
+        )
+        parser.add_argument(
+            f"--{prefix}oracle-username",
+            type=str,
+            default=os.getenv("ATP_USER"),
+            help="Oracle database username (env: ATP_USER)",
+        )
+        parser.add_argument(
+            f"--{prefix}oracle-password",
+            type=str,
+            default=os.getenv("ATP_PASSWORD"),
+            help="Oracle database password (env: ATP_PASSWORD)",
+        )
+        parser.add_argument(
+            f"--{prefix}oracle-pool-min",
+            type=int,
+            default=int(os.getenv("ATP_POOL_MIN", RouterArgs.oracle_pool_min)),
+            help="Minimum Oracle connection pool size (default: 1, env: ATP_POOL_MIN)",
+        )
+        parser.add_argument(
+            f"--{prefix}oracle-pool-max",
+            type=int,
+            default=int(os.getenv("ATP_POOL_MAX", RouterArgs.oracle_pool_max)),
+            help="Maximum Oracle connection pool size (default: 16, env: ATP_POOL_MAX)",
+        )
+        parser.add_argument(
+            f"--{prefix}oracle-pool-timeout-secs",
+            type=int,
+            default=int(
+                os.getenv("ATP_POOL_TIMEOUT_SECS", RouterArgs.oracle_pool_timeout_secs)
+            ),
+            help="Oracle connection pool timeout in seconds (default: 30, env: ATP_POOL_TIMEOUT_SECS)",
+        )
+        # mTLS configuration
+        parser.add_argument(
+            f"--{prefix}client-cert-path",
+            type=str,
+            default=None,
+            help="Path to client certificate for mTLS authentication with workers",
+        )
+        parser.add_argument(
+            f"--{prefix}client-key-path",
+            type=str,
+            default=None,
+            help="Path to client private key for mTLS authentication with workers",
+        )
+        parser.add_argument(
+            f"--{prefix}ca-cert-paths",
+            type=str,
+            nargs="*",
+            default=[],
+            help="Path(s) to CA certificate(s) for verifying worker TLS certificates. Can specify multiple CAs.",
+        )
+        parser.add_argument(
+            f"--{prefix}enable-trace",
+            action="store_true",
+            help="Enable opentelemetry trace",
+        )
+        parser.add_argument(
+            f"--{prefix}otlp-traces-endpoint",
+            type=str,
+            default="localhost:4317",
+            help="Config opentelemetry collector endpoint if --enable-trace is set. format: <ip>:<port>",
+        )
+
+    @classmethod
+    def from_cli_args(
+        cls, args: argparse.Namespace, use_router_prefix: bool = False
+    ) -> "RouterArgs":
+        """
+        Create RouterArgs instance from parsed command line arguments.
+
+        Args:
+            args: Parsed command line arguments
+            use_router_prefix: If True, look for arguments with 'router-' prefix
+        """
+        prefix = "router_" if use_router_prefix else ""
+        cli_args_dict = vars(args)
+        args_dict = {}
+
+        for attr in dataclasses.fields(cls):
+            # Auto strip prefix from args
+            if f"{prefix}{attr.name}" in cli_args_dict:
+                args_dict[attr.name] = cli_args_dict[f"{prefix}{attr.name}"]
+            elif attr.name in cli_args_dict:
+                args_dict[attr.name] = cli_args_dict[attr.name]
+
+        # parse special arguments and remove "--prefill" and "--decode" from cli_args_dict
+        args_dict["prefill_urls"] = cls._parse_prefill_urls(
+            cli_args_dict.get(f"{prefix}prefill", None)
+        )
+        args_dict["decode_urls"] = cls._parse_decode_urls(
+            cli_args_dict.get(f"{prefix}decode", None)
+        )
+        args_dict["selector"] = cls._parse_selector(
+            cli_args_dict.get(f"{prefix}selector", None)
+        )
+        args_dict["prefill_selector"] = cls._parse_selector(
+            cli_args_dict.get(f"{prefix}prefill_selector", None)
+        )
+        args_dict["decode_selector"] = cls._parse_selector(
+            cli_args_dict.get(f"{prefix}decode_selector", None)
+        )
+
+        # Mooncake-specific annotation
+        args_dict["bootstrap_port_annotation"] = "sglang.ai/bootstrap-port"
+
+        return cls(**args_dict)
+
+    def _validate_router_args(self):
+        # Validate configuration based on mode
+        if self.pd_disaggregation:
+            # Allow empty URLs even without service discovery to support dynamic worker addition
+            # URLs will be validated separately if provided
+            pass
+
+            # Warn about policy usage in PD mode
+            if self.prefill_policy and self.decode_policy and self.policy:
+                logger.warning(
+                    "Both --prefill-policy and --decode-policy are specified. "
+                    "The main --policy flag will be ignored for PD mode."
+                )
+            elif self.prefill_policy and not self.decode_policy and self.policy:
+                logger.info(
+                    f"Using --prefill-policy '{self.prefill_policy}' for prefill nodes "
+                    f"and --policy '{self.policy}' for decode nodes."
+                )
+            elif self.decode_policy and not self.prefill_policy and self.policy:
+                logger.info(
+                    f"Using --policy '{self.policy}' for prefill nodes "
+                    f"and --decode-policy '{self.decode_policy}' for decode nodes."
+                )
+
+    @staticmethod
+    def _parse_selector(selector_list):
+        if not selector_list:
+            return {}
+
+        selector = {}
+        for item in selector_list:
+            if "=" in item:
+                key, value = item.split("=", 1)
+                selector[key] = value
+        return selector
+
+    @staticmethod
+    def _parse_prefill_urls(prefill_list):
+        """Parse prefill URLs from --prefill arguments.
+
+        Format: --prefill URL [BOOTSTRAP_PORT]
+        Example:
+            --prefill http://prefill1:8080 9000  # With bootstrap port
+            --prefill http://prefill2:8080 none  # Explicitly no bootstrap port
+            --prefill http://prefill3:8080       # Defaults to no bootstrap port
+        """
+        if not prefill_list:
+            return []
+
+        prefill_urls = []
+        for prefill_args in prefill_list:
+
+            url = prefill_args[0]
+
+            # Handle optional bootstrap port
+            if len(prefill_args) >= 2:
+                bootstrap_port_str = prefill_args[1]
+                # Handle 'none' as None
+                if bootstrap_port_str.lower() == "none":
+                    bootstrap_port = None
+                else:
+                    try:
+                        bootstrap_port = int(bootstrap_port_str)
+                    except ValueError:
+                        raise ValueError(
+                            f"Invalid bootstrap port: {bootstrap_port_str}. Must be a number or 'none'"
+                        )
+            else:
+                # No bootstrap port specified, default to None
+                bootstrap_port = None
+
+            prefill_urls.append((url, bootstrap_port))
+
+        return prefill_urls
+
+    @staticmethod
+    def _parse_decode_urls(decode_list):
+        """Parse decode URLs from --decode arguments.
+
+        Format: --decode URL
+        Example: --decode http://decode1:8081 --decode http://decode2:8081
+        """
+        if not decode_list:
+            return []
+
+        # decode_list is a list of single-element lists due to nargs=1
+        return [url[0] for url in decode_list]

sgl-router/bindings/python/sglang_router/version.py

@@ -0,0 +1 @@
+__version__ = "0.2.3"

sgl-router/bindings/python/src/lib.rs

@@ -0,0 +1,716 @@
+use pyo3::prelude::*;
+use sglang_router::*;
+use std::collections::HashMap;
+
+// Define the enums with PyO3 bindings
+#[pyclass(eq)]
+#[derive(Clone, PartialEq, Debug)]
+pub enum PolicyType {
+    Random,
+    RoundRobin,
+    CacheAware,
+    PowerOfTwo,
+    Bucket,
+}
+
+#[pyclass(eq)]
+#[derive(Clone, PartialEq, Debug)]
+pub enum BackendType {
+    Sglang,
+    Openai,
+}
+
+#[pyclass(eq)]
+#[derive(Clone, PartialEq, Debug)]
+pub enum HistoryBackendType {
+    Memory,
+    None,
+    Oracle,
+    Postgres,
+}
+
+#[pyclass]
+#[derive(Clone, PartialEq)]
+pub struct PyOracleConfig {
+    #[pyo3(get, set)]
+    pub wallet_path: Option<String>,
+    #[pyo3(get, set)]
+    pub connect_descriptor: Option<String>,
+    #[pyo3(get, set)]
+    pub username: Option<String>,
+    #[pyo3(get, set)]
+    pub password: Option<String>,
+    #[pyo3(get, set)]
+    pub pool_min: usize,
+    #[pyo3(get, set)]
+    pub pool_max: usize,
+    #[pyo3(get, set)]
+    pub pool_timeout_secs: u64,
+}
+
+impl std::fmt::Debug for PyOracleConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("PyOracleConfig")
+            .field("wallet_path", &self.wallet_path)
+            .field("connect_descriptor", &"<redacted>")
+            .field("username", &self.username)
+            .field("password", &"<redacted>")
+            .field("pool_min", &self.pool_min)
+            .field("pool_max", &self.pool_max)
+            .field("pool_timeout_secs", &self.pool_timeout_secs)
+            .finish()
+    }
+}
+
+#[pymethods]
+impl PyOracleConfig {
+    #[new]
+    #[pyo3(signature = (
+        password = None,
+        username = None,
+        connect_descriptor = None,
+        wallet_path = None,
+        pool_min = 1,
+        pool_max = 16,
+        pool_timeout_secs = 30,
+    ))]
+    fn new(
+        password: Option<String>,
+        username: Option<String>,
+        connect_descriptor: Option<String>,
+        wallet_path: Option<String>,
+        pool_min: usize,
+        pool_max: usize,
+        pool_timeout_secs: u64,
+    ) -> PyResult<Self> {
+        if pool_min == 0 {
+            return Err(pyo3::exceptions::PyValueError::new_err(
+                "pool_min must be at least 1",
+            ));
+        }
+        if pool_max < pool_min {
+            return Err(pyo3::exceptions::PyValueError::new_err(
+                "pool_max must be >= pool_min",
+            ));
+        }
+
+        Ok(PyOracleConfig {
+            wallet_path,
+            connect_descriptor,
+            username,
+            password,
+            pool_min,
+            pool_max,
+            pool_timeout_secs,
+        })
+    }
+}
+
+impl PyOracleConfig {
+    pub fn to_config_oracle(&self) -> config::OracleConfig {
+        config::OracleConfig {
+            wallet_path: self.wallet_path.clone(),
+            connect_descriptor: self.connect_descriptor.clone().unwrap_or_default(),
+            username: self.username.clone().unwrap_or_default(),
+            password: self.password.clone().unwrap_or_default(),
+            pool_min: self.pool_min,
+            pool_max: self.pool_max,
+            pool_timeout_secs: self.pool_timeout_secs,
+        }
+    }
+}
+
+#[pyclass]
+#[derive(Debug, Clone, PartialEq)]
+pub struct PyPostgresConfig {
+    #[pyo3(get, set)]
+    pub db_url: Option<String>,
+
+    #[pyo3(get, set)]
+    pub pool_max: usize,
+}
+
+#[pymethods]
+impl PyPostgresConfig {
+    #[new]
+    #[pyo3(signature = (db_url = None,pool_max = 16,))]
+    fn new(db_url: Option<String>, pool_max: usize) -> PyResult<Self> {
+        Ok(PyPostgresConfig { db_url, pool_max })
+    }
+}
+
+impl PyPostgresConfig {
+    pub fn to_config_postgres(&self) -> config::PostgresConfig {
+        config::PostgresConfig {
+            db_url: self.db_url.clone().unwrap_or_default(),
+            pool_max: self.pool_max,
+        }
+    }
+}
+
+#[pyclass]
+#[derive(Debug, Clone, PartialEq)]
+struct Router {
+    host: String,
+    port: u16,
+    worker_urls: Vec<String>,
+    policy: PolicyType,
+    worker_startup_timeout_secs: u64,
+    worker_startup_check_interval: u64,
+    cache_threshold: f32,
+    balance_abs_threshold: usize,
+    balance_rel_threshold: f32,
+    eviction_interval_secs: u64,
+    max_tree_size: usize,
+    max_payload_size: usize,
+    dp_aware: bool,
+    api_key: Option<String>,
+    log_dir: Option<String>,
+    log_level: Option<String>,
+    service_discovery: bool,
+    selector: HashMap<String, String>,
+    service_discovery_port: u16,
+    service_discovery_namespace: Option<String>,
+    prefill_selector: HashMap<String, String>,
+    decode_selector: HashMap<String, String>,
+    bootstrap_port_annotation: String,
+    prometheus_port: Option<u16>,
+    prometheus_host: Option<String>,
+    request_timeout_secs: u64,
+    request_id_headers: Option<Vec<String>>,
+    pd_disaggregation: bool,
+    bucket_adjust_interval_secs: usize,
+    prefill_urls: Option<Vec<(String, Option<u16>)>>,
+    decode_urls: Option<Vec<String>>,
+    prefill_policy: Option<PolicyType>,
+    decode_policy: Option<PolicyType>,
+    max_concurrent_requests: i32,
+    cors_allowed_origins: Vec<String>,
+    retry_max_retries: u32,
+    retry_initial_backoff_ms: u64,
+    retry_max_backoff_ms: u64,
+    retry_backoff_multiplier: f32,
+    retry_jitter_factor: f32,
+    disable_retries: bool,
+    cb_failure_threshold: u32,
+    cb_success_threshold: u32,
+    cb_timeout_duration_secs: u64,
+    cb_window_duration_secs: u64,
+    disable_circuit_breaker: bool,
+    health_failure_threshold: u32,
+    health_success_threshold: u32,
+    health_check_timeout_secs: u64,
+    health_check_interval_secs: u64,
+    health_check_endpoint: String,
+    enable_igw: bool,
+    queue_size: usize,
+    queue_timeout_secs: u64,
+    rate_limit_tokens_per_second: Option<i32>,
+    connection_mode: core::ConnectionMode,
+    model_path: Option<String>,
+    tokenizer_path: Option<String>,
+    chat_template: Option<String>,
+    tokenizer_cache_enable_l0: bool,
+    tokenizer_cache_l0_max_entries: usize,
+    tokenizer_cache_enable_l1: bool,
+    tokenizer_cache_l1_max_memory: usize,
+    reasoning_parser: Option<String>,
+    tool_call_parser: Option<String>,
+    mcp_config_path: Option<String>,
+    backend: BackendType,
+    history_backend: HistoryBackendType,
+    oracle_config: Option<PyOracleConfig>,
+    postgres_config: Option<PyPostgresConfig>,
+    client_cert_path: Option<String>,
+    client_key_path: Option<String>,
+    ca_cert_paths: Vec<String>,
+}
+
+impl Router {
+    fn determine_connection_mode(worker_urls: &[String]) -> core::ConnectionMode {
+        for url in worker_urls {
+            if url.starts_with("grpc://") || url.starts_with("grpcs://") {
+                return core::ConnectionMode::Grpc { port: None };
+            }
+        }
+        core::ConnectionMode::Http
+    }
+
+    pub fn to_router_config(&self) -> config::ConfigResult<config::RouterConfig> {
+        use config::{
+            DiscoveryConfig, MetricsConfig, PolicyConfig as ConfigPolicyConfig, RoutingMode,
+        };
+
+        let convert_policy = |policy: &PolicyType| -> ConfigPolicyConfig {
+            match policy {
+                PolicyType::Random => ConfigPolicyConfig::Random,
+                PolicyType::RoundRobin => ConfigPolicyConfig::RoundRobin,
+                PolicyType::CacheAware => ConfigPolicyConfig::CacheAware {
+                    cache_threshold: self.cache_threshold,
+                    balance_abs_threshold: self.balance_abs_threshold,
+                    balance_rel_threshold: self.balance_rel_threshold,
+                    eviction_interval_secs: self.eviction_interval_secs,
+                    max_tree_size: self.max_tree_size,
+                },
+                PolicyType::PowerOfTwo => ConfigPolicyConfig::PowerOfTwo {
+                    load_check_interval_secs: 5,
+                },
+                PolicyType::Bucket => ConfigPolicyConfig::Bucket {
+                    balance_abs_threshold: self.balance_abs_threshold,
+                    balance_rel_threshold: self.balance_rel_threshold,
+                    bucket_adjust_interval_secs: self.bucket_adjust_interval_secs,
+                },
+            }
+        };
+
+        let mode = if self.enable_igw {
+            RoutingMode::Regular {
+                worker_urls: vec![],
+            }
+        } else if matches!(self.backend, BackendType::Openai) {
+            RoutingMode::OpenAI {
+                worker_urls: self.worker_urls.clone(),
+            }
+        } else if self.pd_disaggregation {
+            RoutingMode::PrefillDecode {
+                prefill_urls: self.prefill_urls.clone().unwrap_or_default(),
+                decode_urls: self.decode_urls.clone().unwrap_or_default(),
+                prefill_policy: self.prefill_policy.as_ref().map(convert_policy),
+                decode_policy: self.decode_policy.as_ref().map(convert_policy),
+            }
+        } else {
+            RoutingMode::Regular {
+                worker_urls: self.worker_urls.clone(),
+            }
+        };
+
+        let policy = convert_policy(&self.policy);
+
+        let discovery = if self.service_discovery {
+            Some(DiscoveryConfig {
+                enabled: true,
+                namespace: self.service_discovery_namespace.clone(),
+                port: self.service_discovery_port,
+                check_interval_secs: 60,
+                selector: self.selector.clone(),
+                prefill_selector: self.prefill_selector.clone(),
+                decode_selector: self.decode_selector.clone(),
+                bootstrap_port_annotation: self.bootstrap_port_annotation.clone(),
+            })
+        } else {
+            None
+        };
+
+        let metrics = match (self.prometheus_port, self.prometheus_host.as_ref()) {
+            (Some(port), Some(host)) => Some(MetricsConfig {
+                port,
+                host: host.clone(),
+            }),
+            _ => None,
+        };
+
+        let history_backend = match self.history_backend {
+            HistoryBackendType::Memory => config::HistoryBackend::Memory,
+            HistoryBackendType::None => config::HistoryBackend::None,
+            HistoryBackendType::Oracle => config::HistoryBackend::Oracle,
+            HistoryBackendType::Postgres => config::HistoryBackend::Postgres,
+        };
+
+        let oracle = if matches!(self.history_backend, HistoryBackendType::Oracle) {
+            self.oracle_config
+                .as_ref()
+                .map(|cfg| cfg.to_config_oracle())
+        } else {
+            None
+        };
+
+        let postgres_config = if matches!(self.history_backend, HistoryBackendType::Postgres) {
+            self.postgres_config
+                .as_ref()
+                .map(|cfg| cfg.to_config_postgres())
+        } else {
+            None
+        };
+
+        config::RouterConfig::builder()
+            .mode(mode)
+            .policy(policy)
+            .host(&self.host)
+            .port(self.port)
+            .connection_mode(self.connection_mode.clone())
+            .max_payload_size(self.max_payload_size)
+            .request_timeout_secs(self.request_timeout_secs)
+            .worker_startup_timeout_secs(self.worker_startup_timeout_secs)
+            .worker_startup_check_interval_secs(self.worker_startup_check_interval)
+            .max_concurrent_requests(self.max_concurrent_requests)
+            .queue_size(self.queue_size)
+            .queue_timeout_secs(self.queue_timeout_secs)
+            .cors_allowed_origins(self.cors_allowed_origins.clone())
+            .retry_config(config::RetryConfig {
+                max_retries: self.retry_max_retries,
+                initial_backoff_ms: self.retry_initial_backoff_ms,
+                max_backoff_ms: self.retry_max_backoff_ms,
+                backoff_multiplier: self.retry_backoff_multiplier,
+                jitter_factor: self.retry_jitter_factor,
+            })
+            .circuit_breaker_config(config::CircuitBreakerConfig {
+                failure_threshold: self.cb_failure_threshold,
+                success_threshold: self.cb_success_threshold,
+                timeout_duration_secs: self.cb_timeout_duration_secs,
+                window_duration_secs: self.cb_window_duration_secs,
+            })
+            .health_check_config(config::HealthCheckConfig {
+                failure_threshold: self.health_failure_threshold,
+                success_threshold: self.health_success_threshold,
+                timeout_secs: self.health_check_timeout_secs,
+                check_interval_secs: self.health_check_interval_secs,
+                endpoint: self.health_check_endpoint.clone(),
+            })
+            .tokenizer_cache(config::TokenizerCacheConfig {
+                enable_l0: self.tokenizer_cache_enable_l0,
+                l0_max_entries: self.tokenizer_cache_l0_max_entries,
+                enable_l1: self.tokenizer_cache_enable_l1,
+                l1_max_memory: self.tokenizer_cache_l1_max_memory,
+            })
+            .history_backend(history_backend)
+            .maybe_api_key(self.api_key.as_ref())
+            .maybe_discovery(discovery)
+            .maybe_metrics(metrics)
+            .maybe_log_dir(self.log_dir.as_ref())
+            .maybe_log_level(self.log_level.as_ref())
+            .maybe_request_id_headers(self.request_id_headers.clone())
+            .maybe_rate_limit_tokens_per_second(self.rate_limit_tokens_per_second)
+            .maybe_model_path(self.model_path.as_ref())
+            .maybe_tokenizer_path(self.tokenizer_path.as_ref())
+            .maybe_chat_template(self.chat_template.as_ref())
+            .maybe_oracle(oracle)
+            .maybe_postgres(postgres_config)
+            .maybe_reasoning_parser(self.reasoning_parser.as_ref())
+            .maybe_tool_call_parser(self.tool_call_parser.as_ref())
+            .maybe_mcp_config_path(self.mcp_config_path.as_ref())
+            .dp_aware(self.dp_aware)
+            .retries(!self.disable_retries)
+            .circuit_breaker(!self.disable_circuit_breaker)
+            .igw(self.enable_igw)
+            .maybe_client_cert_and_key(
+                self.client_cert_path.as_ref(),
+                self.client_key_path.as_ref(),
+            )
+            .add_ca_certificates(self.ca_cert_paths.clone())
+            .build()
+    }
+}
+
+#[pymethods]
+impl Router {
+    #[new]
+    #[pyo3(signature = (
+        worker_urls,
+        policy = PolicyType::RoundRobin,
+        host = String::from("0.0.0.0"),
+        port = 3001,
+        worker_startup_timeout_secs = 600,
+        worker_startup_check_interval = 30,
+        cache_threshold = 0.3,
+        balance_abs_threshold = 64,
+        balance_rel_threshold = 1.5,
+        eviction_interval_secs = 120,
+        max_tree_size = 2usize.pow(26),
+        max_payload_size = 512 * 1024 * 1024,
+        dp_aware = false,
+        api_key = None,
+        log_dir = None,
+        log_level = None,
+        service_discovery = false,
+        selector = HashMap::new(),
+        service_discovery_port = 80,
+        service_discovery_namespace = None,
+        prefill_selector = HashMap::new(),
+        decode_selector = HashMap::new(),
+        bootstrap_port_annotation = String::from("sglang.ai/bootstrap-port"),
+        prometheus_port = None,
+        prometheus_host = None,
+        request_timeout_secs = 1800,
+        request_id_headers = None,
+        pd_disaggregation = false,
+        bucket_adjust_interval_secs = 5,
+        prefill_urls = None,
+        decode_urls = None,
+        prefill_policy = None,
+        decode_policy = None,
+        max_concurrent_requests = -1,
+        cors_allowed_origins = vec![],
+        retry_max_retries = 5,
+        retry_initial_backoff_ms = 50,
+        retry_max_backoff_ms = 30_000,
+        retry_backoff_multiplier = 1.5,
+        retry_jitter_factor = 0.2,
+        disable_retries = false,
+        cb_failure_threshold = 10,
+        cb_success_threshold = 3,
+        cb_timeout_duration_secs = 60,
+        cb_window_duration_secs = 120,
+        disable_circuit_breaker = false,
+        health_failure_threshold = 3,
+        health_success_threshold = 2,
+        health_check_timeout_secs = 5,
+        health_check_interval_secs = 60,
+        health_check_endpoint = String::from("/health"),
+        enable_igw = false,
+        queue_size = 100,
+        queue_timeout_secs = 60,
+        rate_limit_tokens_per_second = None,
+        model_path = None,
+        tokenizer_path = None,
+        chat_template = None,
+        tokenizer_cache_enable_l0 = false,
+        tokenizer_cache_l0_max_entries = 10000,
+        tokenizer_cache_enable_l1 = false,
+        tokenizer_cache_l1_max_memory = 52428800,
+        reasoning_parser = None,
+        tool_call_parser = None,
+        mcp_config_path = None,
+        backend = BackendType::Sglang,
+        history_backend = HistoryBackendType::Memory,
+        oracle_config = None,
+        postgres_config = None,
+        client_cert_path = None,
+        client_key_path = None,
+        ca_cert_paths = vec![],
+    ))]
+    #[allow(clippy::too_many_arguments)]
+    fn new(
+        worker_urls: Vec<String>,
+        policy: PolicyType,
+        host: String,
+        port: u16,
+        worker_startup_timeout_secs: u64,
+        worker_startup_check_interval: u64,
+        cache_threshold: f32,
+        balance_abs_threshold: usize,
+        balance_rel_threshold: f32,
+        eviction_interval_secs: u64,
+        max_tree_size: usize,
+        max_payload_size: usize,
+        dp_aware: bool,
+        api_key: Option<String>,
+        log_dir: Option<String>,
+        log_level: Option<String>,
+        service_discovery: bool,
+        selector: HashMap<String, String>,
+        service_discovery_port: u16,
+        service_discovery_namespace: Option<String>,
+        prefill_selector: HashMap<String, String>,
+        decode_selector: HashMap<String, String>,
+        bootstrap_port_annotation: String,
+        prometheus_port: Option<u16>,
+        prometheus_host: Option<String>,
+        request_timeout_secs: u64,
+        request_id_headers: Option<Vec<String>>,
+        pd_disaggregation: bool,
+        bucket_adjust_interval_secs: usize,
+        prefill_urls: Option<Vec<(String, Option<u16>)>>,
+        decode_urls: Option<Vec<String>>,
+        prefill_policy: Option<PolicyType>,
+        decode_policy: Option<PolicyType>,
+        max_concurrent_requests: i32,
+        cors_allowed_origins: Vec<String>,
+        retry_max_retries: u32,
+        retry_initial_backoff_ms: u64,
+        retry_max_backoff_ms: u64,
+        retry_backoff_multiplier: f32,
+        retry_jitter_factor: f32,
+        disable_retries: bool,
+        cb_failure_threshold: u32,
+        cb_success_threshold: u32,
+        cb_timeout_duration_secs: u64,
+        cb_window_duration_secs: u64,
+        disable_circuit_breaker: bool,
+        health_failure_threshold: u32,
+        health_success_threshold: u32,
+        health_check_timeout_secs: u64,
+        health_check_interval_secs: u64,
+        health_check_endpoint: String,
+        enable_igw: bool,
+        queue_size: usize,
+        queue_timeout_secs: u64,
+        rate_limit_tokens_per_second: Option<i32>,
+        model_path: Option<String>,
+        tokenizer_path: Option<String>,
+        chat_template: Option<String>,
+        tokenizer_cache_enable_l0: bool,
+        tokenizer_cache_l0_max_entries: usize,
+        tokenizer_cache_enable_l1: bool,
+        tokenizer_cache_l1_max_memory: usize,
+        reasoning_parser: Option<String>,
+        tool_call_parser: Option<String>,
+        mcp_config_path: Option<String>,
+        backend: BackendType,
+        history_backend: HistoryBackendType,
+        oracle_config: Option<PyOracleConfig>,
+        postgres_config: Option<PyPostgresConfig>,
+        client_cert_path: Option<String>,
+        client_key_path: Option<String>,
+        ca_cert_paths: Vec<String>,
+    ) -> PyResult<Self> {
+        let mut all_urls = worker_urls.clone();
+
+        if let Some(ref prefill_urls) = prefill_urls {
+            for (url, _) in prefill_urls {
+                all_urls.push(url.clone());
+            }
+        }
+
+        if let Some(ref decode_urls) = decode_urls {
+            all_urls.extend(decode_urls.clone());
+        }
+
+        let connection_mode = Self::determine_connection_mode(&all_urls);
+
+        Ok(Router {
+            host,
+            port,
+            worker_urls,
+            policy,
+            worker_startup_timeout_secs,
+            worker_startup_check_interval,
+            cache_threshold,
+            balance_abs_threshold,
+            balance_rel_threshold,
+            eviction_interval_secs,
+            max_tree_size,
+            max_payload_size,
+            dp_aware,
+            api_key,
+            log_dir,
+            log_level,
+            service_discovery,
+            selector,
+            service_discovery_port,
+            service_discovery_namespace,
+            prefill_selector,
+            decode_selector,
+            bootstrap_port_annotation,
+            prometheus_port,
+            prometheus_host,
+            request_timeout_secs,
+            request_id_headers,
+            pd_disaggregation,
+            bucket_adjust_interval_secs,
+            prefill_urls,
+            decode_urls,
+            prefill_policy,
+            decode_policy,
+            max_concurrent_requests,
+            cors_allowed_origins,
+            retry_max_retries,
+            retry_initial_backoff_ms,
+            retry_max_backoff_ms,
+            retry_backoff_multiplier,
+            retry_jitter_factor,
+            disable_retries,
+            cb_failure_threshold,
+            cb_success_threshold,
+            cb_timeout_duration_secs,
+            cb_window_duration_secs,
+            disable_circuit_breaker,
+            health_failure_threshold,
+            health_success_threshold,
+            health_check_timeout_secs,
+            health_check_interval_secs,
+            health_check_endpoint,
+            enable_igw,
+            queue_size,
+            queue_timeout_secs,
+            rate_limit_tokens_per_second,
+            connection_mode,
+            model_path,
+            tokenizer_path,
+            chat_template,
+            tokenizer_cache_enable_l0,
+            tokenizer_cache_l0_max_entries,
+            tokenizer_cache_enable_l1,
+            tokenizer_cache_l1_max_memory,
+            reasoning_parser,
+            tool_call_parser,
+            mcp_config_path,
+            backend,
+            history_backend,
+            oracle_config,
+            postgres_config,
+            client_cert_path,
+            client_key_path,
+            ca_cert_paths,
+        })
+    }
+
+    fn start(&self) -> PyResult<()> {
+        use metrics::PrometheusConfig;
+
+        let router_config = self.to_router_config().map_err(|e| {
+            pyo3::exceptions::PyValueError::new_err(format!("Configuration error: {}", e))
+        })?;
+
+        router_config.validate().map_err(|e| {
+            pyo3::exceptions::PyValueError::new_err(format!(
+                "Configuration validation failed: {}",
+                e
+            ))
+        })?;
+
+        let service_discovery_config = if self.service_discovery {
+            Some(service_discovery::ServiceDiscoveryConfig {
+                enabled: true,
+                selector: self.selector.clone(),
+                check_interval: std::time::Duration::from_secs(60),
+                port: self.service_discovery_port,
+                namespace: self.service_discovery_namespace.clone(),
+                pd_mode: self.pd_disaggregation,
+                prefill_selector: self.prefill_selector.clone(),
+                decode_selector: self.decode_selector.clone(),
+                bootstrap_port_annotation: self.bootstrap_port_annotation.clone(),
+            })
+        } else {
+            None
+        };
+
+        let prometheus_config = Some(PrometheusConfig {
+            port: self.prometheus_port.unwrap_or(29000),
+            host: self
+                .prometheus_host
+                .clone()
+                .unwrap_or_else(|| "127.0.0.1".to_string()),
+        });
+
+        let runtime = tokio::runtime::Runtime::new()
+            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
+
+        runtime.block_on(async move {
+            server::startup(server::ServerConfig {
+                host: self.host.clone(),
+                port: self.port,
+                router_config,
+                max_payload_size: self.max_payload_size,
+                log_dir: self.log_dir.clone(),
+                log_level: self.log_level.clone(),
+                service_discovery_config,
+                prometheus_config,
+                request_timeout_secs: self.request_timeout_secs,
+                request_id_headers: self.request_id_headers.clone(),
+            })
+            .await
+            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
+        })
+    }
+}
+
+#[pymodule]
+fn sglang_router_rs(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_class::<PolicyType>()?;
+    m.add_class::<BackendType>()?;
+    m.add_class::<HistoryBackendType>()?;
+    m.add_class::<PyOracleConfig>()?;
+    m.add_class::<PyPostgresConfig>()?;
+    m.add_class::<Router>()?;
+    Ok(())
+}