[sgl-router] Serve the Anthropic Messages API on /v1/messages

experimental/sgl-router/README.md

@@ -547,6 +547,27 @@ forwarded as sent, since the engine renders and tokenizes each query-document
 pair. Routing is on load, with each pair's size estimated from its text. As for
 embeddings, a PD fleet answers 400 and `--dp-aware` pins no rank.
 
+## Anthropic Messages
+
+`/v1/messages` converts each request to a chat completion at ingress, so it is
+routed exactly like `/v1/chat/completions`, and converts the reply (or stream)
+back to the Messages format. Consecutive turns of the same role are merged, as
+the Messages API does, and a final assistant turn is sent as a prefill
+(`continue_final_message`). Besides the Messages fields, only
+`chat_template_kwargs` and `continue_final_message` reach the chat request.
+Every error uses the Anthropic error envelope; 5xx messages are replaced with a
+generic one, and a failed stream still closes its open block and ends with
+`message_stop`.
+
+`/v1/messages/count_tokens` is answered by the router from the model's chat
+template, without calling an engine. Image tokens are not counted, and a model
+without a chat template answers 500 rather than a raw-text estimate.
+
+`thinking` sets the `thinking` / `enable_thinking` chat template kwargs. The
+engine's own `/v1/messages` also adapts it per reasoning parser (for example
+`reasoning_effort` for Hunyuan), which the router cannot see; on such models
+use `output_config.effort` or `chat_template_kwargs` directly.
+
 ## DeepSeek V4 and V4.1
 
 Native V4 and V4.1 rendering comes from `rust/sglang-processor`, shared with SGLang's Rust

experimental/sgl-router/src/lib.rs

@@ -14,6 +14,7 @@ pub mod discovery;
 pub mod health;
 pub mod policies;
 pub mod policies_reorg;
+pub(crate) mod protocol;
 pub mod proxy;
 pub mod server;
 pub mod state;

experimental/sgl-router/src/protocol/anthropic/mod.rs

