From bf8534f13d2da0fc9e55c403f60b3d08f20468fc Mon Sep 17 00:00:00 2001 From: infosave2007 Date: Tue, 30 Jun 2026 16:50:22 +0300 Subject: [PATCH] feat: Anthropic provider/inbound + embeddings - Anthropic outbound provider: /v1/messages, system extraction, x-api-key + anthropic-version; streaming translates Anthropic SSE -> unified OpenAI wire format - Anthropic inbound adapter (POST /v1/messages): non-streaming JSON + streaming SSE (OpenAI -> Anthropic event translation) - Embeddings: POST /v1/embeddings inbound + Provider::embed(); embedding models resolvable from the pool (registry no longer skips them) - protocols mounted always, gated on live flags; admin meta marks them implemented --- CHANGELOG.md | 7 + src/admin/mod.rs | 6 +- src/protocols/anthropic_messages.rs | 209 +++++++++++++++++++++++++ src/protocols/mod.rs | 11 +- src/protocols/openai_embeddings.rs | 52 +++++++ src/providers/anthropic.rs | 228 ++++++++++++++++++++++++++-- src/providers/mod.rs | 10 +- src/providers/openai.rs | 26 ++++ src/registry.rs | 3 - 9 files changed, 528 insertions(+), 24 deletions(-) create mode 100644 src/protocols/anthropic_messages.rs create mode 100644 src/protocols/openai_embeddings.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 727fbc3..b94327d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/). byte. Live streaming output in the admin Playground (`/admin/api/test/stream`). - Cross-platform Windows binaries in releases (`x86_64-pc-windows-msvc`) and a cross-platform CI test matrix (ubuntu/windows/macos); secure token RNG via `getrandom`. +- **Anthropic provider** (outbound): `POST /v1/messages` with `system` extraction, + `x-api-key` + `anthropic-version` headers, and streaming that translates Anthropic + SSE events into the unified OpenAI wire format. +- **Anthropic inbound** adapter: `POST /v1/messages` (non-streaming JSON and streaming + SSE), so Claude-native clients can target the gateway. +- **Embeddings**: `POST /v1/embeddings` inbound + provider `embed()`; embedding models + are now resolvable from the pool. ## [0.1.0] - 2026-06-30 diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 1774bb1..3e2adf5 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -271,7 +271,7 @@ async fn meta() -> ApiResult { ok(json!({ "version": env!("CARGO_PKG_VERSION"), "providers": ["openai", "anthropic", "ollama", "http"], - "provider_impl": { "openai": true, "ollama": true, "http": true, "anthropic": false }, + "provider_impl": { "openai": true, "ollama": true, "http": true, "anthropic": true }, "cost_tiers": ["cheap", "mid", "expensive"], "kinds": ["chat", "embedding"], "profiles": ["cost-saver", "balanced", "quality-first"], @@ -280,8 +280,8 @@ async fn meta() -> ApiResult { "tiers": ["low", "medium", "high"], "caps": ["tools", "vision"], "protocols_impl": { - "openai_chat": true, "openai_completions": false, "openai_embeddings": false, - "openai_models": false, "anthropic_messages": false, "mcp": false, "native_passthrough": false + "openai_chat": true, "openai_completions": false, "openai_embeddings": true, + "openai_models": false, "anthropic_messages": true, "mcp": false, "native_passthrough": false }, "languages": ["en", "ru", "de", "fr", "es", "zh", "tr"], })) diff --git a/src/protocols/anthropic_messages.rs b/src/protocols/anthropic_messages.rs new file mode 100644 index 0000000..65c04b9 --- /dev/null +++ b/src/protocols/anthropic_messages.rs @@ -0,0 +1,209 @@ +//! Inbound Anthropic Messages adapter: `POST /v1/messages`. +//! Translates the Anthropic request into the canonical [`ChatRequest`], runs the +//! pipeline, and converts the result back to the Anthropic Messages format +//! (non-streaming JSON or streaming SSE). + +use crate::error::{GatewayError, Result}; +use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective}; +use crate::protocols::openai_chat::cortiq_headers; +use crate::state::SharedState; +use axum::response::{IntoResponse, Response}; +use axum::{extract::State, routing::post, Json, Router}; +use serde::Deserialize; + +pub fn routes() -> Router { + Router::new().route("/v1/messages", post(handler)) +} + +#[derive(Deserialize)] +struct MessagesRequest { + model: String, + #[serde(default)] + system: serde_json::Value, + #[serde(default)] + messages: Vec, + #[serde(default)] + max_tokens: Option, + #[serde(default)] + temperature: Option, + #[serde(default)] + top_p: Option, + #[serde(default)] + stream: bool, + #[serde(default)] + tools: Vec, +} + +/// Anthropic content is a string or an array of blocks; flatten to text. +fn content_to_string(v: &serde_json::Value) -> String { + if let Some(s) = v.as_str() { + return s.to_string(); + } + if let Some(arr) = v.as_array() { + return arr + .iter() + .filter_map(|b| b["text"].as_str()) + .collect::>() + .join(""); + } + String::new() +} + +fn parse_routing(model: &str) -> RoutingDirective { + if let Some(rest) = model.strip_prefix("cortiq-auto") { + RoutingDirective::Auto { + profile: rest.strip_prefix(':').map(|p| p.to_string()), + } + } else { + RoutingDirective::Pinned { + model_id: model.to_string(), + } + } +} + +async fn handler( + State(state): State, + Json(req): Json, +) -> Result { + if !state.live().cfg.protocols.anthropic_messages { + return Err(GatewayError::InvalidRequest( + "anthropic_messages protocol is disabled".into(), + )); + } + if req.messages.is_empty() { + return Err(GatewayError::InvalidRequest( + "messages must not be empty".into(), + )); + } + + let mut messages = Vec::new(); + let sys = content_to_string(&req.system); + if !sys.is_empty() { + messages.push(Message { + role: "system".into(), + content: sys, + tool_calls: vec![], + }); + } + for m in &req.messages { + messages.push(Message { + role: m["role"].as_str().unwrap_or("user").to_string(), + content: content_to_string(&m["content"]), + tool_calls: vec![], + }); + } + + let canonical = ChatRequest { + routing: parse_routing(&req.model), + messages, + tools: req.tools, + params: GenParams { + temperature: req.temperature, + max_tokens: req.max_tokens, + top_p: req.top_p, + stop: Vec::new(), + passthrough: Default::default(), + }, + stream: req.stream, + meta: RequestMeta { + protocol: "anthropic_messages".into(), + ..Default::default() + }, + }; + + if req.stream { + let (info, stream) = state.pipeline.run_stream(canonical, &state).await?; + let mut headers = cortiq_headers(&info); + headers.insert( + axum::http::header::CONTENT_TYPE, + axum::http::HeaderValue::from_static("text/event-stream"), + ); + let translated = openai_to_anthropic_sse(stream); + return Ok((headers, axum::body::Body::from_stream(translated)).into_response()); + } + + let resp = state.pipeline.run(canonical, &state).await?; + let answer = resp + .choices + .first() + .map(|c| c.message.content.clone()) + .unwrap_or_default(); + let headers = cortiq_headers(&resp.cortiq); + let body = serde_json::json!({ + "id": resp.id, + "type": "message", + "role": "assistant", + "model": resp.model_used, + "content": [{ "type": "text", "text": answer }], + "stop_reason": "end_turn", + "stop_sequence": serde_json::Value::Null, + "usage": { + "input_tokens": resp.usage.prompt_tokens, + "output_tokens": resp.usage.completion_tokens, + }, + }); + Ok((headers, Json(body)).into_response()) +} + +/// Translate the gateway's internal OpenAI-format SSE into Anthropic Messages SSE +/// events, so Anthropic-native clients can stream regardless of the upstream provider. +fn openai_to_anthropic_sse( + stream: crate::providers::ChatStream, +) -> impl futures::Stream> + Send + 'static { + use futures::StreamExt; + async_stream::stream! { + fn evt(event: &str, data: serde_json::Value) -> bytes::Bytes { + bytes::Bytes::from(format!("event: {event}\ndata: {data}\n\n")) + } + + yield Ok(evt("message_start", serde_json::json!({ + "type": "message_start", + "message": {"id": "msg_gw", "type": "message", "role": "assistant", + "content": [], "model": "cortiq-auto", "stop_reason": null, + "usage": {"input_tokens": 0, "output_tokens": 0}} + }))); + yield Ok(evt("content_block_start", serde_json::json!({ + "type": "content_block_start", "index": 0, + "content_block": {"type": "text", "text": ""} + }))); + + let mut buf = String::new(); + let mut completion_tokens = 0u64; + futures::pin_mut!(stream); + while let Some(item) = stream.next().await { + let bytes = match item { + Ok(b) => b, + Err(e) => { yield Err(e); break; } + }; + if let Ok(s) = std::str::from_utf8(&bytes) { + buf.push_str(s); + } + while let Some(idx) = buf.find("\n\n") { + let chunk: String = buf.drain(..idx + 2).collect(); + for line in chunk.lines() { + let Some(data) = line.trim_start().strip_prefix("data:") else { continue }; + let data = data.trim(); + if data.is_empty() || data == "[DONE]" { continue; } + let Ok(v) = serde_json::from_str::(data) else { continue }; + if let Some(u) = v.get("usage").filter(|u| !u.is_null()) { + completion_tokens = u["completion_tokens"].as_u64().unwrap_or(completion_tokens); + } + if let Some(t) = v["choices"][0]["delta"]["content"].as_str() { + yield Ok(evt("content_block_delta", serde_json::json!({ + "type": "content_block_delta", "index": 0, + "delta": {"type": "text_delta", "text": t} + }))); + } + } + } + } + + yield Ok(evt("content_block_stop", serde_json::json!({"type": "content_block_stop", "index": 0}))); + yield Ok(evt("message_delta", serde_json::json!({ + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": null}, + "usage": {"output_tokens": completion_tokens} + }))); + yield Ok(evt("message_stop", serde_json::json!({"type": "message_stop"}))); + } +} diff --git a/src/protocols/mod.rs b/src/protocols/mod.rs index 273bccd..d95b26c 100644 --- a/src/protocols/mod.rs +++ b/src/protocols/mod.rs @@ -1,7 +1,9 @@ //! Incoming protocol adapters. Each one builds its own axum routes and translates //! its protocol to/from the canonical model. Enabled via flags in `[protocols]`. +pub mod anthropic_messages; pub mod openai_chat; +pub mod openai_embeddings; use crate::state::SharedState; use axum::Router; @@ -10,8 +12,9 @@ use axum::Router; /// each handler checks the live `protocols.*` flag from config itself, so /// toggles in the admin panel take effect without a restart (hot switching). pub fn build_router() -> Router { - Router::new().merge(openai_chat::routes()) - // TODO: openai_completions, openai_embeddings, openai_models, - // anthropic_messages, mcp, native_passthrough — add here as well, - // with a check of the corresponding flag inside the handler. + Router::new() + .merge(openai_chat::routes()) + .merge(openai_embeddings::routes()) + .merge(anthropic_messages::routes()) + // TODO: openai_completions, openai_models, mcp, native_passthrough. } diff --git a/src/protocols/openai_embeddings.rs b/src/protocols/openai_embeddings.rs new file mode 100644 index 0000000..89c76f8 --- /dev/null +++ b/src/protocols/openai_embeddings.rs @@ -0,0 +1,52 @@ +//! Inbound OpenAI Embeddings adapter: `POST /v1/embeddings`. +//! Resolves an embedding model (explicit id, or the first `kind = "embedding"`), +//! calls the provider, and forwards the OpenAI-style response body. + +use crate::error::{GatewayError, Result}; +use crate::state::SharedState; +use axum::{extract::State, response::IntoResponse, routing::post, Json, Router}; +use serde::Deserialize; + +pub fn routes() -> Router { + Router::new().route("/v1/embeddings", post(handler)) +} + +#[derive(Deserialize)] +struct EmbeddingsRequest { + #[serde(default)] + model: String, + input: serde_json::Value, +} + +async fn handler( + State(state): State, + Json(req): Json, +) -> Result { + let live = state.live(); + if !live.cfg.protocols.openai_embeddings { + return Err(GatewayError::InvalidRequest( + "openai_embeddings protocol is disabled".into(), + )); + } + + // resolve the embedding model: explicit real id, else first kind == "embedding" + let model_id = if !req.model.is_empty() + && req.model != "cortiq-auto" + && live.registry.get(&req.model).is_some() + { + req.model.clone() + } else { + live.cfg + .models + .iter() + .find(|m| m.kind == "embedding") + .map(|m| m.id.clone()) + .ok_or_else(|| GatewayError::InvalidRequest("no embedding model configured".into()))? + }; + + let provider = live.registry.get(&model_id).ok_or_else(|| { + GatewayError::UpstreamUnavailable(format!("embedding model '{model_id}' is not available")) + })?; + let body = provider.embed(req.input).await?; + Ok(Json(body)) +} diff --git a/src/providers/anthropic.rs b/src/providers/anthropic.rs index 0af8d20..f5934df 100644 --- a/src/providers/anthropic.rs +++ b/src/providers/anthropic.rs @@ -1,23 +1,25 @@ -//! Anthropic Messages provider (v0.2). Translates the canonical ChatRequest to -//! `POST /v1/messages` and back. The `system` message is extracted into a separate field. +//! Anthropic Messages provider. Translates the canonical [`ChatRequest`] to +//! `POST {base_url}/v1/messages` and back. `system` messages are lifted into the +//! top-level `system` field. Streaming translates Anthropic SSE events into the +//! OpenAI wire format, so the gateway's unified SSE passthrough works uniformly. use super::{Caps, Provider}; use crate::config::ModelCfg; use crate::error::{GatewayError, Result}; -use crate::model::{ChatRequest, ChatResponse}; +use crate::model::{ChatRequest, ChatResponse, Choice, Message, RouteInfo, Usage}; use async_trait::async_trait; +const ANTHROPIC_VERSION: &str = "2023-06-01"; + pub struct AnthropicProvider { id: String, - #[allow(dead_code)] base_url: String, - #[allow(dead_code)] model: String, - #[allow(dead_code)] api_key: Option, caps: Caps, price_in: f64, price_out: f64, + http: reqwest::Client, } impl AnthropicProvider { @@ -30,8 +32,61 @@ impl AnthropicProvider { caps: Caps::from_list(&cfg.caps), price_in: cfg.price_in, price_out: cfg.price_out, + http: reqwest::Client::new(), }) } + + /// Canonical request → Anthropic Messages request body. + fn to_wire(&self, req: &ChatRequest, stream: bool) -> serde_json::Value { + let mut system = String::new(); + let mut messages = Vec::new(); + for m in &req.messages { + if m.role == "system" { + if !system.is_empty() { + system.push('\n'); + } + system.push_str(&m.content); + } else { + let role = if m.role == "assistant" { + "assistant" + } else { + "user" + }; + messages.push(serde_json::json!({ "role": role, "content": m.content })); + } + } + let mut body = serde_json::json!({ + "model": self.model, + "messages": messages, + "max_tokens": req.params.max_tokens.unwrap_or(1024), + "stream": stream, + }); + if !system.is_empty() { + body["system"] = serde_json::Value::String(system); + } + if let Some(t) = req.params.temperature { + body["temperature"] = t.into(); + } + if let Some(p) = req.params.top_p { + body["top_p"] = p.into(); + } + if !req.tools.is_empty() { + body["tools"] = serde_json::Value::Array(req.tools.clone()); + } + body + } + + fn request(&self, body: &serde_json::Value) -> reqwest::RequestBuilder { + let mut r = self + .http + .post(format!("{}/v1/messages", self.base_url)) + .header("anthropic-version", ANTHROPIC_VERSION) + .json(body); + if let Some(key) = &self.api_key { + r = r.header("x-api-key", key); + } + r + } } #[async_trait] @@ -46,12 +101,159 @@ impl Provider for AnthropicProvider { (prompt_tokens as f64 * self.price_in + completion_tokens as f64 * self.price_out) / 1_000_000.0 } - async fn chat(&self, _req: ChatRequest) -> Result { - // TODO(v0.2): extract system from messages → "system" field; messages → content blocks; - // add x-api-key + anthropic-version headers; map response → canonical form. - // Returns an error (not a panic) so that failover/playground continue to work normally. - Err(GatewayError::UpstreamUnavailable( - "anthropic provider not yet implemented (planned v0.2)".to_string(), - )) + + async fn chat(&self, req: ChatRequest) -> Result { + let resp = self + .request(&self.to_wire(&req, false)) + .send() + .await + .map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?; + if !resp.status().is_success() { + let status = resp.status(); + let b = resp.text().await.unwrap_or_default(); + return Err(GatewayError::UpstreamUnavailable(format!( + "provider {} returned HTTP {}: {}", + self.id, status, b + ))); + } + let v: serde_json::Value = resp.json().await.map_err(|e| { + GatewayError::Internal(format!("failed to parse anthropic response: {e}")) + })?; + let content = v["content"] + .as_array() + .map(|blocks| { + blocks + .iter() + .filter_map(|b| b["text"].as_str()) + .collect::>() + .join("") + }) + .unwrap_or_default(); + let prompt_tokens = v["usage"]["input_tokens"].as_u64().unwrap_or(0) as u32; + let completion_tokens = v["usage"]["output_tokens"].as_u64().unwrap_or(0) as u32; + Ok(ChatResponse { + id: v["id"].as_str().unwrap_or("").to_string(), + model_used: v["model"].as_str().unwrap_or(&self.model).to_string(), + choices: vec![Choice { + index: 0, + message: Message { + role: "assistant".into(), + content, + tool_calls: vec![], + }, + finish_reason: v["stop_reason"].as_str().unwrap_or("stop").to_string(), + }], + usage: Usage { + prompt_tokens, + completion_tokens, + total_tokens: prompt_tokens + completion_tokens, + }, + cortiq: RouteInfo { + task_label: String::new(), + complexity_score: 0.0, + complexity_tier: String::new(), + selected_model: self.id.clone(), + route_source: String::new(), + router_request_id: None, + cost_usd: 0.0, + failover: false, + }, + }) + } + + async fn chat_stream(&self, req: ChatRequest) -> Result { + let resp = self + .request(&self.to_wire(&req, true)) + .send() + .await + .map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?; + if !resp.status().is_success() { + let status = resp.status(); + let b = resp.text().await.unwrap_or_default(); + return Err(GatewayError::UpstreamUnavailable(format!( + "provider {} returned HTTP {}: {}", + self.id, status, b + ))); + } + Ok(Box::pin(anthropic_to_openai_sse(resp.bytes_stream()))) + } +} + +/// Translate an Anthropic Messages SSE stream into OpenAI `chat.completion.chunk` +/// SSE so the gateway's unified streaming path works for Anthropic too. +fn anthropic_to_openai_sse( + upstream: impl futures::Stream> + Send + 'static, +) -> impl futures::Stream> + Send + 'static { + use futures::StreamExt; + async_stream::stream! { + futures::pin_mut!(upstream); + let mut buf = String::new(); + let mut input_tokens = 0u64; + let mut role_sent = false; + let emit = |obj: serde_json::Value| bytes::Bytes::from(format!("data: {obj}\n\n")); + + while let Some(chunk) = upstream.next().await { + let chunk = match chunk { + Ok(c) => c, + Err(e) => { + yield Err(GatewayError::UpstreamUnavailable(e.to_string())); + return; + } + }; + if let Ok(s) = std::str::from_utf8(&chunk) { + buf.push_str(s); + } + while let Some(idx) = buf.find("\n\n") { + let evt: String = buf.drain(..idx + 2).collect(); + for line in evt.lines() { + let Some(data) = line.trim_start().strip_prefix("data:") else { + continue; + }; + let data = data.trim(); + if data.is_empty() { + continue; + } + let Ok(v) = serde_json::from_str::(data) else { + continue; + }; + match v["type"].as_str().unwrap_or("") { + "message_start" => { + input_tokens = + v["message"]["usage"]["input_tokens"].as_u64().unwrap_or(0); + if !role_sent { + role_sent = true; + yield Ok(emit(serde_json::json!({ + "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": null}] + }))); + } + } + "content_block_delta" => { + if let Some(t) = v["delta"]["text"].as_str() { + yield Ok(emit(serde_json::json!({ + "choices": [{"index": 0, "delta": {"content": t}, "finish_reason": null}] + }))); + } + } + "message_delta" => { + let out = v["usage"]["output_tokens"].as_u64().unwrap_or(0); + let stop = v["delta"]["stop_reason"].as_str().unwrap_or("stop").to_string(); + yield Ok(emit(serde_json::json!({ + "choices": [{"index": 0, "delta": {}, "finish_reason": stop}] + }))); + yield Ok(emit(serde_json::json!({ + "choices": [], + "usage": { + "prompt_tokens": input_tokens, + "completion_tokens": out, + "total_tokens": input_tokens + out + } + }))); + } + _ => {} + } + } + } + } + yield Ok(bytes::Bytes::from("data: [DONE]\n\n")); } } diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 39ee9d5..ca81163 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -57,5 +57,13 @@ pub trait Provider: Send + Sync { ))) } - // TODO(v0.2): async fn embed(...) -> Result. + /// Embeddings call. `input` is the OpenAI `input` field (string or array of + /// strings); returns the provider's OpenAI-style embeddings response body. + /// Default: not supported (override per provider). + async fn embed(&self, _input: serde_json::Value) -> crate::error::Result { + Err(GatewayError::UpstreamUnavailable(format!( + "embeddings are not supported by provider '{}'", + self.id() + ))) + } } diff --git a/src/providers/openai.rs b/src/providers/openai.rs index 50b2b29..99275d3 100644 --- a/src/providers/openai.rs +++ b/src/providers/openai.rs @@ -185,4 +185,30 @@ impl Provider for OpenAiProvider { .map(|res| res.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))); Ok(Box::pin(stream)) } + + async fn embed(&self, input: serde_json::Value) -> Result { + let body = serde_json::json!({ "model": self.model, "input": input }); + let mut r = self + .http + .post(format!("{}/embeddings", self.base_url)) + .json(&body); + if let Some(key) = &self.api_key { + r = r.bearer_auth(key); + } + let resp = r + .send() + .await + .map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?; + if !resp.status().is_success() { + let status = resp.status(); + let err_body = resp.text().await.unwrap_or_default(); + return Err(GatewayError::UpstreamUnavailable(format!( + "provider {} returned HTTP {}: {}", + self.id, status, err_body + ))); + } + resp.json().await.map_err(|e| { + GatewayError::Internal(format!("failed to parse embeddings response: {e}")) + }) + } } diff --git a/src/registry.rs b/src/registry.rs index 9961770..f52ec3a 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -15,9 +15,6 @@ impl Registry { pub fn from_config(cfg: &Config, secrets: &SecretStore) -> anyhow::Result { let mut by_id: HashMap> = HashMap::new(); for m in &cfg.models { - if m.kind == "embedding" { - continue; // embedding models are handled by a separate branch (v0.1) - } // provider key: secret store → environment variable let api_key = m .api_key_env