feat: add distributed mode (experimental)

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

Author: mudler

README.md

@@ -154,6 +154,7 @@ For older news and full release notes, see [GitHub Releases](https://github.com/
 - [Object Detection](https://localai.io/features/object-detection/)
 - [Reranker API](https://localai.io/features/reranker/)
 - [P2P Inferencing](https://localai.io/features/distribute/)
+- [Distributed Mode](https://localai.io/features/distributed-mode/) — Horizontal scaling with PostgreSQL + NATS
 - [Model Context Protocol (MCP)](https://localai.io/docs/features/mcp/)
 - [Built-in Agents](https://localai.io/features/agents/) — Autonomous AI agents with tool use, RAG, skills, SSE streaming, and [Agent Hub](https://agenthub.localai.io)
 - [Backend Gallery](https://localai.io/backends/) — Install/remove backends on the fly via OCI images

backend/backend.proto

@@ -51,6 +51,7 @@ service Backend {
   rpc StartQuantization(QuantizationRequest) returns (QuantizationJobResult) {}
   rpc QuantizationProgress(QuantizationProgressRequest) returns (stream QuantizationProgressUpdate) {}
   rpc StopQuantization(QuantizationStopRequest) returns (Result) {}
+
 }
 
 // Define the empty request
@@ -676,3 +677,4 @@ message QuantizationProgressUpdate {
 message QuantizationStopRequest {
   string job_id = 1;
 }
+

backend/cpp/llama-cpp/grpc-server.cpp

@@ -22,8 +22,10 @@
 #include <grpcpp/ext/proto_server_reflection_plugin.h>
 #include <grpcpp/grpcpp.h>
 #include <grpcpp/health_check_service_interface.h>
+#include <grpcpp/security/server_credentials.h>
 #include <regex>
 #include <atomic>
+#include <cstdlib>
 #include <mutex>
 #include <signal.h>
 #include <thread>
@@ -37,6 +39,47 @@ using grpc::Server;
 using grpc::ServerBuilder;
 using grpc::ServerContext;
 using grpc::Status;
+
+// gRPC bearer token auth via AuthMetadataProcessor for distributed mode.
+// Reads LOCALAI_GRPC_AUTH_TOKEN from the environment. When set, rejects
+// requests without a matching "authorization: Bearer <token>" metadata header.
+class TokenAuthMetadataProcessor : public grpc::AuthMetadataProcessor {
+public:
+    explicit TokenAuthMetadataProcessor(const std::string& token) : token_(token) {}
+
+    bool IsBlocking() const override { return false; }
+
+    grpc::Status Process(const InputMetadata& auth_metadata,
+                         grpc::AuthContext* /*context*/,
+                         OutputMetadata* /*consumed_auth_metadata*/,
+                         OutputMetadata* /*response_metadata*/) override {
+        auto it = auth_metadata.find("authorization");
+        if (it != auth_metadata.end()) {
+            std::string expected = "Bearer " + token_;
+            std::string got(it->second.data(), it->second.size());
+            // Constant-time comparison
+            if (expected.size() == got.size() && ct_memcmp(expected.data(), got.data(), expected.size()) == 0) {
+                return grpc::Status::OK;
+            }
+        }
+        return grpc::Status(grpc::StatusCode::UNAUTHENTICATED, "invalid token");
+    }
+
+private:
+    std::string token_;
+
+    // Minimal constant-time comparison (avoids OpenSSL dependency)
+    static int ct_memcmp(const void* a, const void* b, size_t n) {
+        const unsigned char* pa = static_cast<const unsigned char*>(a);
+        const unsigned char* pb = static_cast<const unsigned char*>(b);
+        unsigned char result = 0;
+        for (size_t i = 0; i < n; i++) {
+            result |= pa[i] ^ pb[i];
+        }
+        return result;
+    }
+};
+
 // END LocalAI
 
 
@@ -2760,11 +2803,24 @@ int main(int argc, char** argv) {
     BackendServiceImpl service(ctx_server);
 
     ServerBuilder builder;
-    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
+    // Add bearer token auth via AuthMetadataProcessor if LOCALAI_GRPC_AUTH_TOKEN is set
+    const char* auth_token = std::getenv("LOCALAI_GRPC_AUTH_TOKEN");
+    std::shared_ptr<grpc::ServerCredentials> creds;
+    if (auth_token != nullptr && auth_token[0] != '\0') {
+        creds = grpc::InsecureServerCredentials();
+        creds->SetAuthMetadataProcessor(
+            std::make_shared<TokenAuthMetadataProcessor>(auth_token));
+        std::cout << "gRPC auth enabled via LOCALAI_GRPC_AUTH_TOKEN" << std::endl;
+    } else {
+        creds = grpc::InsecureServerCredentials();
+    }
+
+    builder.AddListeningPort(server_address, creds);
     builder.RegisterService(&service);
     builder.SetMaxMessageSize(50 * 1024 * 1024); // 50MB
     builder.SetMaxSendMessageSize(50 * 1024 * 1024); // 50MB
     builder.SetMaxReceiveMessageSize(50 * 1024 * 1024); // 50MB
+
     std::unique_ptr<Server> server(builder.BuildAndStart());
    // run the HTTP server in a thread - see comment below
     std::thread t([&]()

backend/python/ace-step/backend.py

@@ -19,6 +19,10 @@
 import backend_pb2
 import backend_pb2_grpc
 import grpc
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'common'))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'common'))
+from grpc_auth import get_auth_interceptors
+
 from acestep.inference import (
     GenerationParams,
     GenerationConfig,
@@ -444,6 +448,8 @@ def serve(address):
             ("grpc.max_send_message_length", 50 * 1024 * 1024),
             ("grpc.max_receive_message_length", 50 * 1024 * 1024),
         ],
+    
+        interceptors=get_auth_interceptors(),
     )
     backend_pb2_grpc.add_BackendServicer_to_server(BackendServicer(), server)
     server.add_insecure_port(address)

backend/python/chatterbox/backend.py

@@ -16,6 +16,10 @@
 from chatterbox.tts import ChatterboxTTS
 from chatterbox.mtl_tts import ChatterboxMultilingualTTS
 import grpc
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'common'))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'common'))
+from grpc_auth import get_auth_interceptors
+
 import tempfile
 
 def is_float(s):
@@ -225,7 +229,9 @@ def serve(address):
             ('grpc.max_message_length', 50 * 1024 * 1024),  # 50MB
             ('grpc.max_send_message_length', 50 * 1024 * 1024),  # 50MB
             ('grpc.max_receive_message_length', 50 * 1024 * 1024),  # 50MB
-        ])
+        ],
+        interceptors=get_auth_interceptors(),
+    )
     backend_pb2_grpc.add_BackendServicer_to_server(BackendServicer(), server)
     server.add_insecure_port(address)
     server.start()

