Files
LocalAI/core/http/routes/openai.go
Ettore Di Giacinto 2f3d5b2e74 feat(distributed): make MCP execution and discovery a selection
mcp.tools.execute and mcp.discovery were the only NATS subjects that
combined a queue group with a reply, and no carrier in this design
provides both. They never needed one: a queue group is a way of choosing
a subscriber, and choosing is a query.

The frontend now lists the approved, non-draining agent nodes, asks the
node_connections table in one joined statement which of those tunnels a
live replica holds, prefers one this replica holds so the call skips the
relay hop, and issues an ordinary control RPC on the path task 4 already
mounted. A peer-held tunnel is reached through the relay. That is a
choice a broker's hidden balancing could not make.

The selection reads presence and nothing else. It is filtered only on
node type and on the two statuses an operator controls, never on a health
verdict written on another clock, because refusing a worker that is
connected and answering is the same defect as picking one that is gone.
An empty fleet answers ErrNoAgentWorker, which is deliberately neither
ErrWorkerUnroutable nor anything cluster.IsWorkerAnswer accepts: nothing
was asked of any worker, so no reap guard may act on it.

A reply carrying an Error is the worker's own answer and is returned
unchanged; it is never offered to a second worker, which would turn "this
MCP server rejected your arguments" into "the fleet is broken" and could
run a tool twice. A call that never reached a worker is retried against a
different pick, at most three times, and whatever error is finally
returned is returned unwrapped so its identity survives the loop.

MCP prompts and resources now answer 501 in distributed mode instead of
an empty 200. They are served only from sessions the frontend holds, and
in distributed mode it holds none. That gap predates the removal of the
bus and is not closed by it; this only stops it being silent.

Agent workers keep every other subject, including nodes.<id>.backend.stop.
Their minted JWT loses the two MCP subjects and keeps a non-empty allow
list, because NATS reads an empty one as no restriction at all.

