Resolve bind sockets in _serve instead of startup

Move the bind socket resolution from startup() to _serve() so startup
doesn't need to know about config.bind. The resolved sockets flow into
both startup and shutdown through the existing parameter.

docs/settings.md

@@ -46,6 +46,12 @@ uvicorn itself.
 * `--port <int>` - Bind to a socket with this port. If set to 0, an available port will be picked. **Default:** *8000*.
 * `--uds <path>` - Bind to a UNIX domain socket, for example `--uds /tmp/uvicorn.sock`. Useful if you want to run Uvicorn behind a reverse proxy.
 * `--fd <int>` - Bind to socket from this file descriptor. Useful if you want to run Uvicorn within a process manager.
+* `--bind <str>` / `-b <str>` - Bind to one or more addresses. May be specified multiple times to listen on multiple sockets simultaneously. Supported formats: `HOST:PORT` (e.g. `0.0.0.0:8000`), `[HOST]:PORT` for IPv6 (e.g. `[::1]:8000`), `unix:PATH` (e.g. `unix:/tmp/uvicorn.sock`), `fd://NUM` (e.g. `fd://3`). Mutually exclusive with `--host`, `--port`, `--uds`, and `--fd`.
+
+!!! note
+    The `--host`, `--port`, `--uds`, and `--fd` options each bind to a single address of a single type.
+    Use `--bind` when you need to listen on multiple addresses (e.g. dual-stack IPv4 + IPv6) or mix
+    transport types (e.g. a TCP port for internal services and a unix socket behind a reverse proxy).
 
 ## Development
 

tests/test_cli.py

@@ -201,3 +201,75 @@ def test_set_app_via_environment_variable():
             args, _ = mock_run.call_args
             assert result.exit_code == 0
             assert args == (app_path,)
+
+
+def test_cli_bind_option() -> None:
+    runner = CliRunner()
+
+    with mock.patch.object(main, "run") as mock_run:
+        result = runner.invoke(cli, ["tests.test_cli:App", "--bind", "0.0.0.0:8000"])
+
+    assert result.output == ""
+    assert result.exit_code == 0
+    mock_run.assert_called_once()
+    assert mock_run.call_args[1]["bind"] == ["0.0.0.0:8000"]
+
+
+def test_cli_bind_multiple() -> None:
+    runner = CliRunner()
+
+    with mock.patch.object(main, "run") as mock_run:
+        result = runner.invoke(cli, ["tests.test_cli:App", "-b", "127.0.0.1:8000", "-b", "127.0.0.1:9000"])
+
+    assert result.output == ""
+    assert result.exit_code == 0
+    mock_run.assert_called_once()
+    assert mock_run.call_args[1]["bind"] == ["127.0.0.1:8000", "127.0.0.1:9000"]
+
+
+@pytest.mark.parametrize(
+    "extra_args",
+    [
+        ["--host", "0.0.0.0"],
+        ["--port", "9000"],
+        ["--uds", "/tmp/test.sock"],
+        ["--fd", "3"],
+    ],
+    ids=["host", "port", "uds", "fd"],
+)
+def test_cli_bind_mutually_exclusive(extra_args: list[str]) -> None:
+    runner = CliRunner()
+
+    with pytest.raises(ValueError, match="'bind' is mutually exclusive with.*"):
+        runner.invoke(cli, ["tests.test_cli:App", "-b", "127.0.0.1:8000", *extra_args], catch_exceptions=False)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
+def test_cli_bind_unix_cleanup() -> None:  # pragma: py-win32
+    sock_path = "/tmp/uvicorn_test_cleanup.sock"
+    runner = CliRunner()
+
+    try:
+        Path(sock_path).touch()
+        with mock.patch.object(Config, "bind_sockets") as mock_bind_sockets:
+            with mock.patch.object(Multiprocess, "run") as mock_run:
+                result = runner.invoke(cli, ["tests.test_cli:App", "--workers=2", "-b", f"unix:{sock_path}"])
+
+        assert result.exit_code == 0
+        mock_bind_sockets.assert_called_once()
+        mock_run.assert_called_once()
+        assert not Path(sock_path).exists()
+    finally:
+        if Path(sock_path).exists():  # pragma: no cover
+            os.remove(sock_path)
+
+
+def test_cli_bind_without_value_passes_none() -> None:
+    runner = CliRunner()
+
+    with mock.patch.object(main, "run") as mock_run:
+        result = runner.invoke(cli, ["tests.test_cli:App"])
+
+    assert result.exit_code in (0, 3)
+    mock_run.assert_called_once()
+    assert mock_run.call_args[1]["bind"] is None

tests/test_config.py

