scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-12 08:42:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 207

fix(messages): make rust messages copy python messages behaviour (#49611)

Browse source 
Closes the Rust vs Python /v1/messages differences the parity suite marks as known gaps, copying Python exactly.
This commit is contained in:
Nate Armstrong 2026-11-02 21:02:52 -08:00 • committed by GitHub
parent 5e08cb0e63
commit 34314a99f0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
46 changed files with 1595 additions and 468 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
litellm-rust/crates/callbacks-legacy-python/python_contract.json
View file

@ -33,6 +33,17 @@
"api_key",
"additional_args"
],
"record_response": [
"logger",
"status",
"headers",
"body"
],
"record_api_window": [
"logger",
"start",
"end"
],
"defers_async_logging": [
"logger"
],
@ -91,6 +102,13 @@
"logger",
"head"
],
"stamp_stream_head": [
"head",
"logger",
"kwargs",
"start_time",
"end_time"
],
"stream_success": [
"logger",
"url_route",

117
litellm-rust/crates/callbacks-legacy-python/src/adapter.rs
View file

@ -57,6 +57,8 @@ pub struct LegacyLogging {
asynchronous: bool,
internal: bool,
cache_key: Option<String>,
/// When the provider request went out, until its window is recorded (Messages only).
api_started: Option<f64>,
}
fn datetime(py: Python<'_>, epoch_seconds: f64) -> PyResult<Py<PyAny>> {
@ -87,6 +89,60 @@ impl LegacyLogging {
asynchronous,
internal: false,
cache_key: None,
api_started: None,
}
}
/// The kwargs `Logging.update_from_kwargs` sees. Python's Messages handler takes `metadata` as
/// its own argument, so the logger never sees the caller's raw metadata, only the validated
/// request metadata merged in through `litellm_params`. Python updates the logger once, before
/// the call; the refresh once the result is ready leaves out `litellm_metadata` too, so the
/// logged metadata is not rebuilt from what the call wrote into it since.
fn logging_kwargs<'py>(&self, py: Python<'py>, refresh: bool) -> PyResult<Bound<'py, PyDict>> {
let kwargs = self.call.kwargs().bind(py).copy()?;
if self.operation != LoggingOperation::Messages {
return Ok(kwargs);
}
let consumed: &[&str] = if refresh {
&["metadata", "litellm_metadata"]
} else {
&["metadata"]
};
for name in consumed {
if kwargs.contains(name)? {
kwargs.del_item(name)?;
}
}
Ok(kwargs)
}
/// Python's Messages handler hands the logger its request params, validated `metadata`
/// included, as `litellm_params`, beside the `aanthropic_messages` call flag.
fn logged_request_params(
&self,
context: &RequestContext,
) -> Option<serde_json::Map<String, Value>> {
if self.operation != LoggingOperation::Messages {
return None;
}
let mut params = context
.optional_params
.as_object()
.cloned()
.unwrap_or_default();
params.insert(
"aanthropic_messages".to_owned(),
Value::Bool(self.asynchronous),
);
Some(params)
}
/// Python times the provider exchange inside the async `AsyncHTTPHandler.post`; its sync
/// client is untimed.
fn close_api_window(&mut self, py: Python<'_>, end: f64) -> PyResult<()> {
match self.api_started.take() {
Some(start) => self.logger()?.record_api_window(py, start, end),
None => Ok(()),
}
}
@ -111,6 +167,16 @@ impl LegacyLogging {
}
}
/// The key `pre_call` logs. Python's Messages handler logs an empty string, never the key.
fn logged_api_key<'a>(&self, context: &'a RequestContext) -> Option<&'a str> {
match self.operation {
LoggingOperation::Messages => Some(""),
LoggingOperation::Completion | LoggingOperation::Responses | LoggingOperation::Ocr => {
context.api_key.as_ref().map(|api_key| api_key.expose())
}
}
}
fn stream_billing(&self) -> Option<PassThroughStream> {
match self.operation {
LoggingOperation::Messages => Some(PassThroughStream {
@ -377,7 +443,13 @@ impl LegacyLogging {
context: &RequestContext,
) -> PyResult<HookStep<Self, Box<WireRequest>>> {
let logger = self.logger()?;
logger.update_from_kwargs(py, self.call.kwargs(), &wire, context)?;
logger.update_from_kwargs(
py,
&self.logging_kwargs(py, false)?,
&wire,
context,
self.logged_request_params(context).as_ref(),
)?;
let body = to_py(py, &wire.body)?
.into_bound(py)
.cast_into::<PyDict>()?;
@ -400,11 +472,14 @@ impl LegacyLogging {
self.logger()?.pre_call(
py,
self.input_description(),
context.api_key.as_ref().map(|api_key| api_key.expose()),
self.logged_api_key(context),
&body,
&headers,
&wire.url,
)?;
if self.operation == LoggingOperation::Messages && self.asynchronous {
self.api_started = Some(epoch_seconds());
}
let headers = headers
.iter()
.map(|(name, value)| Ok((name.extract::<String>()?, value.extract::<String>()?)))
@ -455,7 +530,7 @@ impl LegacyLogging {
py,
(
&logger,
self.call.kwargs(),
self.logging_kwargs(py, true)?,
&facts.provider.model,
logger.getattr("optional_params")?,
params,
@ -483,6 +558,16 @@ impl LegacyLogging {
py: Python<'_>,
raw: &RawResponse,
) -> PyResult<HookStep<Self, ()>> {
if self.operation == LoggingOperation::Messages {
// Python's Messages handler never calls `post_call` at receipt: it stores the raw
// response, and logging replays it through `AnthropicConfig.transform_response`.
self.close_api_window(py, epoch_seconds())?;
if let Some(head) = &raw.head {
self.logger()?
.record_response(py, head.status, &head.headers, &raw.body)?;
}
return Ok(HookStep::Ready(()));
}
let api_key = self
.request
.as_ref()
@ -522,6 +607,9 @@ impl LegacyLogging {
) -> PyResult<HookStep<Self, ()>> {
self.end = Some(datetime(py, timing.end_time)?);
self.error = Some(error.clone_ref(py).into_value(py));
if self.logger.is_some() {
self.close_api_window(py, timing.end_time)?;
}
if self.stream.is_some() {
return self.stream_failure(py);
}
@ -543,6 +631,20 @@ impl LegacyLogging {
head.bind(py).set_item("cache_key", key)?;
head.bind(py).set_item("cache_hit", true)?;
}
self.close_api_window(py, epoch_seconds())?;
if self.operation == LoggingOperation::Messages {
// Python's `@client` stamps the call metadata onto the stream it hands back.
Streaming::StampHead.call(
py,
(
head.bind(py),
self.logger()?.object(py),
self.call.kwargs(),
&self.start,
datetime(py, epoch_seconds())?,
),
)?;
}
Streaming::Opened.call(py, (self.logger()?.object(py), head.bind(py)))?;
self.stream = Some(DeliveredStream {
chunks: PyList::empty(py).unbind(),
@ -1117,9 +1219,7 @@ assert logger.update['custom_llm_provider'] == 'edenai', logger.update
let step = logging
.before_provider_request(py, Box::new(wire), context)
.unwrap();
let raw = RawResponse {
body: "raw response".into(),
};
let raw = RawResponse::new("raw response".into());
assert!(matches!(
logging
.on_event(
@ -1869,9 +1969,10 @@ assert logger.calls[1][1] is response
py,
&locals,
c"
assert logger.names() == ['stream_opened', 'stream_success'], logger.calls
assert logger.names() == ['stamp_stream_head', 'stream_opened', 'stream_success'], logger.calls
assert logger.calls[0][1] is head
chunks = logger.calls[1][1]
assert logger.calls[1][1] is head
chunks = logger.calls[2][1]
assert len(chunks) == 2
assert chunks[0] is first
assert chunks[1] is last

51
litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs
View file

@ -5,6 +5,7 @@
use litellm_host::interceptors::{RequestContext, WireRequest};
use litellm_host_python::to_py;
use pyo3::{exceptions::PyBaseException, prelude::*, types::PyDict};
use serde_json::Value;
use crate::logger::PythonLogger;
use crate::python::{Logging, Wrapper};
@ -15,9 +16,10 @@ pub trait LegacyCallbacks {
fn update_from_kwargs(
&self,
py: Python<'_>,
kwargs: &Py<PyDict>,
kwargs: &Bound<'_, PyDict>,
wire: &WireRequest,
context: &RequestContext,
request_params: Option<&serde_json::Map<String, Value>>,
) -> PyResult<()>;
/// `Logging.pre_call`.
@ -31,6 +33,20 @@ pub trait LegacyCallbacks {
url: &str,
) -> PyResult<()>;
/// Records the provider's raw HTTP response on the logger, as Python's Messages handler
/// does with `httpx_response` and `response_headers`.
fn record_response(
&self,
py: Python<'_>,
status: u16,
headers: &[(String, String)],
body: &str,
) -> PyResult<()>;
/// Records one provider HTTP exchange's window, as Python's timed `AsyncHTTPHandler.post`
/// does with `llm_api_duration_ms` and the metadata's `llm_api_timing_windows`.
fn record_api_window(&self, py: Python<'_>, start: f64, end: f64) -> PyResult<()>;
/// `Logging.post_call`.
fn post_call(
&self,
@ -83,12 +99,13 @@ impl LegacyCallbacks for PythonLogger {
fn update_from_kwargs(
&self,
py: Python<'_>,
kwargs: &Py<PyDict>,
kwargs: &Bound<'_, PyDict>,
wire: &WireRequest,
context: &RequestContext,
request_params: Option<&serde_json::Map<String, Value>>,
) -> PyResult<()> {
let secret_fields: Vec<&str> = context.secret_fields.iter().map(String::as_str).collect();
let redacted_kwargs = redact(py, kwargs.bind(py), &secret_fields)?;
let redacted_kwargs = redact(py, kwargs, &secret_fields)?;
let optional_params = redact(
py,
&to_py(py, &context.optional_params)?
@ -97,23 +114,23 @@ impl LegacyCallbacks for PythonLogger {
&secret_fields,
)?;
let params = PyDict::new(py);
params.set_item(
"litellm_call_id",
kwargs.bind(py).get_item("litellm_call_id")?,
)?;
params.set_item("litellm_call_id", kwargs.get_item("litellm_call_id")?)?;
params.set_item("api_base", &wire.url)?;
for name in ["logger_fn", "litellm_request_debug"] {
if let Some(value) = kwargs.bind(py).get_item(name)? {
if let Some(value) = kwargs.get_item(name)? {
params.set_item(name, value)?;
}
}
for name in custom_pricing_fields(py)? {
if let Some(value) = kwargs.bind(py).get_item(&name)?
if let Some(value) = kwargs.get_item(&name)?
&& !value.is_none()
{
params.set_item(name, value)?;
}
}
for (name, value) in request_params.into_iter().flatten() {
params.set_item(name, to_py(py, value)?)?;
}
Logging::Update.call(
py,
(
@ -145,6 +162,22 @@ impl LegacyCallbacks for PythonLogger {
Ok(())
}
fn record_response(
&self,
py: Python<'_>,
status: u16,
headers: &[(String, String)],
body: &str,
) -> PyResult<()> {
Logging::RecordResponse.call(py, (self.object(py), status, headers.to_vec(), body))?;
Ok(())
}
fn record_api_window(&self, py: Python<'_>, start: f64, end: f64) -> PyResult<()> {
Logging::RecordApiWindow.call(py, (self.object(py), start, end))?;
Ok(())
}
fn post_call(
&self,
py: Python<'_>,

6
litellm-rust/crates/callbacks-legacy-python/src/python.rs
View file

@ -44,6 +44,10 @@ pub(crate) enum Logging {
PreCall,
#[strum(serialize = "post_call")]
PostCall,
#[strum(serialize = "record_response")]
RecordResponse,
#[strum(serialize = "record_api_window")]
RecordApiWindow,
#[strum(serialize = "defers_async_logging")]
DefersAsync,
#[strum(serialize = "defer_success")]
@ -77,6 +81,8 @@ pub(crate) enum DeploymentHooks {
pub(crate) enum Streaming {
#[strum(serialize = "stream_opened")]
Opened,
#[strum(serialize = "stamp_stream_head")]
StampHead,
#[strum(serialize = "stream_success")]
Success,
#[strum(serialize = "stream_failure")]

7
litellm-rust/crates/callbacks-legacy-python/src/test_support.rs
View file

@ -65,6 +65,12 @@ FAKES = {
'post_call': lambda logger, original_response, api_key, additional_args: logger.post_call(
original_response, api_key, additional_args
),
'record_response': lambda logger, status, headers, body: logger.model_call_details.update(
response_headers=dict(headers)
),
'record_api_window': lambda logger, start, end: logger.model_call_details.update(
llm_api_duration_ms=(end - start) * 1000
),
'defers_async_logging': lambda logger: bool(getattr(logger, '_defer_async_logging', False)),
'defer_success': lambda logger, pending: setattr(logger, '_native_pending_logging', pending),
'sync_success_for_async_call': lambda logger, response, start, end: logger.handle_sync_success_callbacks_for_async_calls(
@ -85,6 +91,7 @@ FAKES = {
),
'after_deployment_failure': lambda kwargs, error, call_type: kwargs['logger'].hook('failure', error, call_type),
'stream_opened': lambda logger, head: logger.record('stream_opened', head),
'stamp_stream_head': lambda head, logger, kwargs, start, end: logger.record('stamp_stream_head', head),
'stream_success': lambda logger, url_route, endpoint_type, request_body, chunks, start, end, first_chunk: logger.record(
'stream_success', list(chunks)
),

14
litellm-rust/crates/host-http/tests/serve.rs
View file

@ -215,9 +215,7 @@ async fn projection_custom_operations_and_hooks_feed_the_http_response(intercept
)
.await?;
route_hooks
.after_provider_response(RawResponse {
body: wire.url.clone(),
})
.after_provider_response(RawResponse::new(wire.url.clone()))
.await?;
Ok(CallOutput::Complete(Bytes::from(wire.url)))
},
@ -501,9 +499,7 @@ async fn hook_rejection_stops_execution_and_is_reported_once(
move |_, _, route_hooks, _observations| async move {
if event {
route_hooks
.after_provider_response(RawResponse {
body: "response".into(),
})
.after_provider_response(RawResponse::new("response".into()))
.await?;
} else {
route_hooks
@ -637,9 +633,7 @@ async fn unary_calls_use_into_response_after_hooks_and_before_success(intercepto
)
.await?;
route_hooks
.after_provider_response(RawResponse {
body: wire.url.clone(),
})
.after_provider_response(RawResponse::new(wire.url.clone()))
.await?;
Ok(CallOutput::Complete(json!({"url": wire.url})))
},
@ -689,7 +683,7 @@ async fn unary_failure_preserves_the_error_without_converting(
None,
|_, _, route_hooks, _observations| async move {
route_hooks
.after_provider_response(RawResponse { body: "raw".into() })
.after_provider_response(RawResponse::new("raw".into()))
.await?;
Err(TestError::Provider)
},

8
litellm-rust/crates/host-native/tests/driver.rs
View file

@ -170,9 +170,7 @@ fn dispatching_call() -> TestMachine {
)
.await?;
route_hooks
.after_provider_response(RawResponse {
body: wire.url.clone(),
})
.after_provider_response(RawResponse::new(wire.url.clone()))
.await?;
Ok(CallOutput::Complete(wire.url))
},
@ -660,9 +658,7 @@ async fn response_interception_waits_and_can_reject_after_observation(#[case] re
"input",
Some(sender),
|_, _, interceptors, observers| async move {
let raw = RawResponse {
body: "provider response".into(),
};
let raw = RawResponse::new("provider response".into());
observers.unwrap().emit(CallEvent::Execution(
ExecutionEvent::ProviderResponseReceived { raw: raw.clone() },
));

2
litellm-rust/crates/host-python/tests/support/mod.rs
View file

@ -467,7 +467,7 @@ pub fn success_machine() -> MachineFactory {
.before_provider_request(wire(), context())
.await?;
host.interceptors
.after_provider_response(RawResponse { body: "raw".into() })
.after_provider_response(RawResponse::new("raw".into()))
.await?;
Ok(format!("{projected}|{signed}|{}", wire.url))
})

24
litellm-rust/crates/host/src/interceptors.rs
View file

@ -28,6 +28,28 @@ pub struct RequestContext {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RawResponse {
pub body: String,
/// The provider's HTTP status and headers, when the route forwards them to logging.
pub head: Option<ResponseHead>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ResponseHead {
pub status: u16,
pub headers: Vec<(String, String)>,
}
impl RawResponse {
pub fn new(body: String) -> Self {
Self { body, head: None }
}
#[must_use]
pub fn with_head(self, head: ResponseHead) -> Self {
Self {
head: Some(head),
..self
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
@ -186,7 +208,7 @@ mod tests {
.await?;
Interceptors::after_provider_response(
&channel.interceptors,
RawResponse { body: "raw".into() },
RawResponse::new("raw".into()),
)
.await?;
Ok((sent, ()))

4
litellm-rust/crates/host/tests/observation.rs
View file

@ -74,9 +74,7 @@ fn snapshots_do_not_retain_runtime_objects(capacity: NonZeroUsize, #[case] faile
#[rstest]
fn provider_snapshots_own_the_response_body(capacity: NonZeroUsize) {
let mut raw = RawResponse {
body: "provider response".into(),
};
let mut raw = RawResponse::new("provider response".into());
let event: CallEvent<(), (), &RawResponse> =
CallEvent::Execution(ExecutionEvent::ProviderResponseReceived { raw: &raw });
let expected = raw.clone();

76
litellm-rust/crates/http/src/error.rs
View file

@ -58,8 +58,20 @@ pub(crate) enum Repr {
Network(#[source] reqwest::Error),
#[error("upstream request timed out")]
Timeout(#[source] reqwest::Error),
/// No bytes arrived within the read timeout, which the caller enforces per read; `body` says
/// the response head had already arrived.
#[error("upstream request timed out")]
ReadTimeout {
#[source]
elapsed: Elapsed,
body: bool,
},
#[error("upstream request failed with status {status}: {body}")]
Status { status: u16, body: UpstreamBody },
Status {
status: u16,
body: UpstreamBody,
headers: Vec<(String, String)>,
},
#[error("invalid WebSocket URL")]
WebSocketUrl(#[source] url::ParseError),
@ -136,13 +148,59 @@ impl Repr {
impl Error {
pub fn upstream(status: u16, body: String) -> Self {
Self::upstream_with_headers(status, body, Vec::new())
}
/// An upstream failure that keeps the provider's response headers (`retry-after`,
/// `request-id`) for the caller.
pub fn upstream_with_headers(
status: u16,
body: String,
headers: Vec<(String, String)>,
) -> Self {
Repr::Status {
status,
body: UpstreamBody(body),
headers,
}
.into()
}
/// The provider's response headers on an upstream failure; empty for every other error.
pub fn upstream_headers(&self) -> &[(String, String)] {
match &*self.0 {
Repr::Status { headers, .. } => headers,
_ => &[],
}
}
/// A timeout reached before the connection was established, as opposed to while reading.
pub fn is_connect_timeout(&self) -> bool {
matches!(&*self.0, Repr::Timeout(error) if error.is_connect())
}
/// A read the caller timed itself (httpx's per-read timeout) ran out.
pub fn read_timeout(elapsed: Elapsed) -> Self {
Repr::ReadTimeout {
elapsed,
body: false,
}
.into()
}
/// A read of the response body the caller timed itself ran out, after the head arrived.
pub fn body_read_timeout(elapsed: Elapsed) -> Self {
Repr::ReadTimeout {
elapsed,
body: true,
}
.into()
}
pub fn is_body_read_timeout(&self) -> bool {
matches!(&*self.0, Repr::ReadTimeout { body: true, .. })
}
pub fn computed_header(name: impl Into<String>) -> Self {
Repr::ComputedHeader(name.into()).into()
}
@ -169,6 +227,7 @@ impl Error {
match &*self.0 {
Repr::Network(_)
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
| Repr::Handshake(_)
| Repr::HandshakeTimeout(_)
@ -227,6 +286,7 @@ impl Error {
| Repr::Connect(_)
| Repr::Network(_)
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
| Repr::Handshake(_)
| Repr::HandshakeTimeout(_)
@ -259,6 +319,7 @@ impl Error {
| Repr::Connect(_)
| Repr::Network(_)
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
| Repr::WebSocketUrl(_)
| Repr::WebSocketScheme(_)
@ -295,6 +356,7 @@ impl Error {
| Repr::Connect(_)
| Repr::Network(_)
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
| Repr::WebSocketUrl(_)
| Repr::WebSocketScheme(_)
@ -333,6 +395,7 @@ impl Error {
| Repr::Signature(_)
| Repr::Network(_)
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
| Repr::WebSocketUrl(_)
| Repr::WebSocketScheme(_)
@ -356,7 +419,10 @@ impl Error {
pub fn is_timeout(&self) -> bool {
match &*self.0 {
Repr::Timeout(_) | Repr::HandshakeTimeout(_) | Repr::DownloadTimeout(_) => true,
Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::HandshakeTimeout(_)
| Repr::DownloadTimeout(_) => true,
Repr::Read { .. }
| Repr::InvalidPem { .. }
| Repr::Client(_)
@ -392,7 +458,7 @@ impl Error {
pub fn upstream_response(&self) -> Option<(u16, &str)> {
match &*self.0 {
Repr::Status { status, body } => Some((*status, &body.0)),
Repr::Status { status, body, .. } => Some((*status, &body.0)),
Repr::Read { .. }
| Repr::InvalidPem { .. }
| Repr::Client(_)
@ -408,6 +474,7 @@ impl Error {
| Repr::Connect(_)
| Repr::Network(_)
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::WebSocketUrl(_)
| Repr::WebSocketScheme(_)
| Repr::WebSocketSchemeRewrite(_)
@ -431,7 +498,7 @@ impl Error {
pub fn status(&self) -> Option<u16> {
match &*self.0 {
Repr::Status { status, .. } | Repr::DownloadStatus(status) => Some(*status),
Repr::Timeout(_) | Repr::DownloadTimeout(_) => Some(408),
Repr::Timeout(_) | Repr::ReadTimeout { .. } | Repr::DownloadTimeout(_) => Some(408),
Repr::Read { .. }
| Repr::InvalidPem { .. }
| Repr::Client(_)
@ -482,6 +549,7 @@ impl Error {
| Repr::Connect(_)
| Repr::Network(_)
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
| Repr::WebSocketUrl(_)
| Repr::WebSocketScheme(_)

2
litellm-rust/crates/inference-chat/src/handler.rs
View file

@ -108,7 +108,7 @@ pub(super) async fn execute(
if !status.is_success() {
return Err(litellm_http::Error::upstream(status.as_u16(), text).into());
}
let raw = RawResponse { body: text.clone() };
let raw = RawResponse::new(text.clone());
if let Some(observers) = observers {
observers.emit(litellm_host::lifecycle::CallEvent::Execution(
ExecutionEvent::ProviderResponseReceived { raw: raw.clone() },

2
litellm-rust/crates/inference-messages/Cargo.toml
View file

@ -24,6 +24,7 @@ litellm-tracing.workspace = true
reqwest.workspace = true
serde.workspace = true
serde_json = { workspace = true, features = ["preserve_order"] }
tokio.workspace = true
tracing.workspace = true
[dev-dependencies]
@ -34,5 +35,4 @@ litellm-host-native.workspace = true
litellm-http = { workspace = true, features = ["test-support"] }
litellm-inference-testing.workspace = true
rstest.workspace = true
tokio.workspace = true
wiremock.workspace = true

74
litellm-rust/crates/inference-messages/src/common_utils.rs
View file

@ -35,6 +35,14 @@ impl MessagesProvider {
pub(crate) fn config(self) -> &'static dyn BaseMessagesConfig {
self.config
}
/// Python's `MESSAGES_SERVICE_TIER_PROVIDERS`: everyone else has `service_tier` dropped.
pub(crate) fn accepts_service_tier(self) -> bool {
matches!(
self.provider,
LlmProviders::Anthropic | LlmProviders::AzureAi | LlmProviders::BedrockMantle
)
}
}
/// Python's `get_provider_anthropic_messages_config`: Vertex AI serves only its Claude
@ -119,3 +127,69 @@ mod tests {
);
}
}
/// Python's `_merge_messages_header_sources`: later sources win per header name, except
/// `anthropic-beta` (any casing), whose values from every source are unioned, deduplicated and
/// sorted into one lowercase `anthropic-beta` entry.
pub fn merge_messages_header_sources(
sources: impl IntoIterator<Item = Map<String, Value>>,
) -> Map<String, Value> {
let mut merged = Map::new();
let mut betas = std::collections::BTreeSet::new();
for (name, value) in sources.into_iter().flatten() {
if !name.eq_ignore_ascii_case("anthropic-beta") {
merged.insert(name, value);
continue;
}
let values: Vec<&str> = match &value {
Value::String(text) => vec![text.as_str()],
Value::Array(entries) => entries.iter().filter_map(Value::as_str).collect(),
_ => Vec::new(),
};
betas.extend(
values
.iter()
.flat_map(|value| value.split(','))
.map(str::trim)
.filter(|beta| !beta.is_empty())
.map(str::to_owned),
);
}
if !betas.is_empty() {
merged.insert(
"anthropic-beta".to_owned(),
Value::String(betas.into_iter().collect::<Vec<_>>().join(",")),
);
}
merged
}
#[cfg(test)]
mod merge_tests {
use serde_json::{Map, Value, json};
use super::merge_messages_header_sources;
fn map(value: Value) -> Map<String, Value> {
value.as_object().cloned().unwrap()
}
#[test]
fn later_sources_win_except_anthropic_beta_which_is_unioned_and_sorted() {
let merged = merge_messages_header_sources([
map(json!({"x-a": "1", "anthropic-beta": "b-2, a-1"})),
map(json!({"x-a": "2", "Anthropic-Beta": ["c-3", 7, "a-1"]})),
map(json!({"anthropic-beta": 5})),
]);
assert_eq!(
Value::Object(merged),
json!({"x-a": "2", "anthropic-beta": "a-1,b-2,c-3"})
);
}
#[test]
fn no_beta_values_leave_no_beta_header() {
let merged = merge_messages_header_sources([map(json!({"anthropic-beta": ""}))]);
assert!(merged.is_empty());
}
}

191
litellm-rust/crates/inference-messages/src/handler.rs
View file

@ -10,9 +10,9 @@ use litellm_core_utils::{
get_provider_specific_headers::get_provider_specific_headers, settings::Lookup,
};
use litellm_host::interceptors::{
Cost, ExecutionFacts, Interceptors, ProviderIdentity, RequestContext, ResultSource, WireRequest,
Cost, ExecutionFacts, Interceptors, ProviderIdentity, RawResponse, RequestContext,
ResponseHead, ResultSource, WireRequest,
};
use litellm_http::request::with_default_headers;
use litellm_llms::base_llm::{
auth::{Authenticated, ValidatedEnvironment, resolve_auth},
messages::{
@ -21,14 +21,16 @@ use litellm_llms::base_llm::{
transformation::BaseMessagesConfig,
},
};
use litellm_llms_types::formats::messages::{MessagesRequest, MessagesResponse};
use litellm_llms_types::formats::messages::{
MessagesOptionalParams, MessagesRequest, MessagesResponse,
};
use litellm_secrets::source::Secrets;
use litellm_tracing::ByteChunk;
use serde_json::Value;
use super::{
Error, MessagesCall, MessagesCallResponse, MessagesRoute,
common_utils::{MessagesProvider, string_headers},
common_utils::{MessagesProvider, merge_messages_header_sources, string_headers},
constants::MESSAGES_TIMEOUT_SECS,
route::Messages,
types::invalid_request,
@ -126,14 +128,18 @@ async fn provider_call(
MessagesRequest { model, ..body },
shaping.settings.reasoning_auto_summary,
)?;
let transform_context =
MessagesTransformContext::new(shaping.capabilities, shaping.settings.drop_params);
let filtered = requested_optional_params(sanitized, provider, &transform_context)?;
let trimmed =
without_additional_drop_params(sanitized, &shaping.settings.additional_drop_params)?;
without_additional_drop_params(filtered, &shaping.settings.additional_drop_params)?;
let scoped =
get_provider_specific_headers(provider_specific_header.as_ref(), provider.as_str());
let forwarded = string_headers(Some(
extra_headers.into_iter().flatten().chain(scoped).collect(),
))?;
let forwarded = string_headers(Some(merge_messages_header_sources([
extra_headers.unwrap_or_default(),
scoped,
])))?;
let validated = config.validate_environment(
forwarded,
api_key.as_deref(),
@ -144,16 +150,13 @@ async fn provider_call(
)?;
let body = config.request_body(
config.transform_anthropic_messages_request(
trimmed,
&MessagesTransformContext::new(shaping.capabilities, shaping.settings.drop_params),
)?,
config.transform_anthropic_messages_request(trimmed, &transform_context)?,
&validated.headers,
shaping.settings.drop_params,
)?;
let environment = ValidatedEnvironment {
headers: config.request_headers(
with_default_headers(validated.headers, config.default_headers()),
with_python_default_headers(validated.headers, config.default_headers()),
&body,
),
auth: validated.auth,
@ -184,7 +187,7 @@ async fn provider_call(
.before_provider_request(
WireRequest {
url,
headers: authenticated.headers,
headers: joined_header_values(authenticated.headers),
body: config.wire_body(serde_json::to_value(&body).map_err(serialize_failure)?),
},
request_context,
@ -213,6 +216,64 @@ async fn provider_call(
})
}
/// Python sends its header dict through `httpx.Headers`, which folds names that differ only in
/// case into one lowercase header whose values are joined with `", "`.
fn joined_header_values(headers: Vec<(String, String)>) -> Vec<(String, String)> {
let mut joined: Vec<(String, String)> = Vec::with_capacity(headers.len());
let mut positions: std::collections::HashMap<String, usize> =
std::collections::HashMap::with_capacity(headers.len());
for (name, value) in headers {
match positions.entry(name.to_ascii_lowercase()) {
std::collections::hash_map::Entry::Occupied(position) => {
let present = &mut joined[*position.get()].1;
present.push_str(", ");
present.push_str(&value);
}
std::collections::hash_map::Entry::Vacant(position) => {
position.insert(joined.len());
joined.push((name, value));
}
}
}
joined
}
/// Python's `finalize_messages_headers` adds a default only when the exact lowercase name is
/// absent, so a caller's `Anthropic-Version` is sent alongside the default, not instead of it.
fn with_python_default_headers(
headers: Vec<(String, String)>,
defaults: &[(&str, &str)],
) -> Vec<(String, String)> {
let missing: Vec<(String, String)> = defaults
.iter()
.filter(|(name, _)| !headers.iter().any(|(present, _)| present == name))
.map(|(name, value)| ((*name).to_string(), (*value).to_string()))
.collect();
headers.into_iter().chain(missing).collect()
}
/// Python's `get_requested_anthropic_messages_optional_param`, which gates speed, service tier
/// and sampling params before `additional_drop_params` and the environment check run.
fn requested_optional_params(
request: MessagesRequest,
provider: MessagesProvider,
context: &MessagesTransformContext,
) -> Result<MessagesRequest, Error> {
let request = litellm_llms::anthropic::messages::transformation::drop_unsupported_params(
request, context,
)?;
if provider.accepts_service_tier() {
return Ok(request);
}
Ok(MessagesRequest {
params: MessagesOptionalParams {
service_tier: None,
..request.params
},
..request
})
}
fn without_additional_drop_params(
request: MessagesRequest,
paths: &[String],
@ -245,6 +306,7 @@ async fn call_provider(
} = request;
let provider_name = provider.as_str();
log_request_body(provider_name, stream, &wire.body);
let timeout = timeout.unwrap_or(Duration::from_secs(MESSAGES_TIMEOUT_SECS));
let response = send(
http,
Authenticated {
@ -254,10 +316,11 @@ async fn call_provider(
&wire.url,
&wire.body,
timeout,
stream,
)
.await?;
if !response.status().is_success() {
return Err(provider_error(response).await);
return Err(provider_error(response, timeout).await);
}
let config = provider.config();
if stream {
@ -265,11 +328,24 @@ async fn call_provider(
response,
config.stream_decoder(),
provider_name,
timeout,
));
}
let head = ResponseHead {
status: response.status().as_u16(),
headers: litellm_http::request::header_pairs(response.headers()),
};
let status = head.status;
let text = response.text().await.map_err(litellm_http::Error::from)?;
log_response_body(&text);
context.response_received(&text).await?;
context
.raw_response_received(RawResponse::new(text.clone()).with_head(head))
.await?;
if serde_json::from_str::<serde::de::IgnoredAny>(&text).is_err() {
// Python raises `AnthropicError(status_code=<the response's>, message=<its text>)` for a
// body it cannot parse, and the public mapping classifies it by that status.
return Err(litellm_http::Error::upstream(status, text).into());
}
decode_response(config, &identity.model, &text)
.map(|message| MessagesCallResponse::Complete(Box::new(message)))
}
@ -281,30 +357,53 @@ fn serialize_failure(err: serde_json::Error) -> Error {
))
}
/// Python's httpx timeout bounds each phase and each read, never the whole exchange. A complete
/// response keeps the client's overall deadline; a stream is bounded until its head arrives and
/// then per read (see [`timed_chunks`]), so a long stream that keeps sending is not cut off.
async fn send(
http: &litellm_http::Client,
authenticated: Authenticated,
url: &str,
body: &Value,
timeout: Option<Duration>,
timeout: Duration,
stream: bool,
) -> Result<reqwest::Response, Error> {
let request = outbound_request(
authenticated,
url,
body,
Some(timeout.unwrap_or(Duration::from_secs(MESSAGES_TIMEOUT_SECS))),
)?;
Ok(litellm_inference::outbound::send(request, http).await?)
let request = outbound_request(authenticated, url, body, (!stream).then_some(timeout))?;
if !stream {
return Ok(litellm_inference::outbound::send(request, http).await?);
}
tokio::time::timeout(timeout, litellm_inference::outbound::send(request, http))
.await
.map_err(litellm_http::Error::read_timeout)?
.map_err(Error::from)
}
async fn provider_error(response: reqwest::Response) -> Error {
/// The stream's body, each read bounded by `timeout` as httpx bounds each read.
fn timed_chunks(
response: reqwest::Response,
timeout: Duration,
) -> impl futures_util::Stream<Item = Result<Bytes, litellm_http::Error>> + Send + 'static {
futures_util::stream::try_unfold(response, move |mut response| async move {
let chunk = tokio::time::timeout(timeout, response.chunk())
.await
.map_err(litellm_http::Error::body_read_timeout)?
.map_err(litellm_http::Error::from)?;
Ok(chunk.map(|chunk| (chunk, response)))
})
}
/// The provider's error, its body read under `timeout`: a stream has no overall deadline once
/// its head arrives, so an error body that stalls would otherwise hold the call forever.
async fn provider_error(response: reqwest::Response, timeout: Duration) -> Error {
let status = response.status().as_u16();
match response.text().await {
Ok(text) => {
let headers = litellm_http::request::header_pairs(response.headers());
match tokio::time::timeout(timeout, response.text()).await {
Ok(Ok(text)) => {
log_error_body(status, &text);
litellm_http::Error::upstream(status, text).into()
litellm_http::Error::upstream_with_headers(status, text, headers).into()
}
Err(error) => litellm_http::Error::from(error).into(),
Ok(Err(error)) => litellm_http::Error::from(error).into(),
Err(elapsed) => litellm_http::Error::read_timeout(elapsed).into(),
}
}
@ -329,22 +428,22 @@ fn streaming_response(
response: reqwest::Response,
decoder: Option<StreamDecoder>,
provider: &'static str,
timeout: Duration,
) -> MessagesCallResponse {
let headers = response
.headers()
.iter()
.filter_map(|(name, value)| Some((name.to_string(), value.to_str().ok()?.to_string())))
.collect();
let chunks = timed_chunks(response, timeout)
.inspect_ok(move |chunk| log_chunk(provider, "provider_response", chunk));
let chunks = match decoder {
None => futures_util::stream::try_unfold(response, move |mut response| async move {
let chunk = response.chunk().await.map_err(litellm_http::Error::from)?;
Ok(chunk.map(|chunk| {
log_chunk(provider, "provider_response", &chunk);
(chunk, response)
}))
})
.boxed(),
Some(decode) => decoded_chunks(response, decode, provider),
None => chunks.map_err(Error::from).boxed(),
Some(decode) => decoded_chunks(
chunks.map_err(std::io::Error::other).boxed(),
decode,
provider,
),
};
MessagesCallResponse::Stream {
head: super::route::MessagesStreamHead { headers },
@ -353,15 +452,10 @@ fn streaming_response(
}
fn decoded_chunks(
response: reqwest::Response,
bytes: ByteStream,
decode: StreamDecoder,
provider: &'static str,
) -> BoxStream<'static, Result<Bytes, Error>> {
let bytes: ByteStream = response
.bytes_stream()
.inspect_ok(move |chunk| log_chunk(provider, "provider_response", chunk))
.map_err(|error| std::io::Error::other(litellm_http::Error::from(error)))
.boxed();
futures_util::stream::try_unfold(decode(bytes), move |mut events| async move {
let Some(event) = events.try_next().await? else {
return Ok(None);
@ -964,9 +1058,12 @@ mod tests {
.send()
.await
.unwrap();
let MessagesCallResponse::Stream { mut chunks, .. } =
streaming_response(response, Some(streaming::event_stream), "test")
else {
let MessagesCallResponse::Stream { mut chunks, .. } = streaming_response(
response,
Some(streaming::event_stream),
"test",
Duration::from_secs(5),
) else {
panic!("a streaming response returns chunks");
};

1
litellm-rust/crates/inference-messages/src/lib.rs
View file

@ -14,6 +14,7 @@ use std::sync::Arc;
use crate::common_utils::messages_provider;
pub use common_utils::merge_messages_header_sources;
pub use litellm_inference::Error;
pub use types::{
MessagesCall, MessagesCallResponse, MessagesSettings, MessagesShaping, litellm_params,

14
litellm-rust/crates/inference-messages/tests/messages/request.rs
View file

@ -440,7 +440,12 @@ async fn an_oauth_key_sends_the_browser_access_header_and_the_oauth_beta(call: M
#[case::deepseek("deepseek")]
#[case::minimax("minimax")]
#[tokio::test]
async fn caller_protocol_headers_win_over_the_defaults(call: MessagesCall, #[case] provider: &str) {
/// Python adds a default unless that exact lowercase name is present, then folds the casings
/// into one header, so a capitalised caller header is sent beside the default.
async fn capitalised_caller_protocol_headers_are_joined_with_the_defaults(
call: MessagesCall,
#[case] provider: &str,
) {
let upstream = upstream([message_response()]).await;
run_message(MessagesCall {
@ -457,10 +462,13 @@ async fn caller_protocol_headers_win_over_the_defaults(call: MessagesCall, #[cas
.await;
let request = only_request(&upstream).await;
assert_eq!(request.header_values("anthropic-version"), ["2024-01-01"]);
assert_eq!(
request.header_values("anthropic-version"),
["2024-01-01, 2023-06-01"]
);
assert_eq!(
request.header_values("content-type"),
["application/json; charset=utf-8"]
["application/json; charset=utf-8, application/json"]
);
}

25
litellm-rust/crates/inference-messages/tests/messages/response.rs
View file

@ -225,14 +225,25 @@ async fn an_upstream_error_keeps_its_status_and_body(call: MessagesCall, #[case]
}
#[rstest]
#[case::not_json(ResponseTemplate::new(200).set_body_string("not json"))]
#[case::not_a_message(json_response(json!({"unexpected": true})))]
#[tokio::test]
async fn an_unreadable_success_body_is_an_invalid_response(
call: MessagesCall,
#[case] response: ResponseTemplate,
) {
let upstream = upstream([response]).await;
async fn an_unparseable_success_body_is_reported_as_the_upstream_answer(call: MessagesCall) {
let upstream = upstream([ResponseTemplate::new(200).set_body_string("not json")]).await;
let error = run(MessagesCall {
api_key: Some("sk".into()),
api_base: Some(upstream.uri()),
..call
})
.await
.expect_err("an unparseable body fails");
assert_eq!(error.upstream_response(), Some((200, "not json")));
}
#[rstest]
#[tokio::test]
async fn an_unreadable_success_body_is_an_invalid_response(call: MessagesCall) {
let upstream = upstream([json_response(json!({"unexpected": true}))]).await;
let error = run(MessagesCall {
api_key: Some("sk".into()),

43
litellm-rust/crates/inference-messages/tests/messages/stream.rs
View file

@ -432,6 +432,49 @@ async fn dropping_the_sdk_stream_closes_the_unfinished_upstream(
.unwrap();
}
/// Answers with an error head whose promised body never arrives.
async fn stalling_error_upstream() -> (String, JoinHandle<()>) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let connection = tokio::spawn(async move {
let (mut socket, _) = listener.accept().await.unwrap();
let mut request = vec![0; 4096];
let _ = socket.read(&mut request).await;
socket
.write_all(b"HTTP/1.1 429 Too Many Requests\r\ncontent-type: application/json\r\ncontent-length: 100\r\n\r\n")
.await
.unwrap();
let _ = socket.read_to_end(&mut Vec::new()).await;
});
(base, connection)
}
#[rstest]
#[tokio::test]
async fn the_timeout_covers_a_stalled_stream_error_body(call: MessagesCall) {
let (base, connection) = stalling_error_upstream().await;
let host = RecordingStreamHost::new(
MessagesCall {
timeout: Some(Duration::from_millis(300)),
..streaming(call, base)
},
usize::MAX,
);
let error = tokio::time::timeout(Duration::from_secs(5), stream_through(&host))
.await
.expect("the stalled error body gives up within the timeout")
.expect_err("a stalled error body fails the call");
assert!(error.dispatched(), "{error:?}");
assert_eq!(error.upstream_response(), None);
assert!(host.seen.into_inner().unwrap().is_empty());
tokio::time::timeout(Duration::from_secs(5), connection)
.await
.expect("timing out closes the upstream connection")
.unwrap();
}
#[rstest]
#[tokio::test]
async fn the_sdk_yields_a_body_error_once_after_delivered_chunks(call: MessagesCall) {

4
litellm-rust/crates/inference-ocr/src/handler.rs
View file

@ -91,9 +91,7 @@ impl<H: Interceptors<Error>> CallHooks<Error> for OcrCallHooks<'_, H> {
}
fn response_received<'a>(&'a self, body: &'a [u8]) -> BoxFuture<'a, Result<(), Error>> {
let raw = RawResponse {
body: String::from_utf8_lossy(body).into_owned(),
};
let raw = RawResponse::new(String::from_utf8_lossy(body).into_owned());
if let Some(observers) = self.observers {
observers.emit(litellm_host::lifecycle::CallEvent::Execution(
ExecutionEvent::ProviderResponseReceived { raw: raw.clone() },

2
litellm-rust/crates/inference-responses/src/handler.rs
View file

@ -114,7 +114,7 @@ pub(super) async fn execute(
});
}
let body = response.text().await.map_err(litellm_http::Error::from)?;
let raw = RawResponse { body: body.clone() };
let raw = RawResponse::new(body.clone());
if let Some(observers) = observers {
observers.emit(litellm_host::lifecycle::CallEvent::Execution(
ExecutionEvent::ProviderResponseReceived { raw: raw.clone() },

8
litellm-rust/crates/inference/src/context.rs
View file

@ -42,9 +42,11 @@ impl<'a, I: Interceptors<Error>> CallContext<'a, I> {
}
pub async fn response_received(&self, body: &str) -> Result<(), Error> {
let raw = RawResponse {
body: body.to_owned(),
};
self.raw_response_received(RawResponse::new(body.to_owned()))
.await
}
pub async fn raw_response_received(&self, raw: RawResponse) -> Result<(), Error> {
if let Some(observers) = &self.observers {
observers.emit(CallEvent::Execution(
ExecutionEvent::ProviderResponseReceived { raw: raw.clone() },

28
litellm-rust/crates/inference/src/error.rs
View file

@ -146,6 +146,34 @@ impl Error {
}
}
/// The provider's response headers on an upstream failure; empty for every other error.
pub fn upstream_headers(&self) -> &[(String, String)] {
match self {
Self::Llm(error) => error.upstream_headers(),
Self::Http(error) => error.upstream_headers(),
_ => &[],
}
}
/// A read of a streamed body timed out after the response head arrived.
pub fn body_read_timeout(&self) -> bool {
match self {
Self::Llm(error) => error.is_body_read_timeout(),
Self::Http(error) => error.is_body_read_timeout(),
_ => false,
}
}
/// `Some(connect)` when the call timed out, where `connect` says the connection was never
/// established; `None` for every other error.
pub fn timeout(&self) -> Option<bool> {
match self {
Self::Llm(error) => error.is_timeout(),
Self::Http(error) if error.is_timeout() => Some(error.is_connect_timeout()),
_ => None,
}
}
pub fn upstream_response(&self) -> Option<(u16, &str)> {
match self {
Self::Llm(error) => error.upstream_response(),

7
litellm-rust/crates/llms-types/src/formats/messages/request.rs
View file

@ -264,7 +264,12 @@ pub enum ThinkingDisplay {
#[macro_rules_attribute::apply(crate::wire_type)]
#[derive(Default)]
pub struct EnabledThinking {
#[serde(default, skip_serializing_if = "Option::is_none")]
/// An explicit `null` stays `Unrecognized(Null)`: Python rejects it, but defaults a missing key to 0.
#[serde(
default,
skip_serializing_if = "Option::is_none",
deserialize_with = "crate::recognized::present"
)]
pub budget_tokens: Option<Recognized<u64>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub display: Option<Recognized<ThinkingDisplay>>,

36
litellm-rust/crates/llms-types/src/formats/messages/response.rs
View file

@ -10,14 +10,29 @@ pub struct MessagesResponse {
pub content: Vec<Value>,
pub stop_reason: Option<String>,
pub stop_sequence: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
/// `Some(Value::Null)` when the provider sent an explicit `null`, which Python keeps.
#[serde(
default,
deserialize_with = "present_value",
skip_serializing_if = "Option::is_none"
)]
pub usage: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(
default,
deserialize_with = "present_value",
skip_serializing_if = "Option::is_none"
)]
pub container: Option<Value>,
#[serde(flatten)]
pub extra: Map<String, Value>,
}
fn present_value<'de, D: serde::Deserializer<'de>>(
deserializer: D,
) -> Result<Option<Value>, D::Error> {
<Value as serde::Deserialize<'de>>::deserialize(deserializer).map(Some)
}
#[cfg(test)]
mod tests {
use super::*;
@ -73,4 +88,21 @@ mod tests {
assert_eq!(body.get("usage").cloned(), usage);
assert_eq!(body.get("container").cloned(), container);
}
#[test]
fn explicit_null_usage_and_container_survive_a_round_trip() {
let body = json!({"id": "msg_1", "type": "message", "role": "assistant", "model": "m",
"content": [], "stop_reason": null, "stop_sequence": null,
"usage": null, "container": null});
let parsed: MessagesResponse = serde_json::from_value(body.clone()).expect("parses");
assert_eq!(serde_json::to_value(parsed).expect("serializable"), body);
}
#[test]
fn absent_usage_and_container_stay_absent() {
let body = json!({"id": "msg_1", "type": "message", "role": "assistant", "model": "m",
"content": [], "stop_reason": null, "stop_sequence": null});
let parsed: MessagesResponse = serde_json::from_value(body.clone()).expect("parses");
assert_eq!(serde_json::to_value(parsed).expect("serializable"), body);
}
}

10
litellm-rust/crates/llms-types/src/recognized.rs
View file

@ -7,6 +7,16 @@ pub enum Recognized<T> {
Unrecognized(Value),
}
/// `deserialize_with` for an optional `Recognized` field whose explicit `null` must stay
/// distinguishable from an absent key: `null` becomes `Some(Unrecognized(Null))`.
pub fn present<'de, D, T>(deserializer: D) -> Result<Option<Recognized<T>>, D::Error>
where
D: serde::Deserializer<'de>,
Recognized<T>: serde::Deserialize<'de>,
{
<Recognized<T> as serde::Deserialize<'de>>::deserialize(deserializer).map(Some)
}
impl<T> Recognized<T> {
pub fn known(&self) -> Option<&T> {
match self {

57
litellm-rust/crates/llms/src/anthropic/common_utils.rs
View file

@ -154,11 +154,42 @@ pub fn complete_anthropic_url(
) -> String {
let api_base = resolve_anthropic_api_base(api_base, env_lookup);
let api_base = api_base.trim_end_matches('/');
if api_base.ends_with(MESSAGES_PATH) {
ensure_api_route(&api_base, MESSAGES_PATH)
}
/// Python's `urlsplit(url).path`, for the URLs an `api_base` holds.
fn url_path(url: &str) -> &str {
let before_fragment = url.split('#').next().unwrap_or_default();
let before_query = before_fragment.split('?').next().unwrap_or_default();
match before_query.split_once("://") {
Some((_, rest)) => rest.find('/').map_or("", |slash| &rest[slash..]),
None => before_query,
}
}
/// Python's `_strip_trailing_slashes_keeping_the_scheme`.
fn strip_trailing_slashes(api_base: &str) -> String {
let stripped = api_base.trim_end_matches('/');
if stripped.ends_with(':') && !stripped.contains('/') && !stripped.contains('?') {
return api_base.to_string();
}
format!("{api_base}{MESSAGES_PATH}")
match stripped.split_once('?') {
Some((before, query)) if url_path(stripped).ends_with('/') => {
format!("{}?{query}", before.trim_end_matches('/'))
}
_ => stripped.to_string(),
}
}
/// Python's `ensure_api_route`: an `api_base` whose path already ends in the route is kept as
/// it is, query string included; anything else gets the route appended to the whole string.
fn ensure_api_route(api_base: &str, route: &str) -> String {
let base = strip_trailing_slashes(api_base);
let route = route.trim_start_matches('/');
if url_path(&base).ends_with(&format!("/{route}")) {
return base;
}
format!("{}/{route}", strip_trailing_slashes(&base))
}
pub fn existing_betas(headers: &[(String, String)]) -> BetaSet {
@ -1922,4 +1953,24 @@ mod tests {
expected
);
}
#[rstest]
#[case::bare_host("https://api.example.com", "https://api.example.com/v1/messages")]
#[case::trailing_slash("https://api.example.com/", "https://api.example.com/v1/messages")]
#[case::route_present(
"https://api.example.com/v1/messages/",
"https://api.example.com/v1/messages"
)]
#[case::route_and_query(
"https://api.example.com/v1/messages?beta=true",
"https://api.example.com/v1/messages?beta=true"
)]
#[case::query_without_route(
"https://api.example.com/?x=1",
"https://api.example.com?x=1/v1/messages"
)]
#[case::scheme_only("https://", "https:///v1/messages")]
fn the_messages_url_follows_python_ensure_api_route(#[case] api_base: &str, #[case] url: &str) {
assert_eq!(complete_anthropic_url(Some(api_base), &|_| None), url);
}
}

233
litellm-rust/crates/llms/src/anthropic/messages/thinking.rs
View file

@ -2,8 +2,8 @@ use litellm_llms_types::{
formats::{
chat_completions::ReasoningEffort,
messages::{
EffortLevel, MessagesOptionalParams, MessagesRequest, OutputConfig, ThinkingConfig,
ThinkingDisplay,
AdaptiveThinking, EffortLevel, MessagesOptionalParams, MessagesRequest, OutputConfig,
ThinkingConfig, ThinkingDisplay,
},
},
recognized::Recognized,
@ -36,16 +36,17 @@ fn budget_for_effort(budgets: &ThinkingBudgets, effort: ReasoningEffort) -> Opti
fn effort_for_budget(
budgets: &ThinkingBudgets,
budget_tokens: u64,
budget_tokens: i64,
capabilities: &MessagesModelCapabilities,
) -> EffortLevel {
if budget_tokens >= budgets.xhigh && capabilities.effort_tiers.xhigh {
let reaches = |tier: u64| i128::from(budget_tokens) >= i128::from(tier);
if reaches(budgets.xhigh) && capabilities.effort_tiers.xhigh {
return EffortLevel::Xhigh;
}
if budget_tokens >= budgets.high {
if reaches(budgets.high) {
return EffortLevel::High;
}
if budget_tokens >= budgets.medium {
if reaches(budgets.medium) {
return EffortLevel::Medium;
}
EffortLevel::Low
@ -231,32 +232,129 @@ fn drop_disabled_thinking(request: MessagesRequest, context: &ThinkingContext) -
}
}
/// `_numeric_budget_tokens`: Python reads any int, float (truncated) or int-parseable string
/// as a budget, and nothing else.
fn numeric_budget_tokens(budget: &Recognized<u64>) -> Option<i64> {
let value = match budget {
Recognized::Known(budget) => return Some(i64::try_from(*budget).unwrap_or(i64::MAX)),
Recognized::Unrecognized(value) => value,
};
match value {
Value::Number(number) => number
.as_i64()
.or_else(|| number.as_f64().map(|float| float.trunc() as i64)),
Value::String(text) => python_int(text),
Value::Null | Value::Bool(_) | Value::Array(_) | Value::Object(_) => None,
}
}
/// Python's `int(str)`: surrounding whitespace, a sign, and single underscores between digits.
fn python_int(text: &str) -> Option<i64> {
let trimmed = text.trim();
let digits = trimmed.trim_start_matches(['+', '-']);
let well_formed = trimmed.len() - digits.len() <= 1
&& !digits.is_empty()
&& !digits.starts_with('_')
&& !digits.ends_with('_')
&& !digits.contains("__")
&& digits.chars().all(|ch| ch.is_ascii_digit() || ch == '_');
if !well_formed {
return None;
}
let cleaned: String = trimmed.chars().filter(|ch| *ch != '_').collect();
cleaned.parse::<i64>().ok().or_else(|| {
Some(if cleaned.starts_with('-') {
i64::MIN
} else {
i64::MAX
})
})
}
fn python_type_name(value: &Value) -> &'static str {
match value {
Value::Null => "NoneType",
Value::Bool(_) => "bool",
Value::Number(number) if number.is_f64() => "float",
Value::Number(_) => "int",
Value::String(_) => "str",
Value::Array(_) => "list",
Value::Object(_) => "dict",
}
}
fn kept_display(
display: Option<&Recognized<ThinkingDisplay>>,
context: &ThinkingContext,
) -> Option<Recognized<ThinkingDisplay>> {
match display? {
Recognized::Known(ThinkingDisplay::Summarized | ThinkingDisplay::Omitted) => {
display.cloned()
}
Recognized::Known(ThinkingDisplay::Updates) if context.keeps_display_updates => {
display.cloned()
}
Recognized::Known(ThinkingDisplay::Updates) | Recognized::Unrecognized(_) => None,
}
}
/// Keys Python's `carried_thinking_keys` keeps on the rebuilt adaptive config: the adaptive
/// keys always, and any other key unless `drop_params` is set.
fn carried_thinking_keys(
extra: &serde_json::Map<String, Value>,
context: &ThinkingContext,
) -> serde_json::Map<String, Value> {
extra
.iter()
.filter(|(key, _)| key.as_str() == "block_binding" || !context.drop_params)
.map(|(key, value)| (key.clone(), value.clone()))
.collect()
}
/// `AnthropicModelInfo.translate_legacy_thinking_for_adaptive_model_silently`.
fn translate_legacy_thinking_for_adaptive_model(
request: MessagesRequest,
context: &ThinkingContext,
) -> MessagesRequest {
) -> Result<MessagesRequest, Error> {
let capabilities = &context.capabilities;
if !capabilities.supports_adaptive_thinking || capabilities.supports_legacy_thinking {
return request;
return Ok(request);
}
let Some(ThinkingConfig::Enabled(enabled)) = known_thinking(&request) else {
return request;
return Ok(request);
};
let budget = enabled
.budget_tokens
.as_ref()
.and_then(Recognized::known)
.copied()
.unwrap_or(0);
let level = effort_for_budget(&context.budgets, budget, capabilities);
MessagesRequest {
let budget = match &enabled.budget_tokens {
None => Some(0),
Some(budget) => numeric_budget_tokens(budget),
};
if budget.is_none() && !context.drop_params {
let actual = match &enabled.budget_tokens {
Some(Recognized::Unrecognized(value)) => python_type_name(value),
Some(Recognized::Known(_)) | None => "int",
};
return Err(Error::InvalidRequest(
crate::ErrorDetail::NonNumericThinkingBudget { actual },
));
}
let thinking = ThinkingConfig::Adaptive(AdaptiveThinking {
display: kept_display(enabled.display.as_ref(), context),
extra: carried_thinking_keys(&enabled.extra, context),
});
let output_config = match budget {
Some(budget) => with_default_effort(
request.params.output_config,
effort_for_budget(&context.budgets, budget, capabilities),
),
None => request.params.output_config,
};
Ok(MessagesRequest {
params: MessagesOptionalParams {
thinking: Some(Recognized::Known(ThinkingConfig::adaptive(None))),
output_config: with_default_effort(request.params.output_config, level),
thinking: Some(Recognized::Known(thinking)),
output_config,
..request.params
},
..request
}
})
}
fn translate_adaptive_effort_for_non_adaptive_model(
@ -337,7 +435,7 @@ pub fn translate_thinking(
) -> Result<MessagesRequest, Error> {
let request = translate_reasoning_effort(request, context)?;
let request = drop_disabled_thinking(request, context);
let request = translate_legacy_thinking_for_adaptive_model(request, context);
let request = translate_legacy_thinking_for_adaptive_model(request, context)?;
let request = translate_adaptive_effort_for_non_adaptive_model(request, context)?;
Ok(drop_incompatible_temperature_for_thinking(request, context))
}
@ -363,6 +461,7 @@ mod tests {
ThinkingContext {
capabilities,
budgets: ThinkingBudgets::default(),
..ThinkingContext::default()
}
}
@ -1239,10 +1338,102 @@ mod tests {
let context = ThinkingContext {
capabilities,
budgets: overridden_budgets(overrides),
..ThinkingContext::default()
};
assert_eq!(
translate_thinking(request(input), &context),
Ok(request(expected))
);
}
use serde_json::json;
fn legacy_context(drop_params: bool, keeps_display_updates: bool) -> ThinkingContext {
ThinkingContext {
drop_params,
keeps_display_updates,
..context(opus_4_7())
}
}
#[rstest]
#[case::float_budget(json!(8192.0), "xhigh")]
#[case::float_below_a_tier(json!(1023.9), "low")]
#[case::string_budget(json!(" 16_000 "), "xhigh")]
#[case::negative_budget(json!(-5), "low")]
fn legacy_budgets_read_like_python_numbers(#[case] budget: Value, #[case] effort: &str) {
let translated = translate_thinking(
request(
json!({"max_tokens": 64000, "thinking": {"type": "enabled", "budget_tokens": budget}}),
),
&legacy_context(false, false),
);
assert_eq!(
translated,
Ok(request(json!({
"max_tokens": 64000,
"thinking": {"type": "adaptive"},
"output_config": {"effort": effort},
})))
);
}
#[rstest]
#[case::text(json!("lots"), "str")]
#[case::list(json!([1]), "list")]
#[case::null(Value::Null, "NoneType")]
#[case::boolean(json!(true), "bool")]
fn a_non_numeric_legacy_budget_is_rejected_unless_dropped(
#[case] budget: Value,
#[case] actual: &'static str,
) {
let fields =
json!({"max_tokens": 8192, "thinking": {"type": "enabled", "budget_tokens": budget}});
assert_eq!(
translate_thinking(request(fields.clone()), &legacy_context(false, false)),
Err(Error::InvalidRequest(
crate::ErrorDetail::NonNumericThinkingBudget { actual }
))
);
assert_eq!(
translate_thinking(request(fields), &legacy_context(true, false)),
Ok(request(
json!({"max_tokens": 8192, "thinking": {"type": "adaptive"}})
))
);
}
#[rstest]
#[case::summarized("summarized", false, Some("summarized"))]
#[case::updates_kept("updates", true, Some("updates"))]
#[case::updates_dropped("updates", false, None)]
#[case::unknown("verbose", true, None)]
fn legacy_display_survives_only_where_python_keeps_it(
#[case] display: &str,
#[case] keeps_display_updates: bool,
#[case] kept: Option<&str>,
) {
let translated = translate_thinking(
request(json!({"max_tokens": 4096, "thinking": {"type": "enabled", "budget_tokens": 2048, "display": display}})),
&legacy_context(false, keeps_display_updates),
)
.unwrap();
let thinking = serde_json::to_value(translated.params.thinking).unwrap();
assert_eq!(thinking.get("display").and_then(Value::as_str), kept);
}
#[rstest]
#[case::kept(false, json!({"type": "adaptive", "block_binding": "strict", "note": "x"}))]
#[case::dropped(true, json!({"type": "adaptive", "block_binding": "strict"}))]
fn legacy_extra_keys_follow_drop_params(#[case] drop_params: bool, #[case] expected: Value) {
let translated = translate_thinking(
request(json!({"max_tokens": 4096, "thinking": {"type": "enabled", "budget_tokens": 2048, "block_binding": "strict", "note": "x"}})),
&legacy_context(drop_params, false),
)
.unwrap();
assert_eq!(
serde_json::to_value(translated.params.thinking).unwrap(),
expected
);
}
}

28
litellm-rust/crates/llms/src/anthropic/messages/transformation.rs
View file

@ -56,7 +56,7 @@ impl BaseMessagesConfig for AnthropicMessagesConfig {
request: MessagesRequest,
context: &MessagesTransformContext,
) -> Result<MessagesRequest, Error> {
transform_messages_request(request, context)
transform_messages_request(request, &context.keeping_display_updates())
}
fn secret_names(&self) -> &'static [&'static str] {
@ -206,6 +206,8 @@ fn feature_betas(request: &MessagesRequest) -> BetaSet {
.then_some(AnthropicBeta::PerTurnControl20260701),
uses_thinking_display_updates(params)
.then_some(AnthropicBeta::ThinkingDisplayUpdates20260818),
uses_thinking_block_binding(params)
.then_some(AnthropicBeta::ThinkingBindingControls20260801),
messages_carry_tool_changes(&request.messages)
.then_some(AnthropicBeta::MidConversationToolChanges20260701),
has_advisor_tool(tools).then_some(AnthropicBeta::AdvisorTool20260301),
@ -267,6 +269,25 @@ fn uses_thinking_display_updates(params: &MessagesOptionalParams) -> bool {
display.and_then(Recognized::known) == Some(&ThinkingDisplay::Updates)
}
/// `AnthropicModelInfo.thinking_betas`: any thinking object carrying `block_binding`.
fn uses_thinking_block_binding(params: &MessagesOptionalParams) -> bool {
match params.thinking.as_ref() {
Some(Recognized::Known(ThinkingConfig::Enabled(thinking))) => {
thinking.extra.contains_key("block_binding")
}
Some(Recognized::Known(ThinkingConfig::Adaptive(thinking))) => {
thinking.extra.contains_key("block_binding")
}
Some(Recognized::Known(ThinkingConfig::Disabled(thinking))) => {
thinking.extra.contains_key("block_binding")
}
Some(Recognized::Unrecognized(Value::Object(thinking))) => {
thinking.contains_key("block_binding")
}
Some(Recognized::Unrecognized(_)) | None => false,
}
}
fn is_tool_reference_change(block: &ContentBlock) -> bool {
matches!(
&block.block_type,
@ -297,7 +318,8 @@ fn unsupported_param(model: &str, param: &str, value: &str, hint: &str) -> Error
})
}
fn drop_unsupported_params(
/// The speed and sampling gates of Python's `get_requested_anthropic_messages_optional_param`.
pub fn drop_unsupported_params(
request: MessagesRequest,
context: &MessagesTransformContext,
) -> Result<MessagesRequest, Error> {
@ -838,6 +860,8 @@ mod tests {
low: 2000,
..ThinkingBudgets::default()
},
drop_params: true,
keeps_display_updates: false,
},
drop_params: true,
}

2
litellm-rust/crates/llms/src/azure_ai/messages/transformation.rs
View file

@ -68,7 +68,7 @@ impl BaseMessagesConfig for AzureAnthropicMessagesConfig {
.capabilities
.supports_mid_conversation_system,
)),
context,
&context.keeping_display_updates(),
)
}

19
litellm-rust/crates/llms/src/base_llm/messages/context.rs
View file

@ -110,6 +110,11 @@ impl ThinkingBudgets {
pub struct ThinkingContext {
pub capabilities: MessagesModelCapabilities,
pub budgets: ThinkingBudgets,
/// `drop_params` as Python's thinking translation reads it.
pub drop_params: bool,
/// The provider accepts `display: "updates"` on thinking translated from the legacy shape
/// (Python's `LEGACY_THINKING_DISPLAY_UPDATES_PROVIDERS`).
pub keeps_display_updates: bool,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct MessagesTransformContext {
@ -130,8 +135,22 @@ impl MessagesTransformContext {
thinking: ThinkingContext {
capabilities,
budgets: ThinkingBudgets::from_lookup(env),
drop_params,
keeps_display_updates: false,
},
drop_params,
}
}
/// The same context for a provider in Python's `LEGACY_THINKING_DISPLAY_UPDATES_PROVIDERS`.
#[must_use]
pub fn keeping_display_updates(&self) -> Self {
Self {
thinking: ThinkingContext {
keeps_display_updates: true,
..self.thinking
},
..*self
}
}
}

22
litellm-rust/crates/llms/src/error.rs
View file

@ -92,6 +92,24 @@ impl Error {
}
}
pub fn upstream_headers(&self) -> &[(String, String)] {
match self {
Self::Http(error) => error.upstream_headers(),
_ => &[],
}
}
pub fn is_body_read_timeout(&self) -> bool {
matches!(self, Self::Http(error) if error.is_body_read_timeout())
}
pub fn is_timeout(&self) -> Option<bool> {
match self {
Self::Http(error) if error.is_timeout() => Some(error.is_connect_timeout()),
_ => None,
}
}
pub fn upstream_response(&self) -> Option<(u16, &str)> {
match self {
Self::Http(error) => error.upstream_response(),
@ -158,6 +176,10 @@ pub enum ErrorDetail {
value: String,
hint: String,
},
#[error(
"thinking.budget_tokens must be a number, got {actual}. To drop it, set `litellm.drop_params=True`, or `drop_params: true` under `litellm_settings` on the proxy"
)]
NonNumericThinkingBudget { actual: &'static str },
#[error(
"Missing API Base - A call is being made to LLM Provider but no api base is set either in the environment variables ({{LLM_PROVIDER}}_API_KEY) or via params"
)]

2
litellm-rust/crates/llms/src/vertex_ai/messages/transformation.rs
View file

@ -77,7 +77,7 @@ impl BaseMessagesConfig for VertexAiPartnerModelsAnthropicMessagesConfig {
request: MessagesRequest,
context: &MessagesTransformContext,
) -> Result<MessagesRequest, Error> {
let request = transform_messages_request(request, context)?;
let request = transform_messages_request(request, &context.keeping_display_updates())?;
let request = strip_cache_control_scope(normalize_system_role_messages(
request,
context

59
litellm-rust/crates/python-bridge/src/errors.rs
View file

@ -23,6 +23,17 @@ pyo3::create_exception!(
);
const MESSAGES_REQUEST_ERROR: &str = "messages_request_error";
/// The rejection is a parameter the model does not support, which Python raises as
/// `UnsupportedParamsError`.
const MESSAGES_UNSUPPORTED_PARAM: &str = "messages_unsupported_param";
/// The phase label (`Connection` or `Read`) of a timed-out call, for Python's timeout text, or
/// `Body` for a streamed body read that timed out after the head arrived.
const MESSAGES_TIMEOUT: &str = "messages_timeout";
/// No credential reached the provider, which Python raises as `AuthenticationError`.
const MESSAGES_MISSING_CREDENTIAL: &str = "messages_missing_credential";
/// Python refuses the request only once its logger knows the resolved model and provider (the
/// provider config's request transform), so the failure logs them.
const MESSAGES_LATE_REJECTION: &str = "messages_late_rejection";
pub(crate) fn upstream_error(
py: Python<'_>,
@ -78,17 +89,57 @@ pub(crate) fn route_failure(
return Ok(raised);
}
if let Some((status, body)) = error.upstream_response() {
return public(upstream_error(py, status, body, &HeaderMap::new())?);
let headers = error
.upstream_headers()
.iter()
.filter_map(|(name, value)| {
Some((
reqwest::header::HeaderName::from_bytes(name.as_bytes()).ok()?,
reqwest::header::HeaderValue::from_str(value).ok()?,
))
})
.collect::<HeaderMap>();
return public(upstream_error(py, status, body, &headers)?);
}
let native = match (marker, error.rejected_request()) {
(RequestMarker::Messages, Some(rejection)) => {
let native = PyValueError::new_err(Report::new(rejection).to_string());
native.value(py).setattr(MESSAGES_REQUEST_ERROR, true)?;
let value = native.value(py);
value.setattr(MESSAGES_REQUEST_ERROR, true)?;
if matches!(
rejection.downcast_ref::<litellm_llms::ErrorDetail>(),
Some(
litellm_llms::ErrorDetail::UnsupportedParameter { .. }
| litellm_llms::ErrorDetail::NonNumericThinkingBudget { .. }
)
) {
value.setattr(MESSAGES_UNSUPPORTED_PARAM, true)?;
}
if matches!(
rejection.downcast_ref::<litellm_llms::ErrorDetail>(),
Some(litellm_llms::ErrorDetail::NonNumericThinkingBudget { .. })
) {
value.setattr(MESSAGES_LATE_REJECTION, true)?;
}
native
}
(RequestMarker::Messages, None) | (RequestMarker::Unmarked, _) => {
by_fault(error.is_request(), &error)
(RequestMarker::Messages, None) => {
let native = by_fault(error.is_request(), &error);
let value = native.value(py);
if let Some(connect) = error.timeout() {
let phase = match (connect, error.body_read_timeout()) {
(true, _) => "Connection",
(false, true) => "Body",
(false, false) => "Read",
};
value.setattr(MESSAGES_TIMEOUT, phase)?;
}
if error.is_missing_credential() {
value.setattr(MESSAGES_MISSING_CREDENTIAL, true)?;
}
native
}
(RequestMarker::Unmarked, _) => by_fault(error.is_request(), &error),
};
public(native)
}

21
litellm-rust/crates/python-bridge/src/routes/messages/host.rs
View file

@ -34,11 +34,9 @@ fn merge_headers(
forwarded: Option<Map<String, Value>>,
extra_headers: Option<Map<String, Value>>,
) -> Option<Map<String, Value>> {
let merged: Map<String, Value> = forwarded
.into_iter()
.flatten()
.chain(extra_headers.into_iter().flatten())
.collect();
let merged = litellm_inference_messages::merge_messages_header_sources(
forwarded.into_iter().chain(extra_headers),
);
(!merged.is_empty()).then_some(merged)
}
@ -107,10 +105,19 @@ impl MessagesPythonHost {
.map(|value| python_timeout_seconds(py, value.unbind()))
.transpose()?
.flatten();
let custom_llm_provider = string("custom_llm_provider")?;
let shaping = self.shaping(py, &model, custom_llm_provider.as_deref(), arguments)?;
let api_key = string("api_key")?;
let api_base = string("api_base")?;
// Python's handler resolves a bare model name (`claude-sonnet-4-5`) through
// `get_llm_provider`'s model lists, which the native prefix lookup does not carry.
let custom_llm_provider = match string("custom_llm_provider")? {
Some(provider) => Some(provider),
None => py
.import(ROUTE_HOST_MODULE)?
.getattr("inferred_provider")?
.call1((&model, api_base.as_deref()))?
.extract()?,
};
let shaping = self.shaping(py, &model, custom_llm_provider.as_deref(), arguments)?;
let extra_headers = self.merged_headers(py, arguments)?;
let provider_specific_header = self.provider_specific_header(py, arguments)?;
let litellm_params = project_litellm_params(argument, |name| module_global(py, name))?;

13
litellm-rust/crates/python-bridge/src/routes/messages/mod.rs
View file

@ -50,9 +50,18 @@ fn run_messages(py: Python<'_>, call: NativeCall<'_>, asynchronous: bool) -> PyR
)
}
/// Python's sync Messages handler raises for every native provider before it sends anything
/// or opens a logging object (`BaseLLMHTTPHandler.anthropic_messages_handler`), and Rust copies
/// that until both sides serve sync calls.
const SYNC_NOT_IMPLEMENTED: &str = "anthropic_messages_handler is not implemented for sync calls";
#[pyfunction]
pub(crate) fn messages(py: Python<'_>, call: NativeCall<'_>) -> PyResult<Py<PyAny>> {
run_messages(py, call, false)
#[pyo3(signature = (call))]
pub(crate) fn messages(call: NativeCall<'_>) -> PyResult<Py<PyAny>> {
drop(call);
Err(pyo3::exceptions::PyValueError::new_err(
SYNC_NOT_IMPLEMENTED,
))
}
#[pyfunction]

45
litellm/rust_bridge/host/callbacks_legacy_python.py
View file

@ -12,7 +12,7 @@ import datetime
import traceback
import uuid
import weakref
from collections.abc import Awaitable, Coroutine, Mapping
from collections.abc import Awaitable, Coroutine, Mapping, Sequence
from dataclasses import dataclass
from typing import (
TYPE_CHECKING,
@ -22,6 +22,7 @@ from typing import (
cast, # noqa: TID251 # bounded compatibility calls into legacy Python integrations
)
import httpx
from pydantic import TypeAdapter
if TYPE_CHECKING:
@ -243,6 +244,27 @@ def post_call(
logger.post_call(original_response=original_response, api_key=api_key, additional_args=additional_args)
def record_response(logger: Logging, status: int, headers: Sequence[tuple[str, str]], body: str) -> None:
"""Stores the provider response the way Python's Messages handler does, for logging to replay."""
response: Final = httpx.Response(status_code=status, headers=list(headers), content=body.encode())
details: Final = logger.model_call_details # the call's own logger: recording on it is the point
details["httpx_response"] = response
details["response_headers"] = dict(response.headers)
def record_api_window(logger: Logging, start: float, end: float) -> None:
"""Records the provider exchange's window the way Python's timed `AsyncHTTPHandler.post` does."""
from litellm.litellm_core_utils.logging_utils import (
_set_duration_in_model_call_details, # pyright: ignore[reportPrivateUsage] # the timed post's own recorder
)
_set_duration_in_model_call_details(
logging_obj=logger,
start_time=datetime.datetime.fromtimestamp(start), # noqa: DTZ006 # the timed post records naive local times
end_time=datetime.datetime.fromtimestamp(end), # noqa: DTZ006 # the timed post records naive local times
)
def defers_async_logging(logger: LoggingSurface) -> bool:
return bool(getattr(logger, "defer_async_logging", False))
@ -359,6 +381,27 @@ def after_deployment_failure(kwargs: dict[str, object], error: Exception, call_t
_STREAM_HEADERS: Final = TypeAdapter(Mapping[str, str])
class _HiddenParamsCarrier:
"""Stands in for the stream `@client` stamps, so the head becomes its `_hidden_params`."""
def __init__(self, hidden_params: dict[str, object]) -> None: # mutable-ok: the stamps land in this dict
self._hidden_params: object = hidden_params
def stamp_stream_head(
head: dict[str, object], # mutable-ok: the stream's hidden params, stamped in place
logger: Logging,
kwargs: dict[str, object], # mutable-ok: `finalize` takes the call kwargs as a dict
start_time: datetime.datetime,
end_time: datetime.datetime,
) -> None:
carrier: Final = _HiddenParamsCarrier(head)
finalize(carrier, logger, kwargs, start_time, end_time)
stamped: Final = carrier._hidden_params # pyright: ignore[reportPrivateUsage] # the carrier's own field
if stamped is not head and isinstance(stamped, dict):
head.update(cast("dict[str, object]", stamped)) # cast-ok: update_response_metadata writes a str-keyed dict
def stream_opened(logger: Logging, head: Mapping[str, object]) -> None:
logger.stream = True
logger.model_call_details["stream"] = True

200
litellm/rust_bridge/host/routes/messages.py
View file

@ -1,21 +1,35 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
import datetime
from collections.abc import Callable, Mapping, Sequence
from dataclasses import asdict, dataclass
from typing import Final, cast # noqa: TID251 # narrows the normalized native payload to the public TypedDict
from typing import Final, TypeAlias, cast # noqa: TID251 # narrows native payloads to public TypedDicts
import httpx
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.litellm_logging import Logging as LiteLLMLoggingObj
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
from litellm.llms.anthropic.messages.native import handler
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 (
_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.rust_bridge import failures
from litellm.rust_bridge.host import callbacks_legacy_python
from litellm.rust_bridge.public_call import known_params, optional_str
from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse
_DROP_PATHS: Final = TypeAdapter(list[object])
_METADATA: Final = TypeAdapter(dict[str, object])
_BODY_FIELDS: Final = frozenset(
{
"max_tokens",
@ -40,6 +54,7 @@ _BODY_FIELDS: Final = frozenset(
"cache_control",
"reasoning_effort",
"safeguards",
"service_tier",
}
)
@ -59,8 +74,40 @@ def response(value: Mapping[str, object]) -> AnthropicMessagesResponse:
)
def inferred_provider(model: str, api_base: str | None) -> str | None:
"""The provider Python's handler resolves for a call that names none, or None when the model
carries no provider `get_llm_provider` knows."""
try:
_, provider, _, _ = litellm.get_llm_provider(model=model, api_base=api_base)
except litellm.BadRequestError:
return None
return provider
def body_params(request: Mapping[str, object]) -> Mapping[str, object]:
return known_params(_BODY_FIELDS, request)
params: Final = known_params(_BODY_FIELDS, request)
if deployment_reasoning_effort_levels(request.get("model_info")) is None:
return params
return _with_deployment_effort(params, request)
def _with_deployment_effort(params: Mapping[str, object], request: Mapping[str, object]) -> Mapping[str, object]:
"""Python's handler rounds the requested effort to the levels the deployment declares in
`model_info.reasoning_effort_levels`, the one place a deployment, not the model, decides."""
explicit_provider: Final = optional_str(request.get("custom_llm_provider"))
try:
model, provider, _, _ = litellm.get_llm_provider(
model=str(request["model"]), custom_llm_provider=explicit_provider
)
except litellm.BadRequestError:
return params
return _deployment_output_effort(
params,
model=model,
custom_llm_provider=provider,
kwargs=request,
drop_params=litellm.drop_params is True or request.get("drop_params") is True,
)
def stream_hidden_params(headers: Sequence[tuple[str, str]]) -> Mapping[str, object]:
@ -71,18 +118,147 @@ def stream_hidden_params(headers: Sequence[tuple[str, str]]) -> Mapping[str, obj
return anthropic_messages_stream_hidden_params(httpx.Headers(list(headers)))
def map_failure(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception:
if getattr(error, "messages_request_error", False):
return litellm.BadRequestError(
message=str(error),
model=str(request["model"]).removeprefix(f"{request_provider}/"),
llm_provider=request_provider,
)
return failures.map_native_failure(
error, str(request["model"]), request_provider, request, optional_str(request.get("api_base"))
_UPSTREAM_ARGS: Final = TypeAdapter(tuple[int, str])
_UPSTREAM_HEADERS: Final = TypeAdapter(list[tuple[str, str]])
_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
)
_exception_type: Final = cast( # cast-ok: the legacy public exception mapper is untyped
Callable[..., Exception], exception_mapping_utils.exception_type
)
_BODY_READ_TIMEOUT: Final = "Timeout on reading data from socket" # aiohttp's SocketTimeoutError text
_MISSING_KEY: Final = (
"Missing Anthropic API Key - A call is being made to anthropic but no key is set "
"either in the environment variables or via params. Please set `ANTHROPIC_API_KEY` "
"or `ANTHROPIC_AUTH_TOKEN` in your environment vars"
)
def _rejection(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception:
"""The error Python's handler raises for a request it refuses before sending it."""
metadata: Final = request.get("metadata")
if isinstance(metadata, dict):
try:
_validate_metadata(_METADATA.validate_python(metadata))
except ValidationError as invalid:
return invalid
if getattr(error, "messages_unsupported_param", False):
return litellm.UnsupportedParamsError(message=str(error))
return litellm.BadRequestError(
message=str(error),
model=str(request["model"]).removeprefix(f"{request_provider}/"),
llm_provider=request_provider,
)
def _timeout_text(phase: str, request: Mapping[str, object]) -> str:
"""`litellm.Timeout`'s text as Python's HTTP client raises it for the same timeout."""
seconds: Final = request.get("timeout")
logging_obj: Final = request.get("litellm_logging_obj")
start_time: Final = getattr(logging_obj, "start_time", None)
time_taken: Final = (
round(
(datetime.datetime.now() - start_time).total_seconds(), # noqa: DTZ005 # the logger's start_time is naive local time
3,
)
if isinstance(start_time, datetime.datetime)
else None
)
return str(
litellm.Timeout(
message=f"{phase} timed out. Timeout passed={seconds}, time taken={time_taken} seconds",
model="default-model-name",
llm_provider="litellm-httpx-handler",
)
)
def _upstream_failure(
error: Exception, request: Mapping[str, object]
) -> tuple[int, str, Sequence[tuple[str, str]]] | None:
"""The status, text and headers Python's HTTP client reports for the same failure."""
timeout_phase: Final = getattr(error, "messages_timeout", None)
if isinstance(timeout_phase, str):
return 408, _timeout_text(timeout_phase, request), ()
try:
status, text = _UPSTREAM_ARGS.validate_python(error.args)
return status, text, _UPSTREAM_HEADERS.validate_python(getattr(error, "headers", None))
except ValidationError:
return None
def _provider_error(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception | None:
"""The provider exception Python's `_handle_error` builds for the same failure."""
failure: Final = _upstream_failure(error, request)
if failure is None:
return None
status, text, headers = failure
api_base: Final = request.get("api_base")
config: Final = native_messages_config(
str(request["model"]).removeprefix(f"{request_provider}/"),
request_provider,
request.get("model_info"),
api_base if isinstance(api_base, str) else None,
)
return (
cast(_ERROR_CLASS, config.get_error_class)( # cast-ok: the config's untyped headers parameter
error_message=text, status_code=status, headers=dict(headers)
)
if config is not None
else AnthropicError(status_code=status, message=text, headers=httpx.Headers(headers))
)
def _log_resolved_call(request: Mapping[str, object], request_provider: str) -> None:
"""Tells the logger the resolved call, as Python's handler does before its provider config
transforms the request."""
logger: Final = request.get("litellm_logging_obj")
if not isinstance(logger, LiteLLMLoggingObj):
return
callbacks_legacy_python.update_logging(
logger,
kwargs={name: value for name, value in request.items() if name != "metadata"},
model=str(request["model"]).removeprefix(f"{request_provider}/"),
optional_params=dict(body_params(request)),
litellm_params={"model_info": request.get("model_info"), **body_params(request)},
custom_llm_provider=request_provider,
)
def map_failure(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception:
"""Raises what Python's Messages handler raises for the same failure."""
if getattr(error, "messages_timeout", None) == "Body":
# Python reads a stream's body outside its HTTP client's error mapping, so the transport's
# own timeout reaches the caller unmapped.
return httpx.ReadTimeout(_BODY_READ_TIMEOUT)
if getattr(error, "messages_late_rejection", False):
_log_resolved_call(request, request_provider)
if getattr(error, "messages_request_error", False):
return _rejection(error, request, request_provider)
if getattr(error, "messages_missing_credential", False):
return litellm.AuthenticationError(
message=_MISSING_KEY,
llm_provider=request_provider,
model=str(request["model"]).removeprefix(f"{request_provider}/"),
)
provider_error: Final = _provider_error(error, request, request_provider)
if provider_error is None:
return failures.map_native_failure(
error, str(request["model"]), request_provider, request, optional_str(request.get("api_base"))
)
try:
return _exception_type(
model=str(request["model"]),
custom_llm_provider=request_provider,
original_exception=provider_error,
extra_kwargs=dict(request),
)
except Exception as mapped: # noqa: BLE001 # exception_type raises the mapped public error
mapped.__context__ = error
return mapped
def _drop_params(kwargs: Mapping[str, object]) -> bool:
return bool(litellm.drop_params) or normalize_drop_params(kwargs.get("drop_params")) is True

321
tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py
View file

@ -20,6 +20,7 @@ R request, S response, C callbacks, F covered by a fallback detector today)
"""
import asyncio
import copy
import json
import os
import re
@ -335,38 +336,12 @@ def _kw(**kwargs: object) -> Mapping[str, object]:
return MappingProxyType(kwargs)
A2_BARE_MODEL: Final = "A2: native resolves the provider from the model prefix only, so a bare Claude model name fails"
A10_NO_MAX_TOKENS: Final = "A10: native admission declines a call without an int max_tokens and Python serves it"
R3_LEGACY_THINKING: Final = "R3: native maps a float legacy thinking budget to effort low instead of Python's tier"
R4_BINDING_BETA: Final = "R4: native omits the thinking-binding-controls beta for block_binding"
R5_DEPLOYMENT_EFFORT: Final = "R5: native ignores the deployment's reasoning_effort_levels from model_info"
R7_DROP_ORDER: Final = "R7: native applies additional_drop_params before Python's unsupported-param check"
R8_VERSION_HEADER: Final = "R8: Python sends anthropic-version twice when the caller capitalises it; native once"
R9_QUERY_PATH: Final = "R9: native appends /v1/messages again to an api_base that ends in /v1/messages?query"
R11_PARAM_ERROR: Final = "R11: native raises BadRequestError where Python raises UnsupportedParamsError"
R12_METADATA_ERROR: Final = (
"R12: Python raises a pydantic ValidationError for a non-string metadata.user_id; native a BadRequestError"
)
S1_S2_ERROR_HEADERS: Final = (
"S1/S2: native drops upstream error headers and wraps the error body in a (status, body) tuple"
)
S2_ERROR_MESSAGE: Final = (
"S2: native wraps the upstream error body in a (status, body) tuple and strips the provider from model"
)
S2_S12_ERROR_STATUS: Final = "S2/S12: native keeps the upstream status where Python maps it (413 to 400, 529 to 500)"
S3_TIMEOUT: Final = "S3: native raises APIConnectionError (500) for a timeout where Python raises Timeout (408)"
S5_STREAM_DEADLINE: Final = "S5: native applies the timeout to the whole stream instead of to each read"
A10_NO_MAX_TOKENS: Final = "A10: native admission hands a call without an int max_tokens to Python, whose signature raises the TypeError; only dispatch differs"
S6_PRIVATE_KEYS: Final = "S6: native writes _hidden_params into the response dict"
S8_TYPED_RESPONSE: Final = "S8: native re-serializes the response and drops null usage and container"
S9_SYNC: Final = "S9: Python's sync handler raises ValueError for native providers; native serves the call"
S13_INVALID_JSON: Final = "S13: native reports an unparseable success body as an internal error, not AnthropicException"
S14_MISSING_KEY: Final = (
"S14: native raises APIConnectionError for a missing API key where Python raises AuthenticationError"
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"
)
C4_METADATA: Final = "C4: native callbacks get the raw Anthropic metadata, not Python's merged litellm metadata"
C6_RESPONSE_HEADERS: Final = "C6: native success logs carry no response_headers"
C7_API_KEY: Final = "C7: native callbacks see the caller's api_key where Python logs an empty string"
C8_STREAM_HIDDEN: Final = "C8: native streams lack the call metadata @client stamps into _hidden_params"
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)"
@ -377,176 +352,10 @@ F_STREAM_COST: Final = "F: cost injection into streamed usage is Python-only (de
KNOWN_GAPS: Final[Mapping[str, Mapping[str, str | None]]] = MappingProxyType(
{
"plain_text": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"system_string": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"system_blocks_with_cache_control": _gaps(
logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"multi_turn_tool_result": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"tools_with_tool_choice": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"tool_choice_none": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"tool_input_examples": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"sampling_params": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"stop_sequences": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"metadata_user_id": _gaps(
logged_metadata=C4_METADATA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"metadata_extra_keys_are_filtered": _gaps(
logged_metadata=C4_METADATA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"metadata_user_id_not_a_string": _gaps(outcome=R12_METADATA_ERROR, logged=R12_METADATA_ERROR),
"service_tier": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"thinking_enabled": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"thinking_disabled": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"legacy_thinking_on_adaptive_model_keeps_display": _gaps(
logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"legacy_thinking_float_budget_on_adaptive_model": _gaps(
request=R3_LEGACY_THINKING, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"adaptive_thinking_with_block_binding": _gaps(
request=R4_BINDING_BETA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"reasoning_effort": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"reasoning_effort_on_adaptive_model": _gaps(
logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"output_config_effort": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"output_format_json_schema": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"speed_on_model_without_speed": _gaps(outcome=R11_PARAM_ERROR, logged=R11_PARAM_ERROR),
"speed_on_model_without_speed_drop_params": _gaps(
logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"speed_on_fast_model": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"temperature_on_adaptive_model": _gaps(outcome=R11_PARAM_ERROR, logged=R11_PARAM_ERROR),
"top_k_on_adaptive_model": _gaps(outcome=R11_PARAM_ERROR, logged=R11_PARAM_ERROR),
"top_k_on_adaptive_model_drop_params": _gaps(
logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"additional_drop_params_hides_rejected_temperature": _gaps(
request=R7_DROP_ORDER,
outcome=R7_DROP_ORDER,
hidden_params=R7_DROP_ORDER,
logged=R7_DROP_ORDER,
logged_metadata=R7_DROP_ORDER,
logged_api_key=R7_DROP_ORDER,
logged_call=R7_DROP_ORDER,
),
"additional_drop_params_nested_path": _gaps(
logged_metadata=C4_METADATA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"context_management_edits": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"container_skills": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"remote_mcp_servers": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"inference_geo": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"top_level_cache_control": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"null_safeguards_are_stripped": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"empty_text_blocks_are_stripped": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"tool_use_ids_are_sanitized": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"provider_specific_fields_are_stripped": _gaps(
logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"image_block": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"document_block": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"server_web_search_tool": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"extra_headers": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"anthropic_beta_from_headers_and_extra_headers": _gaps(
logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"caller_anthropic_version_capitalised": _gaps(
request=R8_VERSION_HEADER, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"provider_specific_header": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"explicit_custom_llm_provider": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"bare_model_name": _gaps(**dict.fromkeys(ASPECTS[1:], A2_BARE_MODEL) | {"logged_model_info": None}),
"api_base_with_messages_path": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"api_base_with_messages_path_and_query": _gaps(
request=R9_QUERY_PATH,
logged=R9_QUERY_PATH,
logged_response_headers=C6_RESPONSE_HEADERS,
logged_api_key=C7_API_KEY,
),
"api_key_from_environment": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"auth_token_from_environment": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"reasoning_auto_summary": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"global_drop_params": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"request_timeout_setting": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"litellm_metadata_and_metadata": _gaps(
logged_metadata=C4_METADATA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"model_info_with_id": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"deployment_effort_levels": _gaps(
request=R5_DEPLOYMENT_EFFORT, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"response_with_tool_use": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"response_with_thinking": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"response_with_cache_usage": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"response_with_null_usage_and_container": _gaps(outcome=S8_TYPED_RESPONSE),
"response_without_stop_sequence": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"response_with_unknown_field": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"response_key_order": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"response_upstream_headers": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"response_private_keys": _gaps(
outcome=S6_PRIVATE_KEYS, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"upstream_400": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY),
"upstream_401": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY),
"upstream_403": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY),
"upstream_404": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY),
"upstream_413": _gaps(outcome=S2_S12_ERROR_STATUS, logged_api_key=C7_API_KEY),
"upstream_429_with_retry_after": _gaps(outcome=S1_S2_ERROR_HEADERS, logged_api_key=C7_API_KEY),
"upstream_500": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY),
"upstream_529_overloaded": _gaps(outcome=S2_S12_ERROR_STATUS, logged_api_key=C7_API_KEY),
"upstream_400_context_window": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY),
"upstream_non_json_error": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY),
"upstream_invalid_json_success": _gaps(
outcome=S13_INVALID_JSON, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"upstream_timeout": _gaps(outcome=S3_TIMEOUT, logged=S3_TIMEOUT, logged_api_key=C7_API_KEY),
"response_private_keys": _gaps(outcome=S6_PRIVATE_KEYS),
"missing_max_tokens": _gaps(dispatch=A10_NO_MAX_TOKENS),
"missing_api_key": _gaps(outcome=S14_MISSING_KEY, logged=S14_MISSING_KEY),
"stream_text": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY),
"stream_tool_use": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY),
"stream_thinking": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY),
"stream_error_event_mid_stream": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY),
"stream_longer_than_timeout": _gaps(
outcome=S5_STREAM_DEADLINE,
hidden_params=S5_STREAM_DEADLINE,
logged=S5_STREAM_DEADLINE,
logged_metadata=S5_STREAM_DEADLINE,
logged_api_key=S5_STREAM_DEADLINE,
),
"stream_upstream_429": _gaps(outcome=S1_S2_ERROR_HEADERS, logged_api_key=C7_API_KEY),
"stream_false_explicit": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"sync_text": _gaps(
request=S9_SYNC,
outcome=S9_SYNC,
hidden_params=S9_SYNC,
logged=S9_SYNC,
logged_metadata=S9_SYNC,
logged_response_headers=S9_SYNC,
logged_api_key=S9_SYNC,
logged_call=S9_SYNC,
),
"sync_stream": _gaps(
request=S9_SYNC,
outcome=S9_SYNC,
hidden_params=S9_SYNC,
logged=S9_SYNC,
logged_metadata=S9_SYNC,
logged_response_headers=S9_SYNC,
logged_api_key=S9_SYNC,
logged_call=S9_SYNC,
),
"sync_upstream_429": _gaps(
request=S9_SYNC,
outcome=S9_SYNC,
logged=S9_SYNC,
logged_response_headers=S9_SYNC,
logged_api_key=S9_SYNC,
logged_call=S9_SYNC,
),
"mock_response": _gaps(
logged_response_headers=F_MOCK,
request=F_MOCK,
outcome=F_MOCK,
hidden_params=F_MOCK,
@ -556,32 +365,25 @@ KNOWN_GAPS: Final[Mapping[str, Mapping[str, str | None]]] = MappingProxyType(
logged_call=F_MOCK,
),
"mock_response_rate_limit": _gaps(
request=F_MOCK, outcome=F_MOCK, logged=F_MOCK, logged_api_key=F_MOCK, logged_call=F_MOCK
logged_response_headers=F_MOCK,
request=F_MOCK,
outcome=F_MOCK,
logged=F_MOCK,
logged_api_key=F_MOCK,
logged_call=F_MOCK,
),
"thinking_replay": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"thinking_signature_rejected_then_retried": _gaps(
request=F_THINKING_RETRY,
outcome=F_THINKING_RETRY,
hidden_params=F_THINKING_RETRY,
logged=F_THINKING_RETRY,
logged_metadata=C4_METADATA,
logged_response_headers=C6_RESPONSE_HEADERS,
logged_api_key=C7_API_KEY,
logged_metadata=F_THINKING_RETRY,
logged_response_headers=F_THINKING_RETRY,
),
"cache_control_injection_points": _gaps(
request=F_CACHE_INJECTION, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"tool_custom_field": _gaps(
request=F_TOOL_CUSTOM, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"strict_tools_rejected_then_retried": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY),
"modify_params": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"stream_cost_injection": _gaps(outcome=F_STREAM_COST, hidden_params=F_STREAM_COST, logged_api_key=C7_API_KEY),
"enable_prompt_caching": _gaps(
request=F_CACHE_INJECTION, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY
),
"request_timeout_kwarg": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY),
"stream_timeout_kwarg": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY),
"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),
}
)
@ -606,14 +408,55 @@ REQUEST_CASES: Final = (
Case("metadata_user_id", _kw(metadata={"user_id": "user-123"})),
Case("metadata_extra_keys_are_filtered", _kw(metadata={"user_id": "user-123", "session": "s-1"})),
Case("metadata_user_id_not_a_string", _kw(metadata={"user_id": 123})),
Case("service_tier", _kw(service_tier="auto"), gaps=_gaps(request="R1: native drops service_tier")),
Case("service_tier", _kw(service_tier="auto")),
Case("thinking_enabled", _kw(thinking={"type": "enabled", "budget_tokens": 2048}, max_tokens=4096)),
Case("thinking_disabled", _kw(thinking={"type": "disabled"})),
Case(
"legacy_thinking_on_adaptive_model_keeps_display",
_kw(thinking={"type": "enabled", "budget_tokens": 2048, "display": "summarized"}, max_tokens=4096),
model=OPUS_ADAPTIVE,
gaps=_gaps(request="R3: native replaces legacy thinking with a bare adaptive block and drops display"),
),
Case(
"legacy_thinking_display_updates_on_adaptive_model",
_kw(thinking={"type": "enabled", "budget_tokens": 2048, "display": "updates"}, max_tokens=4096),
model=OPUS_ADAPTIVE,
),
Case(
"legacy_thinking_string_budget_on_adaptive_model",
_kw(thinking={"type": "enabled", "budget_tokens": " 4_096 "}, max_tokens=8192),
model=OPUS_ADAPTIVE,
),
Case(
"legacy_thinking_non_numeric_budget_on_adaptive_model",
_kw(thinking={"type": "enabled", "budget_tokens": "lots"}, max_tokens=8192),
model=OPUS_ADAPTIVE,
),
Case(
"legacy_thinking_null_budget_on_adaptive_model",
_kw(thinking={"type": "enabled", "budget_tokens": None}, max_tokens=8192),
model=OPUS_ADAPTIVE,
),
Case(
"legacy_thinking_non_numeric_budget_dropped",
_kw(thinking={"type": "enabled", "budget_tokens": [1]}, max_tokens=8192, drop_params=True),
model=OPUS_ADAPTIVE,
),
Case(
"legacy_thinking_carries_extra_keys",
_kw(
thinking={"type": "enabled", "budget_tokens": 2048, "block_binding": "strict", "note": "kept"},
max_tokens=4096,
),
model=OPUS_ADAPTIVE,
),
Case(
"legacy_thinking_drops_extra_keys_with_drop_params",
_kw(
thinking={"type": "enabled", "budget_tokens": 2048, "block_binding": "strict", "note": "dropped"},
max_tokens=4096,
drop_params=True,
),
model=OPUS_ADAPTIVE,
),
Case(
"legacy_thinking_float_budget_on_adaptive_model",
@ -731,7 +574,6 @@ REQUEST_CASES: Final = (
Case(
"anthropic_beta_from_headers_and_extra_headers",
_kw(headers={"anthropic-beta": "client-beta-1"}, extra_headers={"anthropic-beta": "deploy-beta-2"}),
gaps=_gaps(request="R2: native keeps the last anthropic-beta source instead of the union"),
),
Case("caller_anthropic_version_capitalised", _kw(extra_headers={"Anthropic-Version": "2023-06-01"})),
Case(
@ -774,6 +616,24 @@ REQUEST_CASES: Final = (
),
model=OPUS_ADAPTIVE,
),
Case(
"deployment_effort_levels_from_legacy_thinking",
_kw(
thinking={"type": "enabled", "budget_tokens": 30000},
max_tokens=40000,
model_info={"id": "deployment-1", "reasoning_effort_levels": ["low", "medium"]},
),
model=OPUS_ADAPTIVE,
),
Case(
"deployment_effort_levels_below_request_with_drop_params",
_kw(
output_config={"effort": "high"},
drop_params=True,
model_info={"id": "deployment-1", "reasoning_effort_levels": ["low"]},
),
model=OPUS_ADAPTIVE,
),
)
RESPONSE_CASES: Final = (
@ -820,6 +680,18 @@ FAILURE_CASES: Final = (
Case("upstream_non_json_error", upstream=(ResponseSpec(body=b"<html>bad gateway</html>", status=502),)),
Case("upstream_invalid_json_success", upstream=(ResponseSpec(body=b"not json", status=200),)),
Case("upstream_timeout", _kw(timeout=0.5), upstream=(ResponseSpec(body=dict(TEXT_RESPONSE), delay=1.5),)),
Case(
"response_trickles_slower_than_timeout",
_kw(timeout=1),
upstream=(ResponseSpec(body=dict(TEXT_RESPONSE), chunk_delay=0.5, body_chunks=4),),
gaps=_gaps(
outcome=S15_COMPLETE_DEADLINE,
hidden_params=S15_COMPLETE_DEADLINE,
logged=S15_COMPLETE_DEADLINE,
logged_metadata=S15_COMPLETE_DEADLINE,
logged_response_headers=S15_COMPLETE_DEADLINE,
),
),
Case("missing_max_tokens", _kw(max_tokens=None), upstream=()),
Case("missing_api_key", _kw(api_key=None), upstream=()),
)
@ -838,6 +710,11 @@ STREAM_CASES: Final = (
_kw(stream=True, timeout=1),
upstream=(_stream(SLOW_STREAM, chunk_delay=0.25),),
),
Case(
"stream_read_stalls_past_timeout",
_kw(stream=True, timeout=0.5),
upstream=(_stream(SLOW_STREAM, chunk_delay=1.0),),
),
Case(
"stream_upstream_429",
_kw(stream=True),
@ -967,7 +844,9 @@ def _settings(settings: Mapping[str, object]) -> Generator[None]:
def _kwargs(case: Case, server: RecordingServer) -> dict[str, object]:
supplied: Final = dict(case.kwargs)
# Deep: Python writes into the caller's own `litellm_metadata`, which must not leak into the
# other leg.
supplied: Final = copy.deepcopy(dict(case.kwargs))
suffix: Final = supplied.pop("api_base_suffix", "")
kwargs: Final[dict[str, object]] = {
"model": case.model,

48
tests/test_litellm_rust/cache/test_python_cache.py vendored
View file

@ -18,10 +18,7 @@ from litellm.proxy.hooks.model_max_budget_limiter import (
)
from litellm.proxy.hooks.parallel_request_limiter_v3 import PROXY_MaxParallelRequestsHandler_v3
from litellm.proxy.utils import InternalUsageCache
from litellm.rust_bridge import runtime
from litellm.rust_bridge.catalog import Route, RouteContext, RouteRule
from litellm.rust_bridge.configuration import Rollout
from litellm.rust_bridge.public_call import NativeCall
from litellm.rust_bridge.catalog import Route
from litellm.types.caching import CachingSupportedCallTypes
from tests.test_litellm_rust.cache.support import cache_key, collect, invoke, payload
from tests.test_litellm_rust.messages.data import MESSAGES, MESSAGES_EVENTS, MESSAGES_MODEL, MESSAGES_RESPONSE
@ -430,49 +427,6 @@ async def test_python_cache_operations_stay_in_the_rust_callers_task(
assert len(recording_server.requests) == 1
def test_sync_rust_messages_calls_python_cache(recording_server: RecordingServer) -> None:
from litellm.rust_bridge.messages.entrypoints import NATIVE_MESSAGES
litellm.cache = Cache()
recording_server.default_response = ResponseSpec(body=MESSAGES_RESPONSE)
arguments: Final = {
"model": MESSAGES_MODEL,
"messages": list(MESSAGES),
"max_tokens": 32,
"api_key": "test-key",
"api_base": recording_server.base_url,
}
request: Final = NativeCall(
args=(),
kwargs=arguments,
base={
"model": MESSAGES_MODEL,
"messages": list(MESSAGES),
"max_tokens": 32,
"stream": None,
"api_key": "test-key",
"api_base": recording_server.base_url,
"custom_llm_provider": "anthropic",
**arguments,
},
)
def call() -> object:
return runtime.run(
RouteContext(Route.MESSAGES),
binding=NATIVE_MESSAGES,
native=lambda hook: hook(request),
python=runtime.NO_PYTHON,
rules=(RouteRule(Route.MESSAGES, Rollout.RUST_REQUIRED),),
)
first: Final = call()
second: Final = call()
assert payload(first) == payload(second)
assert cache_key(second)
assert len(recording_server.requests) == 1
@pytest.mark.asyncio
@pytest.mark.parametrize("namespace_source", ("cache", "metadata"))
async def test_rust_messages_legacy_cache_honors_request_namespaces(

25
tests/test_litellm_rust/messages/test_callbacks.py
View file

@ -14,7 +14,6 @@ from litellm.types.integrations.argilla import ArgillaItem
from litellm.types.utils import ModelResponse, StandardLoggingPayload
from tests.test_litellm_rust.messages.data import BAD_REQUEST, MESSAGES_RESPONSE, RATE_LIMITED, call
from tests.test_litellm_rust.support.callback_recorder import HookRecorder, RecordingLogger, request_body
from tests.test_litellm_rust.support.gaps import gap
from tests.test_litellm_rust.support.isolation import rebound
from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec, relaxed
from tests.test_litellm_rust.support.results import response_id, with_id
@ -22,7 +21,6 @@ from tests.test_litellm_rust.support.routing import assert_served_natively
pytestmark = pytest.mark.requires_rust_extension
SYNC_HANDLERS_ON_ASYNC_CALLS: Final = gap("native async calls also run the sync success and failure handlers")
_OBJECT: Final = TypeAdapter(dict[str, object])
_HEADERS: Final = TypeAdapter(dict[str, str])
@ -211,7 +209,7 @@ async def test_pre_call_deployment_rejection_logs_failure_without_provider_io(se
failures: Final = await recorder.wait_for_async("async_log_failure_event")
assert raised.value is rejection
assert server.requests == []
assert not server.requests
assert len(failures) == 1
assert _OBJECT.validate_python(failures[0].kwargs)["exception"] is raised.value
assert not {"log_pre_api_call", "log_success_event", "async_log_success_event"} & set(recorder.names)
@ -341,8 +339,8 @@ async def test_a_failing_logger_does_not_fail_the_call(server: RecordingServer)
pytest.param("input", False, False, 1, id="global-input"),
pytest.param("success", True, False, 1, id="global-async-success"),
pytest.param("success", True, True, 1, id="request-async-success"),
pytest.param("success", False, False, 0, id="global-sync-success", marks=SYNC_HANDLERS_ON_ASYNC_CALLS),
pytest.param("success", False, True, 0, id="request-sync-success", marks=SYNC_HANDLERS_ON_ASYNC_CALLS),
pytest.param("success", False, False, 0, id="global-sync-success"),
pytest.param("success", False, True, 0, id="request-sync-success"),
pytest.param("failure", True, False, 1, id="global-async-failure"),
pytest.param("failure", False, False, 1, id="global-sync-failure"),
pytest.param("failure", False, True, 1, id="request-sync-failure"),
@ -384,12 +382,11 @@ async def test_pre_call_rule_rejects_before_provider_io(server: RecordingServer)
checked.append(text)
return False
with rebound(litellm, "pre_call_rules", [reject]):
with pytest.raises(litellm.APIResponseValidationError):
await call(server)
with rebound(litellm, "pre_call_rules", [reject]), pytest.raises(litellm.APIResponseValidationError):
await call(server)
assert checked == ["Hello"]
assert server.requests == []
assert not server.requests
@pytest.mark.asyncio
@ -419,11 +416,11 @@ async def test_dataset_hook_returning_none_drops_the_record_from_the_success_sin
@pytest.mark.asyncio
async def test_sync_call_returns_the_provider_message(server: RecordingServer) -> None:
async def test_sync_call_raises_like_python_without_logging(server: RecordingServer) -> None:
recorder: Final = RecordingLogger()
response: Final = await call(server, asynchronous=False, callbacks=[recorder])
with pytest.raises(ValueError, match="anthropic_messages_handler is not implemented for sync calls"):
await call(server, asynchronous=False, callbacks=[recorder])
assert_served_natively(server, response)
assert response_id(response) == MESSAGES_RESPONSE["id"]
assert len(await recorder.wait_for_async("log_success_event")) == 1
assert not server.requests
assert not recorder.names

13
tests/test_litellm_rust/messages/test_streaming.py
View file

@ -6,7 +6,6 @@ import pytest
from tests.test_litellm_rust.messages.data import MESSAGES_EVENTS, MESSAGES_RESPONSE, call
from tests.test_litellm_rust.support.callback_recorder import RecordingLogger, drain_logging
from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec, relaxed
from tests.test_litellm_rust.support.results import consume
from tests.test_litellm_rust.support.routing import assert_served_natively
pytestmark = pytest.mark.requires_rust_extension
@ -46,16 +45,14 @@ async def test_stream_relays_provider_events_and_logs_success_once_after_the_las
@pytest.mark.asyncio
async def test_sync_stream_relays_provider_events_and_logs_success_once(server: RecordingServer) -> None:
server.enqueue(STREAM)
async def test_sync_stream_raises_like_python_before_sending(server: RecordingServer) -> None:
recorder: Final = RecordingLogger()
stream: Final = await call(server, asynchronous=False, stream=True, callbacks=[recorder])
chunks: Final = await consume(stream)
with pytest.raises(ValueError, match="anthropic_messages_handler is not implemented for sync calls"):
await call(server, asynchronous=False, stream=True, callbacks=[recorder])
assert_served_natively(server)
assert b"".join(chunks) == b"".join(STREAM.payloads())
assert len(await recorder.wait_for_async("async_log_success_event")) == 1
assert not server.requests
assert not recorder.names
@pytest.mark.asyncio

8
tests/test_litellm_rust/support/recording_server.py
View file

@ -27,12 +27,14 @@ class ResponseSpec:
delay: float = 0
chunk_delay: float = 0
events: tuple[tuple[str, object], ...] = ()
# Splits a non-stream body into this many writes, `chunk_delay` apart, so it trickles in.
body_chunks: int = 1
def payloads(self) -> tuple[bytes, ...]:
if isinstance(self.body, bytes):
return (self.body,)
if not self.events:
return (json.dumps(self.body).encode(),)
body: Final = self.body if isinstance(self.body, bytes) else json.dumps(self.body).encode()
size: Final = max(-(-len(body) // max(self.body_chunks, 1)), 1)
return tuple(body[start : start + size] for start in range(0, len(body), size)) or (body,)
return tuple(f"event: {event}\ndata: {json.dumps(data)}\n\n".encode() for event, data in self.events)

107
tests/unit/rust_bridge/host/routes/test_messages.py
View file

@ -138,3 +138,110 @@ def test_stream_hidden_params_projects_upstream_headers_the_way_the_python_handl
assert additional["llm_provider-request-id"] == "req_upstream_123"
assert additional["x-ratelimit-remaining-requests"] == "41"
assert additional["request-id"] == additional["llm_provider-request-id"]
_ADAPTIVE_OPUS: Final = "anthropic/claude-opus-4-7"
@pytest.mark.parametrize(
("model", "expected"),
[("claude-sonnet-4-5", "anthropic"), ("gpt-4o", "openai"), ("not-a-known-model", None)],
)
def test_inferred_provider_is_the_one_the_python_handler_resolves(model: str, expected: str | None) -> None:
assert messages_host.inferred_provider(model, None) == expected
@pytest.mark.parametrize(
("request_effort", "levels", "drop_params", "expected"),
[
("max", ["low", "high"], False, "high"),
("high", ["low"], True, "low"),
],
)
def test_body_params_round_effort_to_the_levels_the_deployment_declares(
request_effort: str, levels: list[str], drop_params: bool, expected: str
) -> None:
request: Final[dict[str, object]] = {
"model": _ADAPTIVE_OPUS,
"messages": [],
"max_tokens": 8,
"output_config": {"effort": request_effort},
"drop_params": drop_params,
"model_info": {"id": "deployment-1", "reasoning_effort_levels": levels},
}
assert messages_host.body_params(request)["output_config"] == {"effort": expected}
def test_body_params_keep_the_requested_effort_without_deployment_levels() -> None:
request: Final[dict[str, object]] = {
"model": _ADAPTIVE_OPUS,
"messages": [],
"max_tokens": 8,
"output_config": {"effort": "max"},
}
assert messages_host.body_params(request)["output_config"] == {"effort": "max"}
def _marked(phase: str | None = None, *args: object) -> ValueError:
error: Final = ValueError(*args)
if phase is not None:
error.messages_timeout = phase # pyright: ignore[reportAttributeAccessIssue] # marker the native host sets
return error
def test_a_stalled_stream_body_reaches_the_caller_as_the_transport_timeout() -> None:
import httpx
mapped: Final = messages_host.map_failure(_marked("Body"), {"model": _ADAPTIVE_OPUS}, "anthropic")
assert isinstance(mapped, httpx.ReadTimeout)
assert str(mapped) == "Timeout on reading data from socket"
def test_a_native_read_timeout_maps_to_the_public_timeout_with_python_text() -> None:
mapped: Final = messages_host.map_failure(_marked("Read"), {"model": _ADAPTIVE_OPUS, "timeout": 1.0}, "anthropic")
assert isinstance(mapped, litellm.Timeout)
assert "Read timed out. Timeout passed=1.0" in str(mapped)
def test_an_upstream_status_maps_through_the_provider_error_class() -> None:
upstream: Final = ValueError(429, "slow down")
upstream.headers = [("retry-after", "3")] # pyright: ignore[reportAttributeAccessIssue] # set by the native host
mapped: Final = messages_host.map_failure(upstream, {"model": _ADAPTIVE_OPUS}, "anthropic")
assert isinstance(mapped, litellm.RateLimitError)
assert "slow down" in str(mapped)
assert mapped.__context__ is upstream
def test_a_late_rejection_tells_the_logger_the_resolved_call_first() -> None:
import datetime
from litellm.litellm_core_utils.litellm_logging import Logging
logger: Final = Logging(
model=_ADAPTIVE_OPUS,
messages=[],
stream=False,
call_type="anthropic_messages",
start_time=datetime.datetime.now(),
litellm_call_id="late",
function_id="late",
)
rejected: Final = ValueError("budget_tokens must be a number")
rejected.messages_late_rejection = True # pyright: ignore[reportAttributeAccessIssue] # marker the native host sets
rejected.messages_request_error = True # pyright: ignore[reportAttributeAccessIssue] # marker the native host sets
mapped: Final = messages_host.map_failure(
rejected,
{"model": _ADAPTIVE_OPUS, "messages": [], "max_tokens": 8, "litellm_logging_obj": logger},
"anthropic",
)
assert isinstance(mapped, litellm.BadRequestError)
assert logger.model_call_details["custom_llm_provider"] == "anthropic"
assert logger.model_call_details["optional_params"] == {"max_tokens": 8}

54
tests/unit/rust_bridge/host/test_callbacks_legacy_python.py
View file

@ -5,6 +5,7 @@ from pathlib import Path
from types import MappingProxyType
from typing import Final
import httpx
import pytest
from pydantic import TypeAdapter
@ -244,3 +245,56 @@ def test_the_rust_contract_matches_the_shim_signatures() -> None:
contract: Final = TypeAdapter(dict[str, list[str]]).validate_json(CONTRACT_PATH.read_text())
assert contract == {name: list(inspect.signature(getattr(legacy, name)).parameters) for name in contract}
def _messages_logger() -> Logging:
return Logging(
model="claude-sonnet-4-5",
messages=[],
stream=False,
call_type="anthropic_messages",
start_time=datetime.datetime.now(),
litellm_call_id="messages",
function_id="messages",
)
def test_native_provider_response_is_recorded_for_logging_replay() -> None:
logger: Final = _messages_logger()
legacy.record_response(logger, 200, (("request-id", "req_native"),), '{"id": "msg_native"}')
recorded: Final[object] = logger.model_call_details["httpx_response"]
assert isinstance(recorded, httpx.Response)
assert recorded.status_code == 200
assert recorded.content == b'{"id": "msg_native"}'
assert logger.model_call_details["response_headers"]["request-id"] == "req_native"
def test_native_provider_exchange_window_is_recorded_like_the_timed_post() -> None:
logger: Final = _messages_logger()
logger.model_call_details["litellm_params"] = {"litellm_metadata": {"user_api_key": "hashed"}}
legacy.record_api_window(logger, 1_700_000_000.0, 1_700_000_000.25)
assert logger.model_call_details["llm_api_duration_ms"] == 250.0
metadata: Final = logger.model_call_details["litellm_params"]["litellm_metadata"]
assert metadata["llm_api_timing_windows"] == ((1_700_000_000.0, 1_700_000_000.25),)
def test_stream_head_is_stamped_in_place_with_the_call_hidden_params() -> None:
logger: Final = _messages_logger()
head: Final[dict[str, object]] = {"additional_headers": {"request-id": "req_native"}}
started: Final = datetime.datetime.now()
legacy.stamp_stream_head(
head,
logger,
{"model": "anthropic/claude-sonnet-4-5", "litellm_call_id": "messages"},
started,
started + datetime.timedelta(milliseconds=5),
)
assert head["additional_headers"] == {"request-id": "req_native", "llm_provider-request-id": "req_native"}
assert head["litellm_call_id"] == "messages"
assert "_response_ms" in head

12
tests/unit/rust_bridge/messages/test_secrets.py
View file

@ -11,7 +11,7 @@ import litellm
from litellm.integrations.custom_secret_manager import CustomSecretManager
from litellm.llms.anthropic.messages.native.handler import anthropic_messages
from litellm.rust_bridge.host import settings
from litellm.rust_bridge.messages.entrypoints import NATIVE_AMESSAGES, NATIVE_MESSAGES
from litellm.rust_bridge.messages.entrypoints import NATIVE_AMESSAGES
from litellm.rust_bridge.public_call import NativeCall
from litellm.types.secret_managers.main import KeyManagementSettings, KeyManagementSystem
from tests.test_litellm_rust.messages.data import MESSAGES, MESSAGES_MODEL, MESSAGES_RESPONSE
@ -72,21 +72,13 @@ async def _python_messages() -> object:
return await anthropic_messages(**_public_kwargs())
async def _rust_messages() -> object:
route: Final = NATIVE_MESSAGES.load()
assert route is not None
return route(_native_request())
async def _rust_amessages() -> object:
route: Final = NATIVE_AMESSAGES.load()
assert route is not None
return await route(_native_request())
@pytest.fixture(
params=(_python_messages, _rust_messages, _rust_amessages), ids=("python-async", "rust-sync", "rust-async")
)
@pytest.fixture(params=(_python_messages, _rust_amessages), ids=("python-async", "rust-async"))
def messages(request: pytest.FixtureRequest) -> Messages:
return cast(Messages, request.param)