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.
This commit is contained in:
Nate Armstrong 2026-10-16 12:43:01 -07:00 • committed by GitHub
parent bce2ced779
commit 99997d1e3a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 1250 additions and 20 deletions

View file

@ -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",

View file

@ -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

View file

@ -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(

View file

@ -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(),

View file

@ -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,

View file

@ -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<u8> {
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<u8> = 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::<Vec<_>>().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);
}

View file

@ -6,3 +6,6 @@ mod edenai;
#[path = "live/vertex_ai.rs"]
mod vertex_ai;
#[path = "live/bedrock.rs"]
mod bedrock;

View file

@ -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<String> {
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<String>) {
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<String>) {
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<String>, #[case] stream: bool) {
if model.is_none() {
return;
}
super::support::tool_round_trip(route, PROVIDER, stream, json!({})).await;
}

View file

@ -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";

View file

@ -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),
}
);

View file

@ -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(),
}
}

View file

@ -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 {

View file

@ -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

View file

@ -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<MessagesRequest, Error> {
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<StreamDecoder> {
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<String, Value>,
model: &str,
context: &MessagesTransformContext,
) -> Result<Map<String, Value>, 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<String, Value>) -> impl Iterator<Item = &Map<String, Value>> {
body.get("tools")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter_map(Value::as_object)
}
fn tools_mut(body: &mut Map<String, Value>) -> impl Iterator<Item = &mut Map<String, Value>> {
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<String, Value>) -> 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<String, Value>, 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<String, Value>,
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<String, Value> = 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<String, Value>,
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<String, Value>, 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<String, Value>, 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<String, Value>, 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<String, Value>, 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::<Vec<_>>()
.join(", ")
),
Value::Object(fields) => format!(
"{{{}}}",
fields
.iter()
.map(|(key, value)| format!(
"{}: {}",
Value::from(key.as_str()),
python_json_dumps(value)
))
.collect::<Vec<_>>()
.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<String, Value>,
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<String, Value>) {
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<String, Value>) {
let Some(Value::Object(management)) = body.get_mut("context_management") else {
return;
};
let retained: Vec<Value> = 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::<Vec<Value>>(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::<Vec<_>>(),
}
})
.filter(|beta| !beta.is_empty())
.filter_map(|beta| {
beta.parse::<AnthropicBeta>()
.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::<AnthropicBeta>().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<Value, Error> {
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<u8> = 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::<Vec<_>>(),
)
.boxed();
let events = bedrock_anthropic_messages_event_stream(bytes)
.try_collect::<Vec<_>>()
.await
.unwrap();
let types: Vec<Value> = 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\""));
}
}

View file

@ -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,

View file