@@ -593,3 +593,88 @@ def test_setup_event_loop_is_removed(caplog: pytest.LogCaptureFixture) -> None:
         AttributeError, match="The `setup_event_loop` method was replaced by `get_loop_factory` in uvicorn 0.36.0."
     ):
         config.setup_event_loop()
+
+
+@pytest.mark.parametrize(
+    "bind_str, expected_family",
+    [
+        ("127.0.0.1:0", socket.AF_INET),
+        ("0.0.0.0:0", socket.AF_INET),
+        ("[::1]:0", socket.AF_INET6),
+        ("[::]:0", socket.AF_INET6),
+        ("localhost:0", socket.AF_INET),
+    ],
+    ids=["ipv4", "ipv4-wildcard", "ipv6", "ipv6-wildcard", "hostname"],
+)
+def test_bind_sockets_address_formats(bind_str: str, expected_family: socket.AddressFamily) -> None:
+    config = Config(app=asgi_app, bind=[bind_str])
+    sockets = config.bind_sockets()
+    assert len(sockets) == 1
+    assert sockets[0].family == expected_family
+    sockets[0].close()
+
+
+def test_bind_sockets_multiple() -> None:
+    config = Config(app=asgi_app, bind=["127.0.0.1:0", "127.0.0.1:0"])
+    sockets = config.bind_sockets()
+    assert len(sockets) == 2
+    for sock in sockets:
+        assert sock.family == socket.AF_INET
+        sock.close()
+
+
+def test_bind_sockets_default_port() -> None:
+    config = Config(app=asgi_app, bind=["127.0.0.1"])
+    sockets = config.bind_sockets()
+    assert len(sockets) == 1
+    assert sockets[0].getsockname()[1] == 8000
+    sockets[0].close()
+
+
+def test_bind_sockets_fallback() -> None:
+    config = Config(app=asgi_app, bind=None)
+    sockets = config.bind_sockets()
+    assert len(sockets) == 1
+    sockets[0].close()
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"host": "0.0.0.0"},
+        {"port": 9000},
+        {"uds": "/tmp/test.sock"},
+        {"fd": 3},
+    ],
+    ids=["host", "port", "uds", "fd"],
+)
+def test_bind_mutually_exclusive_with_other_params(kwargs: dict[str, Any]) -> None:
+    with pytest.raises(ValueError, match="'bind' is mutually exclusive with"):
+        Config(app=asgi_app, bind=["127.0.0.1:0"], **kwargs)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="requires unix sockets")
+def test_bind_sockets_unix() -> None:  # pragma: py-win32
+    sock_path = "/tmp/uvicorn_test_bind.sock"
+    try:
+        config = Config(app=asgi_app, bind=[f"unix:{sock_path}"])
+        sockets = config.bind_sockets()
+        assert len(sockets) == 1
+        assert sockets[0].family == socket.AF_UNIX
+        sockets[0].close()
+    finally:
+        if os.path.exists(sock_path):
+            os.unlink(sock_path)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="requires unix sockets")
+def test_bind_sockets_fd(tmp_path: Path) -> None:  # pragma: py-win32
+    # Create a socket, then bind via its file descriptor.
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.bind(("127.0.0.1", 0))
+    fd = listener.fileno()
+    config = Config(app=asgi_app, bind=[f"fd://{fd}"])
+    sockets = config.bind_sockets()
+    assert len(sockets) == 1
+    sockets[0].close()
+    listener.close()

tests/test_main.py

@@ -117,6 +117,20 @@ async def test_exit_on_create_server_with_invalid_host() -> None:
     assert exc_info.value.code == 1
 
 
