Add granian as a ASGI compliant web server. Provider better throughput stability,

litellm/proxy/proxy_cli.py

@@ -6,6 +6,7 @@
 import subprocess
 import sys
 import urllib.parse as urlparse
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional, Union
 
 import click
@@ -293,6 +294,62 @@ def _init_hypercorn_server(
         # hypercorn serve raises a type warning when passing a fast api app - even though fast API is a valid type
         asyncio.run(serve(app, config))  # type: ignore
 
+    @staticmethod
+    def _init_granian_server(
+        host: str,
+        port: int,
+        num_workers: int,
+        ssl_certfile_path: Optional[str],
+        ssl_keyfile_path: Optional[str],
+        max_requests_before_restart: Optional[int],
+        ciphers: Optional[str],
+        granian_runtime_threads: Optional[int] = None,
+    ) -> None:
+        """
+        Run the proxy with Granian (Rust-backed ASGI server, HTTP/1 + HTTP/2).
+
+        Uses a string import path so workers load ``litellm.proxy.proxy_server:app``
+        the same way as uvicorn's ``app=`` string target.
+        """
+        from granian import Granian
+        from granian.constants import Interfaces
+
+        print(  # noqa
+            f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} using Granian\033[0m\n"
+        )
+        if max_requests_before_restart is not None:
+            print(  # noqa
+                "\033[1;33mLiteLLM: --max_requests_before_restart is not supported by Granian "
+                "(Granian uses workers_lifetime in seconds, not a per-request limit).\033[0m\n"
+            )
+        if ciphers is not None:
+            print(  # noqa
+                "\033[1;33mLiteLLM: --ciphers is not applied when using --run_granian.\033[0m\n"
+            )
+
+        kwargs: dict[str, Any] = {
+            "target": "litellm.proxy.proxy_server:app",
+            "address": host,
+            "port": port,
+            "workers": max(1, num_workers),
+            "interface": Interfaces.ASGI,
+            "websockets": True,
+        }
+        if granian_runtime_threads is not None:
+            kwargs["runtime_threads"] = granian_runtime_threads
+        if ssl_certfile_path is not None and ssl_keyfile_path is not None:
+            print(  # noqa
+                f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
+            )
+            kwargs["ssl_cert"] = Path(ssl_certfile_path)
+            kwargs["ssl_key"] = Path(ssl_keyfile_path)
+        elif ssl_certfile_path is not None or ssl_keyfile_path is not None:
+            raise click.ClickException(
+                "Both --ssl_certfile_path and --ssl_keyfile_path are required for SSL."
+            )
+
+        Granian(**kwargs).serve()
+
     @staticmethod
     def _run_gunicorn_server(
         host: str,
@@ -483,9 +540,23 @@ def _maybe_setup_prometheus_multiproc_dir(
 @click.option(
     "--num_workers",
     default=DEFAULT_NUM_WORKERS_LITELLM_PROXY,
-    help="Number of uvicorn / gunicorn workers to spin up. Default is 1 (from DEFAULT_NUM_WORKERS_LITELLM_PROXY)",
+    help=(
+        "Number of worker processes for uvicorn / gunicorn, or Granian worker processes "
+        "(--workers). Default is 1 (from DEFAULT_NUM_WORKERS_LITELLM_PROXY). "
+        "With --run_granian, use --granian_threads for runtime threads per worker."
+    ),
     envvar="NUM_WORKERS",
 )
+@click.option(
+    "--granian_threads",
+    default=None,
+    type=click.IntRange(min=1),
+    help=(
+        "Only with --run_granian: runtime threads per worker process "
+        "(Granian --runtime-threads / GRANIAN_RUNTIME_THREADS). Omit to use Granian's default (1)."
+    ),
+    envvar="GRANIAN_RUNTIME_THREADS",
+)
 @click.option("--api_base", default=None, help="API base URL.")
 @click.option(
     "--api_version",
@@ -624,6 +695,15 @@ def _maybe_setup_prometheus_multiproc_dir(
     is_flag=True,
     help="Starts proxy via hypercorn, instead of uvicorn (supports HTTP/2)",
 )
+@click.option(
+    "--run_granian",
+    default=False,
+    is_flag=True,
+    help=(
+        "Starts proxy via Granian (Rust ASGI server) instead of uvicorn. "
+        "Requires Python 3.10+ and the `granian` package."
+    ),
+)
 @click.option(
     "--ssl_keyfile_path",
     default=None,
@@ -728,6 +808,7 @@ def run_server(  # noqa: PLR0915
     test,
     local,
     num_workers,
+    granian_threads,
     test_async,
     iam_token_db_auth,
     num_requests,
@@ -737,6 +818,7 @@ def run_server(  # noqa: PLR0915
     version,
     run_gunicorn,
     run_hypercorn,
+    run_granian,
     ssl_keyfile_path,
     ssl_certfile_path,
     ciphers,
@@ -821,12 +903,22 @@ def run_server(  # noqa: PLR0915
             config=config,
             use_queue=use_queue,
         )
-        try:
-            import uvicorn
-        except Exception:
-            raise ImportError(
-                "uvicorn, gunicorn needs to be imported. Run - `pip install 'litellm[proxy]'`"
-            )
+        if run_granian:
+            try:
+                import granian  # noqa: F401
+            except ImportError as e:
+                raise ImportError(
+                    "granian must be installed to use --run_granian. "
+                    "Run `pip install granian` or `pip install 'litellm[proxy]'` "
+                    "(Granian requires Python 3.10+)."
+                ) from e
+        else:
+            try:
+                import uvicorn
+            except Exception:
+                raise ImportError(
+                    "uvicorn, gunicorn needs to be imported. Run - `pip install 'litellm[proxy]'`"
+                )
 
         db_connection_pool_limit = 100
         # Starts optional due to config fallback checks; guaranteed non-None before use.
@@ -1112,7 +1204,7 @@ def run_server(  # noqa: PLR0915
         # Optional: recycle uvicorn workers after N requests
         if max_requests_before_restart is not None:
             uvicorn_args["limit_max_requests"] = max_requests_before_restart
-        if run_gunicorn is False and run_hypercorn is False:
+        if run_gunicorn is False and run_hypercorn is False and run_granian is False:
             if ssl_certfile_path is not None and ssl_keyfile_path is not None:
                 print(  # noqa
                     f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"  # noqa
@@ -1154,6 +1246,17 @@ def run_server(  # noqa: PLR0915
                 ssl_keyfile_path=ssl_keyfile_path,
                 ciphers=ciphers,
             )
+        elif run_granian is True:
+            ProxyInitializationHelpers._init_granian_server(
+                host=host,
+                port=port,
+                num_workers=num_workers,
+                ssl_certfile_path=ssl_certfile_path,
+                ssl_keyfile_path=ssl_keyfile_path,
+                max_requests_before_restart=max_requests_before_restart,
+                ciphers=ciphers,
+                granian_runtime_threads=granian_threads,
+            )
 
 
 if __name__ == "__main__":

pyproject.toml

@@ -40,6 +40,7 @@ Documentation = "https://docs.litellm.ai"
 proxy = [
     "gunicorn==23.0.0",
     "uvicorn==0.33.0",
+    "granian==2.5.7",
     "uvloop==0.21.0; sys_platform != 'win32'",
     "fastapi==0.124.4",
     "backoff==2.2.1",

tests/test_litellm/proxy/test_proxy_cli.py

@@ -1,7 +1,11 @@
 import os
 import sys
+from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import click
+import fastapi
 import pytest
 
 sys.path.insert(
@@ -231,6 +235,96 @@ def test_init_hypercorn_server(self, mock_print, mock_asyncio_run):
             mock_app, "localhost", 8000, "cert.pem", "key.pem", "ECDHE"
         )
 
+    @patch("granian.Granian")
+    @patch("builtins.print")
+    def test_init_granian_server(self, mock_print, mock_granian_cls):
+        pytest.importorskip("granian")
+        mock_server = MagicMock()
+        mock_granian_cls.return_value = mock_server
+        fake_interfaces = SimpleNamespace(ASGI="asgi")
+        with patch("granian.constants.Interfaces", fake_interfaces):
+            ProxyInitializationHelpers._init_granian_server(
+                host="0.0.0.0",
+                port=4000,
+                num_workers=2,
+                ssl_certfile_path=None,
+                ssl_keyfile_path=None,
+                max_requests_before_restart=None,
+                ciphers=None,
+                granian_runtime_threads=None,
+            )
+        mock_granian_cls.assert_called_once()
+        call_kwargs = mock_granian_cls.call_args.kwargs
+        assert call_kwargs["target"] == "litellm.proxy.proxy_server:app"
+        assert call_kwargs["address"] == "0.0.0.0"
+        assert call_kwargs["port"] == 4000
+        assert call_kwargs["workers"] == 2
+        assert call_kwargs["interface"] == "asgi"
+        assert call_kwargs["websockets"] is True
+        assert "runtime_threads" not in call_kwargs
+        mock_server.serve.assert_called_once()
+
+    @patch("granian.Granian")
+    @patch("builtins.print")
+    def test_init_granian_server_runtime_threads(self, mock_print, mock_granian_cls):
+        pytest.importorskip("granian")
+        mock_server = MagicMock()
+        mock_granian_cls.return_value = mock_server
+        fake_interfaces = SimpleNamespace(ASGI="asgi")
+        with patch("granian.constants.Interfaces", fake_interfaces):
+            ProxyInitializationHelpers._init_granian_server(
+                host="0.0.0.0",
+                port=4000,
+                num_workers=1,
+                ssl_certfile_path=None,
+                ssl_keyfile_path=None,
+                max_requests_before_restart=None,
+                ciphers=None,
+                granian_runtime_threads=4,
+            )
+        assert mock_granian_cls.call_args.kwargs["runtime_threads"] == 4
+
+    @patch("granian.Granian")
+    @patch("builtins.print")
+    def test_init_granian_server_ssl(self, mock_print, mock_granian_cls):
+        pytest.importorskip("granian")
+        mock_server = MagicMock()
+        mock_granian_cls.return_value = mock_server
+        fake_interfaces = SimpleNamespace(ASGI="asgi")
+        with patch("granian.constants.Interfaces", fake_interfaces):
+            ProxyInitializationHelpers._init_granian_server(
+                host="0.0.0.0",
+                port=4000,
+                num_workers=1,
+                ssl_certfile_path="/path/to/cert.pem",
+                ssl_keyfile_path="/path/to/key.pem",
+                max_requests_before_restart=None,
+                ciphers=None,
+                granian_runtime_threads=None,
+            )
+        call_kwargs = mock_granian_cls.call_args.kwargs
+        assert call_kwargs["ssl_cert"] == Path("/path/to/cert.pem")
+        assert call_kwargs["ssl_key"] == Path("/path/to/key.pem")
+        mock_server.serve.assert_called_once()
+
+    @patch("granian.Granian")
+    def test_init_granian_server_ssl_requires_cert_and_key(self, mock_granian_cls):
+        pytest.importorskip("granian")
+        fake_interfaces = SimpleNamespace(ASGI="asgi")
+        with patch("granian.constants.Interfaces", fake_interfaces):
+            with pytest.raises(click.ClickException, match="Both --ssl_certfile_path"):
+                ProxyInitializationHelpers._init_granian_server(
+                    host="0.0.0.0",
+                    port=4000,
+                    num_workers=1,
+                    ssl_certfile_path="/path/to/cert.pem",
+                    ssl_keyfile_path=None,
+                    max_requests_before_restart=None,
+                    ciphers=None,
+                    granian_runtime_threads=None,
+                )
+        mock_granian_cls.assert_not_called()
+
     @patch("subprocess.Popen")
     def test_run_ollama_serve(self, mock_popen):
         # Execute