--- title: router_serve ---

fastvideo router-serve CLI subcommand.

Launches the streaming router from a YAML config. Separate from fastvideo serve because the router is an orthogonal process: it fronts one or more running servers rather than hosting a generator itself.

fastvideo.entrypoints.cli.router_serve.load_raw_config

load_raw_config(path: str | Path) -> dict[str, Any]

fastvideo.entrypoints.cli.router_serve.CLISubcommand

CLISubcommand()

Base class for CLI subcommands

fastvideo.entrypoints.cli.router_serve.CLISubcommand.cmd

cmd(args: argparse.Namespace) -> None

Execute the command with the given arguments

fastvideo.entrypoints.cli.router_serve.CLISubcommand.validate

validate(args: argparse.Namespace) -> None

Validate the arguments for this command

fastvideo.entrypoints.cli.router_serve.CLISubcommand.subparser_init

subparser_init(subparsers: argparse._SubParsersAction) -> FlexibleArgumentParser

Initialize the subparser for this command

fastvideo.entrypoints.cli.router_serve.ReplicaEndpoint

ReplicaEndpoint(url: str, name: str | None = None, primary: bool = False, weight: float = 1.0)

One backend replica the router can route to.

fastvideo.entrypoints.cli.router_serve.ReplicaEndpoint.url

url: str

HTTP base URL, e.g. http://host:8000. WebSocket URL is derived automatically by replacing the scheme.

fastvideo.entrypoints.cli.router_serve.ReplicaEndpoint.primary

primary: bool = False

True = prefer this replica over others in steady state.

fastvideo.entrypoints.cli.router_serve.RouterConfig

RouterConfig(host: str = '0.0.0.0', port: int = 9000, replicas: list[ReplicaEndpoint] = list(), health_check_path: str = '/health', health_check_interval_seconds: float = 5.0, health_check_timeout_seconds: float = 2.0, failure_threshold: int = 3, recovery_threshold: int = 2)

Typed router config loaded from a YAML file.

Example::

router:
  host: 0.0.0.0
  port: 9000
  replicas:
    - url: http://streamer-a:8000
      primary: true
    - url: http://streamer-b:8000
  health_check:
    path: /health
    interval_seconds: 5
    failure_threshold: 3

Validation runs in __post_init__: empty replicas, non-positive intervals/timeouts, thresholds < 1, non-http(s) URLs, and more than one primary all raise ValueError so misconfigurations surface at load time rather than as confusing runtime failures.

fastvideo.entrypoints.cli.router_serve.init_logger

init_logger(name: str) -> _FastvideoLogger

The main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.

fastvideo.entrypoints.cli.router_serve.FlexibleArgumentParser

FlexibleArgumentParser(*args, **kwargs)

ArgumentParser that allows both underscore and dash in names.

fastvideo.entrypoints.cli.router_serve.FlexibleArgumentParser.parse_args

parse_args(args = None, namespace = None) -> argparse.Namespace

fastvideo.entrypoints.cli.router_serve.FlexibleArgumentParser.parse_known_args

parse_known_args(args = None, namespace = None) -> tuple[argparse.Namespace, list[str]]

fastvideo.entrypoints.cli.router_serve.RouterServeSubcommand

RouterServeSubcommand()

Start the multi-replica WebSocket router.

Source
class RouterServeSubcommand(CLISubcommand):
    """Start the multi-replica WebSocket router."""

    def __init__(self) -> None:
        self.name = "router-serve"
        super().__init__()

    def cmd(self, args: argparse.Namespace) -> None:
        config = _load_router_config(args.config)
        logger.info(
            "router listening on %s:%d (%d replicas, %d primary)",
            config.host,
            config.port,
            len(config.replicas),
            sum(1 for r in config.replicas if r.primary),
        )
        from fastvideo.entrypoints.streaming.router.main import run_router

        run_router(config)

    def validate(self, args: argparse.Namespace) -> None:
        if not args.config:
            raise ValueError("fastvideo router-serve requires --config PATH")
        if not os.path.exists(args.config):
            raise ValueError(f"Router config file not found: {args.config}")

    def subparser_init(
        self,
        subparsers: argparse._SubParsersAction,
    ) -> FlexibleArgumentParser:
        parser = subparsers.add_parser(
            "router-serve",
            help="Start the streaming router (multi-replica load balancer)",
            usage="fastvideo router-serve --config ROUTER_CONFIG",
        )
        parser.add_argument(
            "--config",
            type=str,
            default="",
            required=False,
            help="Path to a YAML/JSON router config. Required.",
        )
        return cast(FlexibleArgumentParser, parser)

fastvideo.entrypoints.cli.router_serve.RouterServeSubcommand.cmd

cmd(args: argparse.Namespace) -> None
Source
def cmd(self, args: argparse.Namespace) -> None:
    config = _load_router_config(args.config)
    logger.info(
        "router listening on %s:%d (%d replicas, %d primary)",
        config.host,
        config.port,
        len(config.replicas),
        sum(1 for r in config.replicas if r.primary),
    )
    from fastvideo.entrypoints.streaming.router.main import run_router

    run_router(config)

fastvideo.entrypoints.cli.router_serve.RouterServeSubcommand.validate

validate(args: argparse.Namespace) -> None
Source
def validate(self, args: argparse.Namespace) -> None:
    if not args.config:
        raise ValueError("fastvideo router-serve requires --config PATH")
    if not os.path.exists(args.config):
        raise ValueError(f"Router config file not found: {args.config}")

fastvideo.entrypoints.cli.router_serve.RouterServeSubcommand.subparser_init

subparser_init(subparsers: argparse._SubParsersAction) -> FlexibleArgumentParser
Source
def subparser_init(
    self,
    subparsers: argparse._SubParsersAction,
) -> FlexibleArgumentParser:
    parser = subparsers.add_parser(
        "router-serve",
        help="Start the streaming router (multi-replica load balancer)",
        usage="fastvideo router-serve --config ROUTER_CONFIG",
    )
    parser.add_argument(
        "--config",
        type=str,
        default="",
        required=False,
        help="Path to a YAML/JSON router config. Required.",
    )
    return cast(FlexibleArgumentParser, parser)

fastvideo.entrypoints.cli.router_serve.cmd_init

cmd_init() -> list[CLISubcommand]
Source
def cmd_init() -> list[CLISubcommand]:
    return [RouterServeSubcommand()]