diff --git a/litellm-rust/crates/inference-messages/src/handler.rs b/litellm-rust/crates/inference-messages/src/handler.rs index c7a7d93bf98..5acf0bdda8f 100644 --- a/litellm-rust/crates/inference-messages/src/handler.rs +++ b/litellm-rust/crates/inference-messages/src/handler.rs @@ -44,6 +44,7 @@ struct ProviderCall { signer: Option, timeout: Option, stream: bool, + whole_sse_frames: bool, } impl ProviderCall { @@ -158,14 +159,6 @@ async fn provider_call( &validated.headers, shaping.settings.drop_params, )?; - let environment = ValidatedEnvironment { - headers: config.request_headers( - with_python_default_headers(validated.headers, config.default_headers()), - &body, - ), - auth: validated.auth, - }; - let url = config.get_complete_url( api_base.as_deref(), &body.model, @@ -173,6 +166,16 @@ async fn provider_call( body.params.stream == Some(true), &env_lookup, )?; + let environment = ValidatedEnvironment { + headers: config.headers_for_url( + config.request_headers( + with_python_default_headers(validated.headers, config.default_headers()), + &body, + ), + &url, + ), + auth: validated.auth, + }; let request_context = RequestContext { model: body.model.clone(), @@ -217,6 +220,7 @@ async fn provider_call( signer: authenticated.signer, timeout: timeout.or_else(|| shaping.settings.request_timeout()), stream, + whole_sse_frames: shaping.settings.whole_sse_frames, }) } @@ -307,6 +311,7 @@ async fn call_provider( signer, timeout, stream, + whole_sse_frames, } = request; let provider_name = provider.as_str(); log_request_body(provider_name, stream, &wire.body); @@ -328,12 +333,13 @@ async fn call_provider( } let config = provider.config(); if stream { - return Ok(streaming_response( - response, - config.stream_decoder(), - provider_name, - timeout, - )); + let streamed = + streaming_response(response, config.stream_decoder(), provider_name, timeout); + return Ok(if whole_sse_frames { + framed(streamed) + } else { + streamed + }); } let head = ResponseHead { status: response.status().as_u16(), @@ -469,6 +475,52 @@ fn streaming_response( } } +fn framed(response: MessagesCallResponse) -> MessagesCallResponse { + let MessagesCallResponse::Stream { head, chunks } = response else { + return response; + }; + MessagesCallResponse::Stream { + head, + chunks: whole_sse_frames(chunks), + } +} + +fn whole_sse_frames( + chunks: BoxStream<'static, Result>, +) -> BoxStream<'static, Result> { + futures_util::stream::try_unfold( + (Some(chunks), Vec::new()), + |(chunks, mut pending)| async move { + let Some(mut chunks) = chunks else { + return Ok(None); + }; + while let Some(chunk) = chunks.try_next().await? { + pending.extend_from_slice(&chunk); + let end = complete_frames_end(&pending); + if end > 0 { + let rest = pending.split_off(end); + return Ok(Some((Bytes::from(pending), (Some(chunks), rest)))); + } + } + Ok((!pending.is_empty()).then(|| (Bytes::from(pending), (None, Vec::new())))) + }, + ) + .boxed() +} + +fn complete_frames_end(pending: &[u8]) -> usize { + [b"\r\n\r\n".as_slice(), b"\n\n", b"\r\r"] + .iter() + .filter_map(|delimiter| { + pending + .windows(delimiter.len()) + .rposition(|window| window == *delimiter) + .map(|at| at + delimiter.len()) + }) + .max() + .unwrap_or(0) +} + fn decoded_chunks( bytes: ByteStream, decode: StreamDecoder, @@ -1092,4 +1144,48 @@ mod tests { } assert!(chunks.next().await.is_none()); } + + #[rstest] + #[case::no_delimiter(b"data: a".as_slice(), 0)] + #[case::lf(b"data: a\n\ndata: b", 9)] + #[case::crlf(b"data: a\r\n\r\ndata: b", 11)] + #[case::cr(b"data: a\r\rdata: b", 9)] + #[case::last_of_several(b"data: a\n\ndata: b\n\ndata: c", 18)] + fn complete_frames_end_is_after_the_last_delimiter(#[case] pending: &[u8], #[case] end: usize) { + assert_eq!(complete_frames_end(pending), end); + } + + #[rstest] + #[case::split_frame( + vec![&b"data: a\n"[..], b"\ndata: ", b"b\n\n"], + vec![&b"data: a\n\n"[..], b"data: b\n\n"] + )] + #[case::several_frames_in_one_chunk( + vec![&b"data: a\n\ndata: b\n\nda"[..], b"ta: c\n\n"], + vec![&b"data: a\n\ndata: b\n\n"[..], b"data: c\n\n"] + )] + #[case::unterminated_tail_is_flushed( + vec![&b"data: a\n\nda"[..], b"ta: b"], + vec![&b"data: a\n\n"[..], b"data: b"] + )] + #[tokio::test] + async fn whole_sse_frames_regroups_chunks_at_frame_boundaries( + #[case] chunks: Vec<&'static [u8]>, + #[case] expected: Vec<&'static [u8]>, + ) { + let upstream = futures_util::stream::iter( + chunks + .into_iter() + .map(|chunk| Ok::<_, Error>(Bytes::from_static(chunk))), + ) + .boxed(); + let framed: Vec = whole_sse_frames(upstream).try_collect().await.unwrap(); + assert_eq!( + framed, + expected + .into_iter() + .map(Bytes::from_static) + .collect::>() + ); + } } diff --git a/litellm-rust/crates/inference-messages/src/lib.rs b/litellm-rust/crates/inference-messages/src/lib.rs index 25ebeaacb49..23dd8f4a920 100644 --- a/litellm-rust/crates/inference-messages/src/lib.rs +++ b/litellm-rust/crates/inference-messages/src/lib.rs @@ -75,6 +75,11 @@ impl MessagesRoute { call: MessagesCall, context: CallContext<'_, impl Interceptors>, ) -> Result { + if let Some(mock) = &call.shaping.settings.mock_response { + return Ok(MessagesCallResponse::Complete(MessagesBody::Unchecked( + mock.clone(), + ))); + } litellm_inference::diagnostic::call(async { let provider = resolve_llm_provider( &call.body.model, diff --git a/litellm-rust/crates/inference-messages/src/types.rs b/litellm-rust/crates/inference-messages/src/types.rs index 772ea6f7292..b1ac375f7ab 100644 --- a/litellm-rust/crates/inference-messages/src/types.rs +++ b/litellm-rust/crates/inference-messages/src/types.rs @@ -71,6 +71,10 @@ pub struct MessagesSettings { pub additional_drop_params: Vec, #[serde(default)] pub request_timeout: Option, + #[serde(default)] + pub mock_response: Option>, + #[serde(default)] + pub whole_sse_frames: bool, } impl MessagesSettings { @@ -148,6 +152,7 @@ mod tests { reasoning_auto_summary: true, additional_drop_params: vec!["metadata.user_id".to_string(), "thinking".to_string()], request_timeout: Some(1800.0), + ..MessagesSettings::default() }, capabilities: MessagesModelCapabilities { supports_reasoning: true, diff --git a/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs b/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs index 18789e5b2be..ab622f1a1ab 100644 --- a/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs @@ -1,10 +1,12 @@ use litellm_auth::CredentialPlacement; +use litellm_http::request::{with_header, without_headers}; use litellm_llms_types::{ formats::messages::{ ContentBlock, ContentBlockType, ContextEdit, ContextManagement, Message, - MessagesOptionalParams, MessagesRequest, Speed, ThinkingConfig, ThinkingDisplay, + MessagesOptionalParams, MessagesRequest, MessagesTool, Speed, ThinkingConfig, + ThinkingDisplay, }, - providers::anthropic::{AnthropicBeta, BetaProvider, BetaSet, DEFAULT_HEADERS}, + providers::anthropic::{AnthropicBeta, BETA_HEADER, BetaProvider, BetaSet, DEFAULT_HEADERS}, recognized::Recognized, }; use litellm_router_types::LitellmParams; @@ -16,10 +18,10 @@ use crate::{ Error, anthropic::common_utils::{ ANTHROPIC_API_BASE_ENV, ANTHROPIC_API_KEY_ENV, ANTHROPIC_AUTH_TOKEN_ENV, - ANTHROPIC_BASE_URL_ENV, OauthHandling, complete_anthropic_url, get_auth_header, - has_advisor_tool, has_anthropic_credential, is_tool_search_used, merge_beta_headers, - optionally_handle_anthropic_oauth, requires_native_compaction_beta, strip_advisor_blocks, - strip_encrypted_reasoning_blocks, + ANTHROPIC_BASE_URL_ENV, OauthHandling, complete_anthropic_url, existing_betas, + get_auth_header, has_advisor_tool, has_anthropic_credential, is_tool_search_used, + merge_beta_headers, optionally_handle_anthropic_oauth, requires_native_compaction_beta, + strip_advisor_blocks, strip_encrypted_reasoning_blocks, }, base_llm::{ auth::AuthScheme, @@ -57,6 +59,7 @@ impl BaseMessagesConfig for AnthropicMessagesConfig { context: &MessagesTransformContext, ) -> Result { transform_messages_request(request, &context.keeping_display_updates()) + .map(|request| tools_without_custom_field(request, context.drop_params)) } fn secret_names(&self) -> &'static [&'static str] { @@ -116,6 +119,113 @@ impl BaseMessagesConfig for AnthropicMessagesConfig { provider_feature_betas(request, BetaProvider::Anthropic), ) } + + fn headers_for_url(&self, headers: Headers, url: &str) -> Headers { + if !is_foundry_url(url) { + return headers; + } + let betas = existing_betas(&headers); + if betas.is_empty() { + return headers; + } + let kept: BetaSet = betas + .into_iter() + .filter_map(|beta| match beta { + AnthropicBeta::Other(_) => Some(beta), + known => known.on(BetaProvider::AzureAi), + }) + .collect(); + if kept.is_empty() { + return without_headers(headers, &[BETA_HEADER]); + } + with_header(headers, BETA_HEADER, kept.to_string()) + } +} + +const FOUNDRY_HOST_SUFFIXES: [&str; 3] = [ + "services.ai.azure.com", + "openai.azure.com", + "cognitiveservices.azure.com", +]; + +fn is_foundry_url(url: &str) -> bool { + let absolute = if url.contains("://") { + url.to_string() + } else { + format!("https://{url}") + }; + url::Url::parse(&absolute) + .ok() + .and_then(|parsed| parsed.host_str().map(str::to_ascii_lowercase)) + .is_some_and(|host| { + FOUNDRY_HOST_SUFFIXES + .iter() + .any(|suffix| host == *suffix || host.ends_with(&format!(".{suffix}"))) + }) +} + +fn tools_without_custom_field(request: MessagesRequest, drop_custom: bool) -> MessagesRequest { + let hoist = request.model.to_lowercase().contains("claude"); + let tools = request.params.tools.map(|tools| { + tools + .into_iter() + .map(|tool| tool_without_custom_field(tool, hoist, drop_custom)) + .collect() + }); + MessagesRequest { + params: MessagesOptionalParams { + tools, + ..request.params + }, + ..request + } +} + +fn tool_without_custom_field( + tool: Recognized, + hoist: bool, + drop_custom: bool, +) -> Recognized { + let Ok(Value::Object(fields)) = serde_json::to_value(&tool) else { + return tool; + }; + let Some(custom) = fields.get("custom").cloned() else { + return tool; + }; + let defer_loading = hoist + .then(|| hoistable_defer_loading(&fields, &custom)) + .flatten(); + let without_custom = fields + .iter() + .filter(|(key, _)| *key != "custom") + .map(|(key, value)| (key.clone(), value.clone())); + let rebuilt: Map = match defer_loading { + None if drop_custom => without_custom.collect(), + None => return tool, + Some(defer_loading) => { + let unhoisted: Map = custom + .as_object() + .into_iter() + .flatten() + .filter(|(key, _)| *key != "defer_loading") + .map(|(key, value)| (key.clone(), value.clone())) + .collect(); + let kept_custom = (!unhoisted.is_empty() && !drop_custom) + .then(|| ("custom".to_string(), Value::Object(unhoisted))); + without_custom + .chain([("defer_loading".to_string(), Value::Bool(defer_loading))]) + .chain(kept_custom) + .collect() + } + }; + serde_json::from_value(Value::Object(rebuilt)).unwrap_or(tool) +} + +fn hoistable_defer_loading(tool: &Map, custom: &Value) -> Option { + if tool.contains_key("defer_loading") { + return None; + } + custom.as_object()?.get("defer_loading")?.as_bool() } pub(crate) fn transform_messages_request( @@ -437,6 +547,7 @@ mod tests { use super::*; use crate::anthropic::common_utils::ENCRYPTED_REASONING_SIGNATURE_PREFIX; + use litellm_http::request::header_values; type Env = &'static [(&'static str, &'static str)]; @@ -1352,4 +1463,126 @@ mod tests { .collect(); assert_eq!(undeclared, Vec::<&String>::new()); } + + const FOUNDRY: &str = "https://team.services.ai.azure.com/anthropic/v1/messages"; + + #[rstest] + #[case::drops_a_beta_foundry_lacks( + FOUNDRY, + "advisor-tool-2026-03-01,context-1m-2025-08-07", + Some("context-1m-2025-08-07") + )] + #[case::forwards_unknown_betas_sorted( + FOUNDRY, + "zzz,files-api-2025-04-14,aaa", + Some("aaa,files-api-2025-04-14,zzz") + )] + #[case::removes_the_header_when_nothing_is_left(FOUNDRY, "advisor-tool-2026-03-01", None)] + #[case::cognitive_services_host( + "team.cognitiveservices.azure.com/anthropic", + "advisor-tool-2026-03-01", + None + )] + #[case::anthropic_host_is_untouched( + "https://api.anthropic.com/v1/messages", + "advisor-tool-2026-03-01", + Some("advisor-tool-2026-03-01") + )] + #[case::lookalike_host_is_untouched( + "https://notservices.ai.azure.com.example/v1/messages", + "advisor-tool-2026-03-01", + Some("advisor-tool-2026-03-01") + )] + fn foundry_urls_filter_betas_through_the_azure_ai_row( + #[case] url: &str, + #[case] betas: &str, + #[case] expected: Option<&str>, + ) { + let filtered = ANTHROPIC_MESSAGES_CONFIG + .headers_for_url(headers(&[("x-a", "1"), (BETA_HEADER, betas)]), url); + assert_eq!( + header_values(&filtered, BETA_HEADER).collect::>(), + expected.into_iter().collect::>() + ); + assert_eq!(header_values(&filtered, "x-a").collect::>(), ["1"]); + } + + #[rstest] + fn foundry_url_without_betas_keeps_the_headers() { + let sent = headers(&[("x-a", "1")]); + assert_eq!( + ANTHROPIC_MESSAGES_CONFIG.headers_for_url(sent.clone(), FOUNDRY), + sent + ); + } + + const WEATHER: &str = "get_weather"; + + fn tool(fields: Value) -> Value { + merged( + json!({"name": WEATHER, "input_schema": {"type": "object"}}), + fields, + ) + } + + fn tools_after(model: &str, tools: Value, drop_custom: bool) -> Value { + let request = request(json!({"model": model, "tools": [tools]})); + serde_json::to_value(tools_without_custom_field(request, drop_custom)).unwrap()["tools"][0] + .clone() + } + + #[rstest] + #[case::hoists_defer_loading( + "claude-sonnet-4-5", + json!({"custom": {"defer_loading": true}}), + false, + json!({"defer_loading": true}) + )] + #[case::keeps_other_custom_keys( + "claude-sonnet-4-5", + json!({"custom": {"defer_loading": false, "note": "kept"}}), + false, + json!({"defer_loading": false, "custom": {"note": "kept"}}) + )] + #[case::drops_other_custom_keys_with_drop_params( + "claude-sonnet-4-5", + json!({"custom": {"defer_loading": true, "note": "dropped"}}), + true, + json!({"defer_loading": true}) + )] + #[case::keeps_an_opaque_custom( + "claude-sonnet-4-5", + json!({"custom": "opaque"}), + false, + json!({"custom": "opaque"}) + )] + #[case::drops_an_opaque_custom_with_drop_params( + "claude-sonnet-4-5", + json!({"custom": "opaque"}), + true, + json!({}) + )] + #[case::own_defer_loading_wins( + "claude-sonnet-4-5", + json!({"defer_loading": false, "custom": {"defer_loading": true}}), + false, + json!({"defer_loading": false, "custom": {"defer_loading": true}}) + )] + #[case::non_claude_model_is_not_hoisted( + "other-model", + json!({"custom": {"defer_loading": true}}), + false, + json!({"custom": {"defer_loading": true}}) + )] + fn tool_custom_field_follows_the_python_handler( + #[case] model: &str, + #[case] fields: Value, + #[case] drop_custom: bool, + #[case] expected: Value, + ) { + assert_eq!( + tools_after(model, tool(fields), drop_custom), + tool(expected) + ); + } } 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 744ce89afdb..1cb42dfb92f 100644 --- a/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs @@ -77,6 +77,10 @@ pub trait BaseMessagesConfig: Sync { headers } + fn headers_for_url(&self, headers: Headers, _url: &str) -> Headers { + headers + } + fn request_body( &self, request: MessagesRequest, diff --git a/litellm-rust/crates/python-bridge/src/http.rs b/litellm-rust/crates/python-bridge/src/http.rs index c6a8b98747c..5a828d1465f 100644 --- a/litellm-rust/crates/python-bridge/src/http.rs +++ b/litellm-rust/crates/python-bridge/src/http.rs @@ -129,6 +129,22 @@ pub(crate) fn settings_config( Ok(resolved(py, settings)) } +#[pyfunction] +pub(crate) fn unsupported_tls_settings(py: Python<'_>) -> PyResult> { + let settings = HttpSettings::from_layers([ + HttpSettingsLayer::from_environment(&ProcessEnvironment), + match PythonSettings::Http.read_or_unset(py)? { + Some(snapshot) => configured(&snapshot)?, + None => HttpSettingsLayer::default(), + }, + ]); + Ok(Resolution::from(&settings) + .unsupported + .iter() + .map(ToString::to_string) + .collect()) +} + /// The client config `settings` resolve to, warning once per process about each setting the client cannot apply fn resolved(py: Python<'_>, settings: HttpSettings) -> HttpClientConfig { let resolution = diff --git a/litellm-rust/crates/python-bridge/src/lib.rs b/litellm-rust/crates/python-bridge/src/lib.rs index 7edf13035c1..45ea95560e1 100644 --- a/litellm-rust/crates/python-bridge/src/lib.rs +++ b/litellm-rust/crates/python-bridge/src/lib.rs @@ -35,6 +35,8 @@ mod _native { #[pymodule_export] use crate::errors::{RustBridgeDeclined, RustUpstreamError}; #[pymodule_export] + use crate::http::unsupported_tls_settings; + #[pymodule_export] use crate::logger::NativeDiagnosticProcessor; #[pymodule_export] use crate::model_cost::{ @@ -191,6 +193,7 @@ mod tests { "gil_stats", "process_state_started", "reserve_process_for_forking", + "unsupported_tls_settings", ]; #[cfg(feature = "huggingface")] expected.push("HuggingFaceEncoding"); diff --git a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs index 2a5f79ca393..ce117a039ea 100644 --- a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs +++ b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs @@ -67,11 +67,20 @@ fn project_litellm_params<'py>( Ok(litellm_params(fields.into_iter().chain(folded).collect())) } +#[derive(serde::Deserialize)] +struct CallSettings { + model: String, + mock_response: Option>, + stream_cost_model: Option, + provider_affinity_headers: Map, +} + /// The Python side of the Messages route: projects the prepared arguments and builds the /// public response, chunks and exceptions. pub(super) struct MessagesPythonHost { view: RequestView, cache: PythonCache, + stream_cost_model: Option, } impl MessagesPythonHost { @@ -79,19 +88,29 @@ impl MessagesPythonHost { Self { view, cache: PythonCache::new(asynchronous), + stream_cost_model: None, } } + fn call_settings(&self, py: Python<'_>) -> PyResult { + from_py( + &py.import(ROUTE_HOST_MODULE)? + .getattr("call_settings")? + .call1((self.view.request(py)?,))?, + ) + } + fn projection( &self, py: Python<'_>, arguments: &Bound<'_, PyDict>, + settings: CallSettings, ) -> PyResult> { let argument = |name: &str| self.view.argument(py, arguments, name); let string = |name: &str| -> PyResult> { argument(name)?.map(|value| value.extract()).transpose() }; - let model = string("model")?.ok_or(ArgumentError::Required("model"))?; + let model = settings.model; let messages = argument("messages")?.ok_or(ArgumentError::Required("messages"))?; let fields = self.view.body_params(py, arguments, ROUTE_HOST_MODULE)?; let body = [ @@ -118,7 +137,18 @@ impl MessagesPythonHost { .extract()?, }; let shaping = self.shaping(py, &model, custom_llm_provider.as_deref(), arguments)?; - let extra_headers = self.merged_headers(py, arguments)?; + let shaping = MessagesShaping { + settings: MessagesSettings { + mock_response: settings.mock_response, + whole_sse_frames: settings.stream_cost_model.is_some(), + ..shaping.settings + }, + ..shaping + }; + let extra_headers = merge_headers( + self.merged_headers(py, arguments)?, + Some(settings.provider_affinity_headers), + ); let provider_specific_header = self.provider_specific_header(py, arguments)?; let litellm_params = project_litellm_params(argument, |name| module_global(py, name))?; Ok(messages_body(body).and_then(|body| { @@ -230,7 +260,9 @@ impl PythonBinding for MessagesPythonHost { let selection = crate::cache::configure(&mut self.cache, py, arguments, "anthropic_messages") .map_err(InvokeError::Python)?; - self.projection(py, arguments) + let settings = self.call_settings(py).map_err(InvokeError::Python)?; + self.stream_cost_model = settings.stream_cost_model.clone(); + self.projection(py, arguments, settings) .map_err(|error| InvokeError::Python(self.map_failure(py, error)))? .map_err(InvokeError::Native) .map(|request| (request, selection)) @@ -260,7 +292,14 @@ impl PythonBinding for MessagesPythonHost { } fn encode_chunk(&mut self, py: Python<'_>, chunk: Bytes) -> PyResult> { - Ok(PyBytes::new(py, &chunk).into_any().unbind()) + let frames = PyBytes::new(py, &chunk).into_any(); + let Some(model) = &self.stream_cost_model else { + return Ok(frames.unbind()); + }; + py.import(ROUTE_HOST_MODULE)? + .getattr("inject_stream_cost")? + .call1((frames, model, self.view.request(py)?)) + .map(Bound::unbind) } fn map_error(&self, py: Python<'_>, error: Error) -> PyResult { diff --git a/litellm/__init__.py b/litellm/__init__.py index 8a73f3447b8..049e6a44dbd 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -27,7 +27,7 @@ from typing import ( from collections.abc import Mapping from litellm.types.integrations.datadog import DatadogInitParams from litellm.types.integrations.newrelic import NewRelicInitParams -from litellm.litellm_core_utils.core_helpers import drop_params_env_flag +from litellm.litellm_core_utils.core_helpers import drop_params_env_flag, env_flag from litellm.types.integrations.pointfive import PointFiveInitParams from litellm.types.integrations.zerobus import ZerobusInitParams from litellm._logging import ( @@ -227,7 +227,7 @@ token: Optional[str] = ( ) max_tokens: int = DEFAULT_MAX_TOKENS # OpenAI Defaults drop_params = drop_params_env_flag(os.environ, verbose_logger) -modify_params = bool(os.getenv("LITELLM_MODIFY_PARAMS", False)) +modify_params = env_flag(os.environ, "LITELLM_MODIFY_PARAMS", verbose_logger) bedrock_neutralize_orphaned_tool_blocks: bool = True use_chat_completions_url_for_anthropic_messages: bool = bool( os.getenv("LITELLM_USE_CHAT_COMPLETIONS_URL_FOR_ANTHROPIC_MESSAGES", False) diff --git a/litellm/litellm_core_utils/core_helpers.py b/litellm/litellm_core_utils/core_helpers.py index 8dec2941e02..8f6b57537fe 100644 --- a/litellm/litellm_core_utils/core_helpers.py +++ b/litellm/litellm_core_utils/core_helpers.py @@ -242,19 +242,21 @@ def drop_params_flag(value: object, source: str, logger: logging.Logger) -> bool DROP_PARAMS_ENV_VAR: Final = "LITELLM_DROP_PARAMS" -def drop_params_env_flag(environ: Mapping[str, str], logger: logging.Logger) -> bool: - configured: Final = environ.get(DROP_PARAMS_ENV_VAR, "").strip() +def env_flag(environ: Mapping[str, str], name: str, logger: logging.Logger) -> bool: + configured: Final = environ.get(name, "").strip() if configured == "": return False - normalized: Final = normalize_drop_params(configured) + normalized: Final = lax_bool(configured) if normalized is None: - logger.warning( - "%s=%r is not a flag value, treating it as on. Set it to true or false", DROP_PARAMS_ENV_VAR, configured - ) + logger.warning("%s=%r is not a flag value, treating it as on. Set it to true or false", name, configured) return True return normalized +def drop_params_env_flag(environ: Mapping[str, str], logger: logging.Logger) -> bool: + return env_flag(environ, DROP_PARAMS_ENV_VAR, logger) + + def safe_divide( numerator: float, denominator: float, diff --git a/litellm/messages/dispatch.py b/litellm/messages/dispatch.py index faa604a25ce..14e23545472 100644 --- a/litellm/messages/dispatch.py +++ b/litellm/messages/dispatch.py @@ -133,26 +133,25 @@ def messages_target(model: str, custom_llm_provider: str | None, model_info: obj return _route_target(model, resolved_model, resolved_provider, model_info, api_base, compacting=False) +_NATIVE_COMMON_FEATURES: Final = frozenset({Feature.TIMEOUT_VARIANTS, Feature.LATEST_ALIAS}) + _FEATURES: Final[detectors.Detectors] = MappingProxyType( { - **{feature: detect for feature, detect in detectors.COMMON.items() if feature is not Feature.TIMEOUT_VARIANTS}, - Feature.PROVIDER_AFFINITY: route_detectors.provider_affinity, + **{feature: detect for feature, detect in detectors.COMMON.items() if feature not in _NATIVE_COMMON_FEATURES}, Feature.SAFEGUARDS_CLASSIFIER: route_detectors.safeguards_classifier, - Feature.RESPONSE_CACHE: detectors.response_cache, + Feature.RESPONSE_CACHE: route_detectors.response_cache, Feature.MODIFY_PARAMS: detectors.modify_params, Feature.WORKLOAD_IDENTITY: route_detectors.anthropic_workload_identity, Feature.CACHE_CONTROL_INJECTION: detectors.cache_control_injection, - Feature.STREAM_COST_INJECTION: route_detectors.stream_cost_injection, Feature.THINKING_SIGNATURE_RETRY: route_detectors.thinking_signature_retry, - Feature.TOOL_CUSTOM_FIELD: route_detectors.tool_custom_field, } ) _ASYNC_FEATURES: Final[detectors.Detectors] = MappingProxyType( { **_FEATURES, + Feature.MOCK_RESPONSE: route_detectors.streamed_mock_response, Feature.PRE_REQUEST_HOOK: route_detectors.pre_request_hook, - Feature.FOUNDRY_BETA_ROW: route_detectors.foundry_beta_row, Feature.WEBSEARCH_INTERCEPTION: route_detectors.websearch_interception, Feature.MESSAGES_INTERCEPTOR: route_detectors.intercepted, Feature.STRICT_TOOLS_RETRY: route_detectors.strict_tools_retry, diff --git a/litellm/rust_bridge/_native.pyi b/litellm/rust_bridge/_native.pyi index 221968e07ed..7015890cc46 100644 --- a/litellm/rust_bridge/_native.pyi +++ b/litellm/rust_bridge/_native.pyi @@ -22,6 +22,7 @@ class NativeRedisStoreError(Exception): ... def trace_encode_error(message: str) -> bytes: ... def bundled_model_cost_map() -> bytes: ... def native_redis_breaker_states() -> list[str]: ... +def unsupported_tls_settings() -> list[str]: ... @final class NativeCostMapSource: diff --git a/litellm/rust_bridge/catalog.py b/litellm/rust_bridge/catalog.py index c493617003d..210df7b0c10 100644 --- a/litellm/rust_bridge/catalog.py +++ b/litellm/rust_bridge/catalog.py @@ -66,8 +66,6 @@ class Feature(str, Enum): """Extra body fields that Python transforms and merges into the provider request.""" PROVIDER_AFFINITY = "provider_affinity" """A provider affinity header that Python validates and populates from the session id.""" - FOUNDRY_BETA_ROW = "foundry_beta_row" - """A Foundry api_base, where Python drops the anthropic-beta values Foundry rejects.""" PRE_REQUEST_HOOK = "pre_request_hook" """A logger overriding async_pre_request_hook, which may rewrite the request.""" WEBSEARCH_INTERCEPTION = "websearch_interception" @@ -94,12 +92,8 @@ class Feature(str, Enum): """No static key, so Python mints a workload identity federation token.""" CACHE_CONTROL_INJECTION = "cache_control_injection" """Prompt caching injection that Python applies to messages and system.""" - STREAM_COST_INJECTION = "stream_cost_injection" - """litellm.include_cost_in_streaming_usage or the x-litellm-include-cost-in-streaming-usage header, where Python adds cost to streamed usage.""" THINKING_SIGNATURE_RETRY = "thinking_signature_retry" """Replayed thinking blocks, which Python strips and retries once if Anthropic rejects their signature.""" - TOOL_CUSTOM_FIELD = "tool_custom_field" - """Tools carrying a non-spec custom object, which Python drops or hoists for the provider.""" STRICT_TOOLS_RETRY = "strict_tools_retry" """Strict tools under drop_params, which Python sends again without strict if Claude rejects their limits.""" CLIENT_HEADERS = "client_headers" @@ -172,9 +166,7 @@ CHAT_PYTHON_FEATURES: Final = COMMON_PYTHON_FEATURES | { Feature.MODEL_ALIAS_MAP, } -MESSAGES_PYTHON_FEATURES: Final = (COMMON_PYTHON_FEATURES - {Feature.TIMEOUT_VARIANTS}) | { - Feature.PROVIDER_AFFINITY, - Feature.FOUNDRY_BETA_ROW, +MESSAGES_PYTHON_FEATURES: Final = (COMMON_PYTHON_FEATURES - {Feature.TIMEOUT_VARIANTS, Feature.LATEST_ALIAS}) | { Feature.SAFEGUARDS_CLASSIFIER, Feature.PRE_REQUEST_HOOK, Feature.WEBSEARCH_INTERCEPTION, @@ -183,9 +175,7 @@ MESSAGES_PYTHON_FEATURES: Final = (COMMON_PYTHON_FEATURES - {Feature.TIMEOUT_VAR Feature.MODIFY_PARAMS, Feature.WORKLOAD_IDENTITY, Feature.CACHE_CONTROL_INJECTION, - Feature.STREAM_COST_INJECTION, Feature.THINKING_SIGNATURE_RETRY, - Feature.TOOL_CUSTOM_FIELD, Feature.STRICT_TOOLS_RETRY, } diff --git a/litellm/rust_bridge/detectors.py b/litellm/rust_bridge/detectors.py index c38c3ea7135..3c8ee370be1 100644 --- a/litellm/rust_bridge/detectors.py +++ b/litellm/rust_bridge/detectors.py @@ -2,7 +2,6 @@ from __future__ import annotations -import os from collections.abc import Callable, Iterable, Mapping from types import MappingProxyType from typing import TYPE_CHECKING, Final, TypeAlias, cast # noqa: TID251 # tool lists stay caller-owned @@ -168,15 +167,11 @@ def response_cache(request: NativeCall) -> bool: return litellm.cache is not None and cache.get("no-cache") is not True -_NATIVE_ECDH_CURVES: Final = frozenset({"x25519", "prime256v1", "secp256r1", "p-256", "secp384r1", "p-384"}) - - def tls_policy(_request: NativeCall) -> bool: """Python applies TLS settings or ECDH curves the native client cannot configure.""" - curve: Final = os.getenv("SSL_ECDH_CURVE", litellm.ssl_ecdh_curve) - return bool(os.getenv("SSL_SECURITY_LEVEL", litellm.ssl_security_level)) or ( - bool(curve) and str(curve).strip().lower() not in _NATIVE_ECDH_CURVES - ) + from litellm.rust_bridge import _native + + return bool(_native.unsupported_tls_settings()) COMMON: Final[Detectors] = MappingProxyType( diff --git a/litellm/rust_bridge/host/routes/messages.py b/litellm/rust_bridge/host/routes/messages.py index c6568264edb..2bed5d6a182 100644 --- a/litellm/rust_bridge/host/routes/messages.py +++ b/litellm/rust_bridge/host/routes/messages.py @@ -2,6 +2,7 @@ from __future__ import annotations import datetime import errno +import inspect import ipaddress import os import socket @@ -15,25 +16,43 @@ from pydantic import TypeAdapter, ValidationError import litellm from litellm.litellm_core_utils import exception_mapping_utils -from litellm.litellm_core_utils.core_helpers import normalize_drop_params +from litellm.litellm_core_utils.core_helpers import ( + as_str_mapping, + normalize_drop_params, + proxy_request_headers, + stream_cost_injection_requested, +) +from litellm.litellm_core_utils.get_llm_provider_logic import get_provider_affinity_header +from litellm.litellm_core_utils.get_provider_specific_headers import ProviderSpecificHeaderUtils from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.litellm_core_utils.model_latest_aliases import is_model_latest_alias +from litellm.litellm_core_utils.provider_affinity import add_provider_affinity_header from litellm.litellm_core_utils.reasoning_effort_capability import deployment_reasoning_effort_levels from litellm.litellm_core_utils.request_timeout_resolver import get_configured_request_timeout -from litellm.llms.anthropic.common_utils import AnthropicError, anthropic_pass_through_api_base +from litellm.llms.anthropic.common_utils import AnthropicError, AnthropicModelInfo, anthropic_pass_through_api_base from litellm.llms.anthropic.messages.native import handler +from litellm.llms.anthropic.messages.native import utils as native_utils from litellm.llms.anthropic.messages.native.handler import ( _native_messages_config as native_messages_config, # pyright: ignore[reportPrivateUsage] # the handler's own config lookup ) +from litellm.llms.anthropic.messages.native.handler import ( + _resolve_requested_provider as _requested_provider, # pyright: ignore[reportPrivateUsage] # the handler's own provider resolution +) from litellm.llms.anthropic.messages.native.handler import ( _with_deployment_output_effort as _deployment_output_effort, # pyright: ignore[reportPrivateUsage] # the handler's own effort rounding ) from litellm.llms.anthropic.messages.utils import is_reasoning_auto_summary_enabled from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.llms.custom_httpx.http_handler import get_ssl_configuration +from litellm.llms.custom_httpx.llm_http_handler import ( + _merge_messages_header_sources as _merged_headers, # pyright: ignore[reportPrivateUsage] # the handler's own header merge +) from litellm.rust_bridge import failures from litellm.rust_bridge.host import callbacks_legacy_python, http_transport from litellm.rust_bridge.public_call import known_params, optional_str from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse +from litellm.types.router import GenericLiteLLMParams +from litellm.types.utils import ProviderSpecificHeader _METADATA: Final = TypeAdapter(dict[str, object]) _BODY_FIELDS: Final = frozenset( @@ -65,6 +84,14 @@ _BODY_FIELDS: Final = frozenset( ) +@dataclass(frozen=True, slots=True) +class CallSettings: + model: str + mock_response: Mapping[str, object] | None + stream_cost_model: str | None + provider_affinity_headers: Mapping[str, str] + + @dataclass(frozen=True, slots=True) class MessagesSettings: drop_params: bool @@ -129,7 +156,10 @@ _UPSTREAM_HEADERS: Final = TypeAdapter(list[tuple[str, str]]) _BODY_CUT: Final = TypeAdapter(tuple[int, int | None, bool]) _ERROR_CLASS: TypeAlias = Callable[..., Exception] _validate_metadata: Final = cast( # cast-ok: the handler's untyped metadata validator - Callable[[dict[str, object]], object], handler.validate_anthropic_api_metadata + Callable[[object], object], handler.validate_anthropic_api_metadata +) +_mock_response: Final = cast( # cast-ok: the handler's mock builder takes untyped messages + Callable[..., Mapping[str, object]], native_utils.mock_response ) _exception_type: Final = cast( # cast-ok: the legacy public exception mapper is untyped Callable[..., Exception], exception_mapping_utils.exception_type @@ -394,3 +424,111 @@ def settings(kwargs: Mapping[str, object]) -> dict[str, object]: request_timeout=get_configured_request_timeout(), ) ) + + +_HANDLER_PARAMS: Final = frozenset( + name + for name, parameter in inspect.signature( + cast(Callable[..., object], handler.anthropic_messages_handler) # noqa: TID251 # cast-ok: reads the handler's parameter names, never calls it + ).parameters.items() + if parameter.kind is not inspect.Parameter.VAR_KEYWORD +) +_HEADER_VALUES: Final = TypeAdapter(dict[str, str]) + + +def _handler_params(request: Mapping[str, object]) -> GenericLiteLLMParams: + return GenericLiteLLMParams( + **{name: value for name, value in request.items() if name not in _HANDLER_PARAMS}, + api_key=optional_str(request.get("api_key")), + api_base=optional_str(request.get("api_base")), + custom_llm_provider=optional_str(request.get("custom_llm_provider")), + ) + + +def _requested_model(request: Mapping[str, object]) -> str: + model, _, _, _ = _requested_provider( + str(request["model"]), optional_str(request.get("custom_llm_provider")), _handler_params(request) + ) + return model + + +def _model_without_latest_alias(model: str, custom_llm_provider: str | None) -> str: + if not is_model_latest_alias(model, custom_llm_provider): + return model + resolved, provider, _, _ = litellm.get_llm_provider(model=model, custom_llm_provider=custom_llm_provider) + prefix, separator, _ = model.partition("/") + return f"{prefix}/{resolved}" if separator and prefix == provider else resolved + + +def _mock(request: Mapping[str, object]) -> Mapping[str, object] | None: + mock: Final = request.get("mock_response") + if not mock or not isinstance(mock, str): + return None + _validate_metadata(request.get("metadata")) + messages: Final = request.get("messages") + max_tokens: Final = request.get("max_tokens") + return dict( + _mock_response( + model=_requested_model(request), + messages=messages if isinstance(messages, list) else [], + max_tokens=max_tokens if isinstance(max_tokens, int) else 0, + mock_response=mock, + ) + ) + + +def _stream_cost_model(request: Mapping[str, object]) -> str | None: + if request.get("stream") is not True or not stream_cost_injection_requested(proxy_request_headers(request)): + return None + return _requested_model(request) + + +def _provider_affinity_headers(request: Mapping[str, object], provider: str | None) -> Mapping[str, str]: + api_base: Final = optional_str(request.get("api_base")) + affinity_api_base: Final = AnthropicModelInfo.get_api_base(api_base) if provider == "anthropic" else api_base + default_header: Final = get_provider_affinity_header(affinity_api_base) + if default_header is None: + return {} + merged: Final = _merged_headers( + as_str_mapping(request.get("headers")), + as_str_mapping(request.get("extra_headers")), + ProviderSpecificHeaderUtils.get_provider_specific_headers( + provider_specific_header=cast( # cast-ok: the handler hands the caller's value on unvalidated + ProviderSpecificHeader | None, request.get("provider_specific_header") + ), + custom_llm_provider=provider, + native_anthropic_messages=True, + ), + ) + added: Final = add_provider_affinity_header(merged, _handler_params(request), default_header) + return _HEADER_VALUES.validate_python({name: value for name, value in added.items() if name not in merged}) + + +def call_settings(request: Mapping[str, object]) -> Mapping[str, object]: + explicit_provider: Final = optional_str(request.get("custom_llm_provider")) + model: Final = _model_without_latest_alias(str(request["model"]), explicit_provider) + mock: Final = _mock(request) + provider: Final = explicit_provider or inferred_provider(model, optional_str(request.get("api_base"))) + return asdict( + CallSettings( + model=model, + mock_response=mock, + stream_cost_model=None if mock is not None else _stream_cost_model(request), + provider_affinity_headers={} if mock is not None else _provider_affinity_headers(request, provider), + ) + ) + + +def inject_stream_cost(frames: bytes, model: str, request: Mapping[str, object]) -> object: + from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing + + logger: Final = request.get("litellm_logging_obj") + logging_obj: Final = logger if isinstance(logger, LiteLLMLoggingObj) else None + return ProxyBaseLLMRequestProcessing.process_chunk_with_cost_injection( + frames, + model, + logging_obj, + request_headers=proxy_request_headers( + as_str_mapping(logging_obj.model_call_details.get("litellm_params")) if logging_obj else None + ), + ) diff --git a/litellm/rust_bridge/messages/detectors.py b/litellm/rust_bridge/messages/detectors.py index 99555bb4852..7f83e34e22f 100644 --- a/litellm/rust_bridge/messages/detectors.py +++ b/litellm/rust_bridge/messages/detectors.py @@ -5,25 +5,27 @@ from __future__ import annotations from typing import Final import litellm -from litellm.anthropic_beta_headers_manager import is_foundry_api_base -from litellm.litellm_core_utils.core_helpers import ( - drops_unsupported_params, - proxy_request_headers, - stream_cost_injection_requested, -) -from litellm.llms.opencode.common_utils import is_opencode_api_base -from litellm.rust_bridge.detectors import api_base, overridden, present, streaming +from litellm.litellm_core_utils.core_helpers import drops_unsupported_params +from litellm.rust_bridge.detectors import overridden, present from litellm.rust_bridge.public_call import NativeCall, optional_mapping, optional_sequence, optional_str - -def provider_affinity(request: NativeCall) -> bool: - return is_opencode_api_base(api_base(request)) +_CALL_TYPE: Final = "anthropic_messages" -def foundry_beta_row(request: NativeCall) -> bool: - from litellm.llms.anthropic.common_utils import AnthropicModelInfo +def response_cache(request: NativeCall) -> bool: + cache: Final = litellm.cache + controls: Final = optional_mapping(request.resolved.get("cache")) or {} + return ( + cache is not None + and controls.get("no-cache") is not True + and cache.supported_call_types is not None + and _CALL_TYPE in cache.supported_call_types + and cache.should_use_cache(**request.resolved) is True # pyright: ignore[reportUnknownMemberType] # Cache.should_use_cache takes untyped kwargs + ) - return is_foundry_api_base(AnthropicModelInfo.get_api_base(api_base(request))) + +def streamed_mock_response(request: NativeCall) -> bool: + return present(request, "mock_response") and request.resolved.get("stream") is True def anthropic_workload_identity(request: NativeCall) -> bool: @@ -55,11 +57,6 @@ def thinking_signature_retry(request: NativeCall) -> bool: return any(_replays_thinking(message) for message in optional_sequence(request.resolved.get("messages")) or ()) -def tool_custom_field(request: NativeCall) -> bool: - tools: Final = optional_sequence(request.resolved.get("tools")) or () - return any("custom" in (optional_mapping(tool) or {}) for tool in tools) - - def strict_tools_retry(request: NativeCall) -> bool: tools: Final = optional_sequence(request.resolved.get("tools")) or () return drops_unsupported_params(request.resolved.get("drop_params")) and any( @@ -67,11 +64,6 @@ def strict_tools_retry(request: NativeCall) -> bool: ) -def stream_cost_injection(request: NativeCall) -> bool: - """Python adds response cost to streaming usage when the global setting or the request header asks for it.""" - return streaming(request) and stream_cost_injection_requested(proxy_request_headers(request.resolved)) - - def pre_request_hook(_request: NativeCall) -> bool: """The async Messages handler runs callbacks that can rewrite the request.""" return overridden("async_pre_request_hook", litellm.callbacks) diff --git a/tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py b/tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py index 41008f0d57f..8b3e979ffe3 100644 --- a/tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py +++ b/tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py @@ -34,6 +34,7 @@ import pytest from pydantic import JsonValue, TypeAdapter import litellm +from litellm.caching.caching import Cache, CacheMode from litellm.litellm_core_utils.add_retry_fallback_headers import get_hidden_params_dict from litellm.messages import dispatch from litellm.rust_bridge import catalog @@ -342,36 +343,18 @@ S15_COMPLETE_DEADLINE: Final = ( "S15: by design, native bounds a complete response by one overall deadline where Python bounds each read, " "so a body trickling in slower than the timeout succeeds in Python and times out natively" ) -F_MOCK: Final = "F: mock_response is Python-only (detector mock_response)" F_THINKING_RETRY: Final = ( "F: retry without thinking after a rejected signature is Python-only (detector thinking_signature_retry)" ) F_CACHE_INJECTION: Final = "F: prompt cache_control injection is Python-only (detector cache_control_injection)" -F_TOOL_CUSTOM: Final = "F: hoisting a tool's custom field is Python-only (detector tool_custom_field)" -F_STREAM_COST: Final = "F: cost injection into streamed usage is Python-only (detector stream_cost_injection)" +F_STREAMED_MOCK: Final = ( + "F: a mock with stream=True returns the mock dict and logs no success in Python (detector streamed_mock_response)" +) KNOWN_GAPS: Final[Mapping[str, Mapping[str, str | None]]] = MappingProxyType( { "response_private_keys": _gaps(outcome=S6_PRIVATE_KEYS), "missing_max_tokens": _gaps(dispatch=A10_NO_MAX_TOKENS), - "mock_response": _gaps( - logged_response_headers=F_MOCK, - request=F_MOCK, - outcome=F_MOCK, - hidden_params=F_MOCK, - logged=F_MOCK, - logged_metadata=F_MOCK, - logged_api_key=F_MOCK, - logged_call=F_MOCK, - ), - "mock_response_rate_limit": _gaps( - logged_response_headers=F_MOCK, - request=F_MOCK, - outcome=F_MOCK, - logged=F_MOCK, - logged_api_key=F_MOCK, - logged_call=F_MOCK, - ), "thinking_signature_rejected_then_retried": _gaps( request=F_THINKING_RETRY, outcome=F_THINKING_RETRY, @@ -381,9 +364,13 @@ KNOWN_GAPS: Final[Mapping[str, Mapping[str, str | None]]] = MappingProxyType( logged_response_headers=F_THINKING_RETRY, ), "cache_control_injection_points": _gaps(request=F_CACHE_INJECTION), - "tool_custom_field": _gaps(request=F_TOOL_CUSTOM), - "stream_cost_injection": _gaps(outcome=F_STREAM_COST), "enable_prompt_caching": _gaps(request=F_CACHE_INJECTION), + "mock_response_streamed": _gaps( + logged=F_STREAMED_MOCK, + logged_metadata=F_STREAMED_MOCK, + logged_response_headers=F_STREAMED_MOCK, + logged_call=F_STREAMED_MOCK, + ), } ) @@ -729,9 +716,63 @@ SYNC_CASES: Final = ( Case("sync_upstream_429", upstream=(_failure(429, "rate_limit_error", "slow down"),), sync=True), ) -FALLBACK_CASES: Final = ( +LATEST_ALIAS: Final = "anthropic/claude-opus-latest" +CUSTOM_TOOL: Final[Mapping[str, JsonValue]] = MappingProxyType({**WEATHER_TOOL, "custom": {"defer_loading": True}}) +COST_HEADER: Final[Mapping[str, JsonValue]] = MappingProxyType( + {"headers": {"x-litellm-include-cost-in-streaming-usage": "true"}} +) + +SERVED_NATIVELY_CASES: Final = ( Case("mock_response", _kw(mock_response="Hi from the mock"), upstream=()), Case("mock_response_rate_limit", _kw(mock_response="litellm.RateLimitError"), upstream=()), + Case("mock_response_context_window", _kw(mock_response="litellm.ContextWindowExceededError"), upstream=()), + Case("mock_response_empty_calls_the_provider", _kw(mock_response="")), + Case("mock_response_bad_metadata", _kw(mock_response="Hi", metadata={"user_id": 123}), upstream=()), + Case("latest_alias", model=LATEST_ALIAS), + Case("latest_alias_bare_with_provider", _kw(custom_llm_provider="anthropic"), model="claude-opus-latest"), + Case("latest_alias_stream", _kw(stream=True), upstream=(_stream(TEXT_STREAM),), model=LATEST_ALIAS), + Case( + "latest_alias_upstream_400", + upstream=(_failure(400, "invalid_request_error", "max_tokens: too large"),), + model=LATEST_ALIAS, + ), + Case("tool_custom_field", _kw(tools=[dict(CUSTOM_TOOL)])), + Case( + "tool_custom_field_keeps_other_custom_keys", + _kw(tools=[{**WEATHER_TOOL, "custom": {"defer_loading": False, "note": "kept"}}]), + ), + Case( + "tool_custom_field_dropped_with_drop_params", + _kw(tools=[{**WEATHER_TOOL, "custom": {"defer_loading": True, "note": "dropped"}}], drop_params=True), + ), + Case("tool_custom_field_not_an_object", _kw(tools=[{**WEATHER_TOOL, "custom": "opaque"}])), + Case( + "tool_custom_field_under_own_defer_loading", + _kw(tools=[{**WEATHER_TOOL, "defer_loading": False, "custom": {"defer_loading": True}}]), + ), + Case( + "stream_cost_injection", + _kw(stream=True), + upstream=(_stream(TEXT_STREAM),), + settings=MappingProxyType({"include_cost_in_streaming_usage": True}), + ), + Case( + "stream_cost_injection_from_proxy_header", + _kw(stream=True, proxy_server_request=dict(COST_HEADER)), + upstream=(_stream(TOOL_STREAM, chunk_delay=0.01),), + ), + Case( + "cache_off_by_default", + settings=MappingProxyType({"cache": Cache(mode=CacheMode.default_off)}), + ), + Case( + "cache_skips_messages_call_type", + settings=MappingProxyType({"cache": Cache(supported_call_types=["acompletion"])}), + ), +) + +FALLBACK_CASES: Final = ( + Case("mock_response_streamed", _kw(stream=True, mock_response="Hi from the mock"), upstream=()), Case( "thinking_replay", _kw(messages=list(THINKING_REPLAY), thinking={"type": "enabled", "budget_tokens": 2048}, max_tokens=4096), @@ -748,7 +789,6 @@ FALLBACK_CASES: Final = ( "cache_control_injection_points", _kw(system="Answer in one word.", cache_control_injection_points=[{"location": "message", "role": "system"}]), ), - Case("tool_custom_field", _kw(tools=[{**WEATHER_TOOL, "custom": {"defer_loading": True}}])), Case( "strict_tools_rejected_then_retried", _kw(tools=[{**WEATHER_TOOL, "strict": True}], drop_params=True), @@ -758,18 +798,20 @@ FALLBACK_CASES: Final = ( ), ), Case("modify_params", _kw(max_tokens=1), settings=MappingProxyType({"modify_params": True})), - Case( - "stream_cost_injection", - _kw(stream=True), - upstream=(_stream(TEXT_STREAM),), - settings=MappingProxyType({"include_cost_in_streaming_usage": True}), - ), Case("enable_prompt_caching", _kw(system="Answer in one word.", enable_prompt_caching=True)), Case("request_timeout_kwarg", _kw(request_timeout=30)), Case("stream_timeout_kwarg", _kw(stream=True, stream_timeout=30), upstream=(_stream(TEXT_STREAM),)), ) -CASES: Final = (*REQUEST_CASES, *RESPONSE_CASES, *FAILURE_CASES, *STREAM_CASES, *SYNC_CASES, *FALLBACK_CASES) +CASES: Final = ( + *REQUEST_CASES, + *RESPONSE_CASES, + *FAILURE_CASES, + *STREAM_CASES, + *SYNC_CASES, + *SERVED_NATIVELY_CASES, + *FALLBACK_CASES, +) @dataclass(frozen=True, slots=True) @@ -1077,3 +1119,12 @@ async def test_native_messages_match_python( assert native.logged == python.logged case _: assert native.callback_kwargs.get(aspect) == python.callback_kwargs.get(aspect) + + +@pytest.mark.parametrize("case", [pytest.param(case, id=case.name) for case in SERVED_NATIVELY_CASES]) +def test_catalog_serves_the_case_natively(case: Case, parity_server: RecordingServer) -> None: + with isolated_callback_registries(), _environment(case.env), _settings(case.settings): + request: Final = dispatch._public_request(dispatch._AMESSAGES, (), _kwargs(case, parity_server)) # pyright: ignore[reportPrivateUsage] # the public binding the async entry routes by + assert request is not None + context: Final = dispatch._context(request, dispatch._ASYNC_FEATURES) # pyright: ignore[reportPrivateUsage] # the async entry's own route context + assert catalog.rollout(context) is Rollout.RUST_OPT_IN, context.features diff --git a/tests/test_litellm_rust/messages/test_dispatch.py b/tests/test_litellm_rust/messages/test_dispatch.py index b580eca2e42..86d12d57652 100644 --- a/tests/test_litellm_rust/messages/test_dispatch.py +++ b/tests/test_litellm_rust/messages/test_dispatch.py @@ -492,18 +492,6 @@ STRICT_TOOL: Final = {"name": "a", "input_schema": {"type": "object"}, "strict": {"messages": [{"role": "assistant", "content": [{"type": "redacted_thinking", "data": "x"}]}]}, {}, ), - ( - Feature.TOOL_CUSTOM_FIELD, - "anthropic/claude-sonnet-4-5", - {"tools": [{"name": "a", "input_schema": {"type": "object"}, "custom": {"eager_input_streaming": True}}]}, - {}, - ), - ( - Feature.STREAM_COST_INJECTION, - "anthropic/claude-sonnet-4-5", - {"stream": True}, - {"include_cost_in_streaming_usage": True}, - ), ), ) def test_each_detected_feature_reaches_its_rule( @@ -619,12 +607,6 @@ async def test_strict_tools_stay_native_unless_python_could_drop_strict( ASYNC_ONLY_FEATURES: Final = ( pytest.param(Feature.PRE_REQUEST_HOOK, {}, {"callbacks": [_Rewrite()]}, id="pre-request-hook"), - pytest.param( - Feature.FOUNDRY_BETA_ROW, - {"api_base": "https://team.services.ai.azure.com/anthropic"}, - {}, - id="foundry-beta-row", - ), pytest.param( Feature.WEBSEARCH_INTERCEPTION, {}, {"callbacks": [WebSearchInterceptionLogger()]}, id="websearch-interception" ), @@ -686,6 +668,51 @@ async def test_handler_hooks_the_sync_entry_never_runs_keep_only_async_calls_on_ assert served_sync is native_result +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("kwargs", "async_on_python"), + ( + pytest.param({"mock_response": "hi"}, False, id="unary"), + pytest.param({"mock_response": "hi", "stream": True}, True, id="streamed"), + ), +) +async def test_async_mock_response_falls_back_only_when_streamed( + monkeypatch: pytest.MonkeyPatch, kwargs: Mapping[str, object], async_on_python: bool +) -> None: + rules: Final[Rules] = ( + RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, features=frozenset({Feature.MOCK_RESPONSE})), + RouteRule(Route.MESSAGES, Rollout.RUST_REQUIRED), + ) + python_result: Final = response() + native_result: Final = response() + + async def python(*call_args: object, **python_kwargs: object) -> AnthropicMessagesResponse: # kwargs-ok: stub + return python_result + + async def native(request: NativeCall) -> AnthropicMessagesResponse: + return native_result + + monkeypatch.setattr(litellm, "callbacks", []) + served_async: Final = await _ADISPATCH.arun( + (16, MESSAGES, "anthropic/claude-sonnet-4-5"), + kwargs, + python=python, + binding=amessages_binding(native), + native=lambda hook, request, args, native_kwargs: hook(request), + rules=rules, + ) + served_sync: Final = _DISPATCH.run( + (16, MESSAGES, "anthropic/claude-sonnet-4-5"), + kwargs, + python=lambda *call_args, **python_kwargs: python_result, + binding=messages_binding(lambda request: native_result), + native=lambda hook, request, args, native_kwargs: hook(request), + rules=rules, + ) + assert served_async is (python_result if async_on_python else native_result) + assert served_sync is python_result + + @pytest.mark.asyncio async def test_interceptor_feature_follows_the_resolved_provider() -> None: rules: Final[Rules] = ( diff --git a/tests/unit/litellm_core_utils/test_core_helpers.py b/tests/unit/litellm_core_utils/test_core_helpers.py index 02476545cae..ef2048998b1 100644 --- a/tests/unit/litellm_core_utils/test_core_helpers.py +++ b/tests/unit/litellm_core_utils/test_core_helpers.py @@ -17,6 +17,7 @@ from litellm.litellm_core_utils.core_helpers import ( bind_budget_reservation_to_callbacks, budget_reservation_from_metadata, drop_params_env_flag, + env_flag, drop_params_flag, get_litellm_metadata_from_kwargs, get_or_create_metadata_bucket, @@ -497,6 +498,32 @@ def test_drop_params_env_flag_keeps_a_non_flag_value_on_with_a_warning(configure ) +@pytest.mark.parametrize( + ("configured", "expected"), + [("", False), ("false", False), ("False", False), ("0", False), ("no", False), ("true", True), (" 1 ", True)], +) +def test_env_flag_parses_modify_params_as_a_bool(configured: str, expected: bool, caplog: pytest.LogCaptureFixture): + with caplog.at_level(logging.WARNING, logger="env-flag-test"): + assert ( + env_flag({"LITELLM_MODIFY_PARAMS": configured}, "LITELLM_MODIFY_PARAMS", logging.getLogger("env-flag-test")) + is expected + ) + assert caplog.text == "" + + +def test_env_flag_treats_a_non_flag_value_as_on_with_a_warning(caplog: pytest.LogCaptureFixture): + with caplog.at_level(logging.WARNING, logger="env-flag-test"): + assert ( + env_flag({"LITELLM_MODIFY_PARAMS": "enabled"}, "LITELLM_MODIFY_PARAMS", logging.getLogger("env-flag-test")) + is True + ) + assert "LITELLM_MODIFY_PARAMS='enabled' is not a flag value, treating it as on" in caplog.text + + +def test_env_flag_is_off_when_unset(): + assert env_flag({}, "LITELLM_MODIFY_PARAMS", logging.getLogger("env-flag-test")) is False + + class TestIsExpectedClientError: def test_status_ranges(self): from litellm.litellm_core_utils.core_helpers import is_expected_client_error diff --git a/tests/unit/rust_bridge/host/routes/test_messages.py b/tests/unit/rust_bridge/host/routes/test_messages.py index aa7dcc74841..8b0c540e3ac 100644 --- a/tests/unit/rust_bridge/host/routes/test_messages.py +++ b/tests/unit/rust_bridge/host/routes/test_messages.py @@ -5,6 +5,7 @@ import pytest import litellm from litellm.constants import DEFAULT_REQUEST_TIMEOUT_SECONDS +from litellm.litellm_core_utils.model_latest_aliases import get_model_latest_aliases from litellm.rust_bridge.host.routes import messages as messages_host from litellm.rust_bridge.host.routes.messages import response @@ -445,3 +446,71 @@ def test_a_failure_before_any_response_maps_like_the_python_transport_error( f"litellm.InternalServerError: AnthropicException - {text}. Handle with `litellm.InternalServerError`." ) assert "This looks like a bug in LiteLLM" not in str(mapped) + + +_CALL: Final = MappingProxyType({"messages": [{"role": "user", "content": "hi"}], "max_tokens": 8}) +_OPENCODE: Final = "https://opencode.ai/inference/go/anthropic/v1" +_COST_HEADER: Final = MappingProxyType({"headers": {"x-litellm-include-cost-in-streaming-usage": "true"}}) + + +@pytest.mark.parametrize( + ("model", "custom_llm_provider", "expected"), + [ + ( + "anthropic/claude-opus-latest", + None, + f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}", + ), + ("claude-opus-latest", "anthropic", get_model_latest_aliases()["anthropic"]["claude-opus-latest"]), + ("anthropic/claude-sonnet-4-5", None, "anthropic/claude-sonnet-4-5"), + ], +) +def test_call_settings_resolve_a_latest_alias_keeping_the_provider_prefix( + model: str, custom_llm_provider: str | None, expected: str +) -> None: + settings: Final = messages_host.call_settings({**_CALL, "model": model, "custom_llm_provider": custom_llm_provider}) + + assert settings["model"] == expected + + +@pytest.mark.parametrize( + ("extra", "expected"), + [ + ({"api_base": _OPENCODE, "session_id": "s-1"}, {"x-opencode-session": "s-1"}), + ({"api_base": _OPENCODE, "session_id": "s-1", "extra_headers": {"x-opencode-session": "mine"}}, {}), + ({"api_base": "https://other.example/v1", "session_id": "s-1"}, {}), + ({"api_base": _OPENCODE, "session_id": "s-1", "mock_response": "hi"}, {}), + ], +) +def test_call_settings_add_only_the_affinity_header_the_caller_did_not_send( + extra: dict[str, object], expected: dict[str, str] +) -> None: + settings: Final = messages_host.call_settings({**_CALL, "model": "anthropic/claude-sonnet-4-5", **extra}) + + assert settings["provider_affinity_headers"] == expected + + +@pytest.mark.parametrize( + ("extra", "expected"), + [ + ({"stream": True, "proxy_server_request": dict(_COST_HEADER)}, "claude-sonnet-4-5"), + ({"proxy_server_request": dict(_COST_HEADER)}, None), + ({"stream": True}, None), + ({"stream": True, "proxy_server_request": dict(_COST_HEADER), "mock_response": "hi"}, None), + ], +) +def test_call_settings_name_the_cost_model_only_for_an_opted_in_stream( + extra: dict[str, object], expected: str | None +) -> None: + settings: Final = messages_host.call_settings({**_CALL, "model": "anthropic/claude-sonnet-4-5", **extra}) + + assert settings["stream_cost_model"] == expected + + +def test_call_settings_carry_the_mock_the_python_handler_returns() -> None: + settings: Final = messages_host.call_settings( + {**_CALL, "model": "anthropic/claude-sonnet-4-5", "mock_response": "hi"} + ) + + assert isinstance(settings["mock_response"], dict) + assert settings["mock_response"]["content"] == [{"type": "text", "text": "hi"}] diff --git a/tests/unit/rust_bridge/test_catalog.py b/tests/unit/rust_bridge/test_catalog.py index 1994dd00709..a15a9c9511b 100644 --- a/tests/unit/rust_bridge/test_catalog.py +++ b/tests/unit/rust_bridge/test_catalog.py @@ -20,16 +20,10 @@ from litellm.rust_bridge.public_call import NativeCall @pytest.mark.parametrize( - ("model", "expected"), - ( - ("anthropic/claude-opus-latest", Rollout.PYTHON_ONLY), - (f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}", Rollout.RUST_OPT_IN), - ), + "model", + ("anthropic/claude-opus-latest", f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}"), ) -def test_latest_alias_messages_falls_back_without_changing_the_pinned_model_rollout( - model: str, - expected: Rollout, -) -> None: +def test_latest_alias_messages_stays_native_like_the_pinned_model(model: str) -> None: request: Final = NativeCall(args=(), kwargs={"model": model}, base={}) context: Final = RouteContext( Route.MESSAGES, @@ -41,7 +35,7 @@ def test_latest_alias_messages_falls_back_without_changing_the_pinned_model_roll RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, features=catalog.MESSAGES_PYTHON_FEATURES), RouteRule(Route.MESSAGES, Rollout.RUST_OPT_IN, providers=frozenset({"anthropic"})), ) - assert catalog.rollout(context, rules) is expected + assert catalog.rollout(context, rules) is Rollout.RUST_OPT_IN @pytest.fixture(autouse=True) @@ -152,6 +146,7 @@ def test_empty_constraints_match_nothing() -> None: ) assert catalog.decision(RouteContext(Route.OCR), rules) is Decision.PYTHON - assert catalog.decision( - RouteContext(Route.OCR, features=frozenset({Feature.SAFEGUARDS_CLASSIFIER})), rules - ) is Decision.PYTHON + assert ( + catalog.decision(RouteContext(Route.OCR, features=frozenset({Feature.SAFEGUARDS_CLASSIFIER})), rules) + is Decision.PYTHON + ) diff --git a/tests/unit/rust_bridge/test_detectors.py b/tests/unit/rust_bridge/test_detectors.py index fccc01dd8e6..46e4cb5c2b7 100644 --- a/tests/unit/rust_bridge/test_detectors.py +++ b/tests/unit/rust_bridge/test_detectors.py @@ -8,7 +8,7 @@ import httpx import pytest import litellm -from litellm.caching.caching import Cache +from litellm.caching.caching import Cache, CacheMode from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.litellm_logging import Logging from litellm.responses.utils import ResponsesAPIRequestUtils @@ -136,8 +136,6 @@ def _wrap_nested_level(value: object, depth: int) -> object: (responses_detectors.provider_affinity, {"provider_affinity_header": None, "session_id": "session-1"}, False), (responses_detectors.provider_affinity, {}, False), (responses_detectors.provider_affinity, {"api_base": "https://opencode.ai/inference/go/openai/v1"}, True), - (messages_detectors.provider_affinity, {"api_base": "https://opencode.ai/inference/go/anthropic/v1"}, True), - (messages_detectors.provider_affinity, {"api_base": "https://other.example/v1"}, False), (detectors.streaming, {"stream": True}, True), (detectors.streaming, {"stream": False}, False), (detectors.timeout_variants, {"request_timeout": 5}, True), @@ -208,9 +206,6 @@ def _wrap_nested_level(value: object, depth: int) -> object: {"messages": [{"role": "user", "content": [{"type": "text"}]}]}, False, ), - (messages_detectors.tool_custom_field, {"tools": [{"name": "a"}, {"name": "b", "custom": {}}]}, True), - (messages_detectors.tool_custom_field, {"tools": [{"name": "a", "defer_loading": True}]}, False), - (messages_detectors.tool_custom_field, {}, False), (messages_detectors.strict_tools_retry, {"drop_params": True, "tools": [{"name": "a"}, STRICT_TOOL]}, True), ( messages_detectors.strict_tools_retry, @@ -323,8 +318,6 @@ def test_input_rewrite_preserves_caller_input(input_value: object, expected: boo (detectors.modify_params, "modify_params", True, {}), (detectors.retries_fallbacks, "num_retries", 3, {}), (detectors.cache_control_injection, "enable_anthropic_prompt_caching", True, {}), - (messages_detectors.stream_cost_injection, "include_cost_in_streaming_usage", True, {"stream": True}), - (messages_detectors.stream_cost_injection, "include_cost_in_streaming_usage", "true", {"stream": True}), (chat_detectors.chat_retries_fallbacks, "model_fallbacks", [{"gpt-5": ["gpt-5-mini"]}], {}), (chat_detectors.chat_retries_fallbacks, "num_retries", 3, {}), (chat_detectors.model_alias_map, "model_alias_map", {"gpt-5": "openai/gpt-5"}, {}), @@ -347,30 +340,6 @@ def test_global_settings_are_detected( assert detector(call(**kwargs)) -def test_stream_cost_injection_needs_a_stream(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True) - assert not messages_detectors.stream_cost_injection(call()) - - -def test_stream_cost_injection_reads_the_request_header_the_proxy_reads(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False) - opted_in: Final = {"proxy_server_request": {"headers": {"x-litellm-include-cost-in-streaming-usage": "true"}}} - assert messages_detectors.stream_cost_injection(call(stream=True, **opted_in)) - assert not messages_detectors.stream_cost_injection(call(**opted_in)) - assert not messages_detectors.stream_cost_injection(call(stream=True, proxy_server_request={"headers": {}})) - - -def test_foundry_beta_row_reads_the_api_base_the_anthropic_handler_reads(monkeypatch: pytest.MonkeyPatch) -> None: - for name in ("ANTHROPIC_API_BASE", "ANTHROPIC_BASE_URL"): - monkeypatch.delenv(name, raising=False) - assert messages_detectors.foundry_beta_row(call(api_base="https://team.services.ai.azure.com/anthropic")) - assert messages_detectors.foundry_beta_row(call(base_url="https://team.cognitiveservices.azure.com/anthropic")) - assert not messages_detectors.foundry_beta_row(call(api_base="https://api.anthropic.com")) - assert not messages_detectors.foundry_beta_row(call()) - monkeypatch.setenv("ANTHROPIC_BASE_URL", "https://team.services.ai.azure.com/anthropic") - assert messages_detectors.foundry_beta_row(call()) - - def test_model_alias_map_applies_only_to_the_requested_model(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(litellm, "model_alias_map", {"fast": "openai/gpt-5-mini"}) assert chat_detectors.model_alias_map(call(model="fast")) @@ -549,9 +518,50 @@ def test_response_cache_is_skipped_only_by_no_cache( assert detectors.response_cache(call(**kwargs)) is expected +@pytest.mark.parametrize( + ("cache", "kwargs", "expected"), + ( + pytest.param(None, {}, False, id="no-cache-configured"), + pytest.param(Cache(), {}, True, id="default-cache"), + pytest.param(Cache(), {"cache": {"no-cache": True}}, False, id="no-cache-control"), + pytest.param(Cache(mode=CacheMode.default_off), {}, False, id="default-off"), + pytest.param( + Cache(mode=CacheMode.default_off), {"cache": {"use-cache": True}}, True, id="default-off-opted-in" + ), + pytest.param(Cache(supported_call_types=["acompletion"]), {}, False, id="messages-not-cached"), + pytest.param(Cache(supported_call_types=["anthropic_messages"]), {}, True, id="messages-cached"), + ), +) +def test_messages_response_cache_fires_only_when_python_touches_the_cache( + monkeypatch: pytest.MonkeyPatch, cache: Cache | None, kwargs: Mapping[str, object], expected: bool +) -> None: + monkeypatch.setattr(litellm, "cache", cache) + assert messages_detectors.response_cache(call(**kwargs)) is expected + + +@pytest.mark.parametrize( + ("kwargs", "expected"), + ( + pytest.param({"mock_response": "hi", "stream": True}, True, id="streamed"), + pytest.param({"mock_response": "hi"}, False, id="unary"), + pytest.param({"mock_response": "hi", "stream": False}, False, id="stream-false"), + pytest.param({"stream": True}, False, id="no-mock"), + ), +) +def test_streamed_mock_response_needs_a_mock_and_stream_true(kwargs: Mapping[str, object], expected: bool) -> None: + assert messages_detectors.streamed_mock_response(call(**kwargs)) is expected + + @pytest.mark.parametrize( ("environment", "setting", "expected"), - ((None, None, False), ("DEFAULT@SECLEVEL=1", None, True), (None, "DEFAULT@SECLEVEL=1", True)), + ( + (None, None, False), + ("DEFAULT@SECLEVEL=1", None, True), + (None, "DEFAULT@SECLEVEL=1", True), + (None, "ECDHE-RSA-AES128-GCM-SHA256", False), + (None, "ECDHE-RSA-AES128-GCM-SHA256:RC4-SHA", True), + ("HIGH", None, False), + ), ) def test_tls_policy_reads_the_security_level_environment_then_the_setting( monkeypatch: pytest.MonkeyPatch, environment: str | None, setting: str | None, expected: bool