backend/python/common/grpc_auth.py

@@ -0,0 +1,78 @@
+"""Shared gRPC bearer token authentication interceptor for LocalAI Python backends.
+
+When the environment variable LOCALAI_GRPC_AUTH_TOKEN is set, requests without
+a valid Bearer token in the 'authorization' metadata header are rejected with
+UNAUTHENTICATED. When the variable is empty or unset, no authentication is
+performed (backward compatible).
+"""
+
+import hmac
+import os
+
+import grpc
+
+
+class _AbortHandler(grpc.RpcMethodHandler):
+    """A method handler that immediately aborts with UNAUTHENTICATED."""
+
+    def __init__(self):
+        self.request_streaming = False
+        self.response_streaming = False
+        self.request_deserializer = None
+        self.response_serializer = None
+        self.unary_unary = self._abort
+        self.unary_stream = None
+        self.stream_unary = None
+        self.stream_stream = None
+
+    @staticmethod
+    def _abort(request, context):
+        context.abort(grpc.StatusCode.UNAUTHENTICATED, "invalid token")
+
+
+class TokenAuthInterceptor(grpc.ServerInterceptor):
+    """Sync gRPC server interceptor that validates a bearer token."""
+
+    def __init__(self, token: str):
+        self._token = token
+        self._abort_handler = _AbortHandler()
+
+    def intercept_service(self, continuation, handler_call_details):
+        metadata = dict(handler_call_details.invocation_metadata)
+        auth = metadata.get("authorization", "")
+        expected = "Bearer " + self._token
+        if not hmac.compare_digest(auth, expected):
+            return self._abort_handler
+        return continuation(handler_call_details)
+
+
+class AsyncTokenAuthInterceptor(grpc.aio.ServerInterceptor):
+    """Async gRPC server interceptor that validates a bearer token."""
+
+    def __init__(self, token: str):
+        self._token = token
+
+    async def intercept_service(self, continuation, handler_call_details):
+        metadata = dict(handler_call_details.invocation_metadata)
+        auth = metadata.get("authorization", "")
+        expected = "Bearer " + self._token
+        if not hmac.compare_digest(auth, expected):
+            return _AbortHandler()
+        return await continuation(handler_call_details)
+
+
+def get_auth_interceptors(*, aio: bool = False):
+    """Return a list of gRPC interceptors for bearer token auth.
+
+    Args:
+        aio: If True, return async-compatible interceptors for grpc.aio.server().
+             If False (default), return sync interceptors for grpc.server().
+
+    Returns an empty list when LOCALAI_GRPC_AUTH_TOKEN is not set.
+    """
+    token = os.environ.get("LOCALAI_GRPC_AUTH_TOKEN", "")
+    if not token:
+        return []
+    if aio:
+        return [AsyncTokenAuthInterceptor(token)]
+    return [TokenAuthInterceptor(token)]