+async def test_run_with_bind(unused_tcp_port: int) -> None:
+    config = Config(app=app, bind=[f"127.0.0.1:{unused_tcp_port}"], loop="asyncio", limit_max_requests=1)
+    async with run_server(config):
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
+    assert response.status_code == 204
+
+
+async def test_run_with_bind_multiple() -> None:
+    config = Config(app=app, bind=["127.0.0.1:0", "127.0.0.1:0"], loop="asyncio", limit_max_requests=1)
+    async with run_server(config):
+        pass  # Startup itself validates multiple sockets work
+
+
 def test_deprecated_server_state_from_main() -> None:
     with pytest.deprecated_call(
         match="uvicorn.main.ServerState is deprecated, use uvicorn.server.ServerState instead."

uvicorn/config.py

@@ -183,6 +183,7 @@ class Config:
         port: int = 8000,
         uds: str | None = None,
         fd: int | None = None,
+        bind: list[str] | None = None,
         loop: LoopFactoryType | str = "auto",
         http: type[asyncio.Protocol] | HTTPProtocolType | str = "auto",
         ws: type[asyncio.Protocol] | WSProtocolType | str = "auto",
@@ -233,6 +234,7 @@ class Config:
         self.port = port
         self.uds = uds
         self.fd = fd
+        self.bind = bind
         self.loop = loop
         self.http = http
         self.ws = ws
@@ -342,13 +344,23 @@ class Config:
         if self.reload and self.workers > 1:
             logger.warning('"workers" flag is ignored when reloading is enabled.')
 
+        if self.bind is not None:
+            # Only flag options explicitly set to non-default values.
+            conflicting: list[str] = []
+            if self.host != "127.0.0.1":  # default host
+                conflicting.append("host")
+            if self.port != 8000:  # default port
+                conflicting.append("port")
+            if self.uds is not None:
+                conflicting.append("uds")
+            if self.fd is not None:
+                conflicting.append("fd")
+            if conflicting:
+                raise ValueError(f"'bind' is mutually exclusive with {', '.join(map(repr, conflicting))}")
+
     @property
     def asgi_version(self) -> Literal["2.0", "3.0"]:
-        mapping: dict[str, Literal["2.0", "3.0"]] = {
-            "asgi2": "2.0",
-            "asgi3": "3.0",
-            "wsgi": "3.0",
-        }
+        mapping: dict[str, Literal["2.0", "3.0"]] = {"asgi2": "2.0", "asgi3": "3.0", "wsgi": "3.0"}
         return mapping[self.interface]
 
     @property
@@ -496,25 +508,29 @@ class Config:
             return None
         return loop_factory(use_subprocess=self.use_subprocess)
 
-    def bind_socket(self) -> socket.socket:
+    def _bind_one(
+        self,
+        *,
+        uds: str | None = None,
+        fd: int | None = None,
+        host: str = "127.0.0.1",
+        port: int = 8000,
+    ) -> socket.socket:
         logger_args: list[str | int]
-        if self.uds:  # pragma: py-win32
-            path = self.uds
+        if uds:  # pragma: py-win32
             sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
             try:
-                sock.bind(path)
-                uds_perms = 0o666
-                os.chmod(self.uds, uds_perms)
+                sock.bind(uds)
+                os.chmod(uds, 0o666)
             except OSError as exc:  # pragma: full coverage
                 logger.error(exc)
                 sys.exit(1)
-
             message = "Uvicorn running on unix socket %s (Press CTRL+C to quit)"
             sock_name_format = "%s"
             color_message = "Uvicorn running on " + click.style(sock_name_format, bold=True) + " (Press CTRL+C to quit)"
-            logger_args = [self.uds]
-        elif self.fd:  # pragma: py-win32
-            sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
+            logger_args = [uds]
+        elif fd is not None:  # pragma: py-win32
+            sock = socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM)
             message = "Uvicorn running on socket %s (Press CTRL+C to quit)"
             fd_name_format = "%s"
             color_message = "Uvicorn running on " + click.style(fd_name_format, bold=True) + " (Press CTRL+C to quit)"
@@ -522,28 +538,53 @@ class Config:
         else:
             family = socket.AF_INET
             addr_format = "%s://%s:%d"
-
-            if self.host and ":" in self.host:  # pragma: full coverage
-                # It's an IPv6 address.
+            if host and ":" in host:  # pragma: full coverage
                 family = socket.AF_INET6
                 addr_format = "%s://[%s]:%d"
-
             sock = socket.socket(family=family)
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             try:
-                sock.bind((self.host, self.port))
+                sock.bind((host, port))
             except OSError as exc:  # pragma: full coverage
                 logger.error(exc)
                 sys.exit(1)
-
             message = f"Uvicorn running on {addr_format} (Press CTRL+C to quit)"
             color_message = "Uvicorn running on " + click.style(addr_format, bold=True) + " (Press CTRL+C to quit)"
             protocol_name = "https" if self.is_ssl else "http"
-            logger_args = [protocol_name, self.host, sock.getsockname()[1]]
+            logger_args = [protocol_name, host, sock.getsockname()[1]]
         logger.info(message, *logger_args, extra={"color_message": color_message})
         sock.set_inheritable(True)
         return sock
 
+    def bind_socket(self) -> socket.socket:
+        return self._bind_one(uds=self.uds, fd=self.fd, host=self.host, port=self.port)
+
+    def bind_sockets(self) -> list[socket.socket]:
+        if self.bind is None:
+            return [self.bind_socket()]
+
+        sockets: list[socket.socket] = []
+        for bind_str in self.bind:
+            if bind_str.startswith("unix:"):  # pragma: py-win32
+                sock = self._bind_one(uds=bind_str[5:])
+            elif bind_str.startswith("fd://"):  # pragma: py-win32
+                sock = self._bind_one(fd=int(bind_str[5:]))
+            else:
+                # Strip brackets for IPv6, then rsplit on last colon.
+                raw = bind_str.replace("[", "").replace("]", "")
+                try:
+                    host, port_str = raw.rsplit(":", 1)
+                    port = int(port_str)
+                except (ValueError, IndexError):
+                    host, port = raw, 8000
+                sock = self._bind_one(host=host, port=port)
+            sockets.append(sock)
+        return sockets
+
+    @property
+    def bind_unix_paths(self) -> list[str]:  # pragma: py-win32
+        return [b[5:] for b in (self.bind or []) if b.startswith("unix:")]
+
     @property
     def should_reload(self) -> bool:
         return isinstance(self.app, str) and self.reload

