From 99997d1e3ac0710046b84ef2d9ffb35606c59902 Mon Sep 17 00:00:00 2001 From: Nate Armstrong Date: Fri, 16 Oct 2026 12:43:01 -0700 Subject: [PATCH] feat(gateway): native Bedrock invoke for /v1/messages (#46814) Implements the Bedrock InvokeModel request shaping that returned 501: anthropic_version, the top-level allowlist, model/stream moved to the URL, header and feature betas in the body's anthropic_beta, clear_thinking, cache_control, structured output and tool fixes in Python's order. Streams decode the AWS event stream into Anthropic SSE. genconfig and Terraform now default to half Bedrock deployments signed with SigV4. --- litellm-rust/Cargo.lock | 3 + .../crates/inference-messages/Cargo.toml | 3 + .../crates/inference-messages/src/handler.rs | 6 +- .../crates/inference-messages/src/prepare.rs | 1 + .../crates/inference-messages/src/types.rs | 2 + .../tests/bedrock_native/main.rs | 130 +++ .../crates/inference-messages/tests/live.rs | 3 + .../inference-messages/tests/live/bedrock.rs | 83 ++ .../src/providers/bedrock/constants.rs | 2 + .../crates/llms/src/anthropic/common_utils.rs | 2 + .../llms/src/base_llm/messages/context.rs | 9 + .../src/base_llm/messages/transformation.rs | 11 + .../llms/src/bedrock/messages/AGENTS.md | 1 + .../anthropic_claude3_transformation.rs | 895 +++++++++++++++++- litellm-rust/crates/llms/src/error.rs | 9 + .../customer-proxy/bedrock_messages-flow.svg | 82 ++ litellm-rust/loadtest/README.md | 12 +- litellm-rust/loadtest/cmd/genconfig/main.go | 12 +- .../loadtest/terraform/aws/variables.tf | 4 +- 19 files changed, 1250 insertions(+), 20 deletions(-) create mode 100644 litellm-rust/crates/inference-messages/tests/live/bedrock.rs create mode 100644 litellm-rust/docs/customer-proxy/bedrock_messages-flow.svg diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock index e645db0d4b5..c47c8476ab7 100644 --- a/litellm-rust/Cargo.lock +++ b/litellm-rust/Cargo.lock @@ -4024,6 +4024,9 @@ dependencies = [ name = "litellm-inference-messages" version = "0.1.0" dependencies = [ + "aws-smithy-eventstream", + "aws-smithy-types", + "base64 0.22.1", "bytes", "futures-util", "litellm-auth", diff --git a/litellm-rust/crates/inference-messages/Cargo.toml b/litellm-rust/crates/inference-messages/Cargo.toml index e310a0be9b1..17e4c48ba60 100644 --- a/litellm-rust/crates/inference-messages/Cargo.toml +++ b/litellm-rust/crates/inference-messages/Cargo.toml @@ -26,6 +26,9 @@ serde_json = { workspace = true, features = ["preserve_order"] } tracing.workspace = true [dev-dependencies] +aws-smithy-eventstream = "=0.61.4" +aws-smithy-types = "1.6.1" +base64.workspace = true litellm-framer = { workspace = true, features = ["sse"] } litellm-cache-memory.workspace = true litellm-host-native.workspace = true diff --git a/litellm-rust/crates/inference-messages/src/handler.rs b/litellm-rust/crates/inference-messages/src/handler.rs index c4032baa98f..3d72152795f 100644 --- a/litellm-rust/crates/inference-messages/src/handler.rs +++ b/litellm-rust/crates/inference-messages/src/handler.rs @@ -51,6 +51,9 @@ impl MessagesRoute { timeout, api_key, } = request; + // A host that addresses streaming by URL (Bedrock's invoke-with-response-stream) drops + // `stream` from the wire body, so the typed request decides unless the body says so. + let requested_stream = body.params.stream == Some(true); let request_context = RequestContext { model: body.model.clone(), custom_llm_provider: provider.as_str().to_string(), @@ -78,7 +81,8 @@ impl MessagesRoute { ) .await?; let stream = match wire.body.get("stream") { - None | Some(Value::Null) => false, + None => requested_stream, + Some(Value::Null) => false, Some(Value::Bool(stream)) => *stream, Some(value) => { return Err(Error::InvalidRequest( diff --git a/litellm-rust/crates/inference-messages/src/prepare.rs b/litellm-rust/crates/inference-messages/src/prepare.rs index 4b7873b4808..efb1c9fbb24 100644 --- a/litellm-rust/crates/inference-messages/src/prepare.rs +++ b/litellm-rust/crates/inference-messages/src/prepare.rs @@ -103,6 +103,7 @@ fn prepare_provider_request( with_default_headers(forwarded, config.default_headers()), &transformed, ); + let transformed = config.request_with_headers(transformed, &forwarded); let environment = config.validate_environment( forwarded, api_key.as_deref(), diff --git a/litellm-rust/crates/inference-messages/src/types.rs b/litellm-rust/crates/inference-messages/src/types.rs index 21eac0574ed..a2b1e60efd8 100644 --- a/litellm-rust/crates/inference-messages/src/types.rs +++ b/litellm-rust/crates/inference-messages/src/types.rs @@ -149,6 +149,8 @@ mod tests { supports_sampling_params: false, supports_speed: true, supports_mid_conversation_system: false, + supports_prompt_cache_ttl: false, + supports_native_structured_output: false, effort_tiers: SupportedEffortTiers { minimal: false, low: true, diff --git a/litellm-rust/crates/inference-messages/tests/bedrock_native/main.rs b/litellm-rust/crates/inference-messages/tests/bedrock_native/main.rs index 5420477c34a..29165adde13 100644 --- a/litellm-rust/crates/inference-messages/tests/bedrock_native/main.rs +++ b/litellm-rust/crates/inference-messages/tests/bedrock_native/main.rs @@ -321,3 +321,133 @@ async fn deployment_workspace_is_required_and_sent_as_a_header( ); assert_eq!(sent.json().get("workspace_id"), None); } + +const INVOKE_MODEL: &str = "bedrock/us.anthropic.claude-opus-5-5"; + +fn signed_invoke_call(call: MessagesCall, base: String, fields: Value) -> MessagesCall { + MessagesCall { + api_base: Some(base), + custom_llm_provider: None, + litellm_params: litellm_router_types::LitellmParams { + aws: litellm_auth::AwsParams { + aws_access_key_id: Some("static-access".into()), + aws_secret_access_key: Some("static-secret".into()), + aws_region_name: Some("us-east-1".into()), + ..Default::default() + }, + ..Default::default() + }, + provider_specific_header: Some( + serde_json::from_value(json!({"custom_llm_provider": "bedrock", "extra_headers": { + "anthropic-version": "2023-06-01", + "anthropic-beta": "context-1m-2025-08-07,unknown-beta" + }})) + .unwrap(), + ), + ..with_fields(with_model(call, INVOKE_MODEL), fields) + } +} + +fn aws_event(chunk: &Value) -> Vec { + use aws_smithy_types::event_stream::{Header, HeaderValue, Message}; + use base64::Engine; + let payload = + json!({"bytes": base64::engine::general_purpose::STANDARD.encode(chunk.to_string())}); + let message = Message::new(serde_json::to_vec(&payload).unwrap()) + .add_header(Header::new( + ":event-type", + HeaderValue::String("chunk".into()), + )) + .add_header(Header::new( + ":message-type", + HeaderValue::String("event".into()), + )); + let mut wire = Vec::new(); + aws_smithy_eventstream::frame::write_message_to(&message, &mut wire).unwrap(); + wire +} + +#[rstest] +#[tokio::test] +async fn invoke_sends_a_signed_bedrock_body_and_returns_the_message(call: MessagesCall) { + let upstream = upstream([message_response()]).await; + let message = run_message(signed_invoke_call( + call, + upstream.uri(), + json!({"metadata": {"user_id": "u1"}, "speed": "fast"}), + )) + .await; + assert_eq!(message.model, MODEL); + let request = only_request(&upstream).await; + assert_eq!( + request.url.path(), + "/model/us.anthropic.claude-opus-5-5/invoke" + ); + assert!( + request + .header("authorization") + .unwrap() + .starts_with("AWS4-HMAC-SHA256 Credential=static-access/") + ); + assert_eq!( + request.json(), + json!({ + "anthropic_version": "bedrock-2023-05-31", + "anthropic_beta": ["context-1m-2025-08-07"], + "max_tokens": 16, + "metadata": {"user_id": "u1"}, + "messages": [{"role": "user", "content": "hi"}] + }) + ); +} + +#[rstest] +#[tokio::test] +async fn invoke_streams_decode_aws_events_into_anthropic_sse(call: MessagesCall) { + let chunks = [ + json!({"type": "message_start", "message": {"id": "msg_1", "type": "message", "role": "assistant", "model": MODEL, "content": [], "stop_reason": null, "stop_sequence": null, "usage": {"input_tokens": 3, "output_tokens": 1}}}), + json!({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}), + json!({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hi"}}), + json!({"type": "content_block_stop", "index": 0}), + json!({"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 2}}), + json!({"type": "message_stop", "amazon-bedrock-invocationMetrics": {"inputTokenCount": 3, "outputTokenCount": 2}}), + ]; + let body: Vec = chunks.iter().flat_map(aws_event).collect(); + let upstream = upstream([ + ResponseTemplate::new(200).set_body_raw(body, "application/vnd.amazon.eventstream") + ]) + .await; + let response = run(signed_invoke_call( + call, + upstream.uri(), + json!({"stream": true}), + )) + .await + .unwrap(); + let MessagesCallResponse::Stream { chunks, .. } = response else { + panic!("expected SSE"); + }; + let sse = String::from_utf8(chunks.try_collect::>().await.unwrap().concat()).unwrap(); + let events: Vec<&str> = sse + .lines() + .filter_map(|line| line.strip_prefix("event: ")) + .collect(); + assert_eq!( + events, + [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop" + ] + ); + assert!(sse.contains(r#""text":"hi""#)); + let request = only_request(&upstream).await; + assert_eq!( + request.url.path(), + "/model/us.anthropic.claude-opus-5-5/invoke-with-response-stream" + ); + assert_eq!(request.json().get("stream"), None); +} diff --git a/litellm-rust/crates/inference-messages/tests/live.rs b/litellm-rust/crates/inference-messages/tests/live.rs index 24468598141..972b58459b1 100644 --- a/litellm-rust/crates/inference-messages/tests/live.rs +++ b/litellm-rust/crates/inference-messages/tests/live.rs @@ -6,3 +6,6 @@ mod edenai; #[path = "live/vertex_ai.rs"] mod vertex_ai; + +#[path = "live/bedrock.rs"] +mod bedrock; diff --git a/litellm-rust/crates/inference-messages/tests/live/bedrock.rs b/litellm-rust/crates/inference-messages/tests/live/bedrock.rs new file mode 100644 index 00000000000..b8fa6debfd2 --- /dev/null +++ b/litellm-rust/crates/inference-messages/tests/live/bedrock.rs @@ -0,0 +1,83 @@ +//! Claude on Bedrock InvokeModel. Runs when `LITELLM_LIVE_MESSAGES_BEDROCK_MODEL` names a Claude +//! model id (for example a `us.anthropic.*` inference profile) and AWS credentials resolve from the +//! environment or default chain, or `AWS_BEARER_TOKEN_BEDROCK` is set. + +use litellm_inference_messages::MessagesRoute; +use litellm_inference_testing::live::within_deadline; +use rstest::{fixture, rstest}; +use serde_json::json; + +use super::support::{LiveCall, assert_text, call, complete, request, route, stream_events}; + +const PROVIDER: &str = "bedrock"; + +#[fixture] +fn model() -> Option { + std::env::var("LITELLM_LIVE_MESSAGES_BEDROCK_MODEL") + .ok() + .filter(|model| !model.trim().is_empty()) +} + +#[rstest] +#[tokio::test] +async fn completion(route: MessagesRoute, model: Option) { + if model.is_none() { + return; + } + within_deadline(async { + let host = LiveCall::new(PROVIDER); + let message = complete( + request( + &route, + call( + PROVIDER, + json!({"messages": [{"role": "user", "content": "Say hello."}]}), + ), + &host, + ) + .await, + ); + assert_text(&message); + host.assert_provider_result(); + }) + .await; +} + +#[rstest] +#[tokio::test] +async fn streaming(route: MessagesRoute, model: Option) { + if model.is_none() { + return; + } + within_deadline(async { + let host = LiveCall::new(PROVIDER); + let response = request( + &route, + call( + PROVIDER, + json!({"stream": true, "messages": [{"role": "user", "content": "Say hello."}]}), + ), + &host, + ) + .await; + let events = stream_events(response).await; + assert!(events.iter().any(|event| { + event["delta"]["text"] + .as_str() + .is_some_and(|text| !text.is_empty()) + })); + host.assert_provider_result(); + }) + .await; +} + +#[rstest] +#[case::complete(false)] +#[case::streaming(true)] +#[tokio::test] +async fn tool_round_trip(route: MessagesRoute, model: Option, #[case] stream: bool) { + if model.is_none() { + return; + } + super::support::tool_round_trip(route, PROVIDER, stream, json!({})).await; +} diff --git a/litellm-rust/crates/llms-types/src/providers/bedrock/constants.rs b/litellm-rust/crates/llms-types/src/providers/bedrock/constants.rs index 82c248421a3..0b6a24ba77f 100644 --- a/litellm-rust/crates/llms-types/src/providers/bedrock/constants.rs +++ b/litellm-rust/crates/llms-types/src/providers/bedrock/constants.rs @@ -2,3 +2,5 @@ pub const CONVERSE_PATH: &str = "/converse"; pub const INVOKE_PATH: &str = "invoke"; pub const INVOKE_STREAM_PATH: &str = "invoke-with-response-stream"; pub const INVOCATION_METRICS_KEY: &str = "amazon-bedrock-invocationMetrics"; +/// The `anthropic_version` InvokeModel requires in a Claude Messages body. +pub const ANTHROPIC_VERSION: &str = "bedrock-2023-05-31"; diff --git a/litellm-rust/crates/llms/src/anthropic/common_utils.rs b/litellm-rust/crates/llms/src/anthropic/common_utils.rs index 7991a57353e..6bc2f7427f7 100644 --- a/litellm-rust/crates/llms/src/anthropic/common_utils.rs +++ b/litellm-rust/crates/llms/src/anthropic/common_utils.rs @@ -1885,6 +1885,8 @@ mod tests { supports_sampling_params: true, supports_speed: false, supports_mid_conversation_system: false, + supports_prompt_cache_ttl: false, + supports_native_structured_output: false, effort_tiers: tiers(false, false, false, false, false, false), } ); diff --git a/litellm-rust/crates/llms/src/base_llm/messages/context.rs b/litellm-rust/crates/llms/src/base_llm/messages/context.rs index e59a63813b0..7411a139ae5 100644 --- a/litellm-rust/crates/llms/src/base_llm/messages/context.rs +++ b/litellm-rust/crates/llms/src/base_llm/messages/context.rs @@ -41,6 +41,13 @@ pub struct MessagesModelCapabilities { pub supports_speed: bool, #[serde(default)] pub supports_mid_conversation_system: bool, + /// Bedrock accepts `cache_control.ttl` ("5m"/"1h") for the model; Python reads it from + /// `cache_creation_input_token_cost_above_1hr` in the model map. + #[serde(default)] + pub supports_prompt_cache_ttl: bool, + /// Bedrock enforces `output_config.format` for the model; otherwise the schema is inlined. + #[serde(default)] + pub supports_native_structured_output: bool, #[serde(default)] pub effort_tiers: SupportedEffortTiers, } @@ -60,6 +67,8 @@ impl Default for MessagesModelCapabilities { supports_sampling_params: true, supports_speed: false, supports_mid_conversation_system: false, + supports_prompt_cache_ttl: false, + supports_native_structured_output: false, effort_tiers: SupportedEffortTiers::default(), } } diff --git a/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs b/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs index 6f0ea5c8f2a..116057e1ba3 100644 --- a/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs @@ -74,6 +74,17 @@ pub trait BaseMessagesConfig: Sync { headers } + /// The request once the forwarded headers are final, for a host that reads a header's + /// values from the body: Bedrock InvokeModel takes the caller's `anthropic-beta` as the + /// body's `anthropic_beta`. + fn request_with_headers( + &self, + request: MessagesRequest, + _headers: &Headers, + ) -> MessagesRequest { + request + } + /// The JSON that goes on the wire, for a host whose body differs from the typed request: /// Python's configs `pop("model")` when the model is addressed by the URL. fn wire_body(&self, body: Value) -> Value { diff --git a/litellm-rust/crates/llms/src/bedrock/messages/AGENTS.md b/litellm-rust/crates/llms/src/bedrock/messages/AGENTS.md index 48c222a1cdc..94772605dfb 100644 --- a/litellm-rust/crates/llms/src/bedrock/messages/AGENTS.md +++ b/litellm-rust/crates/llms/src/bedrock/messages/AGENTS.md @@ -2,6 +2,7 @@ - Claude on Bedrock InvokeModel: reuses `anthropic/messages` shaping, moves the model into the URL, and decodes the AWS event stream - Invocation metrics in the response map onto Messages usage here +- The wire body is cut to Python's `BedrockInvokeAnthropicMessagesRequest` allowlist, and betas travel in the body's `anthropic_beta` (caller header values join through `request_with_headers`) # references diff --git a/litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.rs b/litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.rs index b467ec723a5..90805b66209 100644 --- a/litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.rs +++ b/litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.rs @@ -1,8 +1,18 @@ use std::convert::Infallible; use crate::{ - anthropic::messages::handler::shape_anthropic_messages_request, - base_llm::messages::context::MessagesTransformContext, + ErrorDetail, + anthropic::{ + common_utils::supports_effort_param, + messages::{ + handler::shape_anthropic_messages_request, + transformation::{provider_feature_betas, transform_messages_request}, + }, + }, + base_llm::messages::{ + context::{MessagesModelCapabilities, MessagesTransformContext, ThinkingBudgets}, + normalization::normalize_system_role_messages, + }, }; use futures_util::StreamExt; use litellm_auth::AwsParams; @@ -15,15 +25,17 @@ use litellm_auth_aws::{ }, resolve_bedrock_region, }; +use litellm_http::request::header_values; use litellm_llms_types::formats::messages::{ - MessagesRequest, + MessagesOptionalParams, MessagesRequest, streaming::{MessagesStreamEvent, MessagesStreamUsage}, }; -use litellm_llms_types::providers::bedrock::{ - INVOCATION_METRICS_KEY, INVOKE_PATH, INVOKE_STREAM_PATH, +use litellm_llms_types::providers::{ + anthropic::{AnthropicBeta, BETA_HEADER, BetaProvider, BetaSet}, + bedrock::{ANTHROPIC_VERSION, INVOCATION_METRICS_KEY, INVOKE_PATH, INVOKE_STREAM_PATH}, }; use litellm_router_types::LitellmParams; -use serde_json::{Map, Value}; +use serde_json::{Map, Value, json}; use crate::{ Error, @@ -55,6 +67,39 @@ const SECRET_NAMES: &[&str] = &[ AWS_DEFAULT_REGION, ]; +/// Python's `BedrockInvokeAnthropicMessagesRequest`. Any other field is a Bedrock 400. +const ALLOWED_TOP_LEVEL_FIELDS: [&str; 16] = [ + "anthropic_version", + "max_tokens", + "messages", + "anthropic_beta", + "system", + "stop_sequences", + "temperature", + "top_p", + "top_k", + "tools", + "tool_choice", + "thinking", + "metadata", + "output_config", + "safeguards", + "context_management", +]; + +/// `clear_thinking_20251015` is absent on purpose, it becomes a `thinking` config instead. +const SUPPORTED_CONTEXT_EDITS: [&str; 2] = ["compact_20260112", "clear_tool_uses_20250919"]; +const CLEAR_THINKING_EDIT: &str = "clear_thinking_20251015"; +const MIN_THINKING_BUDGET_TOKENS: u64 = 1024; +const LEGACY_THINKING_MODEL_PATTERNS: [&str; 12] = [ + "opus-4.5", "opus_4.5", "opus-4-5", "opus_4_5", "opus-4.1", "opus_4.1", "opus-4-1", "opus_4_1", + "opus-4", "opus_4", "sonnet-4", "sonnet_4", +]; +const ADAPTIVE_THINKING_KEYS: [&str; 3] = ["type", "display", "block_binding"]; +const WEB_SEARCH_TOOL_PREFIX: &str = "web_search_"; +const WEB_SEARCH_GUIDANCE: &str = "To use web search with this model, enable LiteLLM's web search interception so the proxy executes the search instead: https://docs.litellm.ai/docs/integrations/websearch_interception. Alternatively, remove the web_search tool from the request."; +const UNNAMED_TOOL_PREFIX: &str = "litellm_unnamed_tool_"; + pub struct AmazonAnthropicClaudeMessagesConfig; pub const BEDROCK_ANTHROPIC_MESSAGES_CONFIG: AmazonAnthropicClaudeMessagesConfig = @@ -125,14 +170,41 @@ impl BaseMessagesConfig for AmazonAnthropicClaudeMessagesConfig { )) } + /// Python's `transform_anthropic_messages_request`: the shared Anthropic shaping, then the + /// InvokeModel body rules. `model` and `stream` stay on the typed request (they pick the + /// URL) and leave in `wire_body`, as does every field outside the allowlist. fn transform_anthropic_messages_request( &self, - _request: MessagesRequest, - _context: &MessagesTransformContext, + request: MessagesRequest, + context: &MessagesTransformContext, ) -> Result { - Err(Error::Unsupported( - "Bedrock invoke messages request shaping", - )) + let capabilities = &context.thinking.capabilities; + // InvokeModel has no fast mode, and Python drops `speed` with the allowlist rather than + // rejecting it the way the shared Anthropic stage does for models without `supports_speed`. + let request = MessagesRequest { + params: MessagesOptionalParams { + speed: None, + ..request.params + }, + ..request + }; + let request = normalize_system_role_messages( + transform_messages_request(request, context)?, + capabilities.supports_mid_conversation_system, + ); + let model = request.model.clone(); + let Value::Object(body) = serde_json::to_value(request).map_err(invalid_request)? else { + return Err(Error::InvalidType { + expected: "a JSON object", + actual: "another JSON value", + }); + }; + let eager_input_streaming = uses_eager_input_streaming(&body); + let body = invoke_body(body, &model, context)?; + let shaped: MessagesRequest = + serde_json::from_value(Value::Object(body)).map_err(invalid_request)?; + let betas = feature_betas(&shaped, eager_input_streaming); + Ok(with_body_betas(shaped, betas)) } fn secret_names(&self) -> &'static [&'static str] { @@ -175,11 +247,501 @@ impl BaseMessagesConfig for AmazonAnthropicClaudeMessagesConfig { &[("content-type", "application/json")] } + /// The caller's `anthropic-beta` values join the feature betas in the body's + /// `anthropic_beta`, which is where InvokeModel reads them. Values Bedrock has no mapping + /// for are dropped, since forwarding them is a 400. + fn request_with_headers(&self, request: MessagesRequest, headers: &Headers) -> MessagesRequest { + let caller = caller_betas(headers); + if caller.is_empty() { + return request; + } + let body = body_betas(&request); + with_body_betas(request, body.union(caller)) + } + + fn wire_body(&self, body: Value) -> Value { + match body { + Value::Object(fields) => Value::Object( + fields + .into_iter() + .filter(|(name, _)| ALLOWED_TOP_LEVEL_FIELDS.contains(&name.as_str())) + .collect(), + ), + other => other, + } + } + fn stream_decoder(&self) -> Option { Some(bedrock_anthropic_messages_event_stream) } } +fn invalid_request(error: serde_json::Error) -> Error { + Error::InvalidRequest(ErrorDetail::invalid( + "Bedrock invoke messages request", + error, + )) +} + +/// The InvokeModel body rules of Python's transformation, in its order. +fn invoke_body( + mut body: Map, + model: &str, + context: &MessagesTransformContext, +) -> Result, Error> { + let capabilities = &context.thinking.capabilities; + reject_web_search_tools(&body, model)?; + body.entry("anthropic_version") + .or_insert_with(|| json!(ANTHROPIC_VERSION)); + ensure_thinking_for_clear_thinking(&mut body, model, context); + sanitize_cache_controls(&mut body, capabilities.supports_prompt_cache_ttl); + apply_structured_output(&mut body, capabilities.supports_native_structured_output); + strip_unsupported_output_config(&mut body, capabilities); + normalize_tools(&mut body); + filter_context_management(&mut body); + let unsupported_effort = body + .get("output_config") + .and_then(Value::as_object) + .is_some_and(|config| config.keys().any(|key| key != "format")) + && !supports_effort_param(capabilities); + if context.drop_params && unsupported_effort { + body.remove("output_config"); + } + Ok(body) +} + +fn tools(body: &Map) -> impl Iterator> { + body.get("tools") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(Value::as_object) +} + +fn tools_mut(body: &mut Map) -> impl Iterator> { + body.get_mut("tools") + .and_then(Value::as_array_mut) + .into_iter() + .flatten() + .filter_map(Value::as_object_mut) +} + +fn uses_eager_input_streaming(body: &Map) -> bool { + tools(body).any(|tool| tool.get("eager_input_streaming") == Some(&Value::Bool(true))) +} + +/// Bedrock cannot run Anthropic's server-side `web_search_*` tool and answers an opaque 400, +/// so the request fails before it is sent with an error that names the fix. +fn reject_web_search_tools(body: &Map, model: &str) -> Result<(), Error> { + let Some(tool) = tools(body) + .filter_map(|tool| tool.get("type")?.as_str()) + .find(|kind| kind.starts_with(WEB_SEARCH_TOOL_PREFIX)) + else { + return Ok(()); + }; + Err(Error::InvalidRequest(ErrorDetail::UnsupportedTool { + provider: "Bedrock", + tool: tool.to_string(), + model: model.to_string(), + guidance: WEB_SEARCH_GUIDANCE, + })) +} + +fn supports_extended_thinking(model: &str, capabilities: &MessagesModelCapabilities) -> bool { + let model = model.to_ascii_lowercase(); + capabilities.supports_adaptive_thinking + || LEGACY_THINKING_MODEL_PATTERNS + .iter() + .any(|pattern| model.contains(pattern)) +} + +fn effort_from_budget(budget: u64, budgets: &ThinkingBudgets) -> &'static str { + match budget { + budget if budget >= budgets.xhigh => "xhigh", + budget if budget >= budgets.high => "high", + budget if budget >= budgets.medium => "medium", + _ => "low", + } +} + +fn inject_adaptive_thinking( + body: &mut Map, + budget: u64, + context: &MessagesTransformContext, +) { + let mut output_config = match body.remove("output_config") { + Some(Value::Object(config)) => config, + _ => Map::new(), + }; + output_config + .entry("effort") + .or_insert_with(|| json!(effort_from_budget(budget, &context.thinking.budgets))); + let carried: Map = match body.remove("thinking") { + Some(Value::Object(thinking)) => thinking + .into_iter() + .filter(|(key, _)| { + !matches!(key.as_str(), "type" | "budget_tokens") + && (!context.drop_params || ADAPTIVE_THINKING_KEYS.contains(&key.as_str())) + }) + .collect(), + _ => Map::new(), + }; + body.insert("output_config".into(), Value::Object(output_config)); + body.insert( + "thinking".into(), + Value::Object( + carried + .into_iter() + .chain([("type".to_string(), json!("adaptive"))]) + .collect(), + ), + ); +} + +/// Bedrock rejects a `clear_thinking_20251015` edit unless thinking is on, and Claude Code +/// sends that edit without a `thinking` field, so a minimal thinking config is added: adaptive +/// with an effort for adaptive models, a minimum budget for the legacy ones. +fn ensure_thinking_for_clear_thinking( + body: &mut Map, + model: &str, + context: &MessagesTransformContext, +) { + let capabilities = &context.thinking.capabilities; + let clears_thinking = body + .get("context_management") + .and_then(|management| management.get("edits")) + .and_then(Value::as_array) + .is_some_and(|edits| { + edits + .iter() + .any(|edit| edit.get("type").and_then(Value::as_str) == Some(CLEAR_THINKING_EDIT)) + }); + if !clears_thinking || !supports_extended_thinking(model, capabilities) { + return; + } + let adaptive_model = capabilities.supports_adaptive_thinking; + let thinking = body.get("thinking").and_then(Value::as_object); + match thinking.and_then(|thinking| thinking.get("type")?.as_str()) { + Some("adaptive") => return, + Some("enabled") if !adaptive_model => return, + Some("enabled") => { + let budget = thinking + .and_then(|thinking| thinking.get("budget_tokens")?.as_u64()) + .unwrap_or(MIN_THINKING_BUDGET_TOKENS); + inject_adaptive_thinking(body, budget, context); + return; + } + _ => {} + } + if body + .get("max_tokens") + .and_then(Value::as_u64) + .is_some_and(|max_tokens| max_tokens <= MIN_THINKING_BUDGET_TOKENS) + { + return; + } + if adaptive_model { + inject_adaptive_thinking(body, MIN_THINKING_BUDGET_TOKENS, context); + return; + } + body.insert( + "thinking".into(), + json!({"type": "enabled", "budget_tokens": MIN_THINKING_BUDGET_TOKENS}), + ); +} + +/// Bedrock takes only `type` and, on models that support it, a "5m" or "1h" `ttl` in a +/// `cache_control`. `scope` always goes. +fn sanitize_cache_control(block: &mut Map, supports_ttl: bool) { + let Some(Value::Object(cache_control)) = block.get_mut("cache_control") else { + return; + }; + cache_control.remove("scope"); + let keeps_ttl = supports_ttl + && cache_control + .get("ttl") + .and_then(Value::as_str) + .is_some_and(|ttl| matches!(ttl, "5m" | "1h")); + if !keeps_ttl { + cache_control.remove("ttl"); + } +} + +fn sanitize_cache_controls(body: &mut Map, supports_ttl: bool) { + tools_mut(body).for_each(|tool| sanitize_cache_control(tool, supports_ttl)); + body.iter_mut() + .flat_map(|(name, value)| match (name.as_str(), value) { + ("system", Value::Array(blocks)) => blocks.iter_mut().collect(), + ("messages", Value::Array(messages)) => messages + .iter_mut() + .filter_map(|message| message.get_mut("content")?.as_array_mut()) + .flatten() + .collect(), + _ => Vec::new(), + }) + .filter_map(Value::as_object_mut) + .for_each(|block| sanitize_cache_control(block, supports_ttl)); +} + +/// Python's `apply_bedrock_invoke_structured_output`: the legacy `output_format` wins over +/// `output_config.format`. Models with native structured output get it as +/// `output_config.format`; for the rest the schema is appended to the last user turn as text. +fn apply_structured_output(body: &mut Map, native: bool) { + let legacy = body.remove("output_format").filter(Value::is_object); + let nested = match body.get_mut("output_config") { + Some(Value::Object(config)) => { + let format = config.remove("format"); + if config.is_empty() { + body.remove("output_config"); + } + format + } + _ => None, + }; + let Some(format) = legacy.or(nested.filter(Value::is_object)) else { + return; + }; + if native { + match body.get_mut("output_config") { + Some(Value::Object(config)) => { + config.insert("format".into(), format); + } + _ => { + body.insert("output_config".into(), json!({"format": format})); + } + } + return; + } + inline_schema(body, &format); +} + +fn inline_schema(body: &mut Map, format: &Value) { + let Some(schema) = format.get("schema").filter(|schema| match schema { + Value::Null => false, + Value::Object(fields) => !fields.is_empty(), + _ => true, + }) else { + return; + }; + let schema_block = json!({"type": "text", "text": python_json_dumps(schema)}); + let Some(last_user) = body + .get_mut("messages") + .and_then(Value::as_array_mut) + .and_then(|messages| { + messages + .iter_mut() + .rev() + .find(|message| message.get("role").and_then(Value::as_str) == Some("user")) + }) + .and_then(Value::as_object_mut) + else { + return; + }; + let content = match last_user.remove("content") { + Some(Value::String(text)) => json!([{"type": "text", "text": text}, schema_block]), + Some(Value::Array(blocks)) => { + Value::Array(blocks.into_iter().chain([schema_block]).collect()) + } + None => json!([schema_block]), + Some(other) => other, + }; + last_user.insert("content".into(), content); +} + +/// `json.dumps` spacing (`", "` and `": "`), so the inlined schema text matches Python's. +fn python_json_dumps(value: &Value) -> String { + match value { + Value::Array(items) => format!( + "[{}]", + items + .iter() + .map(python_json_dumps) + .collect::>() + .join(", ") + ), + Value::Object(fields) => format!( + "{{{}}}", + fields + .iter() + .map(|(key, value)| format!( + "{}: {}", + Value::from(key.as_str()), + python_json_dumps(value) + )) + .collect::>() + .join(", ") + ), + scalar => scalar.to_string(), + } +} + +/// Python's `strip_unsupported_bedrock_invoke_output_config_keys`: `format` always survives, +/// effort-bearing keys only on models that accept `output_config` or an effort tier. +fn strip_unsupported_output_config( + body: &mut Map, + capabilities: &MessagesModelCapabilities, +) { + let Some(Value::Object(config)) = body.get("output_config") else { + return; + }; + if config.keys().all(|key| key == "format") + || capabilities.supports_output_config + || supports_effort_param(capabilities) + { + return; + } + match config.get("format").cloned() { + Some(format) => body.insert("output_config".into(), json!({"format": format})), + None => body.remove("output_config"), + }; +} + +/// Python's `normalize_json_schema_custom_types_to_object`, iterative like the original. +fn custom_schema_types_to_object(schema: &mut Value) { + let mut stack = vec![schema]; + while let Some(Value::Object(node)) = stack.pop() { + if node.get("type").and_then(Value::as_str) == Some("custom") { + node.insert("type".into(), json!("object")); + } + for (key, child) in node.iter_mut() { + match (key.as_str(), child) { + ("items" | "additionalProperties", child @ Value::Object(_)) => stack.push(child), + ("properties", Value::Object(properties)) => { + stack.extend(properties.values_mut().filter(|sub| sub.is_object())); + } + ("allOf" | "anyOf" | "oneOf", Value::Array(subs)) => { + stack.extend(subs.iter_mut().filter(|sub| sub.is_object())); + } + _ => {} + } + } + } +} + +/// Bedrock's tool validation: `custom` is rejected (its boolean `defer_loading` is hoisted +/// first), `input_schema` may not use `type: "custom"`, every tool needs a `name`, and +/// `eager_input_streaming` is carried by the fine-grained streaming beta instead. +fn normalize_tools(body: &mut Map) { + for (index, tool) in tools_mut(body).enumerate() { + if let Some(Value::Object(custom)) = tool.remove("custom") + && !tool.contains_key("defer_loading") + && let Some(deferred @ Value::Bool(_)) = custom.get("defer_loading") + { + tool.insert("defer_loading".into(), deferred.clone()); + } + if let Some(schema) = tool.get_mut("input_schema") { + custom_schema_types_to_object(schema); + } + let named = match tool.get("name") { + None | Some(Value::Null) => false, + Some(Value::String(name)) => !name.trim().is_empty(), + Some(_) => true, + }; + if !named { + tool.insert( + "name".into(), + json!(format!("{UNNAMED_TOOL_PREFIX}{index}")), + ); + } + tool.remove("eager_input_streaming"); + } +} + +/// Keeps the edits InvokeModel accepts; drops `context_management` when none remain. +fn filter_context_management(body: &mut Map) { + let Some(Value::Object(management)) = body.get_mut("context_management") else { + return; + }; + let retained: Vec = management + .get("edits") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter(|edit| { + edit.get("type") + .and_then(Value::as_str) + .is_some_and(|kind| SUPPORTED_CONTEXT_EDITS.contains(&kind)) + }) + .cloned() + .collect(); + if retained.is_empty() { + body.remove("context_management"); + return; + } + management.insert("edits".into(), Value::Array(retained)); +} + +/// The betas the request's own features need, under Bedrock's beta policy. +fn feature_betas(request: &MessagesRequest, eager_input_streaming: bool) -> BetaSet { + let extra = &request.params.extra; + let added: BetaSet = [ + eager_input_streaming.then_some(AnthropicBeta::FineGrainedToolStreaming20250514), + extra + .get("safeguards") + .is_some_and(|value| !value.is_null()) + .then_some(AnthropicBeta::DangerousToolUse20260903), + ] + .into_iter() + .flatten() + .filter_map(|beta| beta.on(BetaProvider::Bedrock)) + .collect(); + provider_feature_betas(request, BetaProvider::Bedrock).union(added) +} + +/// Python's `get_anthropic_beta_from_headers`: a JSON array or comma-separated values, every +/// casing of the header, filtered through Bedrock's beta policy. +fn caller_betas(headers: &Headers) -> BetaSet { + header_values(headers, BETA_HEADER) + .flat_map(|value| { + let value = value.trim(); + match serde_json::from_str::>(value) { + Ok(items) if value.starts_with('[') => items + .iter() + .map(|item| match item { + Value::String(text) => text.trim().to_string(), + other => other.to_string(), + }) + .collect(), + _ => value + .split(',') + .map(|piece| piece.trim().to_string()) + .collect::>(), + } + }) + .filter(|beta| !beta.is_empty()) + .filter_map(|beta| { + beta.parse::() + .ok()? + .on(BetaProvider::Bedrock) + }) + .collect() +} + +fn body_betas(request: &MessagesRequest) -> BetaSet { + request + .params + .extra + .get("anthropic_beta") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(Value::as_str) + .filter_map(|beta| beta.parse::().ok()) + .collect() +} + +fn with_body_betas(request: MessagesRequest, betas: BetaSet) -> MessagesRequest { + if betas.is_empty() { + return request; + } + let mut params = request.params; + params.extra.insert( + "anthropic_beta".into(), + Value::Array(betas.iter().map(|beta| json!(beta.as_str())).collect()), + ); + MessagesRequest { params, ..request } +} + fn with_invocation_usage(chunk: Value) -> Value { match chunk { Value::Object(fields) => Value::Object(with_metrics_usage(fields)), @@ -684,4 +1246,315 @@ mod tests { (other, _) => panic!("unexpected auth {other:?}"), } } + + use crate::base_llm::messages::context::{SupportedEffortTiers, ThinkingContext}; + + const OPUS: &str = "us.anthropic.claude-opus-5-5"; + + /// What Python reads from the model map for Bedrock's Opus 5.5 that the shaping consults, + /// with sampling params left on so the shared Anthropic stage passes them through as + /// Python's messages path does. + fn opus_capabilities() -> MessagesModelCapabilities { + MessagesModelCapabilities { + supports_reasoning: true, + supports_adaptive_thinking: true, + thinking_always_on: true, + supports_output_config: true, + supports_mid_conversation_system: true, + supports_prompt_cache_ttl: true, + effort_tiers: SupportedEffortTiers { + low: true, + medium: true, + high: true, + xhigh: true, + max: true, + ..SupportedEffortTiers::default() + }, + ..MessagesModelCapabilities::default() + } + } + + fn context(capabilities: MessagesModelCapabilities) -> MessagesTransformContext { + MessagesTransformContext { + thinking: ThinkingContext { + capabilities, + budgets: ThinkingBudgets::default(), + }, + drop_params: false, + } + } + + fn wire( + model: &str, + fields: Value, + headers: &[(&str, &str)], + capabilities: MessagesModelCapabilities, + ) -> Result { + let Value::Object(fields) = fields else { + unreachable!("cases are objects") + }; + let request: MessagesRequest = serde_json::from_value(Value::Object( + [("model".to_string(), json!(model))] + .into_iter() + .chain(fields) + .collect(), + )) + .unwrap(); + let headers: Headers = headers + .iter() + .map(|(name, value)| (name.to_string(), value.to_string())) + .collect(); + let config = AmazonAnthropicClaudeMessagesConfig; + let transformed = + config.transform_anthropic_messages_request(request, &context(capabilities))?; + let transformed = config.request_with_headers(transformed, &headers); + Ok(config.wire_body(serde_json::to_value(transformed).unwrap())) + } + + // Expected bodies are Python's `AmazonAnthropicClaudeMessagesConfig(). + // transform_anthropic_messages_request` output for the same input and model. + #[rstest] + #[case::plain( + json!({"max_tokens": 1024, "stream": true, "temperature": 0.5, "metadata": {"user_id": "u1"}, + "messages": [{"role": "user", "content": "Hello"}]}), + &[], + json!({"anthropic_version": "bedrock-2023-05-31", "max_tokens": 1024, "temperature": 0.5, + "metadata": {"user_id": "u1"}, "messages": [{"role": "user", "content": "Hello"}]}), + )] + #[case::tools( + json!({"max_tokens": 1024, "messages": [{"role": "user", "content": "weather?"}], "tool_choice": {"type": "auto"}, + "tools": [ + {"name": "get_weather", "description": "w", "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}}, "eager_input_streaming": true}, + {"input_schema": {"type": "custom", "properties": {"q": {"type": "custom"}}}, "custom": {"defer_loading": true}} + ]}), + &[], + json!({"anthropic_beta": ["fine-grained-tool-streaming-2025-05-14"], "anthropic_version": "bedrock-2023-05-31", + "max_tokens": 1024, "messages": [{"role": "user", "content": "weather?"}], "tool_choice": {"type": "auto"}, + "tools": [ + {"name": "get_weather", "description": "w", "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}}}, + {"name": "litellm_unnamed_tool_1", "defer_loading": true, "input_schema": {"type": "object", "properties": {"q": {"type": "object"}}}} + ]}), + )] + #[case::thinking_and_cache_control( + json!({"max_tokens": 4096, "thinking": {"type": "enabled", "budget_tokens": 2048}, + "system": [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral", "scope": "global", "ttl": "1h"}}], + "messages": [{"role": "user", "content": "Hello"}]}), + &[], + json!({"anthropic_version": "bedrock-2023-05-31", "max_tokens": 4096, "messages": [{"role": "user", "content": "Hello"}], + "output_config": {"effort": "medium"}, "thinking": {"type": "adaptive"}, + "system": [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral", "ttl": "1h"}}]}), + )] + #[case::anthropic_beta_header( + json!({"max_tokens": 1024, "speed": "fast", "mcp_servers": [{"type": "url", "url": "https://x"}], + "messages": [{"role": "user", "content": "Hello"}]}), + &[("anthropic-beta", "context-1m-2025-08-07,claude-code-20250219,unknown-beta-2099-01-01,interleaved-thinking-2025-05-14")], + json!({"anthropic_beta": ["context-1m-2025-08-07"], "anthropic_version": "bedrock-2023-05-31", "max_tokens": 1024, + "messages": [{"role": "user", "content": "Hello"}]}), + )] + #[case::clear_thinking_becomes_adaptive_thinking( + json!({"max_tokens": 4096, "context_management": {"edits": [{"type": "clear_thinking_20251015"}, {"type": "clear_tool_uses_20250919"}]}, + "messages": [{"role": "user", "content": "Hello"}]}), + &[], + json!({"anthropic_beta": ["context-management-2025-06-27"], "anthropic_version": "bedrock-2023-05-31", + "context_management": {"edits": [{"type": "clear_tool_uses_20250919"}]}, "max_tokens": 4096, + "messages": [{"role": "user", "content": "Hello"}], "output_config": {"effort": "low"}, "thinking": {"type": "adaptive"}}), + )] + #[case::safeguards_and_effort( + json!({"max_tokens": 1024, "safeguards": [{"type": "dangerous_tool_use"}], "output_config": {"effort": "high"}, + "messages": [{"role": "user", "content": "Hello"}]}), + &[], + json!({"anthropic_beta": ["dangerous-tool-use-2026-09-03"], "anthropic_version": "bedrock-2023-05-31", "max_tokens": 1024, + "messages": [{"role": "user", "content": "Hello"}], "output_config": {"effort": "high"}, + "safeguards": [{"type": "dangerous_tool_use"}]}), + )] + fn invoke_body_matches_python( + #[case] fields: Value, + #[case] headers: &[(&str, &str)], + #[case] expected: Value, + ) { + assert_eq!( + wire(OPUS, fields, headers, opus_capabilities()), + Ok(expected) + ); + } + + #[rstest] + #[case::comma_separated("context-1m-2025-08-07, unknown-beta")] + #[case::json_array(r#"["context-1m-2025-08-07", "unknown-beta"]"#)] + fn caller_betas_parse_both_header_spellings(#[case] header: &str) { + let betas = caller_betas(&vec![("Anthropic-Beta".into(), header.into())]); + assert_eq!(betas.to_string(), "context-1m-2025-08-07"); + } + + #[rstest] + fn body_betas_from_the_client_join_the_header_betas() { + let body = wire( + OPUS, + json!({"max_tokens": 16, "anthropic_beta": ["context-management-2025-06-27"], "messages": [{"role": "user", "content": "hi"}]}), + &[("anthropic-beta", "context-1m-2025-08-07")], + opus_capabilities(), + ) + .unwrap(); + assert_eq!( + body["anthropic_beta"], + json!(["context-1m-2025-08-07", "context-management-2025-06-27"]) + ); + } + + #[rstest] + fn a_server_side_web_search_tool_fails_before_the_call() { + let error = wire( + OPUS, + json!({"max_tokens": 16, "messages": [{"role": "user", "content": "hi"}], + "tools": [{"type": "web_search_20250305", "name": "web_search"}]}), + &[], + opus_capabilities(), + ) + .unwrap_err(); + assert!(matches!( + &error, + Error::InvalidRequest(ErrorDetail::UnsupportedTool { tool, .. }) if tool == "web_search_20250305" + )); + } + + #[rstest] + #[case::ttl_supported(true, json!({"type": "ephemeral", "ttl": "5m"}))] + #[case::ttl_unsupported(false, json!({"type": "ephemeral"}))] + fn cache_control_keeps_ttl_only_where_bedrock_accepts_it( + #[case] supports_prompt_cache_ttl: bool, + #[case] expected: Value, + ) { + let body = wire( + OPUS, + json!({"max_tokens": 16, "messages": [{"role": "user", "content": [ + {"type": "text", "text": "hi", "cache_control": {"type": "ephemeral", "ttl": "5m", "scope": "global"}} + ]}]}), + &[], + MessagesModelCapabilities { + supports_prompt_cache_ttl, + ..opus_capabilities() + }, + ) + .unwrap(); + assert_eq!(body["messages"][0]["content"][0]["cache_control"], expected); + } + + #[rstest] + #[case::native( + true, + json!({"format": {"type": "json_schema", "schema": {"type": "object"}}}), + json!([{"role": "user", "content": "hi"}]), + )] + #[case::inlined( + false, + Value::Null, + json!([{"role": "user", "content": [{"type": "text", "text": "hi"}, {"type": "text", "text": "{\"type\": \"object\"}"}]}]), + )] + fn structured_output_is_native_or_inlined_into_the_last_user_turn( + #[case] supports_native_structured_output: bool, + #[case] output_config: Value, + #[case] messages: Value, + ) { + let body = wire( + OPUS, + json!({"max_tokens": 16, "messages": [{"role": "user", "content": "hi"}], + "output_config": {"format": {"type": "json_schema", "schema": {"type": "object"}}}}), + &[], + MessagesModelCapabilities { + supports_native_structured_output, + ..opus_capabilities() + }, + ) + .unwrap(); + assert_eq!( + ( + body.get("output_config").cloned().unwrap_or(Value::Null), + &body["messages"] + ), + (output_config, &messages) + ); + } + + #[rstest] + #[case::legacy_model_gets_a_budget( + "anthropic.claude-sonnet-4-20250514-v1:0", + json!({"type": "enabled", "budget_tokens": 1024}) + )] + #[case::model_without_extended_thinking_is_left_alone("anthropic.claude-3-haiku", Value::Null)] + fn clear_thinking_on_non_adaptive_models(#[case] model: &str, #[case] expected: Value) { + let body = wire( + model, + json!({"max_tokens": 4096, "context_management": {"edits": [{"type": "clear_thinking_20251015"}]}, + "messages": [{"role": "user", "content": "hi"}]}), + &[], + MessagesModelCapabilities::default(), + ) + .unwrap(); + assert_eq!( + body.get("thinking").cloned().unwrap_or(Value::Null), + expected + ); + assert_eq!(body.get("context_management"), None); + } + + #[rstest] + fn effort_keys_go_when_the_model_declares_no_output_config() { + let body = wire( + "anthropic.claude-3-haiku", + json!({"max_tokens": 16, "output_config": {"effort": "high"}, "messages": [{"role": "user", "content": "hi"}]}), + &[], + MessagesModelCapabilities::default(), + ) + .unwrap(); + assert_eq!(body.get("output_config"), None); + } + + #[tokio::test] + async fn a_full_bedrock_stream_becomes_anthropic_sse() { + let chunks = [ + json!({"type": "message_start", "message": {"id": "msg_1", "type": "message", "role": "assistant", "model": "m", "content": [], "stop_reason": null, "stop_sequence": null, "usage": {"input_tokens": 3, "output_tokens": 1}}}), + json!({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}), + json!({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hel"}}), + json!({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "lo"}}), + json!({"type": "content_block_stop", "index": 0}), + json!({"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 2}}), + json!({"type": "message_stop", "amazon-bedrock-invocationMetrics": {"inputTokenCount": 3, "outputTokenCount": 2}}), + ]; + let wire: Vec = chunks.iter().flat_map(aws_frame).collect(); + let bytes: ByteStream = futures_util::stream::iter( + wire.chunks(5) + .map(|chunk| Ok(Bytes::copy_from_slice(chunk))) + .collect::>(), + ) + .boxed(); + let events = bedrock_anthropic_messages_event_stream(bytes) + .try_collect::>() + .await + .unwrap(); + let types: Vec = events + .iter() + .map(|event| serde_json::to_value(event).unwrap()["type"].clone()) + .collect(); + assert_eq!( + Value::Array(types), + json!([ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop" + ]) + ); + let sse = String::from_utf8( + events + .iter() + .flat_map(|event| streaming::encode(event).unwrap()) + .collect(), + ) + .unwrap(); + assert!(sse.starts_with("event: message_start\ndata: ")); + assert!(sse.contains("\"text\":\"lo\"")); + } } diff --git a/litellm-rust/crates/llms/src/error.rs b/litellm-rust/crates/llms/src/error.rs index 10a4f696406..06a31222fda 100644 --- a/litellm-rust/crates/llms/src/error.rs +++ b/litellm-rust/crates/llms/src/error.rs @@ -60,6 +60,15 @@ pub enum ErrorDetail { value: String, hint: String, }, + #[error( + "{provider} does not support the server-side tool '{tool}' (model '{model}'). {guidance}" + )] + UnsupportedTool { + provider: &'static str, + tool: String, + model: String, + guidance: &'static str, + }, #[error("invalid {subject} on line {line}: {source}")] InvalidLine { subject: &'static str, diff --git a/litellm-rust/docs/customer-proxy/bedrock_messages-flow.svg b/litellm-rust/docs/customer-proxy/bedrock_messages-flow.svg new file mode 100644 index 00000000000..e8e144e67a3 --- /dev/null +++ b/litellm-rust/docs/customer-proxy/bedrock_messages-flow.svg @@ -0,0 +1,82 @@ + + + + + + + + POST /v1/messages to a bedrock/ deployment + + + Request (inference-messages prepare.rs) + + + Client body + headers + {model, stream, speed, mcp_servers, ...} + anthropic-beta + + + transform_messages_request (shared Anthropic stage) + thinking translation, system-role hoisting + + + invoke_body (Bedrock rules, Python order) + reject server-side web_search_* tools + anthropic_version = bedrock-2023-05-31 + clear_thinking edit -> thinking config, cache_control scope/ttl + structured output native or inlined, output_config keys + tool custom/name/schema fixes, context_management edits + + + request_with_headers (new trait hook) + feature betas + header betas, Bedrock policy -> anthropic_beta + + + wire_body: top-level allowlist + drops model, stream and Anthropic-only fields + + + SigV4 sign (static keys, env chain) or bearer token + POST {api_base}/model/{id}/invoke + .../invoke-with-response-stream + + + + + + + + + + Bedrock runtime + (or mockllm) + + + + Response + + + Non-streaming: Anthropic message JSON + returned to the client as is + (stream flag read from the typed request) + + + Streaming: AWS event-stream frames + prelude | :event-type=chunk | {"bytes": b64} + framer AwsEventStreamCodec, base64 + JSON decode + + + invocationMetrics -> usage + MessageStopUsagePromoter moves cache usage + from message_stop/start onto message_delta + + + Anthropic SSE to the client + event: message_start / data: {...} + + + stream + + non-stream + + + diff --git a/litellm-rust/loadtest/README.md b/litellm-rust/loadtest/README.md index 023abec5fd9..d47a604bb41 100644 --- a/litellm-rust/loadtest/README.md +++ b/litellm-rust/loadtest/README.md @@ -87,15 +87,19 @@ gateway samples and mock hit counts, and prints the summary. `LITELLM_UNREADY_DELAY_SECONDS` (default 0), then open requests and streams get up to `LITELLM_DRAIN_SECONDS` (default 900, the NLB deregistration delay) to finish +## Bedrock deployments + +`genconfig` serves half of each messages group through Bedrock invoke by default +(`-bedrock-fraction 0.5`, `bedrock_fraction` in Terraform; 0 for Anthropic only). Those deployments +carry static dummy AWS keys, so every request is SigV4-signed, and mockllm answers +`/model//invoke` and `/invoke-with-response-stream` with AWS event-stream framing that the +gateway re-encodes as Anthropic SSE. Bedrock hits show up as `bedrock--` in the mock's `/stats`. + ## Until the gateway gains these features - Virtual keys: the Rust gateway only accepts the master key today, so pass `-key sk-loadtest-master` (`KEY=` for `cloud-run.sh`). Once keys are read from Postgres, drop it and set `create_postgres`. - `/v1/messages` auth: `-messages-auth x-api-key` sends the key the way the Anthropic SDK does. The gateway reads only `Authorization: Bearer` today. -- Bedrock: `/v1/messages` to a `bedrock/` deployment returns 501 ("Bedrock invoke messages request - shaping" is unimplemented in Rust). mockllm already serves Bedrock's `/invoke` and - `/invoke-with-response-stream` (AWS event-stream framing), so set `-bedrock-fraction 0.5` - (`bedrock_fraction` in Terraform) once that lands. - Routing: deployments sharing a model name overwrite each other, so all of a group's traffic goes to its last deployment (visible in the mock's `/stats`). diff --git a/litellm-rust/loadtest/cmd/genconfig/main.go b/litellm-rust/loadtest/cmd/genconfig/main.go index fc2a8b385db..69285470ca8 100644 --- a/litellm-rust/loadtest/cmd/genconfig/main.go +++ b/litellm-rust/loadtest/cmd/genconfig/main.go @@ -27,7 +27,7 @@ func main() { chatModel := flag.String("chat-model", "gpt-6-astra", "upstream model for chat deployments; groups are named -0, -1, ...") messagesModel := flag.String("messages-model", "claude-opus-5-5", "upstream model for messages deployments; groups are named -0, ...") bedrockModel := flag.String("bedrock-model", "us.anthropic.claude-opus-5-5", "Bedrock model id for the Bedrock share of messages deployments") - bedrockFraction := flag.Float64("bedrock-fraction", 0, "share of each messages group's deployments served through Bedrock instead of Anthropic (0.5 once the gateway shapes Bedrock invoke requests)") + bedrockFraction := flag.Float64("bedrock-fraction", 0.5, "share of each messages group's deployments served through Bedrock invoke (SigV4-signed, AWS event-stream responses) instead of the Anthropic API; 0 for Anthropic only") chatGroups := flag.Int("chat-groups", 4, "chat model groups") messagesGroups := flag.Int("messages-groups", 4, "messages model groups") deployments := flag.Int("deployments", 4, "deployments per model group") @@ -50,6 +50,14 @@ func main() { group, model, apiBase) n++ } + // Bedrock deployments carry static AWS keys instead of an api_key (which Bedrock reads as a + // bearer token), so every request goes through SigV4 signing as it does with IAM credentials. + bedrockDeployment := func(group, id, model, apiBase string) { + fmt.Fprintf(&cfg, " - model_name: %s\n litellm_params:\n model: %s\n api_base: %s\n"+ + " aws_access_key_id: mock-loadtest-access-key\n aws_secret_access_key: mock-loadtest-secret\n aws_region_name: us-east-1\n", + group, model, apiBase) + n++ + } for g := range *chatGroups { for i := range *deployments { base := strings.TrimRight(bases[n%len(bases)], "/") @@ -64,7 +72,7 @@ func main() { group := fmt.Sprintf("%s-%d", *messagesModel, g) if i < bedrockPerGroup { id := fmt.Sprintf("bedrock-%d-%d", g, i) - deployment(group, id, "bedrock/"+*bedrockModel, fmt.Sprintf("%s/d/%s", base, id)) + bedrockDeployment(group, id, "bedrock/"+*bedrockModel, fmt.Sprintf("%s/d/%s", base, id)) continue } id := fmt.Sprintf("msg-%d-%d", g, i) diff --git a/litellm-rust/loadtest/terraform/aws/variables.tf b/litellm-rust/loadtest/terraform/aws/variables.tf index e9bf6765af8..99e3fbed580 100644 --- a/litellm-rust/loadtest/terraform/aws/variables.tf +++ b/litellm-rust/loadtest/terraform/aws/variables.tf @@ -81,9 +81,9 @@ variable "messages_model" { } variable "bedrock_fraction" { - description = "Share of each Opus group's deployments served through Bedrock. Raise to 0.5 once the Rust gateway shapes Bedrock invoke requests (it returns 501 today)." + description = "Share of each Opus group's deployments served through Bedrock invoke instead of the Anthropic API (0 for Anthropic only)." type = number - default = 0 + default = 0.5 } variable "chat_groups" {