@ -0,0 +1,82 @@
<svg xmlns="http://www.w3.org/2000/svg" width="1000" height="640" viewBox="0 0 1000 640" font-family="Helvetica, Arial, sans-serif" font-size="13">
<defs>
<marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7" markerHeight="7" orient="auto-start-reverse">
<path d="M0,0 L10,5 L0,10 z" fill="#333"/>
</marker>
</defs>
<rect x="0" y="0" width="1000" height="640" fill="#ffffff"/>
<text x="500" y="30" text-anchor="middle" font-size="18" font-weight="bold" fill="#111">POST /v1/messages to a bedrock/ deployment</text>
<!-- request column -->
<text x="250" y="62" text-anchor="middle" font-size="14" font-weight="bold" fill="#1f4e79">Request (inference-messages prepare.rs)</text>
<rect x="40" y="75" width="420" height="58" rx="6" fill="#eef4fb" stroke="#1f4e79"/>
<text x="56" y="97" fill="#111">Client body + headers</text>
<text x="56" y="117" fill="#555" font-family="monospace" font-size="12">{model, stream, speed, mcp_servers, ...} + anthropic-beta</text>
<rect x="40" y="153" width="420" height="58" rx="6" fill="#eef4fb" stroke="#1f4e79"/>
<text x="56" y="175" fill="#111">transform_messages_request (shared Anthropic stage)</text>
<text x="56" y="195" fill="#555" font-size="12">thinking translation, system-role hoisting</text>
<rect x="40" y="231" width="420" height="132" rx="6" fill="#fff4e5" stroke="#b35c00"/>
<text x="56" y="253" fill="#111" font-weight="bold">invoke_body (Bedrock rules, Python order)</text>
<text x="56" y="274" fill="#333" font-size="12">reject server-side web_search_* tools</text>
<text x="56" y="292" fill="#333" font-size="12">anthropic_version = bedrock-2023-05-31</text>
<text x="56" y="310" fill="#333" font-size="12">clear_thinking edit -&gt; thinking config, cache_control scope/ttl</text>
<text x="56" y="328" fill="#333" font-size="12">structured output native or inlined, output_config keys</text>
<text x="56" y="346" fill="#333" font-size="12">tool custom/name/schema fixes, context_management edits</text>
<rect x="40" y="383" width="420" height="58" rx="6" fill="#fff4e5" stroke="#b35c00"/>
<text x="56" y="405" fill="#111">request_with_headers (new trait hook)</text>
<text x="56" y="425" fill="#555" font-size="12">feature betas + header betas, Bedrock policy -&gt; anthropic_beta</text>
<rect x="40" y="461" width="420" height="58" rx="6" fill="#fff4e5" stroke="#b35c00"/>
<text x="56" y="483" fill="#111">wire_body: top-level allowlist</text>
<text x="56" y="503" fill="#555" font-size="12">drops model, stream and Anthropic-only fields</text>
<rect x="40" y="539" width="420" height="72" rx="6" fill="#eef4fb" stroke="#1f4e79"/>
<text x="56" y="561" fill="#111">SigV4 sign (static keys, env chain) or bearer token</text>
<text x="56" y="581" fill="#555" font-family="monospace" font-size="12">POST {api_base}/model/{id}/invoke</text>
<text x="56" y="599" fill="#555" font-family="monospace" font-size="12"> .../invoke-with-response-stream</text>
<line x1="250" y1="133" x2="250" y2="151" stroke="#333" marker-end="url(#arrow)"/>
<line x1="250" y1="211" x2="250" y2="229" stroke="#333" marker-end="url(#arrow)"/>
<line x1="250" y1="363" x2="250" y2="381" stroke="#333" marker-end="url(#arrow)"/>
<line x1="250" y1="441" x2="250" y2="459" stroke="#333" marker-end="url(#arrow)"/>
<line x1="250" y1="519" x2="250" y2="537" stroke="#333" marker-end="url(#arrow)"/>
<!-- upstream -->
<rect x="540" y="539" width="180" height="72" rx="6" fill="#f2f2f2" stroke="#555"/>
<text x="630" y="570" text-anchor="middle" fill="#111" font-weight="bold">Bedrock runtime</text>
<text x="630" y="590" text-anchor="middle" fill="#555" font-size="12">(or mockllm)</text>
<line x1="460" y1="575" x2="538" y2="575" stroke="#333" marker-end="url(#arrow)"/>
<!-- response column -->
<text x="790" y="62" text-anchor="middle" font-size="14" font-weight="bold" fill="#2e6b30">Response</text>
<rect x="600" y="75" width="380" height="80" rx="6" fill="#eef7ee" stroke="#2e6b30"/>
<text x="616" y="97" fill="#111">Non-streaming: Anthropic message JSON</text>
<text x="616" y="117" fill="#555" font-size="12">returned to the client as is</text>
<text x="616" y="137" fill="#555" font-size="12">(stream flag read from the typed request)</text>
<rect x="600" y="175" width="380" height="80" rx="6" fill="#eef7ee" stroke="#2e6b30"/>
<text x="616" y="197" fill="#111">Streaming: AWS event-stream frames</text>
<text x="616" y="217" fill="#555" font-family="monospace" font-size="12">prelude | :event-type=chunk | {"bytes": b64}</text>
<text x="616" y="237" fill="#555" font-size="12">framer AwsEventStreamCodec, base64 + JSON decode</text>
<rect x="600" y="275" width="380" height="80" rx="6" fill="#eef7ee" stroke="#2e6b30"/>
<text x="616" y="297" fill="#111">invocationMetrics -&gt; usage</text>
<text x="616" y="317" fill="#555" font-size="12">MessageStopUsagePromoter moves cache usage</text>
<text x="616" y="337" fill="#555" font-size="12">from message_stop/start onto message_delta</text>
<rect x="600" y="375" width="380" height="58" rx="6" fill="#eef7ee" stroke="#2e6b30"/>
<text x="616" y="397" fill="#111">Anthropic SSE to the client</text>
<text x="616" y="417" fill="#555" font-family="monospace" font-size="12">event: message_start / data: {...}</text>
<path d="M 720 595 L 990 595 L 990 215 L 983 215" fill="none" stroke="#333" marker-end="url(#arrow)"/>
<text x="840" y="588" fill="#555" font-size="12">stream</text>
<path d="M 560 539 L 560 115 L 597 115" fill="none" stroke="#333" stroke-dasharray="4 3" marker-end="url(#arrow)"/>
<text x="568" y="480" fill="#555" font-size="12">non-stream</text>
<line x1="790" y1="255" x2="790" y2="273" stroke="#333" marker-end="url(#arrow)"/>
<line x1="790" y1="355" x2="790" y2="373" stroke="#333" marker-end="url(#arrow)"/>
</svg>

After

Width:  |  Height:  |  Size: 5.7 KiB

View file

@ -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/<id>/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-<g>-<i>` 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`).

View file

@ -27,7 +27,7 @@ func main() {
chatModel := flag.String("chat-model", "gpt-6-astra", "upstream model for chat deployments; groups are named <model>-0, <model>-1, ...")
messagesModel := flag.String("messages-model", "claude-opus-5-5", "upstream model for messages deployments; groups are named <model>-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)

View file

@ -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" {