uvicorn/main.py

@@ -77,6 +77,15 @@ def print_version(ctx: click.Context, param: click.Parameter, value: bool) -> No
 )
 @click.option("--uds", type=str, default=None, help="Bind to a UNIX domain socket.")
 @click.option("--fd", type=int, default=None, help="Bind to socket from this file descriptor.")
+@click.option(
+    "--bind",
+    "-b",
+    "bind",
+    multiple=True,
+    help="Bind to a socket with this address. May be used multiple times. "
+    "Formats: HOST:PORT, [HOST]:PORT (IPv6), unix:PATH, fd://NUM. "
+    "Mutually exclusive with --host, --port, --uds, and --fd.",
+)
 @click.option("--reload", is_flag=True, default=False, help="Enable auto-reload.")
 @click.option(
     "--reload-dir",
@@ -377,6 +386,7 @@ def main(
     port: int,
     uds: str,
     fd: int,
+    bind: tuple[str, ...],
     loop: LoopFactoryType | str,
     http: HTTPProtocolType | str,
     ws: WSProtocolType | str,
@@ -427,6 +437,7 @@ def main(
         port=port,
         uds=uds,
         fd=fd,
+        bind=list(bind) or None,
         loop=loop,
         http=http,
         ws=ws,
@@ -480,6 +491,7 @@ def run(
     port: int = 8000,
     uds: str | None = None,
     fd: int | None = None,
+    bind: list[str] | None = None,
     loop: LoopFactoryType | str = "auto",
     http: type[asyncio.Protocol] | HTTPProtocolType | str = "auto",
     ws: type[asyncio.Protocol] | WSProtocolType | str = "auto",
@@ -533,6 +545,7 @@ def run(
         port=port,
         uds=uds,
         fd=fd,
+        bind=bind,
         loop=loop,
         http=http,
         ws=ws,
@@ -585,11 +598,11 @@ def run(
 
     try:
         if config.should_reload:
-            sock = config.bind_socket()
-            ChangeReload(config, target=server.run, sockets=[sock]).run()
+            socks = config.bind_sockets()
+            ChangeReload(config, target=server.run, sockets=socks).run()
         elif config.workers > 1:
-            sock = config.bind_socket()
-            Multiprocess(config, target=server.run, sockets=[sock]).run()
+            socks = config.bind_sockets()
+            Multiprocess(config, target=server.run, sockets=socks).run()
         else:
             server.run()
     except KeyboardInterrupt:
@@ -597,6 +610,9 @@ def run(
     finally:
         if config.uds and os.path.exists(config.uds):
             os.remove(config.uds)  # pragma: py-win32
+        for path in config.bind_unix_paths:  # pragma: py-win32
+            if os.path.exists(path):
+                os.remove(path)
 
     if not server.started and not config.should_reload and config.workers == 1:
         sys.exit(STARTUP_FAILURE)

uvicorn/server.py

@@ -77,6 +77,9 @@ class Server:
         if not config.loaded:
             config.load()
 
+        if sockets is None and config.bind is not None:
+            sockets = config.bind_sockets()
+
         self.lifespan = config.lifespan_class(config)
 
         message = "Started server process [%d]"
@@ -118,9 +121,7 @@ class Server:
             # Explicitly passed a list of open sockets.
             # We use this when the server is run from a Gunicorn worker.
 
-            def _share_socket(
-                sock: socket.SocketType,
-            ) -> socket.SocketType:  # pragma py-not-win32
+            def _share_socket(sock: socket.SocketType) -> socket.SocketType:  # pragma py-not-win32
                 # Windows requires the socket be explicitly shared across
                 # multiple workers (processes).
                 from socket import fromshare  # type: ignore[attr-defined]
@@ -191,10 +192,7 @@ class Server:
 
         if config.fd is not None:  # pragma: py-win32
             sock = listeners[0]
-            logger.info(
-                "Uvicorn running on socket %s (Press CTRL+C to quit)",
-                sock.getsockname(),
-            )
+            logger.info("Uvicorn running on socket %s (Press CTRL+C to quit)", sock.getsockname())
 
         elif config.uds is not None:  # pragma: py-win32
             logger.info("Uvicorn running on unix socket %s (Press CTRL+C to quit)", config.uds)