Add granian as a ASGI compliant web server. Provider better throughput stability, (#26027)

* Add granian as a ASGI compliant web server. Provides better stability, 10-20 RPS improvement under standard LT conditions.

TODO: Verify poetry lock details and add locust numbers to PR

* Update granian version in license_cache.json and pyproject.toml to 2.5.7

* Enhance proxy CLI tests by adding SSL initialization checks for Granian server. Remove Python version skip conditions and implement tests to ensure SSL certificate and key are required for server initialization.

* update uv lock to fix granian import error

Author: harish-berri

license_cache.json

@@ -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__":

litellm/proxy/proxy_cli.py

@@ -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",

pyproject.toml

@@ -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