Route Anthropic Messages requests through compatible engines

docs/getting-started.mdx

@@ -461,8 +461,19 @@ What that base URL serves depends on the engine behind it:
 
 | Endpoint | Default base URL | Paths it serves |
 | --- | --- | --- |
-| Ollama | `http://127.0.0.1:11434` | Ollama's own API — `/api/chat`, `/api/generate`, `/api/embed`, `/api/tags` — and the OpenAI-compatible `/v1/chat/completions`, `/v1/completions`, `/v1/embeddings`, `/v1/models` |
-| LM Studio | `http://127.0.0.1:1234` | The OpenAI-compatible paths only |
+| Ollama | `http://127.0.0.1:11434` | Ollama's own API — `/api/chat`, `/api/generate`, `/api/embed`, `/api/tags` — and the OpenAI-compatible `/v1/chat/completions`, `/v1/completions`, `/v1/embeddings`, `/v1/models`, plus Anthropic `POST /v1/messages` |
+| LM Studio | `http://127.0.0.1:1234` | The OpenAI-compatible paths and Anthropic `POST /v1/messages` |
+
+Anthropic clients such as Claude Code can use `POST /v1/messages` with Ollama
+0.14.0+, LM Studio 0.4.1+, or oMLX 0.1.0+. PAIR routes the body's `model` to an
+advertised owner and passes JSON or streamed SSE through without translation.
+Use the copied endpoint's API root (without a trailing `/v1`) as
+`ANTHROPIC_BASE_URL`; the client appends `/v1/messages`. Older engines without
+the route return an upstream HTTP error after eligible-owner failover is
+exhausted; upgrade the serving engine. See the proxy references for
+[Ollama](../services/ollama-proxy/README.md#anthropic-messages-api),
+[LM Studio](../services/lmstudio-proxy/README.md#anthropic-messages-api), and
+[oMLX](../services/omlx-proxy/README.md#anthropic-messages-api).
 
 The distinction matters when you fill in an application's settings. A client
 written against the OpenAI API usually wants the `/v1` included, as in

services/lmstudio-proxy/README.md

@@ -9,6 +9,27 @@ A discovery-aware HTTP reverse proxy for LM Studio nodes on the local network. I
 
 > **Clone of `ollama-proxy`.** This proxy is a deliberate clone of [`ollama-proxy`](../ollama-proxy/README.md) so the two share identical routing, failover, CORS, and node-selection behavior — the CORS policy is literally the same code, `nvpair-shared/cors`, and is documented [there](../ollama-proxy/README.md#http-reverse-proxy). The differences are engine-specific: it subscribes to the discovery relay for `lm` nodes, forwards the OpenAI-compatible inference routes (`/v1/chat/completions`, `/v1/completions`, `/v1/embeddings`), tags workloads `lmstudio`, and persists its port to its own file. It has no `--alias-address`, so its self-forward guard covers only its own listener.
 
+## Anthropic Messages API
+
+`POST /v1/messages` is an inference route, supported with **LM Studio
+0.4.1 or newer** (see the engine's [release announcement](https://lmstudio.ai/blog/claudecode)).
+The body’s `model` selects eligible owners through the same selection,
+scheduler, failover, and workload path as other inference routes. PAIR performs
+no protocol translation: the request body and response, including SSE with
+`stream: true`, pass through unchanged. SSE is flushed as it arrives;
+the existing first-body deadline and cooldown rules apply, and no attempt is
+retried after response bytes reach the client.
+
+`x-api-key` and `anthropic-version` follow existing request-header forwarding;
+`x-api-key` is not PAIR authentication. Existing proxy authentication and
+loopback/cluster policies are unchanged.
+
+PAIR does not negotiate engine API versions. An older engine without this
+endpoint returns its HTTP error (typically `404`); that status may try the next
+eligible model owner, but only within the existing finite candidate list.
+If none supports the endpoint, the final engine error is returned, not translated
+or redirected to another model. Upgrade the serving engines to use this route.
+
 ## Worker control health
 
 An id-bearing `worker/health` request (no params) returns `{"ok":true}` with

services/lmstudio-proxy/messages_test.go

@@ -0,0 +1,237 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestMessagesRouteRole(t *testing.T) {
+	for _, tc := range []struct {
+		method, path string
+		inference    bool
+	}{
+		{http.MethodPost, "/v1/messages", true},
+		{http.MethodGet, "/v1/messages", false},
+		{http.MethodPut, "/v1/messages", false},
+		{http.MethodPost, "/v1/messages/count_tokens", false},
+	} {
+		if got := isInferenceRequest(tc.method, tc.path); got != tc.inference {
+			t.Errorf("%s %s inference = %v, want %v", tc.method, tc.path, got, tc.inference)
+		}
+	}
+}
+
+// Exercise real HTTP copying: the first SSE event must arrive while the engine
+// is still waiting to finish, including after an advertised owner's 404.
+func TestMessagesModelRoutingAndFailover(t *testing.T) {
+	for _, streaming := range []bool{false, true} {
+		t.Run(map[bool]string{false: "json", true: "sse"}[streaming], func(t *testing.T) {
+			body := `{"model":"model","max_tokens":7,"messages":[{"role":"user","content":"test"}],"stream":` +
+				map[bool]string{false: "false", true: "true"}[streaming] + `}`
+			var missingHits, goodHits, wrongHits atomic.Int32
+			wrong := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				wrongHits.Add(1)
+			}))
+			defer wrong.Close()
+			check := func(r *http.Request) {
+				t.Helper()
+				data, err := io.ReadAll(r.Body)
+				if err != nil || string(data) != body || r.URL.Path != "/v1/messages" {
+					t.Error("engine did not receive the unchanged Messages request")
+				}
+				if r.Header.Get("x-api-key") != "synthetic-test-key" ||
+					r.Header.Get("anthropic-version") != "2023-06-01" {
+					t.Error("Anthropic request headers were not passed through")
+				}
+			}
+			missing := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				missingHits.Add(1)
+				check(r)
+				w.WriteHeader(http.StatusNotFound)
+			}))
+			defer missing.Close()
+			release := make(chan struct{})
+			var once sync.Once
+			finish := func() { once.Do(func() { close(release) }) }
+			good := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				goodHits.Add(1)
+				check(r)
+				if streaming {
+					w.Header().Set("Content-Type", "text/event-stream")
+					_, _ = io.WriteString(w, "event: message_start\ndata: {}\n\n")
+					w.(http.Flusher).Flush()
+					<-release
+					_, _ = io.WriteString(w, "event: message_stop\ndata: {}\n\n")
+				} else {
+					w.Header().Set("Content-Type", "application/json")
+					_, _ = io.WriteString(w, `{"type":"message","content":[]}`)
+				}
+			}))
+			defer good.Close()
+			defer finish()
+
+			disc := NewDiscovery()
+			disc.AddManual(nodeForModel(t, "wrong", wrong.URL, "different-model"))
+			disc.AddManual(nodeForModel(t, "missing", missing.URL, "model"))
+			disc.AddManual(nodeForModel(t, "good", good.URL, "model"))
+			p := testProxy(disc, 0)
+			defer p.closeIdleTransports()
+			p.SetSelected("wrong")
+			p.SetPriority([]string{"missing", "good"})
+			server := httptest.NewServer(http.HandlerFunc(p.handleHTTP))
+			defer func() {
+				finish()
+				server.Close()
+			}()
+			client := &http.Client{Timeout: 3 * time.Second}
+			defer client.CloseIdleConnections()
+			req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/messages", strings.NewReader(body))
+			if err != nil {
+				t.Fatal(err)
+			}
+			req.Header.Set("Content-Type", "application/json")
+			req.Header.Set("x-api-key", "synthetic-test-key")
+			req.Header.Set("anthropic-version", "2023-06-01")
+			resp, err := client.Do(req)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer resp.Body.Close()
+			if resp.StatusCode != http.StatusOK {
+				t.Fatalf("status = %d, want 200", resp.StatusCode)
+			}
+			if streaming {
+				if resp.Header.Get("Content-Type") != "text/event-stream" {
+					t.Fatal("SSE content type changed")
+				}
+				reader := bufio.NewReader(resp.Body)
+				line, err := reader.ReadString('\n')
+				if err != nil || line != "event: message_start\n" {
+					t.Fatal("first SSE event did not arrive before engine completion")
+				}
+				finish()
+				rest, err := io.ReadAll(reader)
+				if err != nil || string(rest) != "data: {}\n\nevent: message_stop\ndata: {}\n\n" {
+					t.Fatal("SSE response changed")
+				}
+			} else {
+				data, err := io.ReadAll(resp.Body)
+				if err != nil || string(data) != `{"type":"message","content":[]}` {
+					t.Fatal("JSON response changed")
+				}
+			}
+			if missingHits.Load() != 1 || goodHits.Load() != 1 || wrongHits.Load() != 0 {
+				t.Fatal("request did not fail over only between advertised model owners")
+			}
+		})
+	}
+}
+
+func TestMessagesFirstBodyStallCooldown(t *testing.T) {
+	oldFirst, oldHeader := proxyFirstBodyTimeout, proxyResponseTimeout
+	proxyFirstBodyTimeout, proxyResponseTimeout = 50*time.Millisecond, 3*time.Second
+	defer func() { proxyFirstBodyTimeout, proxyResponseTimeout = oldFirst, oldHeader }()
+	var stalledHits atomic.Int32
+	stalled := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		stalledHits.Add(1)
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		w.(http.Flusher).Flush()
+		<-r.Context().Done()
+	}))
+	defer stalled.Close()
+	const event = "event: message_stop\ndata: {}\n\n"
+	good := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, event)
+	}))
+	defer good.Close()
+	disc := NewDiscovery()
+	disc.AddManual(nodeForModel(t, "stalled", stalled.URL, "model"))
+	disc.AddManual(nodeForModel(t, "good", good.URL, "model"))
+	p := testProxy(disc, 0)
+	defer p.closeIdleTransports()
+	for i := 0; i < 2; i++ {
+		p.SetPriority([]string{"stalled", "good"})
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		rec := httptest.NewRecorder()
+		p.handleHTTP(rec, httptest.NewRequest(http.MethodPost, "/v1/messages",
+			strings.NewReader(`{"model":"model","stream":true}`)).WithContext(ctx))
+		cancel()
+		if rec.Code != http.StatusOK || rec.Body.String() != event {
+			t.Fatal("SSE first-body stall did not fail over")
+		}
+		if !p.coolingSet(time.Now())["stalled"] {
+			t.Fatal("first-body stall did not cool candidate")
+		}
+	}
+	if stalledHits.Load() != 1 {
+		t.Fatal("second request retried cooling candidate before a ready peer")
+	}
+}
+
+func TestMessagesCommittedSSENeverRetries(t *testing.T) {
+	var goodHits atomic.Int32
+	const partial = "event: message_start\ndata: {}\n\n"
+	broken := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.Header().Set("Content-Length", "999")
+		_, _ = io.WriteString(w, partial)
+		w.(http.Flusher).Flush()
+	}))
+	defer broken.Close()
+	good := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { goodHits.Add(1) }))
+	defer good.Close()
+	disc := NewDiscovery()
+	disc.AddManual(nodeForModel(t, "broken", broken.URL, "model"))
+	disc.AddManual(nodeForModel(t, "good", good.URL, "model"))
+	p := testProxy(disc, 0)
+	defer p.closeIdleTransports()
+	p.SetPriority([]string{"broken", "good"})
+	server := httptest.NewServer(http.HandlerFunc(p.handleHTTP))
+	defer server.Close()
+	client := &http.Client{Timeout: 3 * time.Second}
+	defer client.CloseIdleConnections()
+	resp, err := client.Post(server.URL+"/v1/messages", "application/json", strings.NewReader(`{"model":"model","stream":true}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	data, err := io.ReadAll(resp.Body)
+	if err == nil || string(data) != partial || goodHits.Load() != 0 {
+		t.Fatal("committed SSE was retried or not truncated")
+	}
+}
+
+func TestMessagesOlderEngineReturnsError(t *testing.T) {
+	var hits atomic.Int32
+	engine := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits.Add(1)
+		w.WriteHeader(http.StatusNotFound)
+		_, _ = io.WriteString(w, `{"error":"unsupported endpoint"}`)
+	}))
+	defer engine.Close()
+	disc := NewDiscovery()
+	disc.AddManual(nodeForModel(t, "old", engine.URL, "model"))
+	p := testProxy(disc, 0)
+	defer p.closeIdleTransports()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	rec := httptest.NewRecorder()
+	p.handleHTTP(rec, httptest.NewRequest(http.MethodPost, "/v1/messages",
+		strings.NewReader(`{"model":"model","stream":true}`)).WithContext(ctx))
+	if rec.Code != http.StatusNotFound || rec.Body.String() != `{"error":"unsupported endpoint"}` || hits.Load() != 1 || ctx.Err() != nil {
+		t.Fatal("older engine error was not returned cleanly after bounded candidates")
+	}
+}

services/lmstudio-proxy/proxy.go

@@ -171,6 +171,7 @@ var inferenceEndpoints = map[string]bool{
 	"/v1/chat/completions": true,
 	"/v1/completions":      true,
 	"/v1/embeddings":       true,
+	"/v1/messages":         true,
 }
 
 // isInferenceRequest reports whether a request should be tracked as a

services/ollama-proxy/README.md

@@ -7,6 +7,27 @@ SPDX-License-Identifier: Apache-2.0
 
 A discovery-aware HTTP reverse proxy for Ollama nodes on the local network. It runs no mDNS browse of its own: its routing targets come from the broker's discovery relay (it sends `discovery:subscribe {services:[ol]}` and replaces its routing overlay from each pushed `discovery:nodes` snapshot) plus user-added manual nodes. It forwards HTTP requests to the selected node, aggregates the model-list routes across candidate nodes, and exposes a bidirectional JSON-RPC 2.0 control channel over stdio (or an IPC socket).
 
+## Anthropic Messages API
+
+`POST /v1/messages` is an inference route, supported with **Ollama
+0.14.0 or newer** (see the engine's [route registration](https://github.com/ollama/ollama/blob/v0.14.0/server/routes.go)).
+The body’s `model` selects eligible owners through the same selection,
+scheduler, failover, and workload path as other inference routes. PAIR performs
+no protocol translation: the request body and response, including SSE with
+`stream: true`, pass through unchanged. SSE is flushed as it arrives;
+the existing first-body deadline and cooldown rules apply, and no attempt is
+retried after response bytes reach the client.
+
+`x-api-key` and `anthropic-version` follow existing request-header forwarding;
+`x-api-key` is not PAIR authentication. Existing proxy authentication and
+loopback/cluster policies are unchanged.
+
+PAIR does not negotiate engine API versions. An older engine without this
+endpoint returns its HTTP error (typically `404`); that status may try the next
+eligible model owner, but only within the existing finite candidate list.
+If none supports the endpoint, the final engine error is returned, not translated
+or redirected to another model. Upgrade the serving engines to use this route.
+
 ## Worker control health
 
 An id-bearing `worker/health` request (no params) returns `{"ok":true}` with