From abb5b90e1d4ac16fdf3f20b22c9be5dd62bf2bbc Mon Sep 17 00:00:00 2001 From: whit3rabbit Date: Wed, 25 Mar 2026 20:21:24 -0500 Subject: [PATCH] feat: add Bedrock backend, OpenTelemetry export, integration tests - AWS Bedrock backend: SigV4 signing, InvokeModel + InvokeModelWithResponseStream with binary event stream decoding, passthrough handler for /v1/messages - OpenTelemetry export: feature-gated (--features otel), OTLP/HTTP with reqwest transport, OtelGuard for graceful shutdown flush - Chat completions integration tests: 6 tests covering non-streaming, error handling, degradation headers, system messages - Updated COMPARISON_LITELLM.md to reflect all closed gaps - Fixed Bedrock match arms across all handler files 549 tests passing, 0 failures, clippy clean. Both `cargo build` and `cargo build --features otel` compile. Co-Authored-By: Claude Opus 4.6 (1M context) --- CLAUDE.md | 27 +- crates/proxy/Cargo.toml | 2 + crates/proxy/admin-ui/index.html | 97 ++++- crates/proxy/src/backend/bedrock_client.rs | 399 ++++++++++++++++++ crates/proxy/src/backend/mod.rs | 35 +- crates/proxy/src/backend/openai_client.rs | 4 +- crates/proxy/src/config/mod.rs | 124 +++++- crates/proxy/src/lib.rs | 3 + crates/proxy/src/main.rs | 16 + crates/proxy/src/otel.rs | 51 +++ .../proxy/src/server/bedrock_passthrough.rs | 240 +++++++++++ crates/proxy/src/server/chat_completions.rs | 8 +- crates/proxy/src/server/mod.rs | 2 + crates/proxy/src/server/routes.rs | 53 ++- crates/proxy/src/server/streaming.rs | 4 +- crates/proxy/tests/chat_completions.rs | 230 ++++++++++ crates/proxy/tests/embeddings.rs | 226 ++++++++++ crates/proxy/tests/live_bedrock.rs | 76 ++++ crates/translator/src/anthropic/messages.rs | 14 +- crates/translator/src/mapping/message_map.rs | 177 +++++++- .../src/mapping/responses_message_map.rs | 4 +- .../translator/src/mapping/streaming_map.rs | 35 ++ crates/translator/src/mapping/tools_map.rs | 36 +- crates/translator/src/mapping/warnings.rs | 60 +++ docs/COMPARISON_LITELLM.md | 147 +++++++ docs/ENV.md | 50 +++ .../20260325-120000-litellm-gap-fill/tasks.md | 28 +- 27 files changed, 2079 insertions(+), 69 deletions(-) create mode 100644 crates/proxy/src/backend/bedrock_client.rs create mode 100644 crates/proxy/src/otel.rs create mode 100644 crates/proxy/src/server/bedrock_passthrough.rs create mode 100644 crates/proxy/tests/chat_completions.rs create mode 100644 crates/proxy/tests/embeddings.rs create mode 100644 crates/proxy/tests/live_bedrock.rs create mode 100644 crates/translator/src/mapping/warnings.rs create mode 100644 docs/COMPARISON_LITELLM.md diff --git a/CLAUDE.md b/CLAUDE.md index 8580e39..03e6748 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,11 +12,13 @@ All 11 implementation phases are complete. **Working (verified):** - Build: `cargo build` clean, `cargo clippy -- -D warnings` clean -- Tests: ~417 tests passing, 4 ignored (live API) +- Tests: ~480 tests passing, 4 ignored (live API) - Full Anthropic Messages API translation: non-streaming, streaming SSE, tool calling, file/document blocks - Proxy middleware: health, auth, request ID, size limits, concurrency limits, retry with backoff - Compatibility endpoints: /v1/models, count_tokens (approximate via tiktoken), batches (stub) - Model mapping and lossy-translation warnings +- `POST /v1/embeddings` passthrough: forwards directly to the backend with no translation; works with OpenAI, Vertex, Gemini (`gemini-embedding-exp-03-07`), and vLLM/HuggingFace models. Not mounted for the Anthropic passthrough backend. +- `x-anyllm-degradation` response header: set when features are silently dropped during translation (e.g., `top_k`, `thinking_config`, `cache_control`, `document_blocks`, `stop_sequences_truncated`) **Not fully validated:** - OpenAI Responses API backend: wired up via `OPENAI_API_FORMAT=responses` but not tested against live API @@ -27,7 +29,8 @@ All 11 implementation phases are complete. ```bash cargo build # build everything -cargo test # run all tests (~417 tests, 4 ignored) +cargo test # run all tests (~480 tests, 4 ignored) +cargo test -p anyllm_client # client crate only cargo test -p anyllm_translate # translator crate only cargo test -p anyllm_proxy # proxy crate only cargo test health_endpoint # single test by name @@ -62,10 +65,22 @@ OPENAI_API_KEY=sk-... cargo run -p anyllm_proxy - `GEMINI_BASE_URL`: Gemini API base URL (default: `https://generativelanguage.googleapis.com/v1beta`) - `PROXY_API_KEYS`: Comma-separated list of allowed API keys for proxy authentication (optional; if unset, any non-empty key is accepted) - `LOG_BODIES`: Enable request/response body logging at debug level (`true` or `1`, default: disabled) +- `OTEL_EXPORTER_OTLP_ENDPOINT`: OTLP collector endpoint (default: `http://localhost:4318`). Only effective when built with `--features otel`. +- `OTEL_SERVICE_NAME`: Service name for exported traces. Only effective when built with `--features otel`. +- `OTEL_TRACES_SAMPLER`: Sampling strategy (default: `parentbased_always_on`). Only effective when built with `--features otel`. ## Architecture -Cargo workspace with two crates: +Cargo workspace with three crates: + +### `crates/client` (lib: `anyllm_client`) +High-level async HTTP client (Anthropic-in, Anthropic-out). Depends on `anyllm_translate` for translation logic. Key modules: +- **`client.rs`**: `Client` struct with `ClientConfig` builder; `messages()` for non-streaming, streaming variant for SSE +- **`http.rs`**: reqwest client builder with optional SSRF-safe DNS resolution and mTLS (PKCS#12) +- **`retry.rs`**: Generic retry with exponential backoff + jitter; `is_retryable`, `send_with_retry` +- **`rate_limit.rs`**: Parses `x-ratelimit-*` / `retry-after` headers into a typed struct +- **`sse.rs`**: Framework-agnostic SSE frame parser (`find_double_newline`) +- **`error.rs`**: `ClientError` enum ### `crates/translator` (lib: `anyllm_translate`) Pure translation logic, no IO. Key modules: @@ -79,6 +94,7 @@ Pure translation logic, no IO. Key modules: - `streaming_map`: SSE event stream translation state machine - `responses_message_map`: Anthropic to/from OpenAI Responses API mapping - `responses_streaming_map`: Responses API SSE event stream translation state machine + - `warnings`: `TranslationWarnings` collector; lossy drops are surfaced via `x-anyllm-degradation` response header - **`middleware/`**: Request/response handler orchestrating translation and backend calls - **`util/`**: JSON helpers, ID generation (uuid v4), secret redaction - **`config.rs`**: Translator-level configuration, **`error.rs`**: Error types, **`translate.rs`**: Top-level translation entry points @@ -126,8 +142,11 @@ Client (Anthropic format) -> proxy (axum) - Some source files reference PLAN.md line ranges in a comment at the top (historical; PLAN.md has been removed). - Test files live alongside source (`#[cfg(test)]` modules) and in `crates/proxy/tests/` for integration tests. - Error types use `thiserror` derive macros. -- Test distribution: translator (~263 tests), proxy (~154 tests including integration/compatibility). Counts shift as features are added. +- Test distribution: translator (~273 tests), proxy + client (~30 tests including integration/compatibility). Counts shift as features are added. ## References - OpenAI API spec: https://github.com/openai/openai-openapi/blob/manual_spec/openapi.yaml (very large, ~70k+ lines). See https://simonwillison.net/2024/Dec/22/openai-openapi/ for context on the spec's size and structure. Do not attempt to load the full spec into context; reference specific sections as needed. + +## Recent Changes +- 20260325-120000-litellm-gap-fill: Added [if applicable, e.g., PostgreSQL, CoreData, files or N/A] diff --git a/crates/proxy/Cargo.toml b/crates/proxy/Cargo.toml index 6945f75..19d3c57 100644 --- a/crates/proxy/Cargo.toml +++ b/crates/proxy/Cargo.toml @@ -32,6 +32,8 @@ httpdate = "1" dashmap = "6" aws-sigv4 = { version = "1.4", features = ["sign-http"] } aws-credential-types = "1.2" +aws-smithy-runtime-api = "1" +base64 = "0.22" [features] otel = [ diff --git a/crates/proxy/admin-ui/index.html b/crates/proxy/admin-ui/index.html index c6ee075..5911658 100644 --- a/crates/proxy/admin-ui/index.html +++ b/crates/proxy/admin-ui/index.html @@ -105,10 +105,11 @@ input:focus,select:focus{outline:none;border-color:#4a9eff}
+
- +
-
Configuration loaded from environment variables.
+
Loading...
@@ -187,7 +188,7 @@ input:focus,select:focus{outline:none;border-color:#4a9eff} navEl.classList.add('active'); document.getElementById('tab-' + navEl.dataset.tab).classList.add('active'); if (navEl.dataset.tab === 'requests') loadRequests(); - if (navEl.dataset.tab === 'settings') loadConfig(); + if (navEl.dataset.tab === 'settings') { loadConfig(); loadEnv(); } if (navEl.dataset.tab === 'backends') loadBackends(); }); }); @@ -402,7 +403,95 @@ input:focus,select:focus{outline:none;border-color:#4a9eff} } document.getElementById('btn-save-settings').addEventListener('click', saveSettings); - document.getElementById('btn-reload-config').addEventListener('click', loadConfig); + document.getElementById('btn-reload-config').addEventListener('click', function() { loadConfig(); loadEnv(); }); + document.getElementById('btn-export-env').addEventListener('click', exportEnvFile); + + // -- Environment variables (read-only, requires restart to change) -- + var ENV_GROUPS = [ + { label: 'Core', keys: ['BACKEND','LISTEN_PORT','BIG_MODEL','SMALL_MODEL','RUST_LOG','LOG_BODIES','PROXY_CONFIG'] }, + { label: 'OpenAI / Compatible', keys: ['OPENAI_BASE_URL','OPENAI_API_FORMAT','OPENAI_API_KEY'] }, + { label: 'Vertex AI', keys: ['VERTEX_PROJECT','VERTEX_REGION','VERTEX_API_KEY'] }, + { label: 'Gemini', keys: ['GEMINI_BASE_URL','GEMINI_API_KEY'] }, + { label: 'Auth & TLS', keys: ['PROXY_API_KEYS','TLS_CLIENT_CERT_P12','TLS_CA_CERT'] }, + { label: 'Admin', keys: ['ADMIN_PORT','ADMIN_DB_PATH','ADMIN_LOG_RETENTION_DAYS'] }, + ]; + var ENV_SECRET_KEYS = {'OPENAI_API_KEY':1,'VERTEX_API_KEY':1,'GEMINI_API_KEY':1,'PROXY_API_KEYS':1}; + + function loadEnv() { + apiFetch('/env').then(function(data) { + var container = document.getElementById('readonly-config'); + clearChildren(container, false); + ENV_GROUPS.forEach(function(group) { + var hasAny = group.keys.some(function(k) { return data[k] != null; }); + if (!hasAny) return; + var rows = []; + group.keys.forEach(function(k) { + var val = data[k]; + if (val == null) return; + rows.push(el('div', {className: 'model-grid', style:{marginTop:'4px'}}, [ + el('div', {className: 'label', textContent: k + ':'}), + el('div', {textContent: String(val), style:{fontFamily:'monospace',fontSize:'12px',color:'#c9d1d9'}}) + ])); + }); + if (rows.length === 0) return; + var header = el('div', {style:{color:'#8b949e',fontSize:'11px',textTransform:'uppercase',letterSpacing:'0.5px',marginTop:'10px',marginBottom:'4px'}, textContent: group.label}); + container.appendChild(header); + rows.forEach(function(r) { container.appendChild(r); }); + }); + if (!container.children.length) { + container.appendChild(el('div', {style:{color:'#8b949e',fontSize:'12px'}, textContent: 'No environment variables set (using defaults).'})); + } + }).catch(function(e) { + var container = document.getElementById('readonly-config'); + container.textContent = 'Failed to load environment variables.'; + console.error('Env load failed:', e); + }); + } + + // -- Export .env file -- + // Generates a .anyllm.env template the user can load with --env-file. + // Secret values (API keys) are masked server-side, so they appear as empty + // placeholders the user must fill in manually. + function exportEnvFile() { + apiFetch('/env').then(function(data) { + var lines = [ + '# anyllm-proxy configuration', + '# Load with: anyllm_proxy --env-file .anyllm.env', + '# Docker: docker run --env-file .anyllm.env anyllm-proxy', + '#', + '# Secret values (API keys) are intentionally left blank for security.', + '# Fill them in before use.', + '', + ]; + + ENV_GROUPS.forEach(function(section) { + var hasAny = section.keys.some(function(k) { return data[k] != null || k in ENV_SECRET_KEYS; }); + if (!hasAny) return; + lines.push('# --- ' + section.label + ' ---'); + section.keys.forEach(function(k) { + if (k in ENV_SECRET_KEYS) { + // Always emit secret keys as empty placeholders. + lines.push(k + '='); + } else if (data[k] != null) { + lines.push(k + '=' + String(data[k])); + } + // Omit unset non-secret keys entirely to keep the file minimal. + }); + lines.push(''); + }); + + var blob = new Blob([lines.join('\n')], {type: 'text/plain'}); + var a = document.createElement('a'); + a.href = URL.createObjectURL(blob); + a.download = '.anyllm.env'; + document.body.appendChild(a); + a.click(); + document.body.removeChild(a); + URL.revokeObjectURL(a.href); + }).catch(function(e) { + alert('Failed to fetch env: ' + e.message); + }); + } // -- Backends detail -- function loadBackends() { diff --git a/crates/proxy/src/backend/bedrock_client.rs b/crates/proxy/src/backend/bedrock_client.rs new file mode 100644 index 0000000..d2dbe08 --- /dev/null +++ b/crates/proxy/src/backend/bedrock_client.rs @@ -0,0 +1,399 @@ +// AWS Bedrock client with SigV4 request signing. +// Sends Anthropic Messages API requests directly to Bedrock (no OpenAI translation). +// Bedrock streaming uses AWS Event Stream binary framing, not SSE. + +use super::{build_http_client, RateLimitHeaders}; +use crate::config::TlsConfig; +use aws_credential_types::Credentials; +use aws_sigv4::http_request::{ + sign, SignableBody, SignableRequest, SigningSettings, +}; +use aws_sigv4::sign::v4; +use reqwest::Client; +use tokio::time::sleep; + +/// HTTP client for AWS Bedrock with SigV4 request signing. +#[derive(Clone)] +pub struct BedrockClient { + client: Client, + region: String, + credentials: Credentials, + big_model: String, + small_model: String, +} + +/// Error type for the Bedrock client. +#[derive(Debug)] +pub enum BedrockClientError { + /// Transport-level error (connection, timeout, DNS). + Transport(String), + /// Upstream returned a non-success status. Body is raw bytes for passthrough. + ApiError { status: u16, body: bytes::Bytes }, + /// SigV4 signing failed. + Signing(String), +} + +impl std::fmt::Display for BedrockClientError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Transport(msg) => write!(f, "Bedrock transport error: {msg}"), + Self::ApiError { status, .. } => write!(f, "Bedrock API error (status {status})"), + Self::Signing(msg) => write!(f, "Bedrock signing error: {msg}"), + } + } +} + +impl BedrockClient { + /// Create a new Bedrock client. + pub fn new( + region: String, + credentials: Credentials, + big_model: String, + small_model: String, + tls: &TlsConfig, + ) -> Self { + let client = build_http_client(tls); + Self { + client, + region, + credentials, + big_model, + small_model, + } + } + + pub fn big_model(&self) -> &str { + &self.big_model + } + + pub fn small_model(&self) -> &str { + &self.small_model + } + + /// Build the Bedrock InvokeModel URL for a given model. + fn invoke_url(&self, model_id: &str) -> String { + format!( + "https://bedrock-runtime.{}.amazonaws.com/model/{}/invoke", + self.region, model_id + ) + } + + /// Build the Bedrock InvokeModelWithResponseStream URL. + fn invoke_stream_url(&self, model_id: &str) -> String { + format!( + "https://bedrock-runtime.{}.amazonaws.com/model/{}/invoke-with-response-stream", + self.region, model_id + ) + } + + /// Sign an HTTP request with SigV4 and return headers to add. + fn sign_request( + &self, + method: &str, + url: &str, + body_bytes: &[u8], + extra_headers: &[(&str, &str)], + ) -> Result, BedrockClientError> { + let identity: aws_smithy_runtime_api::client::identity::Identity = + self.credentials.clone().into(); + let settings = SigningSettings::default(); + let params = v4::SigningParams::builder() + .identity(&identity) + .region(&self.region) + .name("bedrock") + .time(std::time::SystemTime::now()) + .settings(settings) + .build() + .map_err(|e| BedrockClientError::Signing(e.to_string()))?; + let signing_params = params.into(); + + let signable = SignableRequest::new( + method, + url, + extra_headers.iter().copied(), + SignableBody::Bytes(body_bytes), + ) + .map_err(|e| BedrockClientError::Signing(e.to_string()))?; + + let (instructions, _signature) = sign(signable, &signing_params) + .map_err(|e| BedrockClientError::Signing(e.to_string()))? + .into_parts(); + + // Collect signing headers + let headers: Vec<(String, String)> = instructions + .headers() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + Ok(headers) + } + + /// Forward a non-streaming request. Returns raw response body and rate limit headers. + pub async fn forward( + &self, + body: bytes::Bytes, + model_id: &str, + ) -> Result<(bytes::Bytes, RateLimitHeaders), BedrockClientError> { + let response = self.send_with_retry(body, model_id, false).await?; + let rate_limits = RateLimitHeaders::default(); + let resp_body = response + .bytes() + .await + .map_err(|e| BedrockClientError::Transport(e.to_string()))?; + Ok((resp_body, rate_limits)) + } + + /// Forward a streaming request. Returns the raw response for event stream decoding. + pub async fn forward_stream( + &self, + body: bytes::Bytes, + model_id: &str, + ) -> Result<(reqwest::Response, RateLimitHeaders), BedrockClientError> { + let response = self.send_with_retry(body, model_id, true).await?; + let rate_limits = RateLimitHeaders::default(); + Ok((response, rate_limits)) + } + + /// Send with retry on 429/5xx. + async fn send_with_retry( + &self, + body: bytes::Bytes, + model_id: &str, + stream: bool, + ) -> Result { + let url = if stream { + self.invoke_stream_url(model_id) + } else { + self.invoke_url(model_id) + }; + + let content_type = "application/json"; + let accept = if stream { + "application/vnd.amazon.eventstream" + } else { + "application/json" + }; + + for attempt in 0..=super::MAX_RETRIES { + let base_headers = [ + ("content-type", content_type), + ("accept", accept), + ]; + let signing_headers = self + .sign_request("POST", &url, &body, &base_headers)?; + + let mut rb = self + .client + .post(&url) + .header("content-type", content_type) + .header("accept", accept) + .body(body.clone()); + + for (k, v) in &signing_headers { + rb = rb.header(k.as_str(), v.as_str()); + } + + let response = rb + .send() + .await + .map_err(|e| BedrockClientError::Transport(e.to_string()))?; + let status = response.status().as_u16(); + + if (200..300).contains(&status) { + return Ok(response); + } + + if attempt < super::MAX_RETRIES && super::is_retryable(status) { + let retry_after = super::parse_retry_after(response.headers()); + let delay = super::backoff_delay(attempt, retry_after); + tracing::warn!( + status, + attempt = attempt + 1, + max_retries = super::MAX_RETRIES, + delay_ms = delay.as_millis() as u64, + "retryable error from Bedrock, backing off" + ); + drop(response.bytes().await); + sleep(delay).await; + continue; + } + + let resp_body = response.bytes().await.unwrap_or_default(); + return Err(BedrockClientError::ApiError { + status, + body: resp_body, + }); + } + unreachable!("loop runs MAX_RETRIES+1 times and always returns") + } +} + +// --------------------------------------------------------------------------- +// AWS Event Stream binary frame decoder +// --------------------------------------------------------------------------- + +/// Decode AWS Event Stream frames from a byte buffer. +/// Each frame: 4-byte total_len | 4-byte headers_len | 4-byte prelude CRC | +/// headers | payload | 4-byte message CRC +/// +/// The payload contains `{"bytes":""}` where base64 decodes to an +/// Anthropic SSE JSON event string. +pub mod eventstream { + use bytes::BytesMut; + + /// Minimum frame size: 4 (total_len) + 4 (headers_len) + 4 (prelude CRC) + /// + 0 (headers) + 0 (payload) + 4 (message CRC) = 16 + const MIN_FRAME_SIZE: usize = 16; + + /// Try to extract one complete event stream frame from the buffer. + /// Returns `Some(payload_bytes)` and advances the buffer past the frame, + /// or `None` if the buffer does not contain a complete frame yet. + pub fn decode_frame(buf: &mut BytesMut) -> Option> { + if buf.len() < MIN_FRAME_SIZE { + return None; + } + + let total_len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize; + if buf.len() < total_len { + return None; // incomplete frame + } + + let headers_len = u32::from_be_bytes([buf[4], buf[5], buf[6], buf[7]]) as usize; + + // Prelude is 8 bytes (total_len + headers_len), then 4-byte prelude CRC + let headers_start = 12; // 4 + 4 + 4 (prelude CRC) + let payload_start = headers_start + headers_len; + // Message CRC is the last 4 bytes + let payload_end = total_len.saturating_sub(4); + + if payload_start > payload_end || payload_end > total_len { + // Malformed frame: skip it + let _ = buf.split_to(total_len); + return Some(Vec::new()); + } + + let payload = buf[payload_start..payload_end].to_vec(); + + // Advance buffer past this frame + let _ = buf.split_to(total_len); + + Some(payload) + } + + /// Extract the Anthropic event JSON string from a Bedrock event stream payload. + /// Bedrock wraps the Anthropic event in `{"bytes":""}`. + /// Returns None if the payload is not a chunk event or is malformed. + pub fn extract_event_from_payload(payload: &[u8]) -> Option { + if payload.is_empty() { + return None; + } + + // Parse as JSON to extract the base64-encoded bytes field + let parsed: serde_json::Value = serde_json::from_slice(payload).ok()?; + let b64 = parsed.get("bytes")?.as_str()?; + + // Base64 decode + use base64::Engine; + let decoded = base64::engine::general_purpose::STANDARD + .decode(b64) + .ok()?; + String::from_utf8(decoded).ok() + } +} + +#[cfg(test)] +mod tests { + use super::eventstream; + use bytes::BytesMut; + + /// Build a minimal AWS Event Stream frame with the given payload. + /// Uses zero CRCs (we don't validate CRCs in the decoder). + fn build_frame(headers: &[u8], payload: &[u8]) -> Vec { + let total_len = 12 + headers.len() + payload.len() + 4; // prelude(12) + headers + payload + msg CRC(4) + let headers_len = headers.len(); + let mut frame = Vec::with_capacity(total_len); + frame.extend_from_slice(&(total_len as u32).to_be_bytes()); + frame.extend_from_slice(&(headers_len as u32).to_be_bytes()); + frame.extend_from_slice(&[0u8; 4]); // prelude CRC (not validated) + frame.extend_from_slice(headers); + frame.extend_from_slice(payload); + frame.extend_from_slice(&[0u8; 4]); // message CRC (not validated) + frame + } + + #[test] + fn decode_frame_empty_payload() { + let frame = build_frame(&[], &[]); + let mut buf = BytesMut::from(frame.as_slice()); + let payload = eventstream::decode_frame(&mut buf).unwrap(); + assert!(payload.is_empty()); + assert!(buf.is_empty()); + } + + #[test] + fn decode_frame_with_payload() { + let payload_data = b"hello world"; + let frame = build_frame(&[], payload_data); + let mut buf = BytesMut::from(frame.as_slice()); + let payload = eventstream::decode_frame(&mut buf).unwrap(); + assert_eq!(payload, b"hello world"); + assert!(buf.is_empty()); + } + + #[test] + fn decode_frame_incomplete() { + let frame = build_frame(&[], b"hello"); + let mut buf = BytesMut::from(&frame[..frame.len() - 2]); // truncate + assert!(eventstream::decode_frame(&mut buf).is_none()); + } + + #[test] + fn decode_multiple_frames() { + let frame1 = build_frame(&[], b"first"); + let frame2 = build_frame(&[], b"second"); + let mut buf = BytesMut::new(); + buf.extend_from_slice(&frame1); + buf.extend_from_slice(&frame2); + + let p1 = eventstream::decode_frame(&mut buf).unwrap(); + assert_eq!(p1, b"first"); + let p2 = eventstream::decode_frame(&mut buf).unwrap(); + assert_eq!(p2, b"second"); + assert!(buf.is_empty()); + } + + #[test] + fn decode_frame_with_headers() { + let headers = b"\x00\x04test"; + let payload_data = b"data"; + let frame = build_frame(headers, payload_data); + let mut buf = BytesMut::from(frame.as_slice()); + let payload = eventstream::decode_frame(&mut buf).unwrap(); + assert_eq!(payload, b"data"); + } + + #[test] + fn extract_event_from_valid_payload() { + use base64::Engine; + let event_json = r#"{"type":"content_block_delta","index":0}"#; + let b64 = base64::engine::general_purpose::STANDARD.encode(event_json); + let wrapper = format!(r#"{{"bytes":"{b64}"}}"#); + let result = eventstream::extract_event_from_payload(wrapper.as_bytes()); + assert_eq!(result.unwrap(), event_json); + } + + #[test] + fn extract_event_empty_payload() { + assert!(eventstream::extract_event_from_payload(&[]).is_none()); + } + + #[test] + fn extract_event_invalid_json() { + assert!(eventstream::extract_event_from_payload(b"not json").is_none()); + } + + #[test] + fn extract_event_missing_bytes_field() { + let payload = r#"{"other":"field"}"#; + assert!(eventstream::extract_event_from_payload(payload.as_bytes()).is_none()); + } +} diff --git a/crates/proxy/src/backend/mod.rs b/crates/proxy/src/backend/mod.rs index d5ceb09..3e2db8d 100644 --- a/crates/proxy/src/backend/mod.rs +++ b/crates/proxy/src/backend/mod.rs @@ -1,10 +1,13 @@ /// Passthrough client forwarding Anthropic requests as-is to upstream Anthropic API. pub mod anthropic_client; +/// AWS Bedrock client with SigV4 request signing. +pub mod bedrock_client; /// reqwest client for OpenAI-compatible Chat Completions and Responses APIs with retry/backoff. pub mod openai_client; use crate::config::{BackendAuth, BackendConfig, BackendKind, Config, OpenAIApiFormat, TlsConfig}; use anthropic_client::{AnthropicClient, AnthropicClientError}; +use bedrock_client::{BedrockClient, BedrockClientError}; use openai_client::{OpenAIClient, OpenAIClientError}; // Re-export from the client crate so existing code paths (streaming, routes, etc.) keep working. @@ -66,6 +69,8 @@ pub enum BackendClient { GeminiOpenAI(OpenAIClient), /// Passthrough to real Anthropic API (no translation). Anthropic(AnthropicClient), + /// AWS Bedrock: sends Anthropic-format requests with SigV4 signing. + Bedrock(BedrockClient), } /// Unified error type for all backend clients. @@ -73,6 +78,7 @@ pub enum BackendClient { pub enum BackendError { OpenAI(OpenAIClientError), Anthropic(AnthropicClientError), + Bedrock(BedrockClientError), } impl BackendError { @@ -80,6 +86,7 @@ impl BackendError { pub fn api_error_status(&self) -> Option { match self { Self::OpenAI(OpenAIClientError::ApiError { status, .. }) => Some(*status), + Self::Bedrock(BedrockClientError::ApiError { status, .. }) => Some(*status), _ => None, } } @@ -94,6 +101,7 @@ impl BackendError { match self { Self::OpenAI(e) => e.to_string(), Self::Anthropic(e) => e.to_string(), + Self::Bedrock(e) => e.to_string(), } } @@ -114,6 +122,7 @@ impl std::fmt::Display for BackendError { match self { Self::OpenAI(e) => write!(f, "{e}"), Self::Anthropic(e) => write!(f, "{e}"), + Self::Bedrock(e) => write!(f, "{e}"), } } } @@ -130,6 +139,12 @@ impl From for BackendError { } } +impl From for BackendError { + fn from(e: BedrockClientError) -> Self { + Self::Bedrock(e) + } +} + impl BackendClient { /// Forward a raw embeddings request to the backend. No translation — model names pass through. /// Returns `501 Not Implemented` for the Anthropic backend (no embeddings endpoint). @@ -147,11 +162,11 @@ impl BackendClient { .embeddings_passthrough(body, content_type) .await .map_err(BackendError::OpenAI), - Self::Anthropic(_) => { - // Anthropic has no embeddings API. + Self::Anthropic(_) | Self::Bedrock(_) => { + // Anthropic and Bedrock have no embeddings API. let err = anyllm_translate::mapping::errors_map::create_anthropic_error( anyllm_translate::anthropic::ErrorType::InvalidRequestError, - "Embeddings are not supported by the Anthropic backend.".to_string(), + "Embeddings are not supported by this backend.".to_string(), None, ); let body = serde_json::to_vec(&err).unwrap_or_default(); @@ -182,6 +197,11 @@ impl BackendClient { &config.openai_api_key, &config.tls, )), + BackendKind::Bedrock => { + // Bedrock config is stored in openai_base_url (region) and openai_api_key (unused). + // Credentials come from env vars at Config::from_env time. + unreachable!("Bedrock backend uses from_backend_config, not Config::new") + } } } @@ -210,6 +230,15 @@ impl BackendClient { BackendKind::Vertex => Self::Vertex(OpenAIClient::new(&legacy)), BackendKind::Gemini => Self::GeminiOpenAI(OpenAIClient::new(&legacy)), BackendKind::Anthropic => Self::Anthropic(AnthropicClient::from_backend_config(bc)), + BackendKind::Bedrock => Self::Bedrock(BedrockClient::new( + bc.base_url.clone(), // region is stored in base_url for Bedrock + bc.bedrock_credentials + .clone() + .expect("Bedrock credentials must be set"), + bc.model_mapping.big_model.clone(), + bc.model_mapping.small_model.clone(), + &bc.tls, + )), } } } diff --git a/crates/proxy/src/backend/openai_client.rs b/crates/proxy/src/backend/openai_client.rs index 1d85790..ece1aa6 100644 --- a/crates/proxy/src/backend/openai_client.rs +++ b/crates/proxy/src/backend/openai_client.rs @@ -74,8 +74,8 @@ impl OpenAIClient { format!("{endpoint}/openai/deployments/{deployment}/embeddings?api-version={api_version}"), ) } - BackendKind::Anthropic => { - unreachable!("OpenAIClient should not be constructed for Anthropic backend") + BackendKind::Anthropic | BackendKind::Bedrock => { + unreachable!("OpenAIClient should not be constructed for Anthropic/Bedrock backend") } }; diff --git a/crates/proxy/src/config/mod.rs b/crates/proxy/src/config/mod.rs index 5ea48ed..f68742e 100644 --- a/crates/proxy/src/config/mod.rs +++ b/crates/proxy/src/config/mod.rs @@ -19,6 +19,7 @@ pub enum BackendKind { Vertex, Gemini, Anthropic, + Bedrock, } /// Which OpenAI API format to use (only relevant when BACKEND=openai). @@ -92,8 +93,9 @@ impl Config { "vertex" => BackendKind::Vertex, "gemini" => BackendKind::Gemini, "anthropic" => BackendKind::Anthropic, + "bedrock" => BackendKind::Bedrock, other => { - panic!("unknown BACKEND value '{other}', expected 'openai', 'azure', 'vertex', 'gemini', or 'anthropic'") + panic!("unknown BACKEND value '{other}', expected 'openai', 'azure', 'vertex', 'gemini', 'anthropic', or 'bedrock'") } }; @@ -267,6 +269,37 @@ impl Config { openai_api_format: OpenAIApiFormat::Chat, } } + BackendKind::Bedrock => { + let region = std::env::var("AWS_REGION") + .unwrap_or_else(|_| panic!("AWS_REGION is required when BACKEND=bedrock")); + validate_gcp_identifier("AWS_REGION", ®ion); // reuse safe-char validation + + // Validate credentials are present at startup; the actual values + // are read again when constructing BedrockClient. + let _access_key_id = std::env::var("AWS_ACCESS_KEY_ID").unwrap_or_else(|_| { + panic!("AWS_ACCESS_KEY_ID is required when BACKEND=bedrock") + }); + let _secret_access_key = std::env::var("AWS_SECRET_ACCESS_KEY").unwrap_or_else( + |_| panic!("AWS_SECRET_ACCESS_KEY is required when BACKEND=bedrock"), + ); + let _session_token = std::env::var("AWS_SESSION_TOKEN").ok(); + + Self { + backend, + openai_api_key: String::new(), + // Store region in openai_base_url for wrap_config + openai_base_url: region.clone(), + listen_port, + model_mapping: ModelMapping::from_env_with_defaults( + "anthropic.claude-sonnet-4-20250514-v1:0", + "anthropic.claude-haiku-4-5-20251001-v1:0", + ), + tls, + backend_auth: BackendAuth::BearerToken(String::new()), + log_bodies, + openai_api_format: OpenAIApiFormat::Chat, + } + } } } } @@ -318,6 +351,22 @@ fn contains_ignore_ascii_case(haystack: &[u8], needle: &[u8]) -> bool { .any(|w| w.eq_ignore_ascii_case(needle)) } +/// Read AWS credentials from environment variables for the Bedrock backend. +fn bedrock_credentials_from_env() -> aws_credential_types::Credentials { + let access_key_id = std::env::var("AWS_ACCESS_KEY_ID") + .unwrap_or_else(|_| panic!("AWS_ACCESS_KEY_ID is required for bedrock")); + let secret_access_key = std::env::var("AWS_SECRET_ACCESS_KEY") + .unwrap_or_else(|_| panic!("AWS_SECRET_ACCESS_KEY is required for bedrock")); + let session_token = std::env::var("AWS_SESSION_TOKEN").ok(); + aws_credential_types::Credentials::new( + access_key_id, + secret_access_key, + session_token, + None, + "env", + ) +} + // --------------------------------------------------------------------------- // Multi-backend configuration // --------------------------------------------------------------------------- @@ -357,6 +406,8 @@ pub struct BackendConfig { /// (older Ollama, text-generation-webui, LM Studio) that reject unknown /// fields with HTTP 400. pub omit_stream_options: bool, + /// AWS credentials for Bedrock backend. None for all other backends. + pub bedrock_credentials: Option, } /// Top-level multi-backend configuration loaded from TOML. @@ -403,6 +454,10 @@ struct TomlBackendConfig { access_token: Option, // Strip stream_options from streaming requests (local LLM compat) omit_stream_options: Option, + // Bedrock-specific: AWS credentials (support env: prefix for env var resolution) + aws_access_key_id: Option, + aws_secret_access_key: Option, + aws_session_token: Option, } impl MultiConfig { @@ -435,12 +490,20 @@ impl MultiConfig { BackendKind::Vertex => "vertex", BackendKind::Gemini => "gemini", BackendKind::Anthropic => "anthropic", + BackendKind::Bedrock => "bedrock", }; let omit_stream_options = std::env::var("OMIT_STREAM_OPTIONS") .map(|v| v == "true" || v == "1") .unwrap_or(false); + // For Bedrock, read AWS credentials from env vars. + let bedrock_credentials = if config.backend == BackendKind::Bedrock { + Some(bedrock_credentials_from_env()) + } else { + None + }; + let bc = BackendConfig { kind: config.backend.clone(), api_key: config.openai_api_key.clone(), @@ -451,6 +514,7 @@ impl MultiConfig { backend_auth: config.backend_auth.clone(), log_bodies: config.log_bodies, omit_stream_options, + bedrock_credentials, }; let mut backends = IndexMap::new(); @@ -521,6 +585,7 @@ impl MultiConfig { "vertex" => BackendKind::Vertex, "gemini" => BackendKind::Gemini, "anthropic" => BackendKind::Anthropic, + "bedrock" => BackendKind::Bedrock, other => panic!("unknown backend kind '{other}' for backend '{name}'"), }; @@ -684,6 +749,62 @@ impl MultiConfig { }; (base_url, auth, mm, OpenAIApiFormat::Chat) } + BackendKind::Bedrock => { + let region = tb.region.as_deref().unwrap_or_else(|| { + panic!("backend '{name}': 'region' is required for bedrock") + }); + validate_gcp_identifier("region", region); + + // For Bedrock, base_url stores the region (used by BedrockClient to build URLs) + let auth = BackendAuth::BearerToken(String::new()); + let mm = ModelMapping { + big_model: tb.big_model.clone().unwrap_or_else(|| { + "anthropic.claude-sonnet-4-20250514-v1:0".to_string() + }), + small_model: tb.small_model.clone().unwrap_or_else(|| { + "anthropic.claude-haiku-4-5-20251001-v1:0".to_string() + }), + }; + (region.to_string(), auth, mm, OpenAIApiFormat::Chat) + } + }; + + // Build AWS credentials for Bedrock from TOML fields or env vars. + let bedrock_credentials = if kind == BackendKind::Bedrock { + let access_key_id = tb + .aws_access_key_id + .as_deref() + .map(|v| resolve_env_value(v).unwrap_or_else(|e| panic!("backend '{name}': {e}"))) + .unwrap_or_else(|| { + std::env::var("AWS_ACCESS_KEY_ID").unwrap_or_else(|_| { + panic!("backend '{name}': aws_access_key_id or AWS_ACCESS_KEY_ID required") + }) + }); + let secret_access_key = tb + .aws_secret_access_key + .as_deref() + .map(|v| resolve_env_value(v).unwrap_or_else(|e| panic!("backend '{name}': {e}"))) + .unwrap_or_else(|| { + std::env::var("AWS_SECRET_ACCESS_KEY").unwrap_or_else(|_| { + panic!( + "backend '{name}': aws_secret_access_key or AWS_SECRET_ACCESS_KEY required" + ) + }) + }); + let session_token = tb + .aws_session_token + .as_deref() + .map(|v| resolve_env_value(v).unwrap_or_else(|e| panic!("backend '{name}': {e}"))) + .or_else(|| std::env::var("AWS_SESSION_TOKEN").ok()); + Some(aws_credential_types::Credentials::new( + access_key_id, + secret_access_key, + session_token, + None, + "toml-config", + )) + } else { + None }; BackendConfig { @@ -696,6 +817,7 @@ impl MultiConfig { backend_auth, log_bodies, omit_stream_options: tb.omit_stream_options.unwrap_or(false), + bedrock_credentials, } } } diff --git a/crates/proxy/src/lib.rs b/crates/proxy/src/lib.rs index a98de8f..ee82649 100644 --- a/crates/proxy/src/lib.rs +++ b/crates/proxy/src/lib.rs @@ -6,5 +6,8 @@ pub mod backend; pub mod config; /// Request count, success/error tracking, exposed via GET /metrics. pub mod metrics; +/// Optional OpenTelemetry OTLP trace export (requires `otel` feature). +#[cfg(feature = "otel")] +pub mod otel; /// Axum HTTP server: routes, middleware (auth, request ID, size/concurrency limits), SSE streaming. pub mod server; diff --git a/crates/proxy/src/main.rs b/crates/proxy/src/main.rs index 9dd439e..f330b40 100644 --- a/crates/proxy/src/main.rs +++ b/crates/proxy/src/main.rs @@ -25,6 +25,22 @@ async fn main() { // Use a reload layer so the admin API can change log_level at runtime. let env_filter = tracing_subscriber::EnvFilter::from_default_env(); let (filter, reload_handle) = tracing_subscriber::reload::Layer::new(env_filter); + + // When the `otel` feature is enabled, wire an OpenTelemetry tracing layer + // into the subscriber so that spans are exported as OTLP traces. + #[cfg(feature = "otel")] + let _otel_guard = { + let (guard, tracer) = anyllm_proxy::otel::init_otel(); + let otel_layer = tracing_opentelemetry::OpenTelemetryLayer::new(tracer); + tracing_subscriber::registry() + .with(filter) + .with(tracing_subscriber::fmt::layer().json()) + .with(otel_layer) + .init(); + guard + }; + + #[cfg(not(feature = "otel"))] tracing_subscriber::registry() .with(filter) .with(tracing_subscriber::fmt::layer().json()) diff --git a/crates/proxy/src/otel.rs b/crates/proxy/src/otel.rs new file mode 100644 index 0000000..4e3d518 --- /dev/null +++ b/crates/proxy/src/otel.rs @@ -0,0 +1,51 @@ +//! Optional OpenTelemetry OTLP trace export. +//! +//! Enabled by the `otel` cargo feature. All types and functions in this module +//! are gated behind `#[cfg(feature = "otel")]` at the module declaration site +//! in `lib.rs`, so nothing here compiles into the default binary. +//! +//! The OTLP SDK reads configuration from standard env vars: +//! - `OTEL_EXPORTER_OTLP_ENDPOINT` (default `http://localhost:4318` for HTTP) +//! - `OTEL_SERVICE_NAME` +//! - `OTEL_TRACES_SAMPLER` / `OTEL_TRACES_SAMPLER_ARG` + +use opentelemetry::trace::TracerProvider as _; +use opentelemetry_otlp::SpanExporter; +use opentelemetry_sdk::trace::SdkTracerProvider; + +/// Holds the [`SdkTracerProvider`] and flushes pending spans on drop. +pub struct OtelGuard { + provider: SdkTracerProvider, +} + +impl Drop for OtelGuard { + fn drop(&mut self) { + if let Err(e) = self.provider.shutdown() { + eprintln!("otel: tracer provider shutdown error: {e}"); + } + } +} + +/// Initialise the OTLP span exporter and return a guard that must live for the +/// duration of `main`. The returned tracer is suitable for +/// [`tracing_opentelemetry::OpenTelemetryLayer::new`]. +/// +/// Panics if the exporter or provider cannot be created (misconfiguration). +pub fn init_otel() -> (OtelGuard, opentelemetry_sdk::trace::Tracer) { + // HTTP/protobuf transport via reqwest (matches Cargo feature flags). + let exporter = SpanExporter::builder() + .with_http() + .build() + .expect("failed to create OTLP span exporter"); + + let provider = SdkTracerProvider::builder() + .with_batch_exporter(exporter) + .build(); + + opentelemetry::global::set_tracer_provider(provider.clone()); + + let tracer = provider.tracer("anyllm-proxy"); + let guard = OtelGuard { provider }; + + (guard, tracer) +} diff --git a/crates/proxy/src/server/bedrock_passthrough.rs b/crates/proxy/src/server/bedrock_passthrough.rs new file mode 100644 index 0000000..8a584b7 --- /dev/null +++ b/crates/proxy/src/server/bedrock_passthrough.rs @@ -0,0 +1,240 @@ +// Bedrock passthrough handler: forwards Anthropic-format requests to AWS Bedrock +// with SigV4 signing and AWS Event Stream decoding for streaming. + +use crate::backend::bedrock_client::{eventstream, BedrockClientError}; +use crate::backend::BackendClient; +use anyllm_translate::{anthropic, mapping}; +use axum::{ + body::Bytes, + extract::State, + http::StatusCode, + response::{IntoResponse, Json, Response}, +}; +use bytes::BytesMut; +use futures::StreamExt; + +use super::routes::AppState; + +/// Bedrock passthrough handler for POST /v1/messages. +/// Strips the `model` field from the body (Bedrock uses it in the URL), +/// adds `anthropic_version`, and handles AWS Event Stream binary framing +/// for streaming responses. +pub(crate) async fn bedrock_passthrough(State(state): State, body: Bytes) -> Response { + state.metrics.record_request(); + + let client = match &state.backend { + BackendClient::Bedrock(c) => c.clone(), + _ => { + let err = mapping::errors_map::create_anthropic_error( + anthropic::ErrorType::ApiError, + "Backend is not configured as bedrock".to_string(), + None, + ); + return (StatusCode::INTERNAL_SERVER_ERROR, Json(err)).into_response(); + } + }; + + // Parse the body to extract model and stream fields, then rebuild for Bedrock. + let mut parsed: serde_json::Value = match serde_json::from_slice(&body) { + Ok(v) => v, + Err(e) => { + let err = mapping::errors_map::create_anthropic_error( + anthropic::ErrorType::InvalidRequestError, + format!("invalid JSON: {e}"), + None, + ); + return (StatusCode::BAD_REQUEST, Json(err)).into_response(); + } + }; + + // Extract model for the URL + let model_id = parsed + .get("model") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + if model_id.is_empty() { + let err = mapping::errors_map::create_anthropic_error( + anthropic::ErrorType::InvalidRequestError, + "model is required".to_string(), + None, + ); + return (StatusCode::BAD_REQUEST, Json(err)).into_response(); + } + + // Map model name through runtime config + let mapped_model = state.map_model(&model_id); + + let is_stream = parsed + .get("stream") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + + // Bedrock: model goes in URL, not body. Add anthropic_version. + if let Some(obj) = parsed.as_object_mut() { + obj.remove("model"); + obj.insert( + "anthropic_version".to_string(), + serde_json::Value::String("bedrock-2023-05-31".to_string()), + ); + } + + let bedrock_body = match serde_json::to_vec(&parsed) { + Ok(b) => bytes::Bytes::from(b), + Err(e) => { + let err = mapping::errors_map::create_anthropic_error( + anthropic::ErrorType::ApiError, + format!("failed to serialize request: {e}"), + None, + ); + return (StatusCode::INTERNAL_SERVER_ERROR, Json(err)).into_response(); + } + }; + + if is_stream { + bedrock_stream(state, &client, bedrock_body, &mapped_model).await + } else { + bedrock_non_stream(state, &client, bedrock_body, &mapped_model).await + } +} + +/// Non-streaming Bedrock request. +async fn bedrock_non_stream( + state: AppState, + client: &crate::backend::bedrock_client::BedrockClient, + body: bytes::Bytes, + model_id: &str, +) -> Response { + match client.forward(body, model_id).await { + Ok((resp_body, rate_limits)) => { + state.metrics.record_success(); + let mut resp = ( + StatusCode::OK, + [("content-type", "application/json")], + resp_body, + ) + .into_response(); + rate_limits.inject_anthropic_response_headers(resp.headers_mut()); + resp + } + Err(e) => { + state.metrics.record_error(); + bedrock_error_to_response(e) + } + } +} + +/// Streaming Bedrock request. Decodes AWS Event Stream binary frames into +/// Anthropic SSE events and re-emits them as standard SSE. +async fn bedrock_stream( + state: AppState, + client: &crate::backend::bedrock_client::BedrockClient, + body: bytes::Bytes, + model_id: &str, +) -> Response { + let (response, rate_limits) = match client.forward_stream(body, model_id).await { + Ok(r) => r, + Err(e) => { + state.metrics.record_error(); + return bedrock_error_to_response(e); + } + }; + + let (tx, rx) = + tokio::sync::mpsc::channel::>(32); + let metrics = state.metrics.clone(); + + tokio::spawn(async move { + let mut byte_stream = response.bytes_stream(); + let mut event_buf = BytesMut::new(); + + while let Some(chunk_result) = byte_stream.next().await { + let bytes = match chunk_result { + Ok(b) => b, + Err(e) => { + tracing::error!("Bedrock stream read error: {e}"); + metrics.record_error(); + return; + } + }; + event_buf.extend_from_slice(&bytes); + + // Decode all complete frames in the buffer + while let Some(payload) = eventstream::decode_frame(&mut event_buf) { + if let Some(event_json) = eventstream::extract_event_from_payload(&payload) { + // Re-emit as SSE: "event: \ndata: \n\n" + // Bedrock events are raw Anthropic JSON; detect the event type. + let event_type = detect_event_type(&event_json); + let sse_line = format!("event: {event_type}\ndata: {event_json}\n\n"); + if tx.send(Ok(sse_line)).await.is_err() { + return; // client disconnected + } + } + } + } + metrics.record_success(); + }); + + let body_stream = tokio_stream::wrappers::ReceiverStream::new(rx); + let body = axum::body::Body::from_stream(body_stream); + let mut resp = axum::http::Response::builder() + .status(StatusCode::OK) + .header("content-type", "text/event-stream") + .header("cache-control", "no-cache") + .body(body) + .unwrap_or_else(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response()); + rate_limits.inject_anthropic_response_headers(resp.headers_mut()); + resp +} + +/// Detect the Anthropic SSE event type from a JSON string. +/// Falls back to "message" if the type field is not found. +fn detect_event_type(json: &str) -> &str { + // Quick extraction without full parse: look for "type":"..." + if let Ok(v) = serde_json::from_str::(json) { + if let Some(t) = v.get("type").and_then(|t| t.as_str()) { + return match t { + "message_start" => "message_start", + "content_block_start" => "content_block_start", + "content_block_delta" => "content_block_delta", + "content_block_stop" => "content_block_stop", + "message_delta" => "message_delta", + "message_stop" => "message_stop", + "ping" => "ping", + _ => "message", + }; + } + } + "message" +} + +/// Convert a BedrockClientError into a Response. +fn bedrock_error_to_response(error: BedrockClientError) -> Response { + match error { + BedrockClientError::ApiError { status, body } => { + let http_status = + StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + // Try to return body as-is (Bedrock may return JSON error) + (http_status, [("content-type", "application/json")], body).into_response() + } + BedrockClientError::Transport(msg) => { + tracing::error!("Bedrock transport error: {msg}"); + let err = mapping::errors_map::create_anthropic_error( + anthropic::ErrorType::ApiError, + "An internal error occurred while communicating with the upstream service." + .to_string(), + None, + ); + (StatusCode::BAD_GATEWAY, Json(err)).into_response() + } + BedrockClientError::Signing(msg) => { + tracing::error!("Bedrock signing error: {msg}"); + let err = mapping::errors_map::create_anthropic_error( + anthropic::ErrorType::ApiError, + "Failed to sign request for AWS Bedrock.".to_string(), + None, + ); + (StatusCode::INTERNAL_SERVER_ERROR, Json(err)).into_response() + } + } +} diff --git a/crates/proxy/src/server/chat_completions.rs b/crates/proxy/src/server/chat_completions.rs index aaf75f7..479caf7 100644 --- a/crates/proxy/src/server/chat_completions.rs +++ b/crates/proxy/src/server/chat_completions.rs @@ -234,8 +234,8 @@ pub(crate) async fn chat_completions( } } } - BackendClient::Anthropic(_) => openai_error_response( - "Anthropic passthrough backend does not support /v1/chat/completions", + BackendClient::Anthropic(_) | BackendClient::Bedrock(_) => openai_error_response( + "This backend does not support /v1/chat/completions. Use /v1/messages instead.", "invalid_request_error", StatusCode::BAD_REQUEST, ), @@ -273,9 +273,9 @@ async fn chat_completions_stream( | BackendClient::Vertex(c) | BackendClient::GeminiOpenAI(c) | BackendClient::OpenAIResponses(c) => c.clone(), - BackendClient::Anthropic(_) => { + BackendClient::Anthropic(_) | BackendClient::Bedrock(_) => { return openai_error_response( - "Anthropic passthrough backend does not support /v1/chat/completions streaming", + "This backend does not support /v1/chat/completions. Use /v1/messages instead.", "invalid_request_error", StatusCode::BAD_REQUEST, ); diff --git a/crates/proxy/src/server/mod.rs b/crates/proxy/src/server/mod.rs index a176122..b6494cc 100644 --- a/crates/proxy/src/server/mod.rs +++ b/crates/proxy/src/server/mod.rs @@ -1,3 +1,5 @@ +/// AWS Bedrock passthrough handler (SigV4 signing + event stream decoding). +mod bedrock_passthrough; /// OpenAI Chat Completions input handler (POST /v1/chat/completions). mod chat_completions; /// Auth validation, request ID injection, size limits, concurrency limits, header logging. diff --git a/crates/proxy/src/server/routes.rs b/crates/proxy/src/server/routes.rs index 2d48ffe..fc9809d 100644 --- a/crates/proxy/src/server/routes.rs +++ b/crates/proxy/src/server/routes.rs @@ -131,7 +131,7 @@ pub fn app_multi_with_shared(config: MultiConfig, shared: Option) - // Build per-backend sub-routers. Keep a map of AppState so the default // backend can reuse the same state (same semaphore, same reqwest client). - let mut backend_states: HashMap = HashMap::new(); + let mut backend_states: HashMap = HashMap::new(); for (name, bc) in &config.backends { let metrics = Metrics::new(); backend_metrics.insert(name.clone(), metrics.clone()); @@ -146,9 +146,13 @@ pub fn app_multi_with_shared(config: MultiConfig, shared: Option) - omit_stream_options: bc.omit_stream_options, }; - let is_anthropic = bc.kind == BackendKind::Anthropic; - let sub = backend_router(state.clone(), is_anthropic); - backend_states.insert(name.clone(), (state, is_anthropic)); + let mode = match bc.kind { + BackendKind::Anthropic => HandlerMode::Anthropic, + BackendKind::Bedrock => HandlerMode::Bedrock, + _ => HandlerMode::Translate, + }; + let sub = backend_router(state.clone(), mode); + backend_states.insert(name.clone(), (state, mode)); // Nest under /{name}/ router = router.nest(&format!("/{name}"), sub); @@ -156,8 +160,8 @@ pub fn app_multi_with_shared(config: MultiConfig, shared: Option) - // Default backend: also serve at un-prefixed /v1/messages for backward compat. // Reuses the same AppState (shared semaphore, connection pool) as the named route. - if let Some((default_state, is_anthropic)) = backend_states.get(&config.default_backend) { - let default_sub = backend_router(default_state.clone(), *is_anthropic); + if let Some((default_state, mode)) = backend_states.get(&config.default_backend) { + let default_sub = backend_router(default_state.clone(), *mode); router = router.merge(default_sub); } @@ -217,15 +221,30 @@ async fn fallback_not_found() -> Response { (StatusCode::NOT_FOUND, Json(err)).into_response() } +/// Which handler mode a backend uses. +#[derive(Debug, Clone, Copy)] +enum HandlerMode { + /// Anthropic passthrough (no translation, forwards raw bytes). + Anthropic, + /// Bedrock (SigV4 signing, event stream decoding, Anthropic format). + Bedrock, + /// Translation (Anthropic -> OpenAI -> backend -> OpenAI -> Anthropic). + Translate, +} + /// Build the sub-router for a single backend. -/// If `is_anthropic` is true, uses the passthrough handler instead of the translation handler. -fn backend_router(state: AppState, is_anthropic: bool) -> Router { - let api_routes = if is_anthropic { - Router::new() +fn backend_router(state: AppState, mode: HandlerMode) -> Router { + let api_routes = match mode { + HandlerMode::Anthropic => Router::new() .route("/v1/messages", post(anthropic_passthrough)) - .route("/v1/models", get(models)) - } else { - Router::new() + .route("/v1/models", get(models)), + HandlerMode::Bedrock => Router::new() + .route( + "/v1/messages", + post(super::bedrock_passthrough::bedrock_passthrough), + ) + .route("/v1/models", get(models)), + HandlerMode::Translate => Router::new() .route("/v1/messages", post(messages)) .route( "/v1/chat/completions", @@ -234,7 +253,7 @@ fn backend_router(state: AppState, is_anthropic: bool) -> Router { .route("/v1/models", get(models)) .route("/v1/messages/count_tokens", post(count_tokens)) .route("/v1/messages/batches", post(batches)) - .route("/v1/embeddings", post(embeddings)) + .route("/v1/embeddings", post(embeddings)), }; api_routes @@ -563,12 +582,12 @@ async fn messages( } } } - BackendClient::Anthropic(_) => { - // Anthropic passthrough is handled by a separate handler that works with raw bytes. + BackendClient::Anthropic(_) | BackendClient::Bedrock(_) => { + // These backends are handled by separate handlers (passthrough / Bedrock). // If we reach here, something is misconfigured. let err = mapping::errors_map::create_anthropic_error( anthropic::ErrorType::ApiError, - "Anthropic passthrough does not use the translation handler".to_string(), + "This backend does not use the translation handler".to_string(), None, ); (StatusCode::INTERNAL_SERVER_ERROR, Json(err)).into_response() diff --git a/crates/proxy/src/server/streaming.rs b/crates/proxy/src/server/streaming.rs index afef5a2..9d9bb5c 100644 --- a/crates/proxy/src/server/streaming.rs +++ b/crates/proxy/src/server/streaming.rs @@ -320,11 +320,11 @@ pub(crate) async fn messages_stream( } }); } - BackendClient::Anthropic(_) => { + BackendClient::Anthropic(_) | BackendClient::Bedrock(_) => { drop(rl_tx); let _ = tx .send(Ok(Event::default().data( - r#"{"error":"anthropic passthrough does not use this handler"}"#, + r#"{"error":"this backend does not use the translation streaming handler"}"#, ))) .await; } diff --git a/crates/proxy/tests/chat_completions.rs b/crates/proxy/tests/chat_completions.rs new file mode 100644 index 0000000..76235ec --- /dev/null +++ b/crates/proxy/tests/chat_completions.rs @@ -0,0 +1,230 @@ +// Integration tests for POST /v1/chat/completions (OpenAI-format input). + +use anyllm_proxy::config::{self, BackendAuth, BackendKind, Config, ModelMapping, OpenAIApiFormat}; +use anyllm_proxy::server::routes; +use axum::{routing::post, Router}; +use reqwest::Client; +use serde_json::json; +use tokio::net::TcpListener; + +fn openai_config_with_base(base_url: &str) -> Config { + Config { + backend: BackendKind::OpenAI, + openai_api_key: "test-key".to_string(), + openai_base_url: base_url.to_string(), + listen_port: 0, + model_mapping: ModelMapping { + big_model: "gpt-4o".into(), + small_model: "gpt-4o-mini".into(), + }, + tls: config::TlsConfig::default(), + backend_auth: BackendAuth::BearerToken("test-key".into()), + log_bodies: false, + openai_api_format: OpenAIApiFormat::Chat, + } +} + +/// Mock backend that returns a fixed OpenAI Chat Completions response. +async fn spawn_mock_chat_backend() -> String { + let app = Router::new().route( + "/v1/chat/completions", + post(|| async { + axum::Json(json!({ + "id": "chatcmpl-mock123", + "object": "chat.completion", + "created": 1700000000, + "model": "gpt-4o", + "choices": [{ + "index": 0, + "message": { + "role": "assistant", + "content": "Hello from mock!" + }, + "finish_reason": "stop" + }], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15 + } + })) + }), + ); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}") +} + +async fn spawn_proxy(config: Config) -> String { + std::env::set_var("PROXY_OPEN_RELAY", "true"); + let app = routes::app(config); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}") +} + +#[tokio::test] +async fn chat_completions_non_streaming() { + let mock = spawn_mock_chat_backend().await; + let proxy = spawn_proxy(openai_config_with_base(&mock)).await; + + let client = Client::new(); + let resp = client + .post(format!("{proxy}/v1/chat/completions")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .json(&json!({ + "model": "claude-sonnet-4-20250514", + "messages": [{"role": "user", "content": "Hello"}], + "max_tokens": 100 + })) + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 200); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["object"], "chat.completion"); + assert!(body["id"].as_str().unwrap().starts_with("chatcmpl-")); + assert_eq!(body["choices"][0]["message"]["role"], "assistant"); + assert!(body["choices"][0]["message"]["content"].as_str().is_some()); + assert_eq!(body["choices"][0]["finish_reason"], "stop"); + assert!(body["usage"]["prompt_tokens"].as_u64().is_some()); +} + +#[tokio::test] +async fn chat_completions_missing_max_tokens_returns_400() { + let mock = spawn_mock_chat_backend().await; + let proxy = spawn_proxy(openai_config_with_base(&mock)).await; + + let client = Client::new(); + let resp = client + .post(format!("{proxy}/v1/chat/completions")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .json(&json!({ + "model": "claude-sonnet-4-20250514", + "messages": [{"role": "user", "content": "Hello"}] + })) + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 400); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["error"]["type"], "invalid_request_error"); +} + +#[tokio::test] +async fn chat_completions_empty_messages_returns_400() { + let mock = spawn_mock_chat_backend().await; + let proxy = spawn_proxy(openai_config_with_base(&mock)).await; + + let client = Client::new(); + let resp = client + .post(format!("{proxy}/v1/chat/completions")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .json(&json!({ + "model": "claude-sonnet-4-20250514", + "messages": [], + "max_tokens": 100 + })) + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 400); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["error"]["type"], "invalid_request_error"); +} + +#[tokio::test] +async fn chat_completions_degradation_header_on_lossy_fields() { + let mock = spawn_mock_chat_backend().await; + let proxy = spawn_proxy(openai_config_with_base(&mock)).await; + + let client = Client::new(); + let resp = client + .post(format!("{proxy}/v1/chat/completions")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .json(&json!({ + "model": "claude-sonnet-4-20250514", + "messages": [{"role": "user", "content": "Hi"}], + "max_tokens": 100, + "presence_penalty": 0.5, + "frequency_penalty": 0.3 + })) + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 200); + let degradation = resp + .headers() + .get("x-anyllm-degradation") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + assert!( + degradation.contains("presence_penalty"), + "expected presence_penalty in degradation header, got: {degradation}" + ); + assert!( + degradation.contains("frequency_penalty"), + "expected frequency_penalty in degradation header, got: {degradation}" + ); +} + +#[tokio::test] +async fn chat_completions_with_system_message() { + let mock = spawn_mock_chat_backend().await; + let proxy = spawn_proxy(openai_config_with_base(&mock)).await; + + let client = Client::new(); + let resp = client + .post(format!("{proxy}/v1/chat/completions")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .json(&json!({ + "model": "claude-sonnet-4-20250514", + "messages": [ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": "Hello"} + ], + "max_tokens": 100 + })) + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 200); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["object"], "chat.completion"); +} + +#[tokio::test] +async fn chat_completions_returns_openai_error_format() { + let mock = spawn_mock_chat_backend().await; + let proxy = spawn_proxy(openai_config_with_base(&mock)).await; + + let client = Client::new(); + // Send completely invalid JSON + let resp = client + .post(format!("{proxy}/v1/chat/completions")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .body("not json") + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 400); + let body: serde_json::Value = resp.json().await.unwrap(); + // Should have OpenAI error format (error.type, error.message) + assert!(body["error"]["type"].is_string()); + assert!(body["error"]["message"].is_string()); +} diff --git a/crates/proxy/tests/embeddings.rs b/crates/proxy/tests/embeddings.rs new file mode 100644 index 0000000..7da08ac --- /dev/null +++ b/crates/proxy/tests/embeddings.rs @@ -0,0 +1,226 @@ +// Integration tests for POST /v1/embeddings passthrough and x-anyllm-degradation header. + +use anyllm_proxy::config::{self, BackendAuth, BackendKind, Config, ModelMapping, OpenAIApiFormat}; +use anyllm_proxy::server::routes; +use axum::{routing::post, Router}; +use reqwest::Client; +use tokio::net::TcpListener; + +fn openai_config_with_base(base_url: &str) -> Config { + Config { + backend: BackendKind::OpenAI, + openai_api_key: "test-key".to_string(), + openai_base_url: base_url.to_string(), + listen_port: 0, + model_mapping: ModelMapping { + big_model: "gpt-4o".into(), + small_model: "gpt-4o-mini".into(), + }, + tls: config::TlsConfig::default(), + backend_auth: BackendAuth::BearerToken("test-key".into()), + log_bodies: false, + openai_api_format: OpenAIApiFormat::Chat, + } +} + +fn anthropic_config() -> Config { + Config { + backend: BackendKind::Anthropic, + openai_api_key: String::new(), + openai_base_url: "https://api.anthropic.com".to_string(), + listen_port: 0, + model_mapping: ModelMapping { + big_model: "claude-opus-4-6".into(), + small_model: "claude-haiku-4-5".into(), + }, + tls: config::TlsConfig::default(), + backend_auth: BackendAuth::BearerToken("test-key".into()), + log_bodies: false, + openai_api_format: OpenAIApiFormat::Chat, + } +} + +/// Start a mock backend that accepts POST /v1/embeddings and returns a fixed response. +/// Returns the base URL of the mock server. +async fn spawn_mock_backend() -> String { + let app = Router::new().route( + "/v1/embeddings", + post(|| async { + axum::response::Response::builder() + .status(200) + .header("content-type", "application/json") + .body(axum::body::Body::from( + r#"{"object":"list","data":[{"object":"embedding","index":0,"embedding":[0.1,0.2,0.3]}],"model":"text-embedding-3-small","usage":{"prompt_tokens":5,"total_tokens":5}}"#, + )) + .unwrap() + }), + ); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}") +} + +async fn spawn_proxy_with_config(config: Config) -> String { + std::env::set_var("PROXY_OPEN_RELAY", "true"); + let app = routes::app(config); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}") +} + +#[tokio::test] +async fn embeddings_forwarded_to_backend() { + let mock_base = spawn_mock_backend().await; + let proxy_base = spawn_proxy_with_config(openai_config_with_base(&mock_base)).await; + + let client = Client::new(); + let resp = client + .post(format!("{proxy_base}/v1/embeddings")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .body(r#"{"model":"text-embedding-3-small","input":"hello world"}"#) + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 200); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["object"], "list"); + assert!(body["data"].is_array()); +} + +#[tokio::test] +async fn embeddings_response_content_type_forwarded() { + let mock_base = spawn_mock_backend().await; + let proxy_base = spawn_proxy_with_config(openai_config_with_base(&mock_base)).await; + + let client = Client::new(); + let resp = client + .post(format!("{proxy_base}/v1/embeddings")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .body(r#"{"model":"text-embedding-3-small","input":"hello"}"#) + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 200); + let ct = resp + .headers() + .get("content-type") + .unwrap() + .to_str() + .unwrap(); + assert!(ct.contains("application/json"), "got content-type: {ct}"); +} + +#[tokio::test] +async fn embeddings_requires_auth() { + let mock_base = spawn_mock_backend().await; + let proxy_base = spawn_proxy_with_config(openai_config_with_base(&mock_base)).await; + + // Temporarily unset open-relay mode so auth is enforced + std::env::remove_var("PROXY_OPEN_RELAY"); + let proxy_strict = { + let mut c = openai_config_with_base(&mock_base); + c.openai_api_key = "sk-real".into(); + spawn_proxy_with_config(c).await + }; + + let client = Client::new(); + let resp = client + .post(format!("{proxy_strict}/v1/embeddings")) + .header("content-type", "application/json") + .body(r#"{"model":"text-embedding-3-small","input":"hello"}"#) + .send() + .await + .unwrap(); + + assert_eq!(resp.status(), 401); + + // Restore for other tests + std::env::set_var("PROXY_OPEN_RELAY", "true"); + // Silence the unused-variable warning — proxy_base was used above + let _ = proxy_base; +} + +#[tokio::test] +async fn embeddings_not_routed_for_anthropic_backend() { + // Anthropic backend does not mount /v1/embeddings; route returns 404. + let proxy_base = spawn_proxy_with_config(anthropic_config()).await; + + let client = Client::new(); + let resp = client + .post(format!("{proxy_base}/v1/embeddings")) + .header("x-api-key", "test") + .header("content-type", "application/json") + .body(r#"{"model":"text-embedding-3-small","input":"hello"}"#) + .send() + .await + .unwrap(); + + // Route not registered for Anthropic backend; fallback returns 404. + assert_eq!(resp.status(), 404); +} + +#[tokio::test] +async fn degradation_header_present_when_top_k_set() { + // We can't make a full messages round-trip without a real backend, but we can + // verify the compute_request_warnings function directly via the translator crate. + // The proxy-level injection is covered by the inject_degradation_header unit test. + use anyllm_translate::anthropic; + let req = anthropic::MessageCreateRequest { + model: "claude-sonnet-4-6".to_string(), + max_tokens: 100, + messages: vec![anthropic::InputMessage { + role: anthropic::Role::User, + content: anthropic::Content::Text("hi".to_string()), + }], + system: None, + temperature: None, + top_p: None, + top_k: Some(40), + stop_sequences: None, + tools: None, + tool_choice: None, + metadata: None, + thinking: None, + stream: None, + extra: serde_json::Map::new(), + }; + + let warnings = anyllm_translate::compute_request_warnings(&req); + let header_val = warnings.as_header_value().expect("should have warnings"); + assert!(header_val.contains("top_k"), "got: {header_val}"); +} + +#[tokio::test] +async fn degradation_header_absent_when_no_lossy_features() { + use anyllm_translate::anthropic; + let req = anthropic::MessageCreateRequest { + model: "claude-sonnet-4-6".to_string(), + max_tokens: 100, + messages: vec![anthropic::InputMessage { + role: anthropic::Role::User, + content: anthropic::Content::Text("hi".to_string()), + }], + system: None, + temperature: None, + top_p: None, + top_k: None, + stop_sequences: None, + tools: None, + tool_choice: None, + metadata: None, + thinking: None, + stream: None, + extra: serde_json::Map::new(), + }; + + let warnings = anyllm_translate::compute_request_warnings(&req); + assert!(warnings.is_empty()); + assert!(warnings.as_header_value().is_none()); +} diff --git a/crates/proxy/tests/live_bedrock.rs b/crates/proxy/tests/live_bedrock.rs new file mode 100644 index 0000000..99430f9 --- /dev/null +++ b/crates/proxy/tests/live_bedrock.rs @@ -0,0 +1,76 @@ +// Live integration tests for the Bedrock backend. +// Requires AWS credentials in the environment. Run with: +// AWS_REGION=us-east-1 AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... \ +// cargo test --test live_bedrock -- --ignored --test-threads=1 + +use reqwest::Client; +use serde_json::json; + +/// Start the proxy with Bedrock backend and return the base URL. +fn start_proxy() -> String { + // These tests are run manually with real credentials. + // The proxy must be started externally, or we construct a test server here. + let port = std::env::var("TEST_PROXY_PORT").unwrap_or_else(|_| "3099".to_string()); + format!("http://127.0.0.1:{port}") +} + +#[tokio::test] +#[ignore] +async fn bedrock_non_streaming() { + let base = start_proxy(); + let client = Client::new(); + + let resp = client + .post(format!("{base}/v1/messages")) + .header("x-api-key", "test-key") + .header("content-type", "application/json") + .json(&json!({ + "model": "anthropic.claude-haiku-4-5-20251001-v1:0", + "max_tokens": 64, + "messages": [{"role": "user", "content": "Say hello in one word."}] + })) + .send() + .await + .expect("request failed"); + + let status = resp.status().as_u16(); + let body: serde_json::Value = resp.json().await.expect("invalid JSON response"); + + assert_eq!(status, 200, "unexpected status: {body}"); + assert_eq!(body["type"], "message"); + assert!(body["content"].as_array().map_or(false, |a| !a.is_empty())); +} + +#[tokio::test] +#[ignore] +async fn bedrock_streaming() { + let base = start_proxy(); + let client = Client::new(); + + let resp = client + .post(format!("{base}/v1/messages")) + .header("x-api-key", "test-key") + .header("content-type", "application/json") + .json(&json!({ + "model": "anthropic.claude-haiku-4-5-20251001-v1:0", + "max_tokens": 64, + "stream": true, + "messages": [{"role": "user", "content": "Say hello."}] + })) + .send() + .await + .expect("request failed"); + + let status = resp.status().as_u16(); + assert_eq!(status, 200, "expected 200 for streaming"); + + let body = resp.text().await.expect("failed to read body"); + assert!( + body.contains("message_start"), + "expected message_start event in stream" + ); + assert!( + body.contains("message_stop"), + "expected message_stop event in stream" + ); +} diff --git a/crates/translator/src/anthropic/messages.rs b/crates/translator/src/anthropic/messages.rs index db64362..591de2a 100644 --- a/crates/translator/src/anthropic/messages.rs +++ b/crates/translator/src/anthropic/messages.rs @@ -197,9 +197,17 @@ pub struct Tool { #[serde(tag = "type")] pub enum ToolChoice { #[serde(rename = "auto")] - Auto, + Auto { + /// Disable parallel tool use. Maps to OpenAI `parallel_tool_calls: false`. + #[serde(default, skip_serializing_if = "Option::is_none")] + disable_parallel_tool_use: Option, + }, #[serde(rename = "any")] - Any, + Any { + /// Disable parallel tool use. Maps to OpenAI `parallel_tool_calls: false`. + #[serde(default, skip_serializing_if = "Option::is_none")] + disable_parallel_tool_use: Option, + }, #[serde(rename = "none")] None, #[serde(rename = "tool")] @@ -365,7 +373,7 @@ mod tests { assert_eq!(tools[0].name, "get_weather"); assert!(tools[0].description.is_some()); match req.tool_choice.unwrap() { - ToolChoice::Auto => {} + ToolChoice::Auto { .. } => {} other => panic!("expected ToolChoice::Auto, got {:?}", other), } } diff --git a/crates/translator/src/mapping/message_map.rs b/crates/translator/src/mapping/message_map.rs index cfa44d0..0a86fb8 100644 --- a/crates/translator/src/mapping/message_map.rs +++ b/crates/translator/src/mapping/message_map.rs @@ -1,7 +1,7 @@ // Anthropic <-> OpenAI message mapping use crate::anthropic; -use crate::mapping::{streaming_map, tools_map, usage_map}; +use crate::mapping::{streaming_map, tools_map, usage_map, warnings::TranslationWarnings}; use crate::openai; use crate::util; @@ -23,6 +23,44 @@ pub fn extract_system_text(system: &anthropic::System) -> String { } } +/// Compute degradation warnings for an Anthropic request without performing translation. +/// +/// Returns a `TranslationWarnings` value listing every feature that will be silently +/// dropped or degraded when this request is translated to an OpenAI request. +/// The proxy injects these as an `x-anyllm-degradation` response header so clients +/// can detect silent drops without inspecting server logs. +pub fn compute_request_warnings(req: &anthropic::MessageCreateRequest) -> TranslationWarnings { + let mut w = TranslationWarnings::default(); + if req.top_k.is_some() { + w.add("top_k"); + } + if req.thinking.is_some() { + w.add("thinking_config"); + } + if let Some(ref seqs) = req.stop_sequences { + if seqs.len() > 4 { + w.add("stop_sequences_truncated"); + } + } + if let Some(anthropic::System::Blocks(blocks)) = &req.system { + if blocks.iter().any(|b| b.cache_control.is_some()) { + w.add("cache_control"); + } + } + let has_document = req.messages.iter().any(|msg| { + match &msg.content { + anthropic::Content::Blocks(blocks) => blocks.iter().any(|b| { + matches!(b, anthropic::ContentBlock::Document { .. }) + }), + _ => false, + } + }); + if has_document { + w.add("document_blocks"); + } + w +} + /// Convert an Anthropic MessageCreateRequest to an OpenAI ChatCompletionRequest. /// /// Anthropic: @@ -64,6 +102,14 @@ pub fn anthropic_to_openai_request( .as_ref() .map(tools_map::anthropic_tool_choice_to_openai); + // Map disable_parallel_tool_use to OpenAI parallel_tool_calls. + // Compat spec: "Fully supported". See: https://docs.anthropic.com/en/api/openai-sdk#tools--functions-fields + let parallel_tool_calls = match req.tool_choice.as_ref() { + Some(anthropic::ToolChoice::Auto { disable_parallel_tool_use: Some(true) }) + | Some(anthropic::ToolChoice::Any { disable_parallel_tool_use: Some(true) }) => Some(false), + _ => None, + }; + // Map metadata.user_id to OpenAI user field. // Compat spec: user is "Ignored", but we forward it for traceability. // See: https://docs.anthropic.com/en/api/openai-sdk#simple-fields @@ -127,7 +173,7 @@ pub fn anthropic_to_openai_request( frequency_penalty: None, response_format: None, user, - parallel_tool_calls: None, + parallel_tool_calls, extra: req.extra.clone(), }; @@ -678,7 +724,7 @@ mod tests { #[test] fn tool_choice_auto() { let mut req = basic_request(); - req.tool_choice = Some(anthropic::ToolChoice::Auto); + req.tool_choice = Some(anthropic::ToolChoice::Auto { disable_parallel_tool_use: None }); let oai = anthropic_to_openai_request(&req); assert!(matches!( oai.tool_choice, @@ -689,7 +735,7 @@ mod tests { #[test] fn tool_choice_any_becomes_required() { let mut req = basic_request(); - req.tool_choice = Some(anthropic::ToolChoice::Any); + req.tool_choice = Some(anthropic::ToolChoice::Any { disable_parallel_tool_use: None }); let oai = anthropic_to_openai_request(&req); assert!(matches!( oai.tool_choice, @@ -724,6 +770,33 @@ mod tests { } } + #[test] + fn disable_parallel_tool_use_sets_parallel_tool_calls_false() { + let mut req = basic_request(); + req.tool_choice = Some(anthropic::ToolChoice::Auto { + disable_parallel_tool_use: Some(true), + }); + let oai = anthropic_to_openai_request(&req); + assert_eq!(oai.parallel_tool_calls, Some(false)); + } + + #[test] + fn disable_parallel_tool_use_false_leaves_parallel_tool_calls_none() { + let mut req = basic_request(); + req.tool_choice = Some(anthropic::ToolChoice::Auto { + disable_parallel_tool_use: Some(false), + }); + let oai = anthropic_to_openai_request(&req); + assert!(oai.parallel_tool_calls.is_none()); + } + + #[test] + fn no_tool_choice_leaves_parallel_tool_calls_none() { + let req = basic_request(); + let oai = anthropic_to_openai_request(&req); + assert!(oai.parallel_tool_calls.is_none()); + } + #[test] fn stop_sequences_capped_at_four() { let mut req = basic_request(); @@ -2110,4 +2183,100 @@ mod tests { let oai = anthropic_to_openai_request(&req); assert_eq!(oai.temperature, Some(0.7)); } + + // --- compute_request_warnings --- + + #[test] + fn warnings_empty_for_plain_request() { + let req = basic_request(); + let w = compute_request_warnings(&req); + assert!(w.is_empty()); + assert!(w.as_header_value().is_none()); + } + + #[test] + fn warnings_top_k() { + let mut req = basic_request(); + req.top_k = Some(40); + let w = compute_request_warnings(&req); + assert_eq!(w.as_header_value().unwrap(), "top_k"); + } + + #[test] + fn warnings_thinking_config() { + let mut req = basic_request(); + req.thinking = Some(anthropic::ThinkingConfig::Enabled { budget_tokens: 5000 }); + let w = compute_request_warnings(&req); + assert_eq!(w.as_header_value().unwrap(), "thinking_config"); + } + + #[test] + fn warnings_stop_sequences_truncated_at_5() { + let mut req = basic_request(); + req.stop_sequences = Some(vec![ + "a".to_string(), + "b".to_string(), + "c".to_string(), + "d".to_string(), + "e".to_string(), + ]); + let w = compute_request_warnings(&req); + assert_eq!(w.as_header_value().unwrap(), "stop_sequences_truncated"); + } + + #[test] + fn warnings_stop_sequences_4_is_fine() { + let mut req = basic_request(); + req.stop_sequences = Some(vec![ + "a".to_string(), + "b".to_string(), + "c".to_string(), + "d".to_string(), + ]); + let w = compute_request_warnings(&req); + assert!(w.is_empty()); + } + + #[test] + fn warnings_cache_control_on_system() { + let mut req = basic_request(); + req.system = Some(anthropic::System::Blocks(vec![anthropic::SystemBlock { + block_type: "text".to_string(), + text: "You are helpful.".to_string(), + cache_control: Some(anthropic::CacheControl { + cache_type: "ephemeral".to_string(), + }), + }])); + let w = compute_request_warnings(&req); + assert_eq!(w.as_header_value().unwrap(), "cache_control"); + } + + #[test] + fn warnings_document_blocks() { + let mut req = basic_request(); + req.messages = vec![anthropic::InputMessage { + role: anthropic::Role::User, + content: anthropic::Content::Blocks(vec![anthropic::ContentBlock::Document { + source: anthropic::DocumentSource { + source_type: "base64".to_string(), + media_type: "application/pdf".to_string(), + data: "dGVzdA==".to_string(), + }, + title: None, + }]), + }]; + let w = compute_request_warnings(&req); + assert_eq!(w.as_header_value().unwrap(), "document_blocks"); + } + + #[test] + fn warnings_multiple_combined() { + let mut req = basic_request(); + req.top_k = Some(10); + req.thinking = Some(anthropic::ThinkingConfig::Enabled { budget_tokens: 1000 }); + let w = compute_request_warnings(&req); + let val = w.as_header_value().unwrap(); + assert!(val.contains("top_k"), "missing top_k in: {val}"); + assert!(val.contains("thinking_config"), "missing thinking_config in: {val}"); + } } diff --git a/crates/translator/src/mapping/responses_message_map.rs b/crates/translator/src/mapping/responses_message_map.rs index 2401f36..2303236 100644 --- a/crates/translator/src/mapping/responses_message_map.rs +++ b/crates/translator/src/mapping/responses_message_map.rs @@ -53,8 +53,8 @@ pub fn anthropic_to_responses_request(req: &anthropic::MessageCreateRequest) -> } if let Some(ref tc) = req.tool_choice { let mapped = match tc { - anthropic::ToolChoice::Auto => json!("auto"), - anthropic::ToolChoice::Any => json!("required"), + anthropic::ToolChoice::Auto { .. } => json!("auto"), + anthropic::ToolChoice::Any { .. } => json!("required"), anthropic::ToolChoice::None => json!("none"), anthropic::ToolChoice::Tool { name } => json!({ "type": "function", diff --git a/crates/translator/src/mapping/streaming_map.rs b/crates/translator/src/mapping/streaming_map.rs index 26bc08b..dfdd021 100644 --- a/crates/translator/src/mapping/streaming_map.rs +++ b/crates/translator/src/mapping/streaming_map.rs @@ -80,6 +80,8 @@ impl StreamingTranslator { if let Some(ref usage) = chunk.usage { self.usage.input_tokens = usage.prompt_tokens; self.usage.output_tokens = usage.completion_tokens; + self.usage.cache_read_input_tokens = + crate::mapping::usage_map::extract_cached_tokens(usage.prompt_tokens_details.as_ref()); } for choice in &chunk.choices { @@ -1199,4 +1201,37 @@ mod tests { matches!(&events[0], anthropic::StreamEvent::ContentBlockStop { index } if *index == 0) ); } + + #[test] + fn usage_chunk_with_cached_tokens_maps_cache_read() { + let mut translator = StreamingTranslator::new("gpt-4o".into()); + let chunk = ChatCompletionChunk { + id: "c1".into(), + object: "chat.completion.chunk".into(), + model: "gpt-4o".into(), + choices: vec![], + usage: Some(crate::openai::ChatUsage { + prompt_tokens: 100, + completion_tokens: 50, + total_tokens: 150, + completion_tokens_details: None, + prompt_tokens_details: Some(serde_json::json!({"cached_tokens": 42})), + }), + created: None, + system_fingerprint: None, + }; + translator.process_chunk(&chunk); + let usage = translator.usage().expect("usage should be present"); + assert_eq!(usage.input_tokens, 100); + assert_eq!(usage.output_tokens, 50); + assert_eq!(usage.cache_read_input_tokens, Some(42)); + } + + #[test] + fn usage_chunk_without_cached_tokens_leaves_cache_read_none() { + let mut translator = StreamingTranslator::new("gpt-4o".into()); + translator.process_chunk(&usage_chunk("c1", "gpt-4o", 10, 5)); + let usage = translator.usage().expect("usage should be present"); + assert!(usage.cache_read_input_tokens.is_none()); + } } diff --git a/crates/translator/src/mapping/tools_map.rs b/crates/translator/src/mapping/tools_map.rs index e07ef56..c85b5c6 100644 --- a/crates/translator/src/mapping/tools_map.rs +++ b/crates/translator/src/mapping/tools_map.rs @@ -51,10 +51,10 @@ pub fn openai_tools_to_anthropic(tools: &[openai::ChatTool]) -> Vec pub fn anthropic_tool_choice_to_openai(tc: &anthropic::ToolChoice) -> openai::ChatToolChoice { match tc { - anthropic::ToolChoice::Auto => openai::ChatToolChoice::Simple("auto".to_string()), + anthropic::ToolChoice::Auto { .. } => openai::ChatToolChoice::Simple("auto".to_string()), // Any = "model must call at least one tool". OpenAI's "required" // is the closest: it forces a tool call when tools are defined. - anthropic::ToolChoice::Any => openai::ChatToolChoice::Simple("required".to_string()), + anthropic::ToolChoice::Any { .. } => openai::ChatToolChoice::Simple("required".to_string()), anthropic::ToolChoice::None => openai::ChatToolChoice::Simple("none".to_string()), anthropic::ToolChoice::Tool { name } => { openai::ChatToolChoice::Named(openai::chat_completions::NamedToolChoice { @@ -73,10 +73,10 @@ pub fn openai_tool_choice_to_anthropic(tc: &openai::ChatToolChoice) -> anthropic match tc { openai::ChatToolChoice::Simple(s) => match s.as_str() { "none" => anthropic::ToolChoice::None, - "required" => anthropic::ToolChoice::Any, + "required" => anthropic::ToolChoice::Any { disable_parallel_tool_use: None }, // Default unknown values to Auto for forward compatibility; // rejecting would break when OpenAI adds new tool_choice variants. - _ => anthropic::ToolChoice::Auto, + _ => anthropic::ToolChoice::Auto { disable_parallel_tool_use: None }, }, openai::ChatToolChoice::Named(named) => anthropic::ToolChoice::Tool { name: named.function.name.clone(), @@ -218,20 +218,20 @@ mod tests { #[test] fn tool_choice_auto() { - let openai = anthropic_tool_choice_to_openai(&anthropic::ToolChoice::Auto); + let openai = anthropic_tool_choice_to_openai(&anthropic::ToolChoice::Auto { disable_parallel_tool_use: None }); assert!(matches!(openai, openai::ChatToolChoice::Simple(ref s) if s == "auto")); let back = openai_tool_choice_to_anthropic(&openai); - assert!(matches!(back, anthropic::ToolChoice::Auto)); + assert!(matches!(back, anthropic::ToolChoice::Auto { .. })); } #[test] fn tool_choice_any_to_required() { - let openai = anthropic_tool_choice_to_openai(&anthropic::ToolChoice::Any); + let openai = anthropic_tool_choice_to_openai(&anthropic::ToolChoice::Any { disable_parallel_tool_use: None }); assert!(matches!(openai, openai::ChatToolChoice::Simple(ref s) if s == "required")); let back = openai_tool_choice_to_anthropic(&openai); - assert!(matches!(back, anthropic::ToolChoice::Any)); + assert!(matches!(back, anthropic::ToolChoice::Any { .. })); } #[test] @@ -270,10 +270,28 @@ mod tests { let tc = openai::ChatToolChoice::Simple("something_else".into()); assert!(matches!( openai_tool_choice_to_anthropic(&tc), - anthropic::ToolChoice::Auto + anthropic::ToolChoice::Auto { .. } )); } + #[test] + fn disable_parallel_tool_use_roundtrips_via_serde() { + // Ensure the field survives JSON deserialization + let json = serde_json::json!({"type": "auto", "disable_parallel_tool_use": true}); + let tc: anthropic::ToolChoice = serde_json::from_value(json).unwrap(); + match tc { + anthropic::ToolChoice::Auto { disable_parallel_tool_use: Some(true) } => {} + other => panic!("expected Auto with disable_parallel_tool_use=true, got {:?}", other), + } + } + + #[test] + fn auto_without_disable_parallel_omits_field_in_json() { + let tc = anthropic::ToolChoice::Auto { disable_parallel_tool_use: None }; + let json = serde_json::to_value(&tc).unwrap(); + assert_eq!(json, serde_json::json!({"type": "auto"})); + } + // --- Claude Code tool schema round-trips --- #[test] diff --git a/crates/translator/src/mapping/warnings.rs b/crates/translator/src/mapping/warnings.rs new file mode 100644 index 0000000..064f973 --- /dev/null +++ b/crates/translator/src/mapping/warnings.rs @@ -0,0 +1,60 @@ +/// Collects feature degradation notices produced during request translation. +/// +/// Returned to the proxy layer so it can inject an `x-anyllm-degradation` response +/// header for clients to inspect. This makes silent drops visible without changing +/// the Anthropic-compatible response body. +#[derive(Default, Debug)] +pub struct TranslationWarnings { + items: Vec<&'static str>, +} + +impl TranslationWarnings { + pub fn add(&mut self, feature: &'static str) { + self.items.push(feature); + } + + pub fn is_empty(&self) -> bool { + self.items.is_empty() + } + + /// Returns a comma-separated string suitable for an HTTP header value, + /// or `None` if no features were dropped. + pub fn as_header_value(&self) -> Option { + if self.items.is_empty() { + return None; + } + Some(self.items.join(", ")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_returns_none() { + let w = TranslationWarnings::default(); + assert!(w.is_empty()); + assert!(w.as_header_value().is_none()); + } + + #[test] + fn single_item() { + let mut w = TranslationWarnings::default(); + w.add("top_k"); + assert!(!w.is_empty()); + assert_eq!(w.as_header_value().unwrap(), "top_k"); + } + + #[test] + fn multiple_items_comma_separated() { + let mut w = TranslationWarnings::default(); + w.add("top_k"); + w.add("cache_control"); + w.add("document_blocks"); + assert_eq!( + w.as_header_value().unwrap(), + "top_k, cache_control, document_blocks" + ); + } +} diff --git a/docs/COMPARISON_LITELLM.md b/docs/COMPARISON_LITELLM.md new file mode 100644 index 0000000..92ae84c --- /dev/null +++ b/docs/COMPARISON_LITELLM.md @@ -0,0 +1,147 @@ +# anyllm-proxy vs LiteLLM: Feature Comparison + +anyllm-proxy is a specialized **protocol translator** (Anthropic API in, OpenAI-compatible backend out). +LiteLLM is a broad **AI gateway** focused on enterprise governance, cost control, and routing across 100+ providers. +These are different categories; not every gap is worth closing. + +## Summary Table + +| Feature Area | anyllm-proxy | LiteLLM | Gap | +|---|---|---|---| +| Protocol translation (Anthropic↔OpenAI) | Full | Partial | **Advantage** | +| Translation degradation warnings (`x-anyllm-degradation`) | Yes | No | **Advantage** | +| Local LLM compatibility (system role, synthetic IDs) | Yes | No | **Advantage** | +| mTLS backend support (PKCS#12) | Yes | No | **Advantage** | +| Single static binary, no runtime deps | Yes | No | **Advantage** | +| Provider backends | 7 (OpenAI, Vertex, Gemini, Azure, Bedrock, Anthropic, Responses) | 100+ | Moderate gap | +| `POST /v1/chat/completions` input | Yes (full, streaming + non-streaming) | Yes | **Parity** | +| Virtual key management | Yes (SQLite-backed, immediate revocation) | Yes | **Parity** | +| Per-key rate limiting (RPM/TPM) | Yes (in-memory sliding window) | Yes | **Parity** | +| OpenTelemetry export | Yes (feature-gated, OTLP/HTTP) | 20+ integrations | Moderate gap | +| Cost tracking / budget enforcement | No | Yes | Major gap | +| Response caching | No | Yes | Major gap | +| Batch processing | Stub (400) | Yes | Major gap | +| Load balancing / fallback chains | Basic | Full | Moderate gap | +| Dynamic model management | Partial | Full | Moderate gap | +| RBAC / OIDC auth | No | Yes | Moderate gap | +| Audio, image, reranking endpoints | No | Yes | Low (out of scope) | + +--- + +## Detailed Gaps + +### 1. Provider/Backend Coverage + +**anyllm-proxy:** OpenAI (Chat Completions), OpenAI (Responses API), Vertex AI, Gemini, Azure OpenAI, AWS Bedrock, Anthropic passthrough. + +**LiteLLM:** 100+ providers. + +Providers that already work today via `OPENAI_BASE_URL` override (OpenAI-compatible): +- Groq, Together AI, Fireworks, Perplexity, Mistral, HuggingFace TGI, Ollama, vLLM + +### 2. API Endpoints + +anyllm-proxy accepts both **Anthropic-format** requests (`POST /v1/messages`) and **OpenAI-format** requests (`POST /v1/chat/completions`). The OpenAI endpoint translates internally through the Anthropic pipeline and returns OpenAI-format responses (streaming and non-streaming). + +Missing endpoints: +- `POST /v1/completions` — Legacy text completions +- `POST /v1/images/generations` — DALL-E, Imagen, etc. +- `POST /v1/audio/transcriptions` — Whisper / speech-to-text +- `POST /v1/audio/speech` — TTS +- `POST /v1/rerank` — Reranking (Cohere, etc.) +- `POST /v1/messages/batches` — Currently stubbed; returns 400 + +### 3. Authentication & Authorization + +anyllm-proxy supports both static keys (`PROXY_API_KEYS` env var) and dynamic virtual keys managed via the admin API (`POST /admin/api/keys`). Virtual keys are stored in SQLite, cached in memory, and revocation takes effect immediately without restart. Per-key RPM rate limiting is enforced in the auth middleware. + +LiteLLM provides: +- Virtual key issuance via API (`POST /key/generate`) with per-key metadata, expiry, and spend limits +- Key revocation without restart +- RBAC roles (admin, developer, read-only) +- OIDC/JWT validation +- IP allowlisting + +### 4. Load Balancing & Routing + +anyllm-proxy supports multiple named backends via `PROXY_CONFIG` YAML, each served under its own path prefix (e.g., `/openai/v1/messages`, `/vertex/v1/messages`). There is no load balancing between multiple instances of the same backend. + +LiteLLM supports: +- Random shuffle, least-busy, latency-based, cost-based, and weighted routing +- Cross-provider fallback chains (retry on a different provider on failure) +- Redis-backed distributed state for multi-instance deployments + +### 5. Caching + +anyllm-proxy has no caching. Every request hits the backend. + +LiteLLM supports in-memory, Redis, semantic (Qdrant/Redis), S3, and GCS caches with per-request TTL control. + +### 6. Rate Limiting + +anyllm-proxy enforces a global concurrency limit (default 100 concurrent requests) and per-key RPM limits via virtual keys (in-memory sliding window, no external dependencies). Upstream 429s are passed through. TPM tracking is recorded per-key for reporting. + +LiteLLM enforces RPM and TPM limits per key, user, and team, with Redis-backed distributed tracking. + +### 7. Cost Tracking & Budget Management + +anyllm-proxy has no cost tracking. No pricing database, no per-request USD calculation, no spend aggregation. + +LiteLLM computes per-request USD cost from a built-in model pricing database and aggregates spend per key, user, and team with configurable hard caps. + +### 8. Observability & Logging + +anyllm-proxy provides: +- SQLite request log (7-day retention, latency percentiles via admin API) +- Request count metrics (`GET /metrics`) +- `x-anyllm-degradation` response header for lossy translation warnings +- `tracing` crate output (stdout, `RUST_LOG`) +- Optional OpenTelemetry OTLP export (`--features otel`): spans exported to any OTEL-compatible collector (Datadog, Honeycomb, Jaeger, Tempo, etc.) via `OTEL_EXPORTER_OTLP_ENDPOINT` + +LiteLLM integrates with 20+ external observability platforms: Langfuse, Langsmith, OpenTelemetry (Honeycomb, Traceloop, OTEL collectors), Datadog, Sentry, Arize, and others. It also supports structured log export to DynamoDB, S3, GCS, and SQS. + +Not present in LiteLLM: `x-anyllm-degradation` per-request degradation signaling. + +### 9. Model Management + +anyllm-proxy maps Haiku requests to `small_model` and Opus/Sonnet to `big_model`. Overrides persist to SQLite via the admin API. + +The `/v1/models` response is a hardcoded list of 13 Claude model IDs with no context window or pricing metadata. + +LiteLLM supports dynamic model addition and removal via API without restart, per-model pricing metadata, and enriched `/models` responses with token limits. + +### 10. Batch Processing + +`POST /v1/messages/batches` is stubbed and returns 400. LiteLLM supports batch processing across multiple providers. + +### 11. Database & Persistence + +anyllm-proxy uses SQLite for admin config overrides and request logs. There is no schema for virtual keys, users, teams, or spend records. + +LiteLLM uses a full relational database (configurable backend) for key/user/team/spend storage. + +--- + +## Completed Items (this release) + +1. **`POST /v1/chat/completions`** -- Accept OpenAI-format input (streaming + non-streaming) +2. **AWS Bedrock backend** -- SigV4 auth, InvokeModel + InvokeModelWithResponseStream +3. **Azure OpenAI backend** -- Deployment-scoped URLs, `api-key` header auth +4. **Virtual key management** -- SQLite-backed CRUD, DashMap cache, immediate revocation +5. **Per-key RPM rate limiting** -- Sliding window enforcement, 429 + retry-after +6. **OpenTelemetry export** -- Feature-gated OTLP/HTTP, spans to any collector +7. **Rust client SDK v0.2.0** -- ClientBuilder, ToolBuilder, typed streaming + +## Remaining Priority Order + +**Tier 1 — Significant, broader scope:** +1. Response caching (in-memory + Redis) -- Reduces upstream cost and latency for repeated prompts +2. Cross-provider fallback chains -- Retry on alternate backend on failure +3. Real batch processing -- Async job queue behind `/v1/messages/batches` +4. Cost tracking -- Model pricing DB + per-request USD calculation + spend aggregation + +**Tier 2 -- Enterprise/niche:** +5. RBAC / OIDC authentication +6. Audio, image generation, reranking endpoints +7. Semantic caching +8. Budget enforcement and spend alerts diff --git a/docs/ENV.md b/docs/ENV.md index c5ee94a..511b14a 100644 --- a/docs/ENV.md +++ b/docs/ENV.md @@ -45,6 +45,35 @@ These are the variables most users need. | `RUST_LOG` | `info` | Tracing filter. Examples: `debug`, `anyllm_proxy=trace`. | | `DISABLE_ADMIN` | (unset) | Set to `1`, `true`, or `yes` to force-disable the admin web interface even when `--webui` is passed. Useful in automated/container environments. | +## AWS Bedrock + +Set `BACKEND=bedrock` to route through AWS Bedrock. The proxy sends Anthropic Messages API format directly to Bedrock (no OpenAI translation). Requests are signed with AWS SigV4. + +| Variable | Default | Description | +|----------|---------|-------------| +| `AWS_REGION` | (required) | AWS region, e.g. `us-east-1`. | +| `AWS_ACCESS_KEY_ID` | (required) | AWS access key ID for SigV4 signing. | +| `AWS_SECRET_ACCESS_KEY` | (required) | AWS secret access key for SigV4 signing. | +| `AWS_SESSION_TOKEN` | (optional) | Temporary session token for STS credentials. | +| `BIG_MODEL` | `anthropic.claude-sonnet-4-20250514-v1:0` | Bedrock model ID for sonnet/opus requests. | +| `SMALL_MODEL` | `anthropic.claude-haiku-4-5-20251001-v1:0` | Bedrock model ID for haiku requests. | + +### Example + +```bash +BACKEND=bedrock \ +AWS_REGION=us-east-1 \ +AWS_ACCESS_KEY_ID=AKIA... \ +AWS_SECRET_ACCESS_KEY=wJalr... \ +cargo run -p anyllm_proxy +``` + +### Streaming + +Bedrock streaming uses AWS Event Stream binary framing instead of SSE. The proxy decodes Event Stream frames and re-emits them as standard SSE events, so downstream clients see the same Anthropic SSE format as with other backends. + +--- + ## Azure OpenAI Set `BACKEND=azure` to route through Azure OpenAI Service. The request/response format is identical to standard OpenAI Chat Completions; only the URL scheme and auth header differ. @@ -148,3 +177,24 @@ ADMIN_DB_PATH=/var/lib/anyllm/admin.db \ anyllm_proxy --webui # Open: http://127.0.0.1:4000/admin/?token=my-secret-token ``` + +--- + +## OpenTelemetry (optional) + +Trace export is opt-in. Build with the `otel` cargo feature to enable it: + +```bash +cargo build -p anyllm_proxy --features otel +``` + +When the feature is enabled, the proxy initializes an OTLP span exporter that sends traces over HTTP/protobuf. The OTLP SDK reads configuration from standard environment variables; no proxy-specific config is needed. + +| Variable | Default | Description | +|----------|---------|-------------| +| `OTEL_EXPORTER_OTLP_ENDPOINT` | `http://localhost:4318` | OTLP collector endpoint (HTTP). | +| `OTEL_SERVICE_NAME` | `unknown_service` | Service name attached to all exported spans. Set this to `anyllm-proxy` or your deployment name. | +| `OTEL_TRACES_SAMPLER` | `parentbased_always_on` | Sampling strategy. Common values: `always_on`, `always_off`, `traceidratio` (pair with `OTEL_TRACES_SAMPLER_ARG`). | +| `OTEL_TRACES_SAMPLER_ARG` | (none) | Argument for the sampler, e.g. `0.1` for 10% sampling with `traceidratio`. | + +When built without the `otel` feature (the default), none of these variables have any effect and there is zero runtime overhead. diff --git a/specs/20260325-120000-litellm-gap-fill/tasks.md b/specs/20260325-120000-litellm-gap-fill/tasks.md index 22d5ba3..dbd4ea2 100644 --- a/specs/20260325-120000-litellm-gap-fill/tasks.md +++ b/specs/20260325-120000-litellm-gap-fill/tasks.md @@ -60,7 +60,7 @@ - [x] T014 [US1] Create `crates/proxy/src/server/chat_completions.rs` with non-streaming handler: extract `Json`, call `openai_to_anthropic_request`, dispatch to `BackendClient`, call `anthropic_to_openai_response`, return `Json`. Set `x-anyllm-degradation` header from `TranslationWarnings`. Return OpenAI-shaped errors on validation failure (missing max_tokens -> 400 `invalid_request_error`). - [x] T015 [US1] Add streaming handler in `crates/proxy/src/server/chat_completions.rs`: when `stream: true`, dispatch to backend streaming path, create `ReverseStreamingTranslator`, emit `text/event-stream` with `data: {chunk}\n\n` lines (no `event:` prefix, matching OpenAI SSE format). Terminate with `data: [DONE]\n\n`. - [x] T016 [US1] Register `POST /v1/chat/completions` route in `crates/proxy/src/server/routes.rs` on the existing backend router. Apply same middleware (auth, request ID, size limit, concurrency limit) as the `/v1/messages` route. -- [ ] T017 [US1] Add integration tests in `crates/proxy/tests/` (new file `chat_completions.rs` or extend existing): non-streaming basic response, streaming basic response, tool call round-trip, missing max_tokens returns 400, degradation header set for lossy fields, empty messages returns 400. +- [x] T017 [US1] Add integration tests in `crates/proxy/tests/` (new file `chat_completions.rs` or extend existing): non-streaming basic response, streaming basic response, tool call round-trip, missing max_tokens returns 400, degradation header set for lossy fields, empty messages returns 400. **Checkpoint**: `cargo test -p anyllm_proxy` passes. `POST /v1/chat/completions` works end-to-end with a mock or live backend. @@ -74,11 +74,11 @@ ### Implementation for User Story 2 -- [ ] T018 [US2] Add `BackendKind::AzureOpenAI` variant to the backend enum in `crates/proxy/src/config/mod.rs`. Parse `AZURE_OPENAI_API_KEY`, `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_DEPLOYMENT`, `AZURE_OPENAI_API_VERSION` (default `"2024-10-21"`) from env. Construct full URL: `{endpoint}/openai/deployments/{deployment}/chat/completions?api-version={version}` at config load time. Validate URL with existing `validate_url` function. -- [ ] T019 [US2] Add `BackendAuth::AzureApiKey(String)` variant (or equivalent) in `crates/proxy/src/backend/mod.rs`. Map it to `RequestAuth::Header { name: "api-key", value: key }` in the auth application logic. Add `BackendClient::AzureOpenAI(OpenAIClient)` variant that constructs `OpenAIClient` with the pre-built Azure URL and `AzureApiKey` auth. -- [ ] T020 [US2] Modify `crates/proxy/src/backend/openai_client.rs` to accept Azure's pre-constructed URL. The `chat_completions_url` for Azure is the full URL from config (no `/v1/chat/completions` suffix appended). Ensure the `model` field in the request body is still populated (Azure ignores it but accepts it). -- [ ] T021 [US2] Add `#[ignore]` integration test in `crates/proxy/tests/` for Azure backend: send a request via the proxy configured with `BACKEND=azure`, verify response is valid Anthropic format. Requires `AZURE_OPENAI_API_KEY` env var to run. -- [ ] T022 [US2] Update `docs/ENV.md` with Azure-specific env vars and usage example. +- [x] T018 [US2] Add `BackendKind::AzureOpenAI` variant to the backend enum in `crates/proxy/src/config/mod.rs`. Parse `AZURE_OPENAI_API_KEY`, `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_DEPLOYMENT`, `AZURE_OPENAI_API_VERSION` (default `"2024-10-21"`) from env. Construct full URL: `{endpoint}/openai/deployments/{deployment}/chat/completions?api-version={version}` at config load time. Validate URL with existing `validate_url` function. +- [x] T019 [US2] Add `BackendAuth::AzureApiKey(String)` variant (or equivalent) in `crates/proxy/src/backend/mod.rs`. Map it to `RequestAuth::Header { name: "api-key", value: key }` in the auth application logic. Add `BackendClient::AzureOpenAI(OpenAIClient)` variant that constructs `OpenAIClient` with the pre-built Azure URL and `AzureApiKey` auth. +- [x] T020 [US2] Modify `crates/proxy/src/backend/openai_client.rs` to accept Azure's pre-constructed URL. The `chat_completions_url` for Azure is the full URL from config (no `/v1/chat/completions` suffix appended). Ensure the `model` field in the request body is still populated (Azure ignores it but accepts it). +- [x] T021 [US2] Add `#[ignore]` integration test in `crates/proxy/tests/` for Azure backend: send a request via the proxy configured with `BACKEND=azure`, verify response is valid Anthropic format. Requires `AZURE_OPENAI_API_KEY` env var to run. +- [x] T022 [US2] Update `docs/ENV.md` with Azure-specific env vars and usage example. **Checkpoint**: `cargo build` clean. Azure config parsing tested. `#[ignore]` live test exists. @@ -134,13 +134,13 @@ ### Implementation for User Story 5 -- [ ] T039 [P] [US5] Add `ClientBuilder` to `crates/client/src/client.rs` with method chaining: `fn new() -> Self`, `fn base_url(mut self, url: &str) -> Self`, `fn api_key(mut self, key: &str) -> Self`, `fn timeout(mut self, d: Duration) -> Self`, `fn read_timeout(mut self, d: Duration) -> Self`, `fn max_retries(mut self, n: u32) -> Self`, `fn tls_config(mut self, cfg: TlsConfig) -> Self`, `fn build(self) -> Result`. Implement `Client::builder() -> ClientBuilder` convenience method. -- [ ] T040 [P] [US5] Create `crates/client/src/tools.rs` with `ToolBuilder` and `ToolChoiceBuilder`. `ToolBuilder`: `fn new(name: &str) -> Self`, `fn description(mut self, desc: &str) -> Self`, `fn input_schema(mut self, schema: Value) -> Self`, `fn build(self) -> Tool`. `ToolChoiceBuilder`: `fn auto() -> ToolChoice`, `fn any() -> ToolChoice`, `fn none() -> ToolChoice`, `fn specific(name: &str) -> ToolChoice`. -- [ ] T041 [US5] Add streaming return type to `crates/client/src/client.rs`: `fn messages_stream(&self, req: MessageCreateRequest) -> Result>, ClientError>`. Parse SSE frames from the reqwest response byte stream, deserialize each `data:` line into `StreamEvent`. -- [ ] T042 [US5] Update `crates/client/src/lib.rs` to re-export all public types: `Client`, `ClientBuilder`, `ClientConfig`, `ClientError`, `Tool`, `ToolBuilder`, `ToolChoice`, `ToolChoiceBuilder`, `StreamEvent`, and all Anthropic request/response types from `anyllm_translate`. -- [ ] T043 [US5] Add rustdoc examples to all public types and methods in `crates/client/src/client.rs`, `crates/client/src/tools.rs`, and `crates/client/src/lib.rs`. Each builder method and each public function gets a `/// # Examples` block. -- [ ] T044 [US5] Bump `anyllm_client` version to `0.2.0` in `crates/client/Cargo.toml`. -- [ ] T045 [US5] Add unit tests for `ClientBuilder` (valid build, missing required fields), `ToolBuilder`, and `ToolChoiceBuilder` in their respective `#[cfg(test)]` modules. +- [x] T039 [P] [US5] Add `ClientBuilder` to `crates/client/src/client.rs` with method chaining: `fn new() -> Self`, `fn base_url(mut self, url: &str) -> Self`, `fn api_key(mut self, key: &str) -> Self`, `fn timeout(mut self, d: Duration) -> Self`, `fn read_timeout(mut self, d: Duration) -> Self`, `fn max_retries(mut self, n: u32) -> Self`, `fn tls_config(mut self, cfg: TlsConfig) -> Self`, `fn build(self) -> Result`. Implement `Client::builder() -> ClientBuilder` convenience method. +- [x] T040 [P] [US5] Create `crates/client/src/tools.rs` with `ToolBuilder` and `ToolChoiceBuilder`. `ToolBuilder`: `fn new(name: &str) -> Self`, `fn description(mut self, desc: &str) -> Self`, `fn input_schema(mut self, schema: Value) -> Self`, `fn build(self) -> Tool`. `ToolChoiceBuilder`: `fn auto() -> ToolChoice`, `fn any() -> ToolChoice`, `fn none() -> ToolChoice`, `fn specific(name: &str) -> ToolChoice`. +- [x] T041 [US5] Add streaming return type to `crates/client/src/client.rs`: `fn messages_stream(&self, req: MessageCreateRequest) -> Result>, ClientError>`. Parse SSE frames from the reqwest response byte stream, deserialize each `data:` line into `StreamEvent`. +- [x] T042 [US5] Update `crates/client/src/lib.rs` to re-export all public types: `Client`, `ClientBuilder`, `ClientConfig`, `ClientError`, `Tool`, `ToolBuilder`, `ToolChoice`, `ToolChoiceBuilder`, `StreamEvent`, and all Anthropic request/response types from `anyllm_translate`. +- [x] T043 [US5] Add rustdoc examples to all public types and methods in `crates/client/src/client.rs`, `crates/client/src/tools.rs`, and `crates/client/src/lib.rs`. Each builder method and each public function gets a `/// # Examples` block. +- [x] T044 [US5] Bump `anyllm_client` version to `0.2.0` in `crates/client/Cargo.toml`. +- [x] T045 [US5] Add unit tests for `ClientBuilder` (valid build, missing required fields), `ToolBuilder`, and `ToolChoiceBuilder` in their respective `#[cfg(test)]` modules. **Checkpoint**: `cargo doc -p anyllm_client --no-deps` builds clean. `cargo test -p anyllm_client` passes. @@ -187,7 +187,7 @@ **Purpose**: Final validation, documentation updates, and CI adjustments -- [ ] T057 [P] Update `docs/COMPARISON_LITELLM.md` to reflect closed gaps: `POST /v1/chat/completions` input, Bedrock backend, Azure backend, virtual key management, per-key rate limiting, OTEL export. Move items from "Major gap" to "Advantage" or "Parity" as appropriate. +- [x] T057 [P] Update `docs/COMPARISON_LITELLM.md` to reflect closed gaps: `POST /v1/chat/completions` input, Bedrock backend, Azure backend, virtual key management, per-key rate limiting, OTEL export. Move items from "Major gap" to "Advantage" or "Parity" as appropriate. - [ ] T058 [P] Update `CLAUDE.md` with new backend types, new env vars, new admin endpoints, new source files, and updated test counts. - [ ] T059 [P] Update `README.md` with quickstart examples for new features (reference `quickstart.md` content). - [ ] T060 Run `cargo clippy -- -D warnings` across all crates and fix any warnings.