Assisted-by: Claude Opus 5 [claude-code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
2026-09-20 03:05:34 +00:00

281 lines
14 KiB
Go

package routes
import (
"github.com/labstack/echo/v4"
"github.com/mudler/LocalAI/core/application"
"github.com/mudler/LocalAI/core/config"
"github.com/mudler/LocalAI/core/http/endpoints/localai"
"github.com/mudler/LocalAI/core/http/endpoints/openai"
"github.com/mudler/LocalAI/core/http/middleware"
"github.com/mudler/LocalAI/core/schema"
compressionservice "github.com/mudler/LocalAI/core/services/compression"
"github.com/mudler/LocalAI/core/services/routing/pii"
"github.com/mudler/LocalAI/core/services/routing/piiadapter"
"github.com/mudler/LocalAI/core/services/routing/router"
"github.com/mudler/LocalAI/pkg/tokens"
)
func RegisterOpenAIRoutes(app *echo.Echo,
re *middleware.RequestExtractor,
application *application.Application,
) {
// openAI compatible API endpoint
traceMiddleware := middleware.TraceMiddleware(application)
usageMiddleware := middleware.UsageMiddleware(application.StatsRecorder(), application.FallbackUser())
// X-LocalAI-Node attribution middleware: wraps the response writer and
// stamps the header on first write when --expose-node-header is on. No-op
// otherwise. Applied to every inference path that routes through
// ml.Load (chat, completion, embeddings, audio transcriptions/speech,
// image generation/inpainting) so distributed-mode operators can observe
// which worker served each request.
nodeHeaderMiddleware := middleware.ExposeNodeHeader(application.ApplicationConfig())
// realtime
// TODO: Modify/disable the API key middleware for this endpoint to allow ephemeral keys created by sessions
app.GET("/v1/realtime", openai.Realtime(application))
app.POST("/v1/realtime/sessions", openai.RealtimeTranscriptionSession(application), traceMiddleware)
app.POST("/v1/realtime/transcription_session", openai.RealtimeTranscriptionSession(application), traceMiddleware)
app.POST("/v1/realtime/calls", openai.RealtimeCalls(application), traceMiddleware)
// How the MCP endpoints reach an agent worker; nil outside distributed mode.
agentControl := mcpAgentControl(application)
// chat
chatCompressor := compressionservice.New(
compressionservice.CounterFunc(tokens.CountMessages),
compressionservice.NewInferenceSummarizer(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig()),
)
chatHandler := openai.ChatEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.TemplatesEvaluator(), application.ApplicationConfig(), agentControl, application.LocalAIAssistant(), chatCompressor)
chatMiddleware := []echo.MiddlewareFunc{
nodeHeaderMiddleware,
usageMiddleware,
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_CHAT)),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenAIRequest) }),
func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
if err := re.SetOpenAIRequest(c); err != nil {
return err
}
return next(c)
}
},
// RouteModel runs AFTER the schema-specific request parser so
// the classifier sees a populated *schema.OpenAIRequest. When
// the resolved model has a Router config, the middleware
// rewrites input.Model to the chosen candidate, swaps
// MODEL_CONFIG, and stamps RequestedModel/ServedModel for the
// usage log. Models without a Router pass through.
middleware.RouteModel(
application.ModelConfigLoader(),
application.ApplicationConfig(),
application.RouterDecisions(),
application.FallbackUser(),
middleware.OpenAIProbe,
router.SourceChat,
middleware.NewClassifierDeps(application),
),
// Admission control runs after RouteModel so the SERVED
// model's limits apply — a router fanout that lands on a
// saturated downstream gets rejected even when the requested
// router-model has slack.
middleware.AdmissionControl(application.AdmissionLimiter(), application.PIIEvents()),
// PII redaction runs after RouteModel has resolved the actual served
// model and before compression. This makes per-model PII
// configs honour the routed target (e.g., a router fans out to
// claude-strict; that model's pii block applies, not the router
// model's), and prevents the compressor from seeing unredacted input.
pii.RequestMiddleware(application.PIIRedactor(), application.PIIEvents(), piiadapter.OpenAI(), application.FallbackUser(), pii.WithNERResolver(application.PIINERResolver()), pii.WithPolicyResolver(application.PIIPolicyResolver())),
}
app.POST("/v1/chat/completions", chatHandler, chatMiddleware...)
app.POST("/chat/completions", chatHandler, chatMiddleware...)
// edit
editHandler := openai.EditEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.TemplatesEvaluator(), application.ApplicationConfig())
editMiddleware := []echo.MiddlewareFunc{
usageMiddleware,
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_EDIT)),
re.BuildConstantDefaultModelNameMiddleware("gpt-4o"),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenAIRequest) }),
func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
if err := re.SetOpenAIRequest(c); err != nil {
return err
}
return next(c)
}
},
pii.RequestMiddleware(application.PIIRedactor(), application.PIIEvents(), piiadapter.OpenAICompletion(), application.FallbackUser(), pii.WithNERResolver(application.PIINERResolver()), pii.WithPolicyResolver(application.PIIPolicyResolver())),
}
app.POST("/v1/edits", editHandler, editMiddleware...)
app.POST("/edits", editHandler, editMiddleware...)
// completion
completionHandler := openai.CompletionEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.TemplatesEvaluator(), application.ApplicationConfig())
completionMiddleware := []echo.MiddlewareFunc{
nodeHeaderMiddleware,
usageMiddleware,
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_COMPLETION)),
re.BuildConstantDefaultModelNameMiddleware("gpt-4o"),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenAIRequest) }),
func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
if err := re.SetOpenAIRequest(c); err != nil {
return err
}
return next(c)
}
},
pii.RequestMiddleware(application.PIIRedactor(), application.PIIEvents(), piiadapter.OpenAICompletion(), application.FallbackUser(), pii.WithNERResolver(application.PIINERResolver()), pii.WithPolicyResolver(application.PIIPolicyResolver())),
}
app.POST("/v1/completions", completionHandler, completionMiddleware...)
app.POST("/completions", completionHandler, completionMiddleware...)
app.POST("/v1/engines/:model/completions", completionHandler, completionMiddleware...)
// moderation
moderationHandler := openai.ModerationEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.TemplatesEvaluator(), application.ApplicationConfig())
moderationMiddleware := []echo.MiddlewareFunc{
nodeHeaderMiddleware,
usageMiddleware,
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_COMPLETION)),
re.BuildConstantDefaultModelNameMiddleware("gpt-4o"),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.ModerationRequest) }),
middleware.AdmissionControl(application.AdmissionLimiter(), application.PIIEvents()),
}
app.POST("/v1/moderations", moderationHandler, moderationMiddleware...)
app.POST("/moderations", moderationHandler, moderationMiddleware...)
// embeddings
embeddingHandler := openai.EmbeddingsEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.TemplatesEvaluator(), application.ApplicationConfig())
embeddingMiddleware := []echo.MiddlewareFunc{
nodeHeaderMiddleware,
usageMiddleware,
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_EMBEDDINGS)),
re.BuildConstantDefaultModelNameMiddleware("gpt-4o"),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenAIRequest) }),
func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
if err := re.SetOpenAIRequest(c); err != nil {
return err
}
return next(c)
}
},
pii.RequestMiddleware(application.PIIRedactor(), application.PIIEvents(), piiadapter.OpenAICompletion(), application.FallbackUser(), pii.WithNERResolver(application.PIINERResolver()), pii.WithPolicyResolver(application.PIIPolicyResolver())),
}
app.POST("/v1/embeddings", embeddingHandler, embeddingMiddleware...)
app.POST("/embeddings", embeddingHandler, embeddingMiddleware...)
app.POST("/v1/engines/:model/embeddings", embeddingHandler, embeddingMiddleware...)
audioHandler := openai.TranscriptEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig())
audioMiddleware := []echo.MiddlewareFunc{
nodeHeaderMiddleware,
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_TRANSCRIPT)),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenAIRequest) }),
func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
if err := re.SetOpenAIRequest(c); err != nil {
return err
}
return next(c)
}
},
}
// audio
app.POST("/v1/audio/transcriptions", audioHandler, audioMiddleware...)
app.POST("/audio/transcriptions", audioHandler, audioMiddleware...)
diarizationHandler := openai.DiarizationEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig())
diarizationMiddleware := []echo.MiddlewareFunc{
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_DIARIZATION)),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenAIRequest) }),
func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
if err := re.SetOpenAIRequest(c); err != nil {
return err
}
return next(c)
}
},
}
app.POST("/v1/audio/diarization", diarizationHandler, diarizationMiddleware...)
app.POST("/audio/diarization", diarizationHandler, diarizationMiddleware...)
soundClassificationHandler := openai.SoundClassificationEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig())
soundClassificationMiddleware := []echo.MiddlewareFunc{
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_SOUND_CLASSIFICATION)),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenAIRequest) }),
func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
if err := re.SetOpenAIRequest(c); err != nil {
return err
}
return next(c)
}
},
}
app.POST("/v1/audio/classification", soundClassificationHandler, soundClassificationMiddleware...)
app.POST("/audio/classification", soundClassificationHandler, soundClassificationMiddleware...)
audioSpeechHandler := localai.TTSEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig(), application.VoiceProfileStore())
audioSpeechMiddleware := []echo.MiddlewareFunc{
nodeHeaderMiddleware,
traceMiddleware,
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_TTS)),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.TTSRequest) }),
}
app.POST("/v1/audio/speech", audioSpeechHandler, audioSpeechMiddleware...)
app.POST("/audio/speech", audioSpeechHandler, audioSpeechMiddleware...)
app.GET("/v1/audio/voices", localai.TTSVoicesEndpoint(application.ModelConfigLoader(), application.AuthDB()))
app.GET("/audio/voices", localai.TTSVoicesEndpoint(application.ModelConfigLoader(), application.AuthDB()))
// images
imageHandler := openai.ImageEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig())
imageMiddleware := []echo.MiddlewareFunc{
nodeHeaderMiddleware,
traceMiddleware,
// Default: use the first available image generation model
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_IMAGE)),
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenAIRequest) }),
func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
if err := re.SetOpenAIRequest(c); err != nil {
return err
}
return next(c)
}
},
}
app.POST("/v1/images/generations", imageHandler, imageMiddleware...)
app.POST("/images/generations", imageHandler, imageMiddleware...)
// inpainting endpoint (image + mask) - reuse same middleware config as images
inpaintingHandler := openai.InpaintingEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig())
app.POST("/v1/images/inpainting", inpaintingHandler, imageMiddleware...)
app.POST("/images/inpainting", inpaintingHandler, imageMiddleware...)
// upscale endpoint - reuse same middleware config as images
upscaleHandler := openai.UpscaleEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig())
app.POST("/v1/images/upscale", upscaleHandler, imageMiddleware...)
app.POST("/images/upscale", upscaleHandler, imageMiddleware...)
// List models
app.GET("/v1/models", openai.ListModelsEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig(), application.AuthDB()))
app.GET("/models", openai.ListModelsEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig(), application.AuthDB()))
// List models enriched with capabilities + input/output modalities
// (LocalAI-specific, additive superset of /v1/models).
capabilitiesHandler := openai.ListModelCapabilitiesEndpoint(application.ModelConfigLoader(), application.ModelLoader(), application.ApplicationConfig(), application.AuthDB())
app.GET("/v1/models/capabilities", capabilitiesHandler)
app.GET("/models/capabilities", capabilitiesHandler)
}