backend/python/coqui/backend.py

@@ -15,6 +15,10 @@
 from TTS.api import TTS
 
 import grpc
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'common'))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'common'))
+from grpc_auth import get_auth_interceptors
+
 
 
 _ONE_DAY_IN_SECONDS = 60 * 60 * 24
@@ -93,7 +97,9 @@ def serve(address):
             ('grpc.max_message_length', 50 * 1024 * 1024),  # 50MB
             ('grpc.max_send_message_length', 50 * 1024 * 1024),  # 50MB
             ('grpc.max_receive_message_length', 50 * 1024 * 1024),  # 50MB
-        ])
+        ],
+        interceptors=get_auth_interceptors(),
+    )
     backend_pb2_grpc.add_BackendServicer_to_server(BackendServicer(), server)
     server.add_insecure_port(address)
     server.start()

backend/python/diffusers/backend.py

@@ -22,6 +22,10 @@
 import backend_pb2_grpc
 
 import grpc
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'common'))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'common'))
+from grpc_auth import get_auth_interceptors
+
 
 # Import dynamic loader for pipeline discovery
 from diffusers_dynamic_loader import (
@@ -1042,7 +1046,9 @@ def serve(address):
             ('grpc.max_message_length', 50 * 1024 * 1024),  # 50MB
             ('grpc.max_send_message_length', 50 * 1024 * 1024),  # 50MB
             ('grpc.max_receive_message_length', 50 * 1024 * 1024),  # 50MB
-        ])
+        ],
+        interceptors=get_auth_interceptors(),
+    )
     backend_pb2_grpc.add_BackendServicer_to_server(BackendServicer(), server)
     server.add_insecure_port(address)
     server.start()

backend/python/faster-qwen3-tts/backend.py

@@ -15,6 +15,10 @@
 import soundfile as sf
 
 import grpc
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'common'))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'common'))
+from grpc_auth import get_auth_interceptors
+
 
 
 def is_float(s):
@@ -165,6 +169,8 @@ def serve(address):
             ('grpc.max_send_message_length', 50 * 1024 * 1024),
             ('grpc.max_receive_message_length', 50 * 1024 * 1024),
         ]
+    ,
+        interceptors=get_auth_interceptors(),
     )
     backend_pb2_grpc.add_BackendServicer_to_server(BackendServicer(), server)
     server.add_insecure_port(address)

backend/python/faster-whisper/backend.py

@@ -14,6 +14,10 @@
 from faster_whisper import WhisperModel
 
 import grpc
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'common'))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'common'))
+from grpc_auth import get_auth_interceptors
+
 
 
 _ONE_DAY_IN_SECONDS = 60 * 60 * 24
@@ -70,7 +74,9 @@ def serve(address):
             ('grpc.max_message_length', 50 * 1024 * 1024),  # 50MB
             ('grpc.max_send_message_length', 50 * 1024 * 1024),  # 50MB
             ('grpc.max_receive_message_length', 50 * 1024 * 1024),  # 50MB
-        ])
+        ],
+        interceptors=get_auth_interceptors(),
+    )
     backend_pb2_grpc.add_BackendServicer_to_server(BackendServicer(), server)
     server.add_insecure_port(address)
     server.start()