@@ -0,0 +1,224 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Anthropic Messages API (`/v1/messages`) over chat completions.
+
+pub mod request;
+pub mod response;
+pub mod stream;
+
+use serde_json::{json, Value};
+
+pub use request::{to_chat, Converted, EchoContext};
+
+/// `input_tokens` excludes cache reads, as Anthropic defines it.
+pub(crate) fn usage_from_chat(usage: Option<&Value>) -> Value {
+    let int = |p: &str| {
+        usage
+            .and_then(|u| u.pointer(p))
+            .and_then(Value::as_u64)
+            .unwrap_or(0)
+    };
+    let prompt = int("/prompt_tokens");
+    let cached = int("/prompt_tokens_details/cached_tokens").min(prompt);
+    json!({
+        "input_tokens": prompt - cached,
+        "output_tokens": int("/completion_tokens"),
+        "cache_creation_input_tokens": 0,
+        "cache_read_input_tokens": cached,
+    })
+}
+
+/// `finish_reason` + sglang's `matched_stop` → (`stop_reason`, `stop_sequence`).
+pub(crate) fn stop_reason(
+    finish_reason: Option<&str>,
+    matched_stop: Option<&Value>,
+    echo: &EchoContext,
+) -> (&'static str, Value) {
+    match finish_reason {
+        Some("length") => ("max_tokens", Value::Null),
+        Some("tool_calls") => ("tool_use", Value::Null),
+        Some("content_filter") => ("refusal", Value::Null),
+        _ => match matched_stop.and_then(Value::as_str) {
+            Some(s) if echo.stop_sequences.iter().any(|q| q == s) => {
+                ("stop_sequence", Value::String(s.to_owned()))
+            }
+            _ => ("end_turn", Value::Null),
+        },
+    }
+}
+
+pub fn error_type(status: u16) -> &'static str {
+    match status {
+        400 | 422 => "invalid_request_error",
+        401 => "authentication_error",
+        402 => "billing_error",
+        403 => "permission_error",
+        404 => "not_found_error",
+        413 => "request_too_large",
+        429 => "rate_limit_error",
+        503 | 529 => "overloaded_error",
+        504 => "timeout_error",
+        _ => "api_error",
+    }
+}
+
+pub fn error_body(status: u16, message: &str) -> Vec<u8> {
+    serde_json::to_vec(&json!({
+        "type": "error",
+        "error": {"type": error_type(status), "message": message},
+    }))
+    .expect("serialize error")
+}
+
+/// Re-wrap any chat-path error body into the Anthropic envelope.
+pub fn wrap_error_body(body: &[u8], status: u16) -> Vec<u8> {
+    let v: Option<Value> = serde_json::from_slice(body).ok();
+    let message = match &v {
+        Some(v) => v
+            .pointer("/error/message")
+            .or_else(|| v.get("message"))
+            .or_else(|| v.get("detail"))
+            .or_else(|| v.get("error"))
+            .and_then(Value::as_str)
+            .map(str::to_owned),
+        None => Some(String::from_utf8_lossy(body).into_owned()),
+    };
+    let upstream_type = v
+        .as_ref()
+        .filter(|v| v.get("type").and_then(Value::as_str) == Some("error"))
+        .and_then(|v| v.pointer("/error/type"))
+        .and_then(Value::as_str)
+        .filter(|_| status < 500);
+    serde_json::to_vec(&json!({
+        "type": "error",
+        "error": {
+            "type": upstream_type.unwrap_or(error_type(status)),
+            "message": scrub_message(message.as_deref().unwrap_or(""), status),
+        },
+    }))
+    .expect("serialize error")
+}
+
+/// 5xx text can carry stack frames, paths or prompt fragments, so it never
+/// reaches the client; 4xx keeps the validation message.
+pub(crate) fn scrub_message(message: &str, status: u16) -> String {
+    const MAX_CHARS: usize = 500;
+    if status >= 500 {
+        return "Internal server error".into();
+    }
+    let kept: Vec<&str> = message
+        .lines()
+        .filter(|l| !l.starts_with("Traceback") && !l.contains("File \"/"))
+        .collect();
+    let cleaned = kept.join("\n");
+    let cleaned = cleaned.trim();
+    if cleaned.is_empty() {
+        return "Request failed".into();
+    }
+    match cleaned.char_indices().nth(MAX_CHARS) {
+        Some((at, _)) => format!("{}\u{2026}", &cleaned[..at]),
+        None => cleaned.to_owned(),
+    }
+}
+
+pub(crate) fn new_id(prefix: &str) -> String {
+    format!("{prefix}_{}", uuid::Uuid::new_v4().simple())
+}
+
+pub(crate) fn tool_input(arguments: &str) -> Value {
+    match serde_json::from_str::<Value>(arguments) {
+        Ok(v @ Value::Object(_)) => v,
+        _ if arguments.trim().is_empty() => json!({}),
+        _ => {
+            tracing::warn!("messages: tool call arguments are not a JSON object");
+            json!({})
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn echo(stops: &[&str]) -> EchoContext {
+        EchoContext {
+            model: "m".into(),
+            stop_sequences: stops.iter().map(|s| s.to_string()).collect(),
+        }
+    }
+
+    #[test]
+    fn usage_subtracts_cache_reads() {
+        let u = usage_from_chat(Some(&json!({"prompt_tokens": 10, "completion_tokens": 3,
+            "prompt_tokens_details": {"cached_tokens": 4}})));
+        assert_eq!(
+            u,
+            json!({"input_tokens": 6, "output_tokens": 3, "cache_creation_input_tokens": 0,
+                   "cache_read_input_tokens": 4})
+        );
+        assert_eq!(usage_from_chat(None)["cache_read_input_tokens"], 0);
+    }
+
+    #[test]
+    fn stop_reasons() {
+        let e = echo(&["<END>"]);
+        assert_eq!(stop_reason(Some("stop"), None, &e).0, "end_turn");
+        assert_eq!(
+            stop_reason(Some("stop"), Some(&json!("<END>")), &e),
+            ("stop_sequence", json!("<END>"))
+        );
+        assert_eq!(stop_reason(Some("stop"), Some(&json!(2)), &e).0, "end_turn");
+        assert_eq!(stop_reason(Some("length"), None, &e).0, "max_tokens");
+        assert_eq!(stop_reason(Some("tool_calls"), None, &e).0, "tool_use");
+    }
+
+    #[test]
+    fn error_wrapping() {
+        let v: Value = serde_json::from_slice(&wrap_error_body(
+            br#"{"error":{"type":"invalid_request_error","code":"bad_request","message":"bad"}}"#,
+            400,
+        ))
+        .unwrap();
+        assert_eq!(
+            v,
+            json!({"type": "error", "error": {"type": "invalid_request_error", "message": "bad"}})
+        );
+        let v: Value = serde_json::from_slice(&wrap_error_body(
+            br#"{"object":"error","message":"schema","type":"BadRequestError","code":400}"#,
+            400,
+        ))
+        .unwrap();
+        assert_eq!(v["error"]["message"], "schema");
+        let v: Value = serde_json::from_slice(&wrap_error_body(b"", 503)).unwrap();
+        assert_eq!(v["error"]["type"], "overloaded_error");
+        assert!(!v["error"]["message"].as_str().unwrap().is_empty());
+    }
+
+    #[test]
+    fn error_messages_are_scrubbed() {
+        let v: Value = serde_json::from_slice(&wrap_error_body(
+            br#"{"object":"error","message":"CUDA OOM at /opt/sglang/x.py","code":500}"#,
+            500,
+        ))
+        .unwrap();
+        assert_eq!(
+            v,
+            json!({"type": "error", "error": {"type": "api_error",
+                   "message": "Internal server error"}})
+        );
+        let v: Value = serde_json::from_slice(&wrap_error_body(
+            br#"{"type":"error","error":{"type":"api_error","message":"secret"}}"#,
+            502,
+        ))
+        .unwrap();
+        assert_eq!(v["error"]["message"], "Internal server error");
+
+        let traced = "bad input\nTraceback (most recent call last):\n  File \"/a.py\", line 1";
+        assert_eq!(scrub_message(traced, 400), "bad input");
+        let long = scrub_message(&"x".repeat(600), 400);
+        assert_eq!(long.chars().count(), 501);
+        assert!(long.ends_with('\u{2026}'));
+        assert_eq!(scrub_message("", 400), "Request failed");
+    }
+}