add streaming message

2026-09-21 00:21:49 +00:00 · 2026-09-18 15:32:28 -07:00 · 2026-09-18 15:32:28 -07:00 · 1a52bae779
commit 1a52bae779
parent 4627ec4ea8
40 changed files with 1319 additions and 344 deletions
--- a/litellm-rust/crates/callbacks-legacy/python_contract.json
+++ b/litellm-rust/crates/callbacks-legacy/python_contract.json
@ -94,5 +94,22 @@
    "kwargs",
    "error",
    "call_type"
+  ],
+  "stream_opened": [
+    "logger"
+  ],
+  "stream_success": [
+    "logger",
+    "request_body",
+    "chunks",
+    "start",
+    "end",
+    "first_chunk"
+  ],
+  "stream_failure": [
+    "logger",
+    "request_body",
+    "chunks",
+    "error"
  ]
 }
--- a/litellm-rust/crates/callbacks-legacy/src/adapter.rs
+++ b/litellm-rust/crates/callbacks-legacy/src/adapter.rs
@ -2,7 +2,9 @@
 //! raises is answered with the same `Logging` calls, in the same order, as the Python
 //! `@client` path makes them.

-use litellm_callbacks::event::{CallEvent, FailureOrigin, RequestContext, Timing, WireRequest};
+use litellm_callbacks::event::{
+    CallEvent, FailureOrigin, RequestContext, Timing, WireRequest, epoch_seconds,
+};
 use litellm_host_python::{
    LifecycleStep, PublicValue, PythonLifecycle, from_py, missing_state, to_py,
 };
@ -10,14 +12,16 @@ use pyo3::{
    exceptions::{PyBaseException, PyException},
    gc::{PyTraverseError, PyVisit},
    prelude::*,
-    types::PyDict,
+    types::{PyDict, PyList},
 };
 use serde_json::Value;

 use crate::{
    DeploymentHooks, LegacyCallbacks, PublicCall, PythonLogger,
    deferred::{PendingLogging, PendingSuccess},
-    finalize, is_internal_call, prepare, setup,
+    finalize, is_internal_call,
+    legacy_python::Streaming,
+    prepare, setup,
 };

 /// What the legacy contract needs to know about the route it is logging.
@ -28,6 +32,12 @@ pub struct LegacySurface {
    pub input_description: &'static str,
 }

+/// What the Messages stream iterator keeps for its end-of-stream billing.
+struct DeliveredStream {
+    chunks: Py<PyList>,
+    first_chunk: Option<Py<PyAny>>,
+}
+
 enum Pending {
    DeploymentPreCall,
    DeploymentPostCall,
@ -46,6 +56,7 @@ pub struct LegacyLogging {
    body: Option<Py<PyDict>>,
    headers: Option<Py<PyDict>>,
    context: Option<RequestContext>,
+    stream: Option<DeliveredStream>,
    asynchronous: bool,
    internal: bool,
    pending: Option<Pending>,
@ -80,6 +91,7 @@ impl LegacyLogging {
            body: None,
            headers: None,
            context: None,
+            stream: None,
            asynchronous,
            internal: false,
            pending: None,
@ -163,6 +175,49 @@ impl LegacyLogging {
        logger.sync_success_for_async_call(py, &self.response, &self.start, &self.end)
    }

+    fn stream_success(&self, py: Python<'_>, stream: &DeliveredStream) -> PyResult<()> {
+        let logger = self.logger()?;
+        let billed = Streaming::Success.call(
+            py,
+            (
+                logger.object(py),
+                &self.body,
+                &stream.chunks,
+                &self.start,
+                &self.end,
+                &stream.first_chunk,
+            ),
+        );
+        match billed {
+            Err(error) if error.is_instance_of::<PyException>(py) => {
+                error.write_unraisable(py, Some(logger.object(py)));
+                Ok(())
+            }
+            result => result.map(|_| ()),
+        }
+    }
+
+    /// A failure after the stream reached the caller bills the delivered chunks as
+    /// partial usage. The sync path has no loop to schedule that on, so it falls back to
+    /// the plain failure handler.
+    fn stream_failure(&mut self, py: Python<'_>) -> PyResult<LifecycleStep> {
+        let (Some(logger), Some(error), Some(stream)) = (&self.logger, &self.error, &self.stream)
+        else {
+            return Ok(LifecycleStep::Done);
+        };
+        if !self.asynchronous {
+            return self.dispatch_failure(py);
+        }
+        match Streaming::Failure.call(py, (logger.object(py), &self.body, &stream.chunks, error)) {
+            Ok(awaitable) => {
+                self.pending = Some(Pending::AsyncFailure);
+                Ok(LifecycleStep::Await(awaitable.unbind()))
+            }
+            Err(failure) if is_cancellation(py, &failure) => Err(failure),
+            Err(_) => Ok(LifecycleStep::Done),
+        }
+    }
+
    /// The sync failure handler, then the async one for async calls. Ordinary handler
    /// errors never replace the selected failure or suppress the other family; a
    /// cancellation does end the call.
@ -296,6 +351,22 @@ impl PythonLifecycle for LegacyLogging {
    ) -> PyResult<LifecycleStep> {
        match (event, public) {
            (CallEvent::Started { .. }, _) => Ok(LifecycleStep::Done),
+            (CallEvent::Opened, _) => {
+                Streaming::Opened.call(py, (self.logger()?.object(py),))?;
+                self.stream = Some(DeliveredStream {
+                    chunks: PyList::empty(py).unbind(),
+                    first_chunk: None,
+                });
+                Ok(LifecycleStep::Done)
+            }
+            (CallEvent::Delivered, Some(PublicValue::Chunk(chunk))) => {
+                let stream = self.stream.as_mut().ok_or_else(missing_state)?;
+                if stream.first_chunk.is_none() {
+                    stream.first_chunk = Some(datetime(py, epoch_seconds())?);
+                }
+                stream.chunks.bind(py).append(chunk)?;
+                Ok(LifecycleStep::Done)
+            }
            (CallEvent::ResponseReceived { raw }, _) => {
                let api_key = self
                    .context
@ -314,12 +385,18 @@ impl PythonLifecycle for LegacyLogging {
            (CallEvent::Succeeded { timing }, Some(PublicValue::Response(response))) => {
                self.end = Some(datetime(py, timing.end_time)?);
                self.response = Some(response.clone_ref(py));
-                self.dispatch_success(py)?;
+                match &self.stream {
+                    Some(stream) => self.stream_success(py, stream)?,
+                    None => self.dispatch_success(py)?,
+                }
                Ok(LifecycleStep::Done)
            }
            (CallEvent::Failed { timing, origin }, Some(PublicValue::Error(error))) => {
                self.end = Some(datetime(py, timing.end_time)?);
                self.error = Some(error.clone_ref(py).into_value(py));
+                if self.stream.is_some() {
+                    return self.stream_failure(py);
+                }
                if *origin == FailureOrigin::Call
                    && self.logger.is_some()
                    && self.runs_deployment_hooks()
@ -366,6 +443,7 @@ impl PythonLifecycle for LegacyLogging {
        }
        self.body = None;
        self.context = None;
+        self.stream = None;
    }

    fn traverse(&self, visit: &PyVisit<'_>) -> Result<(), PyTraverseError> {
@ -377,6 +455,10 @@ impl PythonLifecycle for LegacyLogging {
        visit.call(&self.end)?;
        visit.call(&self.response)?;
        visit.call(&self.error)?;
+        if let Some(stream) = &self.stream {
+            visit.call(&stream.chunks)?;
+            visit.call(&stream.first_chunk)?;
+        }
        visit.call(&self.body)
    }
 }
--- a/litellm-rust/crates/callbacks-legacy/src/legacy_python.rs
+++ b/litellm-rust/crates/callbacks-legacy/src/legacy_python.rs
@ -16,6 +16,7 @@ pub(crate) enum LegacyPython {
    Wrapper(Wrapper),
    Logging(Logging),
    DeploymentHooks(DeploymentHooks),
+    Streaming(Streaming),
 }

 /// The `@client` wrapper around the call: `function_setup`, limits, credentials,
@ -76,12 +77,25 @@ pub(crate) enum DeploymentHooks {
    AfterDeploymentFailure,
 }

+/// The Messages stream iterator's logging: the stream flag, the end-of-stream billing
+/// from the delivered chunks, and the partial-usage failure path.
+#[derive(Clone, Copy, Debug, IntoStaticStr, PartialEq, Eq, VariantArray)]
+pub(crate) enum Streaming {
+    #[strum(serialize = "stream_opened")]
+    Opened,
+    #[strum(serialize = "stream_success")]
+    Success,
+    #[strum(serialize = "stream_failure")]
+    Failure,
+}
+
 impl LegacyPython {
    fn name(self) -> &'static str {
        match self {
            Self::Wrapper(function) => function.into(),
            Self::Logging(function) => function.into(),
            Self::DeploymentHooks(function) => function.into(),
+            Self::Streaming(function) => function.into(),
        }
    }

@ -111,6 +125,15 @@ impl Logging {
    }
 }

+impl Streaming {
+    pub(crate) fn call<'py, A>(self, py: Python<'py>, args: A) -> PyResult<Bound<'py, PyAny>>
+    where
+        A: pyo3::call::PyCallArgs<'py>,
+    {
+        LegacyPython::Streaming(self).call(py, args)
+    }
+}
+
 impl DeploymentHooks {
    pub(crate) fn call<'py, A>(self, py: Python<'py>, args: A) -> PyResult<Bound<'py, PyAny>>
    where
@ -126,7 +149,7 @@ mod tests {

    use strum::VariantArray;

-    use super::{DeploymentHooks, LegacyPython, Logging, Wrapper};
+    use super::{DeploymentHooks, LegacyPython, Logging, Streaming, Wrapper};
    use crate::test_support::PYTHON_CONTRACT;

    #[test]
@ -147,6 +170,11 @@ mod tests {
                    .iter()
                    .map(|&function| LegacyPython::DeploymentHooks(function)),
            )
+            .chain(
+                Streaming::VARIANTS
+                    .iter()
+                    .map(|&function| LegacyPython::Streaming(function)),
+            )
            .map(LegacyPython::name)
            .collect();
        assert_eq!(called.len(), declared.len(), "a function is borrowed twice");
--- a/litellm-rust/crates/callbacks-legacy/tests/support.rs
+++ b/litellm-rust/crates/callbacks-legacy/tests/support.rs
@ -84,6 +84,11 @@ FAKES = {
        'success', response, call_type
    ),
    'after_deployment_failure': lambda kwargs, error, call_type: kwargs['logger'].hook('failure', error, call_type),
+    'stream_opened': lambda logger: logger.record('stream_opened', None),
+    'stream_success': lambda logger, request_body, chunks, start, end, first_chunk: logger.record(
+        'stream_success', list(chunks)
+    ),
+    'stream_failure': lambda logger, request_body, chunks, error: logger.record('stream_failure', error),
 }
 assert FAKES.keys() == CONTRACT.keys(), sorted(FAKES.keys() ^ CONTRACT.keys())
 for name, fake in FAKES.items():
--- a/litellm-rust/crates/callbacks/src/event.rs
+++ b/litellm-rust/crates/callbacks/src/event.rs
@ -60,6 +60,10 @@ pub enum CallEvent {
    ResponseReceived {
        raw: RawResponse,
    },
+    /// The call streams and its stream was handed to the caller.
+    Opened,
+    /// One chunk of an open stream reached the caller.
+    Delivered,
    Succeeded {
        timing: Timing,
    },
--- a/litellm-rust/crates/callbacks/src/host.rs
+++ b/litellm-rust/crates/callbacks/src/host.rs
@ -11,12 +11,25 @@ pub enum HostOp<R: Route> {
        context: Box<RequestContext>,
    },
    Emit(CallEvent),
+    /// The response streams: the host hands the caller a stream and answers once the
+    /// caller asks for the first chunk or goes away.
+    Open(R::StreamHead),
+    /// The next chunk of an open stream, answered once the caller asks for the one after.
+    Deliver(R::Chunk),
 }

 pub enum HostResult<R: Route> {
    Route(R::OpResult),
    BeforeSend(Box<WireRequest>),
    Emitted,
+    Demand(Demand),
+}
+
+/// Whether the caller of a streamed call still reads it.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Demand {
+    More,
+    Detached,
 }

 /// A host answer that is either available now or arrives once the host's own
@ -42,4 +55,12 @@ pub trait Host<R: Route>: Send + Sync {
    fn emit(&self, _event: &CallEvent) -> impl Future<Output = Result<(), R::Error>> + Send {
        async { Ok(()) }
    }
+
+    fn open(&self, _head: R::StreamHead) -> impl Future<Output = Result<Demand, R::Error>> + Send {
+        async { Ok(Demand::More) }
+    }
+
+    fn deliver(&self, _chunk: R::Chunk) -> impl Future<Output = Result<Demand, R::Error>> + Send {
+        async { Ok(Demand::More) }
+    }
 }
--- a/litellm-rust/crates/callbacks/src/route.rs
+++ b/litellm-rust/crates/callbacks/src/route.rs
@ -6,4 +6,9 @@ pub trait Route: Send + Sync + 'static {
    type Error: Clone + Send + Sync + 'static;
    type Op: Send + 'static;
    type OpResult: Send + 'static;
+    /// One piece of a streamed response, handed to the caller as it arrives. A route
+    /// that never streams uses `Infallible`.
+    type Chunk: Send + 'static;
+    /// What the route knows once a streamed response starts, before its first chunk.
+    type StreamHead: Send + 'static;
 }
--- a/litellm-rust/crates/callbacks/src/run.rs
+++ b/litellm-rust/crates/callbacks/src/run.rs
@ -26,6 +26,8 @@ where
                .await
                .map(|wire| HostResult::BeforeSend(Box::new(wire))),
            HostOp::Emit(event) => host.emit(&event).await.map(|()| HostResult::Emitted),
+            HostOp::Open(head) => host.open(head).await.map(HostResult::Demand),
+            HostOp::Deliver(chunk) => host.deliver(chunk).await.map(HostResult::Demand),
        };
        match answer {
            Ok(answer) => result = Some(answer),
@ -61,6 +63,8 @@ mod tests {
        type Error = &'static str;
        type Op = &'static str;
        type OpResult = ();
+        type Chunk = std::convert::Infallible;
+        type StreamHead = std::convert::Infallible;
    }

    struct Scripted {
--- a/litellm-rust/crates/core/src/machine/mod.rs
+++ b/litellm-rust/crates/core/src/machine/mod.rs
@ -9,7 +9,7 @@ use std::{future::Future, pin::Pin};
 pub use auth::{HostTokenProvider, TokenRoute};
 use litellm_callbacks::{
    event::{CallEvent, RequestContext, WireRequest},
-    host::{HostOp, HostResult},
+    host::{Demand, HostOp, HostResult},
    machine::{HostFailure, Interrupted, Machine, MachineStep, Step},
    route::Route,
 };
@ -88,6 +88,21 @@ where
            _ => Err(MachineFault::Mismatch.into()),
        }
    }
+
+    pub async fn open(&self, head: R::StreamHead) -> Result<Demand, R::Error> {
+        self.demand(HostOp::Open(head)).await
+    }
+
+    pub async fn deliver(&self, chunk: R::Chunk) -> Result<Demand, R::Error> {
+        self.demand(HostOp::Deliver(chunk)).await
+    }
+
+    async fn demand(&self, op: HostOp<R>) -> Result<Demand, R::Error> {
+        match self.invoke(op).await? {
+            HostResult::Demand(demand) => Ok(demand),
+            _ => Err(MachineFault::Mismatch.into()),
+        }
+    }
 }

 enum Execution<R: Route> {
--- a/litellm-rust/crates/core/src/messages/handler.rs
+++ b/litellm-rust/crates/core/src/messages/handler.rs
@ -1,88 +1,54 @@
-use litellm_llms::custom_httpx::http_handler::http_request;
-use litellm_types::llms::anthropic_messages::anthropic_response::AnthropicMessagesResponse;
+use std::time::Duration;

-use super::{
-    Error, client::http_client, common_utils::truncate_error_body,
-    prepare::prepare_provider_request,
+use litellm_llms::{
+    base_llm::anthropic_messages::transformation::BaseAnthropicMessagesConfig,
+    custom_httpx::{http_handler::http_request, transport::Error as TransportError},
 };
-use crate::{constants::ANTHROPIC_MESSAGES_PROVIDER, messages::types::MessagesRequest};
+use litellm_types::llms::anthropic_messages::anthropic_response::AnthropicMessagesResponse;
+use serde_json::Value;

-pub(super) async fn execute_messages_provider_call(
-    request: MessagesRequest<'_>,
+use super::{Error, client::http_client, common_utils::truncate_error_body};
+
+pub(super) fn network(error: reqwest::Error) -> Error {
+    Error::Transport(TransportError::Network(error.to_string()))
+}
+
+pub(super) async fn send(
+    url: &str,
+    headers: &[(String, String)],
+    body: &Value,
+    timeout: Option<Duration>,
+) -> Result<reqwest::Response, Error> {
+    let builder = headers.iter().fold(
+        http_client().post(url).json(body),
+        |builder, (key, value)| builder.header(key, value),
+    );
+    let builder = match timeout {
+        Some(duration) => builder.timeout(duration),
+        None => builder,
+    };
+    http_request(builder).await.map_err(network)
+}
+
+pub(super) async fn provider_error(response: reqwest::Response) -> Error {
+    let status = response.status().as_u16();
+    match response.text().await {
+        Ok(text) => Error::Transport(TransportError::Http {
+            status,
+            body: truncate_error_body(&text),
+        }),
+        Err(error) => network(error),
+    }
+}
+
+pub(super) fn decode_response(
+    config: &dyn BaseAnthropicMessagesConfig,
+    model: &str,
+    text: &str,
 ) -> Result<AnthropicMessagesResponse, Error> {
-    let request = prepare_provider_request(request)?;
-    let mut request_builder = http_client().post(&request.url).json(&request.body);
-    for (key, value) in &request.upstream_headers {
-        request_builder = request_builder.header(key, value);
-    }
-    if let Some(duration) = request.timeout {
-        request_builder = request_builder.timeout(duration);
-    }
-
-    let response = http_request(request_builder).await.map_err(|err| {
-        Error::Transport(litellm_llms::custom_httpx::transport::Error::Network(
-            err.to_string(),
-        ))
-    })?;
-
-    let status = response.status();
-    let text = response.text().await.map_err(|err| {
-        Error::Transport(litellm_llms::custom_httpx::transport::Error::Network(
-            err.to_string(),
-        ))
-    })?;
-
-    if !status.is_success() {
-        return Err(Error::Transport(
-            litellm_llms::custom_httpx::transport::Error::Http {
-                status: status.as_u16(),
-                body: truncate_error_body(&text),
-            },
-        ));
-    }
-
-    let response = serde_json::from_str(&text)
+    let response = serde_json::from_str(text)
        .map_err(|err| Error::InvalidResponse(format!("invalid messages response JSON: {err}")))?;
-    request
-        .config
-        .transform_anthropic_messages_response(&request.model, response)
+    config
+        .transform_anthropic_messages_response(model, response)
        .map_err(Error::from)
 }
-
-pub(super) async fn execute_messages_provider_stream(
-    request: MessagesRequest<'_>,
-) -> Result<reqwest::Response, Error> {
-    let request = prepare_provider_request(request)?;
-    if request.provider != ANTHROPIC_MESSAGES_PROVIDER {
-        return Err(Error::Unsupported("streaming messages for this provider"));
-    }
-
-    let mut request_builder = http_client().post(&request.url).json(&request.body);
-    for (key, value) in &request.upstream_headers {
-        request_builder = request_builder.header(key, value);
-    }
-    if let Some(duration) = request.timeout {
-        request_builder = request_builder.timeout(duration);
-    }
-
-    let response = http_request(request_builder).await.map_err(|err| {
-        Error::Transport(litellm_llms::custom_httpx::transport::Error::Network(
-            err.to_string(),
-        ))
-    })?;
-    let status = response.status();
-    if !status.is_success() {
-        let text = response.text().await.map_err(|err| {
-            Error::Transport(litellm_llms::custom_httpx::transport::Error::Network(
-                err.to_string(),
-            ))
-        })?;
-        return Err(Error::Transport(
-            litellm_llms::custom_httpx::transport::Error::Http {
-                status: status.as_u16(),
-                body: truncate_error_body(&text),
-            },
-        ));
-    }
-    Ok(response)
-}
--- a/litellm-rust/crates/core/src/messages/mod.rs
+++ b/litellm-rust/crates/core/src/messages/mod.rs
@ -1,11 +1,8 @@
 //! The Anthropic Messages call, the Rust equivalent of Python's
 //! `litellm.messages()`.
 //!
-//! [`messages`] is the top-level entrypoint: give it a model, a body, and
-//! credentials, and it resolves the provider, transforms the request, calls the
-//! provider, and returns a typed non-streaming response. [`messages_stream`]
-//! is the streaming variant; it hands the raw upstream response back so a host
-//! can splice the event stream to its own caller.
+//! [`route`] is the call as a machine a host drives, streaming or not. [`messages`] runs
+//! it in process for a caller that already holds the request and wants the message.

 mod error;
 pub mod types;
@ -14,17 +11,34 @@ mod client;
 mod common_utils;
 mod handler;
 mod prepare;
-use handler::{execute_messages_provider_call, execute_messages_provider_stream};
+pub mod route;
 use litellm_types::llms::anthropic_messages::anthropic_response::AnthropicMessagesResponse;
+use route::{LocalMessagesHost, MessagesCall, MessagesOutput, messages_machine};
+use serde_json::Value;

 use crate::messages::types::MessagesRequest;

 pub async fn messages(request: MessagesRequest<'_>) -> Result<AnthropicMessagesResponse, Error> {
-    execute_messages_provider_call(request).await
-}
-
-pub async fn messages_stream(request: MessagesRequest<'_>) -> Result<reqwest::Response, Error> {
-    execute_messages_provider_stream(request).await
+    let Value::Object(body) = request.body else {
+        return Err(Error::InvalidRequest(
+            "messages body must be an object".into(),
+        ));
+    };
+    let call = MessagesCall {
+        model: request.model.into(),
+        body,
+        api_key: request.api_key.map(Into::into),
+        api_base: request.api_base.map(Into::into),
+        custom_llm_provider: request.custom_llm_provider.map(Into::into),
+        extra_headers: request.extra_headers,
+        timeout: request.timeout,
+    };
+    match litellm_callbacks::run::run(messages_machine(), &LocalMessagesHost::new(call)).await? {
+        MessagesOutput::Message(message) => Ok(message),
+        MessagesOutput::Streamed => Err(Error::Unsupported(
+            "streamed responses need a streaming host",
+        )),
+    }
 }

 #[cfg(test)]
--- a/litellm-rust/crates/core/src/messages/prepare.rs
+++ b/litellm-rust/crates/core/src/messages/prepare.rs
@ -2,6 +2,7 @@ use litellm_core_utils::get_llm_provider_logic::{CustomLlmProvider, get_custom_l
 use litellm_llms::base_llm::anthropic_messages::transformation::{
    BaseAnthropicMessagesConfig, MessagesAuthStrategy,
 };
+use litellm_types::llms::anthropic_messages::anthropic_request::AnthropicMessagesRequest;
 use serde_json::{Map, Value};

 use super::{
@ -37,10 +38,14 @@ pub(super) fn prepare_provider_request(
    let headers =
        validate_environment(config, request.extra_headers, request.api_key, &env_lookup)?;

-    let typed_request = serde_json::from_value(request.body).map_err(|err| {
-        Error::InvalidRequest(format!("invalid Anthropic messages request: {err}"))
+    let typed_request: AnthropicMessagesRequest =
+        serde_json::from_value(request.body).map_err(|err| {
+            Error::InvalidRequest(format!("invalid Anthropic messages request: {err}"))
+        })?;
+    let transformed = config.transform_anthropic_messages_request(AnthropicMessagesRequest {
+        model: model.clone(),
+        ..typed_request
    })?;
-    let transformed = config.transform_anthropic_messages_request(typed_request)?;
    let body = serde_json::to_value(transformed).map_err(|err| {
        Error::InvalidRequest(format!(
            "failed to serialize Anthropic messages request: {err}"
--- a/litellm-rust/crates/core/src/messages/route.rs
+++ b/litellm-rust/crates/core/src/messages/route.rs
@ -0,0 +1,197 @@
+use std::{sync::Mutex, time::Duration};
+
+use bytes::Bytes;
+use litellm_auth::SecretValue;
+use litellm_callbacks::{
+    event::{CallEvent, RawResponse, RequestContext, WireRequest},
+    host::{Demand, Host},
+    route::Route,
+};
+use litellm_core_utils::get_llm_provider_logic::get_custom_llm_provider;
+use litellm_types::llms::anthropic_messages::anthropic_response::AnthropicMessagesResponse;
+use serde_json::{Map, Value};
+
+use super::{
+    Error,
+    common_utils::messages_provider_config,
+    handler::{decode_response, network, provider_error, send},
+    prepare::prepare_provider_request,
+    types::MessagesRequest,
+};
+use crate::{
+    constants::ANTHROPIC_MESSAGES_PROVIDER,
+    machine::{HostChannel, MachineFault, RouteMachine},
+};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum MessagesOp {
+    ProjectRequest,
+}
+
+pub enum MessagesOpResult {
+    Request(Box<MessagesCall>),
+}
+
+/// The caller's request as the host projects it.
+pub struct MessagesCall {
+    pub model: String,
+    pub body: Map<String, Value>,
+    pub api_key: Option<String>,
+    pub api_base: Option<String>,
+    pub custom_llm_provider: Option<String>,
+    pub extra_headers: Option<Map<String, Value>>,
+    pub timeout: Option<Duration>,
+}
+
+impl MessagesCall {
+    fn streams(&self) -> bool {
+        self.body.get("stream").and_then(Value::as_bool) == Some(true)
+    }
+}
+
+pub enum MessagesOutput {
+    Message(AnthropicMessagesResponse),
+    /// Every chunk already reached the host through `Deliver`.
+    Streamed,
+}
+
+pub struct Messages;
+
+impl Route for Messages {
+    type Response = MessagesOutput;
+    type Error = Error;
+    type Op = MessagesOp;
+    type OpResult = MessagesOpResult;
+    type Chunk = Bytes;
+    type StreamHead = ();
+}
+
+impl From<MachineFault> for Error {
+    fn from(fault: MachineFault) -> Self {
+        Self::InvalidRequest(match fault {
+            MachineFault::Abandoned => "messages host driver was abandoned".into(),
+            MachineFault::Protocol(message) => format!("messages {message}"),
+            MachineFault::Mismatch => "invalid messages host operation result".into(),
+        })
+    }
+}
+
+pub type MessagesHost = HostChannel<Messages>;
+pub type MessagesMachine = RouteMachine<Messages>;
+
+/// Whether this route serves the request, decided before any callback runs so a host
+/// can still run its own path.
+pub fn supports(model: &str, custom_llm_provider: Option<&str>, stream: bool) -> bool {
+    let provider = get_custom_llm_provider(model, custom_llm_provider)
+        .map(|resolved| resolved.custom_llm_provider)
+        .or(custom_llm_provider);
+    match provider {
+        Some(ANTHROPIC_MESSAGES_PROVIDER) => true,
+        Some(provider) => !stream && messages_provider_config(provider).is_some(),
+        None => false,
+    }
+}
+
+/// The in-process host for a request already in hand. It answers projection once and
+/// observes nothing.
+pub struct LocalMessagesHost {
+    call: Mutex<Option<MessagesCall>>,
+}
+
+impl LocalMessagesHost {
+    pub fn new(call: MessagesCall) -> Self {
+        Self {
+            call: Mutex::new(Some(call)),
+        }
+    }
+}
+
+impl Host<Messages> for LocalMessagesHost {
+    async fn route(&self, op: MessagesOp) -> Result<MessagesOpResult, Error> {
+        match op {
+            MessagesOp::ProjectRequest => self
+                .call
+                .lock()
+                .unwrap_or_else(|error| error.into_inner())
+                .take()
+                .map(|call| MessagesOpResult::Request(Box::new(call)))
+                .ok_or_else(|| {
+                    Error::InvalidRequest("messages request was already projected".into())
+                }),
+        }
+    }
+}
+
+pub fn messages_machine() -> MessagesMachine {
+    RouteMachine::new(|host| Box::pin(execute(host)))
+}
+
+async fn execute(host: MessagesHost) -> Result<MessagesOutput, Error> {
+    let MessagesOpResult::Request(call) = host.route(MessagesOp::ProjectRequest).await?;
+    let stream = call.streams();
+    let request = prepare_provider_request(MessagesRequest {
+        model: &call.model,
+        body: Value::Object(call.body.clone()),
+        api_key: call.api_key.as_deref(),
+        api_base: call.api_base.as_deref(),
+        custom_llm_provider: call.custom_llm_provider.as_deref(),
+        extra_headers: call.extra_headers.clone(),
+        timeout: call.timeout,
+    })?;
+    if stream && request.provider != ANTHROPIC_MESSAGES_PROVIDER {
+        return Err(Error::Unsupported("streaming messages for this provider"));
+    }
+    let context = RequestContext {
+        model: request.model.clone(),
+        custom_llm_provider: request.provider.clone(),
+        optional_params: Value::Object(
+            call.body
+                .iter()
+                .filter(|(name, _)| !matches!(name.as_str(), "model" | "messages"))
+                .map(|(name, value)| (name.clone(), value.clone()))
+                .collect(),
+        ),
+        secret_fields: Vec::new(),
+        api_key: call.api_key.clone().map(SecretValue::new),
+    };
+    let wire = host
+        .before_send(
+            WireRequest {
+                url: request.url,
+                headers: request.upstream_headers,
+                body: request.body,
+            },
+            context,
+        )
+        .await?;
+    let response = send(&wire.url, &wire.headers, &wire.body, request.timeout).await?;
+    if !response.status().is_success() {
+        return Err(provider_error(response).await);
+    }
+    if stream {
+        return relay(&host, response).await;
+    }
+    let text = response.text().await.map_err(network)?;
+    host.emit(CallEvent::ResponseReceived {
+        raw: RawResponse { body: text.clone() },
+    })
+    .await?;
+    decode_response(request.config, &request.model, &text).map(MessagesOutput::Message)
+}
+
+/// Hands each upstream chunk to the caller as it arrives. A caller that stops reading
+/// ends the upstream read, and the call completes with what it delivered.
+async fn relay(
+    host: &MessagesHost,
+    mut response: reqwest::Response,
+) -> Result<MessagesOutput, Error> {
+    if host.open(()).await? == Demand::Detached {
+        return Ok(MessagesOutput::Streamed);
+    }
+    while let Some(chunk) = response.chunk().await.map_err(network)? {
+        if host.deliver(chunk).await? == Demand::Detached {
+            break;
+        }
+    }
+    Ok(MessagesOutput::Streamed)
+}
--- a/litellm-rust/crates/core/src/ocr/route.rs
+++ b/litellm-rust/crates/core/src/ocr/route.rs
@ -39,6 +39,8 @@ impl Route for Ocr {
    type Error = Error;
    type Op = OcrOp;
    type OpResult = OcrOpResult;
+    type Chunk = std::convert::Infallible;
+    type StreamHead = std::convert::Infallible;
 }

 impl TokenRoute for Ocr {
--- a/litellm-rust/crates/core/tests/ocr.rs
+++ b/litellm-rust/crates/core/tests/ocr.rs
@ -198,6 +198,8 @@ fn event_name(event: &CallEvent) -> &'static str {
        CallEvent::ResponseReceived { .. } => "response",
        CallEvent::Succeeded { .. } => "success",
        CallEvent::Failed { .. } => "failure",
+        CallEvent::Opened => "opened",
+        CallEvent::Delivered => "delivered",
    }
 }

--- a/litellm-rust/crates/host-python/src/adapter.rs
+++ b/litellm-rust/crates/host-python/src/adapter.rs
@ -23,6 +23,7 @@ pub enum LifecycleStep {
 pub enum PublicValue<'a> {
    Response(&'a Py<PyAny>),
    Error(&'a PyErr),
+    Chunk(&'a Py<PyAny>),
 }

 /// One consumer of a call's lifecycle on the Python side. The driver calls the steps in
@ -111,6 +112,13 @@ pub trait RouteHost: Send + Sync {
        response: <Self::Route as Route>::Response,
    ) -> PyResult<Py<PyAny>>;

+    /// One streamed chunk as the caller receives it.
+    fn chunk(
+        &mut self,
+        py: Python<'_>,
+        chunk: <Self::Route as Route>::Chunk,
+    ) -> PyResult<Py<PyAny>>;
+
    fn classify(
        &self,
        py: Python<'_>,
--- a/litellm-rust/crates/host-python/src/driver.rs
+++ b/litellm-rust/crates/host-python/src/driver.rs
@ -3,7 +3,7 @@ use std::task::Poll;

 use futures_util::future::{AbortHandle, Abortable};
 use litellm_callbacks::event::{CallEvent, FailureOrigin, Timing, epoch_seconds};
-use litellm_callbacks::host::{HostOp, HostResult, HostStep};
+use litellm_callbacks::host::{Demand, HostOp, HostResult, HostStep};
 use litellm_callbacks::machine::{HostFailure, Machine, MachineStep};
 use litellm_callbacks::route::Route;
 use pyo3::exceptions::{PyBaseException, PyException, PyRuntimeError};
@ -38,6 +38,7 @@ struct MachineState<M: Machine> {
 enum Stage {
    Begin,
    Call,
+    Streaming,
    AfterSuccess,
    Succeeded(Py<PyAny>),
    Failed(Py<PyBaseException>),
@ -56,6 +57,8 @@ enum Expect {
 enum Pending {
    Native,
    Adapter(Expect),
+    /// The stream handed to the caller waits for its next read or its close.
+    Consumer,
 }

 enum Next<H: RouteHost> {
@ -121,7 +124,14 @@ where
    }
    match driver.resume(None)? {
        ExecutionStep::Return(value) => Ok(value),
-        ExecutionStep::Await(_) => Err(PyRuntimeError::new_err("sync call suspended")),
+        ExecutionStep::Open => py
+            .import("litellm.rust_bridge.lifecycle")?
+            .getattr("SyncStream")?
+            .call1((Py::new(py, Execution::suspended(driver))?,))
+            .map(Bound::unbind),
+        ExecutionStep::Await(_) | ExecutionStep::Yield(_) => {
+            Err(PyRuntimeError::new_err("sync call suspended"))
+        }
    }
 }

@ -162,6 +172,14 @@ where
                self.run_steps(py, HostStep::Ready(result))
            }
            (Some(Pending::Native), Some(Err(error))) => self.interrupt(py, error),
+            (Some(Pending::Consumer), Some(read)) => {
+                let demand = if read.is_ok() {
+                    Demand::More
+                } else {
+                    Demand::Detached
+                };
+                self.resume_machine(py, Some(Ok(HostResult::Demand(demand))))
+            }
            (Some(Pending::Adapter(expect)), Some(result)) => {
                match self.adapter.resume(py, result) {
                    Ok(step) => self.on_adapter(py, step, expect),
@ -216,7 +234,7 @@ where
    fn adapter_failed(&mut self, py: Python<'_>, error: PyErr) -> PyResult<ExecutionStep> {
        match self.stage {
            Stage::Begin | Stage::AfterSuccess => self.failure(py, error, FailureOrigin::Host),
-            Stage::Call => self.interrupt(py, error),
+            Stage::Call | Stage::Streaming => self.interrupt(py, error),
            Stage::Succeeded(_) | Stage::Failed(_) => Err(error),
        }
    }
@ -283,6 +301,8 @@ where
                    Err(error) => Err(error),
                }
            }
+            HostOp::Open(_) => return self.opened(py).map(Next::Return),
+            HostOp::Deliver(chunk) => return self.delivered(py, chunk).map(Next::Return),
            HostOp::Emit(event) => match self.adapter.emit(py, &event, None) {
                Ok(LifecycleStep::Done) => Ok(HostResult::Emitted),
                Ok(LifecycleStep::Await(awaitable)) => {
@ -299,6 +319,40 @@ where
        }
    }

+    fn opened(&mut self, py: Python<'_>) -> PyResult<ExecutionStep> {
+        self.stage = Stage::Streaming;
+        match self.adapter.emit(py, &CallEvent::Opened, None) {
+            Ok(LifecycleStep::Done) => {
+                self.pending = Some(Pending::Consumer);
+                Ok(ExecutionStep::Open)
+            }
+            Ok(_) => Err(missing_state()),
+            Err(error) => self.interrupt(py, error),
+        }
+    }
+
+    fn delivered(
+        &mut self,
+        py: Python<'_>,
+        chunk: <RouteOf<H> as Route>::Chunk,
+    ) -> PyResult<ExecutionStep> {
+        let chunk = match self.route.chunk(py, chunk) {
+            Ok(chunk) => chunk,
+            Err(error) => return self.interrupt(py, error),
+        };
+        let observed =
+            self.adapter
+                .emit(py, &CallEvent::Delivered, Some(PublicValue::Chunk(&chunk)));
+        match observed {
+            Ok(LifecycleStep::Done) => {
+                self.pending = Some(Pending::Consumer);
+                Ok(ExecutionStep::Yield(chunk))
+            }
+            Ok(_) => Err(missing_state()),
+            Err(error) => self.interrupt(py, error),
+        }
+    }
+
    fn interrupt(&mut self, py: Python<'_>, error: PyErr) -> PyResult<ExecutionStep> {
        let cancelled = is_cancellation(py, &error);
        let native = H::host_error(&error);
@ -369,6 +423,9 @@ where
            Ok(public) => public,
            Err(error) => return self.failure(py, error, FailureOrigin::Call),
        };
+        if let Stage::Streaming = self.stage {
+            return self.succeeded(py, public);
+        }
        self.stage = Stage::AfterSuccess;
        match self.adapter.after_success(py, public, self.timing()) {
            Ok(step) => self.on_adapter(py, step, Expect::Response),
@ -536,6 +593,8 @@ sys.modules.setdefault('litellm.rust_bridge', types.ModuleType('litellm.rust_bri
        type Error = Error;
        type Op = &'static str;
        type OpResult = String;
+        type Chunk = std::convert::Infallible;
+        type StreamHead = std::convert::Infallible;
    }

    /// Yields the scripted ops in order, then completes or fails as scripted.
@ -574,6 +633,7 @@ sys.modules.setdefault('litellm.rust_bridge', types.ModuleType('litellm.rust_bri
                        HostResult::Route(value) => value,
                        HostResult::BeforeSend(wire) => wire.url,
                        HostResult::Emitted => "emitted".into(),
+                        HostResult::Demand(demand) => format!("{demand:?}"),
                    });
                }
                if !self.ops.is_empty() {
@ -648,6 +708,10 @@ sys.modules.setdefault('litellm.rust_bridge', types.ModuleType('litellm.rust_bri
            }
        }

+        fn chunk(&mut self, _: Python<'_>, chunk: std::convert::Infallible) -> PyResult<Py<PyAny>> {
+            match chunk {}
+        }
+
        fn complete(&mut self, py: Python<'_>, response: String) -> PyResult<Py<PyAny>> {
            self.log.push("complete");
            Ok(pyo3::types::PyString::new(py, &response)
@ -1138,6 +1202,13 @@ sys.modules.setdefault('litellm.rust_bridge', types.ModuleType('litellm.rust_bri
                    )
                    .into())
                }
+                fn chunk(
+                    &mut self,
+                    _: Python<'_>,
+                    chunk: std::convert::Infallible,
+                ) -> PyResult<Py<PyAny>> {
+                    match chunk {}
+                }
                fn complete(&mut self, _: Python<'_>, _: String) -> PyResult<Py<PyAny>> {
                    Err(missing_state())
                }
--- a/litellm-rust/crates/host-python/src/handle.rs
+++ b/litellm-rust/crates/host-python/src/handle.rs
@ -8,6 +8,10 @@ use pyo3::prelude::*;
 pub enum ExecutionStep {
    Return(Py<PyAny>),
    Await(Py<PyAny>),
+    /// The call streams: the caller gets a stream over this execution, which stays
+    /// suspended until the stream asks for a chunk.
+    Open,
+    Yield(Py<PyAny>),
 }

 pub trait ExecutionBody: Send + Sync {
@ -34,6 +38,13 @@ impl Execution {
        }
    }

+    /// An execution already started elsewhere and now waiting for its next input.
+    pub fn suspended(body: impl ExecutionBody + 'static) -> Self {
+        Self {
+            state: ExecutionState::Suspended(Box::new(body)),
+        }
+    }
+
    fn advance(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
@ -64,6 +75,8 @@ impl Execution {
            let step = body.resume(result)?;
            let (tag, value, suspended) = match step {
                ExecutionStep::Await(value) => ("Await", value, true),
+                ExecutionStep::Open => ("Open", py.None(), true),
+                ExecutionStep::Yield(value) => ("Yield", value, true),
                ExecutionStep::Return(value) => ("Complete", value, false),
            };
            let step = py
--- a/litellm-rust/crates/python-bridge/src/marshal.rs
+++ b/litellm-rust/crates/python-bridge/src/marshal.rs
@ -18,10 +18,6 @@ pub(crate) struct RouteOptions {
    pub(crate) timeout: Option<Duration>,
 }

-pub(crate) fn body_argument(value: &Bound<'_, PyAny>) -> PyResult<Map<String, Value>> {
-    required_object("body", from_py_argument(value)?)
-}
-
 pub(crate) fn messages_argument(value: &Bound<'_, PyAny>) -> PyResult<Vec<Value>> {
    match from_py_argument(value)? {
        Value::Array(values) => Ok(values),
@ -192,18 +188,6 @@ mod tests {
                json!([{"role": "user", "content": [{"type": "text", "text": "hi"}]}])
            );

-            let body = py
-                .eval(
-                    c"{'model': 'claude', 'metadata': {'user': '1'}}",
-                    None,
-                    None,
-                )
-                .unwrap();
-            assert_eq!(
-                Value::Object(body_argument(&body).unwrap()),
-                json!({"model": "claude", "metadata": {"user": "1"}})
-            );
-
            let params = py.eval(c"{'temperature': 0.2}", None, None).unwrap();
            assert_eq!(
                optional_params_argument(&params).unwrap(),
--- a/litellm-rust/crates/python-bridge/src/routes/messages.rs
+++ b/litellm-rust/crates/python-bridge/src/routes/messages.rs
@ -1,88 +0,0 @@
-use litellm_core::messages::{Error, messages as run_messages, types::MessagesRequest};
-use litellm_host_python::{run_async, run_sync};
-use litellm_types::llms::anthropic_messages::anthropic_response::AnthropicMessagesResponse;
-use pyo3::prelude::*;
-use serde_json::{Map, Value};
-
-use crate::{
-    errors::messages_error_to_pyerr,
-    marshal::{RouteOptions, body_argument, extra_headers_argument, optional_timeout},
-};
-
-async fn execute(
-    body: Map<String, Value>,
-    options: RouteOptions,
-) -> Result<AnthropicMessagesResponse, Error> {
-    let RouteOptions {
-        model,
-        api_key,
-        api_base,
-        custom_llm_provider,
-        extra_headers,
-        timeout,
-    } = options;
-    run_messages(MessagesRequest {
-        model: &model,
-        body: Value::Object(body),
-        api_key: api_key.as_deref(),
-        api_base: api_base.as_deref(),
-        custom_llm_provider: custom_llm_provider.as_deref(),
-        extra_headers,
-        timeout,
-    })
-    .await
-}
-
-#[pyfunction]
-#[pyo3(signature = (model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None))]
-#[expect(
-    clippy::too_many_arguments,
-    reason = "one parameter per Python keyword"
-)]
-pub(crate) fn messages(
-    py: Python<'_>,
-    model: String,
-    #[pyo3(from_py_with = body_argument)] body: Map<String, Value>,
-    api_key: Option<String>,
-    api_base: Option<String>,
-    custom_llm_provider: Option<String>,
-    #[pyo3(from_py_with = extra_headers_argument)] extra_headers: Option<Map<String, Value>>,
-    timeout_seconds: Option<f64>,
-) -> PyResult<Py<PyAny>> {
-    let options = RouteOptions {
-        model,
-        api_key,
-        api_base,
-        custom_llm_provider,
-        extra_headers,
-        timeout: optional_timeout(timeout_seconds),
-    };
-    run_sync(py, execute(body, options), messages_error_to_pyerr)
-}
-
-#[pyfunction]
-#[pyo3(signature = (model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None))]
-#[expect(
-    clippy::too_many_arguments,
-    reason = "one parameter per Python keyword"
-)]
-pub(crate) fn amessages<'py>(
-    py: Python<'py>,
-    model: String,
-    #[pyo3(from_py_with = body_argument)] body: Map<String, Value>,
-    api_key: Option<String>,
-    api_base: Option<String>,
-    custom_llm_provider: Option<String>,
-    #[pyo3(from_py_with = extra_headers_argument)] extra_headers: Option<Map<String, Value>>,
-    timeout_seconds: Option<f64>,
-) -> PyResult<Bound<'py, PyAny>> {
-    let options = RouteOptions {
-        model,
-        api_key,
-        api_base,
-        custom_llm_provider,
-        extra_headers,
-        timeout: optional_timeout(timeout_seconds),
-    };
-    run_async(py, execute(body, options), messages_error_to_pyerr)
-}
--- a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs
+++ b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs
@ -0,0 +1,186 @@
+use bytes::Bytes;
+use litellm_core::messages::{
+    Error,
+    route::{Messages, MessagesCall, MessagesOp, MessagesOpResult, MessagesOutput},
+};
+use litellm_host_python::{HostOpError, RouteHost, from_py, lookup, to_py};
+use litellm_llms::custom_httpx::transport::Error as TransportError;
+use pyo3::{
+    exceptions::{PyException, PyValueError},
+    gc::{PyTraverseError, PyVisit},
+    prelude::*,
+    types::{PyBytes, PyDict},
+};
+use serde_json::{Map, Value};
+
+use crate::{
+    errors::{RustUpstreamError, messages_error_to_pyerr},
+    marshal::{optional_timeout, python_timeout_seconds},
+};
+
+/// The Anthropic Messages body fields a caller may pass besides `model` and `messages`,
+/// as `AnthropicMessagesRequestOptionalParams` declares them.
+const BODY_FIELDS: [&str; 20] = [
+    "max_tokens",
+    "metadata",
+    "stop_sequences",
+    "stream",
+    "system",
+    "temperature",
+    "thinking",
+    "tool_choice",
+    "tools",
+    "top_k",
+    "inference_geo",
+    "top_p",
+    "mcp_servers",
+    "context_management",
+    "container",
+    "output_format",
+    "speed",
+    "output_config",
+    "cache_control",
+    "reasoning_effort",
+];
+
+/// The Python side of the Messages route: projects the prepared arguments and builds the
+/// public response, chunks and exceptions.
+pub(super) struct MessagesRouteHost {
+    request: Py<PyAny>,
+}
+
+impl MessagesRouteHost {
+    pub(super) fn new(request: Py<PyAny>) -> Self {
+        Self { request }
+    }
+
+    fn project(&self, py: Python<'_>, arguments: &Bound<'_, PyDict>) -> PyResult<MessagesCall> {
+        let request = self.request.bind(py);
+        let argument = |name: &str| -> PyResult<Option<Bound<'_, PyAny>>> {
+            Ok(lookup(arguments, request, name)?.filter(|value| !value.is_none()))
+        };
+        let string = |name: &str| -> PyResult<Option<String>> {
+            argument(name)?.map(|value| value.extract()).transpose()
+        };
+        let model = string("model")?.ok_or_else(|| PyValueError::new_err("model is required"))?;
+        let messages =
+            argument("messages")?.ok_or_else(|| PyValueError::new_err("messages is required"))?;
+        let fields = BODY_FIELDS
+            .iter()
+            .filter_map(|name| match argument(name) {
+                Ok(Some(value)) => Some(from_py(&value).map(|value| ((*name).to_string(), value))),
+                Ok(None) => None,
+                Err(error) => Some(Err(error)),
+            })
+            .collect::<PyResult<Vec<(String, Value)>>>()?;
+        let body = [
+            ("model".to_string(), Value::String(model.clone())),
+            ("messages".to_string(), from_py(&messages)?),
+        ]
+        .into_iter()
+        .chain(fields)
+        .collect::<Map<String, Value>>();
+        let timeout = argument("timeout")?
+            .map(|value| python_timeout_seconds(py, value.unbind()))
+            .transpose()?
+            .flatten();
+        Ok(MessagesCall {
+            model,
+            body,
+            api_key: string("api_key")?,
+            api_base: string("api_base")?,
+            custom_llm_provider: string("custom_llm_provider")?,
+            extra_headers: argument("extra_headers")?
+                .map(|value| from_py(&value))
+                .transpose()?,
+            timeout: optional_timeout(timeout),
+        })
+    }
+
+    fn provider(&self, py: Python<'_>) -> String {
+        self.request
+            .bind(py)
+            .getattr("custom_llm_provider")
+            .and_then(|value| value.extract::<Option<String>>())
+            .ok()
+            .flatten()
+            .unwrap_or_else(|| "anthropic".into())
+    }
+
+    fn map_failure(&self, py: Python<'_>, error: PyErr) -> PyErr {
+        if !error.is_instance_of::<PyException>(py) {
+            return error;
+        }
+        let mapped = py
+            .import("litellm.rust_bridge.messages.route_host")
+            .and_then(|module| module.getattr("map_failure"))
+            .and_then(|map| map.call1((error.value(py), self.request.bind(py), self.provider(py))))
+            .and_then(|mapped| {
+                mapped
+                    .extract::<Py<pyo3::exceptions::PyBaseException>>()
+                    .map_err(PyErr::from)
+            });
+        match mapped {
+            Ok(mapped) => PyErr::from_value(mapped.into_bound(py).into_any()),
+            Err(_) => error,
+        }
+    }
+}
+
+impl RouteHost for MessagesRouteHost {
+    type Route = Messages;
+    type Failure = PyErr;
+
+    fn invoke(
+        &mut self,
+        py: Python<'_>,
+        arguments: &Bound<'_, PyDict>,
+        op: MessagesOp,
+    ) -> Result<MessagesOpResult, HostOpError<Error>> {
+        match op {
+            MessagesOp::ProjectRequest => self
+                .project(py, arguments)
+                .map(|call| MessagesOpResult::Request(Box::new(call)))
+                .map_err(|error| HostOpError::Python(self.map_failure(py, error))),
+        }
+    }
+
+    fn complete(&mut self, py: Python<'_>, response: MessagesOutput) -> PyResult<Py<PyAny>> {
+        match response {
+            MessagesOutput::Message(message) => py
+                .import("litellm.rust_bridge.messages.route_host")?
+                .getattr("response")?
+                .call1((to_py(py, &message)?,))
+                .map(Bound::unbind),
+            MessagesOutput::Streamed => Ok(py.None()),
+        }
+    }
+
+    fn chunk(&mut self, py: Python<'_>, chunk: Bytes) -> PyResult<Py<PyAny>> {
+        Ok(PyBytes::new(py, &chunk).into_any().unbind())
+    }
+
+    fn classify(&self, py: Python<'_>, error: Error) -> PyResult<PyErr> {
+        let native = match error {
+            Error::Transport(TransportError::Http { status, body }) => {
+                let error = RustUpstreamError::new_err((status, body));
+                error
+                    .value(py)
+                    .setattr("headers", Vec::<(String, String)>::new())?;
+                error
+            }
+            other => messages_error_to_pyerr(other),
+        };
+        Ok(self.map_failure(py, native))
+    }
+
+    fn host_error(error: &PyErr) -> Error {
+        Error::InvalidRequest(error.to_string())
+    }
+
+    fn close(&mut self, _: Python<'_>) {}
+
+    fn traverse(&self, visit: &PyVisit<'_>) -> Result<(), PyTraverseError> {
+        visit.call(&self.request)
+    }
+}
--- a/litellm-rust/crates/python-bridge/src/routes/messages/mod.rs
+++ b/litellm-rust/crates/python-bridge/src/routes/messages/mod.rs
@ -0,0 +1,64 @@
+mod host;
+
+use host::MessagesRouteHost;
+use litellm_callbacks_legacy::{LegacySurface, PublicCall, run_legacy_call};
+use litellm_core::messages::route::{messages_machine, supports};
+use pyo3::{
+    prelude::*,
+    types::{PyDict, PyTuple},
+};
+
+use crate::errors::RustBridgeDeclined;
+
+const SURFACE: LegacySurface = LegacySurface {
+    call_type: "anthropic_messages",
+    input_description: "Messages",
+};
+
+fn run_messages(
+    py: Python<'_>,
+    request: Bound<'_, PyAny>,
+    args: Bound<'_, PyTuple>,
+    kwargs: Bound<'_, PyDict>,
+    asynchronous: bool,
+) -> PyResult<Py<PyAny>> {
+    let model: String = request.getattr("model")?.extract()?;
+    let provider: Option<String> = request.getattr("custom_llm_provider")?.extract()?;
+    let stream = request
+        .getattr("stream")?
+        .extract::<Option<bool>>()?
+        .unwrap_or(false);
+    if !supports(&model, provider.as_deref(), stream) {
+        return Err(RustBridgeDeclined::new_err(
+            "the Rust Messages route does not serve this provider",
+        ));
+    }
+    run_legacy_call(
+        py,
+        SURFACE,
+        PublicCall::capture(&request, &args, &kwargs)?,
+        messages_machine(),
+        MessagesRouteHost::new(request.unbind()),
+        asynchronous,
+    )
+}
+
+#[pyfunction]
+pub(crate) fn messages(
+    py: Python<'_>,
+    request: Bound<'_, PyAny>,
+    args: Bound<'_, PyTuple>,
+    kwargs: Bound<'_, PyDict>,
+) -> PyResult<Py<PyAny>> {
+    run_messages(py, request, args, kwargs, false)
+}
+
+#[pyfunction]
+pub(crate) fn amessages(
+    py: Python<'_>,
+    request: Bound<'_, PyAny>,
+    args: Bound<'_, PyTuple>,
+    kwargs: Bound<'_, PyDict>,
+) -> PyResult<Py<PyAny>> {
+    run_messages(py, request, args, kwargs, true)
+}
--- a/litellm-rust/crates/python-bridge/src/routes/mod.rs
+++ b/litellm-rust/crates/python-bridge/src/routes/mod.rs
@ -22,11 +22,6 @@ mod tests {
                    "atranscription",
                    "(model, audio, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, optional_params=None, timeout_seconds=None)",
                ),
-                (
-                    "messages",
-                    "amessages",
-                    "(model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None)",
-                ),
                (
                    "chat_completions",
                    "achat_completions",
@ -113,25 +108,6 @@ value = Broken()
            );
            assert_eq!(async_chat_error.to_string(), sync_chat_error.to_string());

-            let invalid_body = PyList::empty(py);
-            let sync_messages_error = module
-                .getattr("messages")
-                .and_then(|function| function.call1(("model", &invalid_body)))
-                .expect_err("sync Messages should reject a non-dict body");
-            let async_messages_error = module
-                .getattr("amessages")
-                .and_then(|function| function.call1(("model", &invalid_body)))
-                .expect_err("async Messages should reject a non-dict body");
-
-            assert_eq!(
-                sync_messages_error.to_string(),
-                "ValueError: body must be a dict"
-            );
-            assert_eq!(
-                async_messages_error.to_string(),
-                sync_messages_error.to_string()
-            );
-
            let invalid_headers = PyList::empty(py);
            let kwargs = PyDict::new(py);
            kwargs
@ -193,13 +169,6 @@ value = Broken()
            headers_kwargs
                .set_item("extra_headers", &invalid)
                .expect("kwargs should accept extra_headers");
-            let invalid_body = PyList::empty(py);
-            let error = module
-                .getattr("messages")
-                .and_then(|function| function.call(("model", &invalid_body), Some(&headers_kwargs)))
-                .expect_err("body should be validated before headers");
-            assert_eq!(error.to_string(), "ValueError: body must be a dict");
-
            let invalid_payload =
                PyModule::new(py, "invalid_payload").expect("invalid payload should be created");
            let error = module
--- a/litellm-rust/crates/python-bridge/src/routes/ocr/host.rs
+++ b/litellm-rust/crates/python-bridge/src/routes/ocr/host.rs
@ -125,6 +125,10 @@ impl RouteHost for OcrRouteHost {
            .map(Bound::unbind)
    }

+    fn chunk(&mut self, _: Python<'_>, chunk: std::convert::Infallible) -> PyResult<Py<PyAny>> {
+        match chunk {}
+    }
+
    fn classify(&self, py: Python<'_>, error: Error) -> PyResult<PyErr> {
        Ok(self.map_failure(py, ocr_error_to_pyerr(error)))
    }
--- a/litellm/rust_bridge/_native.pyi
+++ b/litellm/rust_bridge/_native.pyi
@ -1,9 +1,11 @@
 from asyncio import Future
-from collections.abc import Coroutine, Mapping, Sequence
+from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
 from typing import Never, final

 from litellm.llms.base_llm.ocr.transformation import OCRResponse
+from litellm.rust_bridge.messages.entrypoints import LiteLLMMessagesRequest
 from litellm.rust_bridge.ocr.entrypoints import LiteLLMOcrRequest
+from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse

 class RustBridgeDeclined(Exception): ...
 class RustUpstreamError(Exception): ...
@ -39,23 +41,15 @@ def atranscription(
    timeout_seconds: float | None = None,
 ) -> Future[dict[str, object]]: ...
 def messages(
-    model: str,
-    body: Mapping[str, object],
-    api_key: str | None = None,
-    api_base: str | None = None,
-    custom_llm_provider: str | None = None,
-    extra_headers: Mapping[str, object] | None = None,
-    timeout_seconds: float | None = None,
-) -> dict[str, object]: ...
+    request: LiteLLMMessagesRequest,
+    args: tuple[object, ...],
+    kwargs: dict[str, object],
+) -> AnthropicMessagesResponse | Iterator[bytes]: ...
 def amessages(
-    model: str,
-    body: Mapping[str, object],
-    api_key: str | None = None,
-    api_base: str | None = None,
-    custom_llm_provider: str | None = None,
-    extra_headers: Mapping[str, object] | None = None,
-    timeout_seconds: float | None = None,
-) -> Future[dict[str, object]]: ...
+    request: LiteLLMMessagesRequest,
+    args: tuple[object, ...],
+    kwargs: dict[str, object],
+) -> Coroutine[object, object, AnthropicMessagesResponse | AsyncIterator[bytes]]: ...
 def chat_completions_decline(
    model: str,
    messages: Sequence[object],
--- a/litellm/rust_bridge/catalog.py
+++ b/litellm/rust_bridge/catalog.py
@ -59,6 +59,7 @@ Rules: TypeAlias = tuple[Rule, ...]

 RULES: Final[Rules] = (
    Rule(Route.OCR, Rollout.RUST_OPT_OUT),
+    Rule(Route.MESSAGES, Rollout.RUST_OPT_IN),
    Rule(Route.TRANSCRIPTION, Rollout.RUST_REQUIRED, providers=frozenset({"bedrock"})),
 )

--- a/litellm/rust_bridge/failures.py
+++ b/litellm/rust_bridge/failures.py
@ -5,8 +5,37 @@ from __future__ import annotations
 from collections.abc import Mapping
 from typing import Final, Protocol, cast  # noqa: TID251  # adapts the public exception mapper

+import httpx
+import openai
+from pydantic import TypeAdapter, ValidationError
+
 import litellm

+_UPSTREAM_ARGS: Final = TypeAdapter(tuple[int, str])
+_UPSTREAM_HEADERS: Final = TypeAdapter(list[tuple[str, str]])
+
+
+class UpstreamFailure(Exception):
+    def __init__(self, response: httpx.Response, cause: Exception) -> None:
+        super().__init__(str(cause))
+        self.message: Final = str(cause)
+        self.response: Final = response
+        self.status_code: Final = response.status_code
+        self.__cause__ = cause
+
+
+def _upstream_failure(error: Exception, api_base: str | None) -> Exception:
+    try:
+        status, body = _UPSTREAM_ARGS.validate_python(error.args)
+        headers: Final = _UPSTREAM_HEADERS.validate_python(getattr(error, "headers", None))
+    except ValidationError:
+        return error
+    http_request: Final = httpx.Request("POST", api_base or "https://docs.litellm.ai/docs")
+    return UpstreamFailure(
+        httpx.Response(status, content=body.encode(), headers=headers, request=http_request),
+        error,
+    )
+

 class ExceptionMapper(Protocol):
    def __call__(
@ -35,3 +64,17 @@ def map_failure(error: Exception, model: str, request_provider: str, kwargs: Map
    except Exception as public_error:
        public_error.__context__ = error
        return public_error
+
+
+def map_native_failure(
+    error: Exception, model: str, request_provider: str, kwargs: Mapping[str, object], api_base: str | None = None
+) -> Exception:
+    """`map_failure`, reading a native `(status, body)` provider failure as the HTTP response it was."""
+    original: Final = _upstream_failure(error, api_base)
+    public_error: Final = map_failure(original, model, request_provider, kwargs)
+    if isinstance(original, UpstreamFailure) and public_error.__context__ is original:
+        public_error.__context__ = error
+        if isinstance(public_error, openai.APIStatusError):
+            public_error.response = original.response
+            public_error.status_code = original.status_code
+    return public_error
--- a/litellm/rust_bridge/legacy_callbacks.py
+++ b/litellm/rust_bridge/legacy_callbacks.py
@ -6,6 +6,7 @@ registries it fans out to. It expires with that contract.

 from __future__ import annotations

+import asyncio
 import contextvars
 import datetime
 import traceback
@ -160,6 +161,21 @@ class LoggingWorker(Protocol):
    def ensure_initialized_and_enqueue(self, async_coroutine: Coroutine[object, object, None]) -> None: ...


+class StreamingLogBuilder(Protocol):
+    def __call__(
+        self,
+        *,
+        litellm_logging_obj: Logging,
+        passthrough_success_handler_obj: object,
+        url_route: str,
+        request_body: dict[str, object],
+        endpoint_type: object,
+        start_time: datetime.datetime,
+        raw_bytes: list[bytes],
+        end_time: datetime.datetime,
+    ) -> Coroutine[object, object, None]: ...
+
+
 class DeploymentHook(Protocol):
    def __call__(self, kwargs: dict[str, object], call_type: str) -> Awaitable[object]: ...

@ -306,3 +322,67 @@ def after_deployment_failure(kwargs: dict[str, object], error: Exception, call_t
        DeploymentFailureHook, utils.async_post_call_failure_deployment_hook
    )
    return hook(kwargs, error, call_type)
+
+
+def stream_opened(logger: Logging) -> None:
+    logger.stream = True
+    logger.model_call_details["stream"] = True
+
+
+def stream_success(
+    logger: Logging,
+    request_body: dict[str, object],
+    chunks: list[bytes],
+    start: datetime.datetime,
+    end: datetime.datetime,
+    first_chunk: datetime.datetime | None,
+) -> None:
+    from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
+        GLOBAL_PASS_THROUGH_SUCCESS_HANDLER_OBJ,
+    )
+    from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler
+    from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
+
+    if first_chunk is not None:
+        logger.completion_start_time = first_chunk
+        logger.model_call_details["completion_start_time"] = first_chunk
+    build: Final = cast(  # cast-ok: bounded adapter for the untyped pass-through logging builder
+        StreamingLogBuilder,
+        PassThroughStreamingHandler._route_streaming_logging_to_handler,  # pyright: ignore[reportPrivateUsage]  # the Messages stream iterator bills through the same builder
+    )
+    coroutine: Final = build(
+        litellm_logging_obj=logger,
+        passthrough_success_handler_obj=GLOBAL_PASS_THROUGH_SUCCESS_HANDLER_OBJ,
+        url_route="/v1/messages",
+        request_body=request_body,
+        endpoint_type=EndpointType.ANTHROPIC,
+        start_time=start,
+        raw_bytes=chunks,
+        end_time=end,
+    )
+    if getattr(logger, "_on_deferred_stream_complete", None) is not None:
+        logger._deferred_stream_complete_args = (coroutine,)  # pyright: ignore[reportAttributeAccessIssue]  # the proxy's deferred stream release reads this slot
+        return
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        from litellm.litellm_core_utils.litellm_logging import executor
+
+        executor.submit(contextvars.copy_context().run, asyncio.run, coroutine)
+        return
+    enqueue_logging(coroutine)
+
+
+def stream_failure(
+    logger: Logging, request_body: dict[str, object], chunks: list[bytes], error: Exception
+) -> Coroutine[object, object, None]:
+    from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler
+    from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
+
+    return PassThroughStreamingHandler.schedule_stream_failure_logging(
+        litellm_logging_obj=logger,
+        endpoint_type=EndpointType.ANTHROPIC,
+        request_body=request_body,
+        raw_bytes=chunks,
+        exception=error,
+    )
--- a/litellm/rust_bridge/lifecycle.py
+++ b/litellm/rust_bridge/lifecycle.py
@ -1,8 +1,8 @@
 from __future__ import annotations

-from collections.abc import Awaitable
+from collections.abc import AsyncIterator, Awaitable, Iterator
 from dataclasses import dataclass
-from typing import Protocol
+from typing import Final, Protocol


@dataclass(frozen=True, slots=True)
@ -15,28 +15,133 @@ class Complete:
    value: object


+@dataclass(frozen=True, slots=True)
+class Open:
+    value: None
+
+
+@dataclass(frozen=True, slots=True)
+class Yield:
+    value: object
+
+
+Settled = Complete | Open | Yield
+Step = Await | Settled
+
+
 class Execution(Protocol):
-    def start(self) -> Await | Complete: ...
+    def start(self) -> Step: ...

-    def resume_value(self, value: object) -> Await | Complete: ...
+    def resume_value(self, value: object) -> Step: ...

-    def resume_error(self, error: BaseException) -> Await | Complete: ...
+    def resume_error(self, error: BaseException) -> Step: ...

    def close(self) -> None: ...


+class StreamClosed(Exception):
+    """Tells a streaming execution that its caller stopped reading."""
+
+
+async def _settle(execution: Execution, step: Step) -> Settled:
+    while isinstance(step, Await):
+        try:
+            value = await step.awaitable  # rebind-ok: each selected await produces the next protocol input
+        except GeneratorExit:
+            raise
+        except BaseException as error:
+            step = execution.resume_error(error)  # rebind-ok: advance the execution protocol
+        else:
+            step = execution.resume_value(value)  # rebind-ok: advance the execution protocol
+    return step
+
+
+def _settled(step: Step) -> Settled:
+    if isinstance(step, Await):
+        raise RuntimeError("sync call suspended")
+    return step
+
+
 async def drive(execution: Execution) -> object:
+    handed_off = False  # rebind-ok: set once the execution belongs to the returned stream
    try:
-        step = execution.start()  # rebind-ok: the execution protocol advances after each selected await
-        while isinstance(step, Await):
-            try:
-                value = await step.awaitable  # rebind-ok: each selected await produces the next protocol input
-            except GeneratorExit:
-                raise
-            except BaseException as error:
-                step = execution.resume_error(error)  # rebind-ok: advance the execution protocol
-            else:
-                step = execution.resume_value(value)  # rebind-ok: advance the execution protocol
+        step: Final = await _settle(execution, execution.start())
+        if isinstance(step, Open):
+            handed_off = True
+            return Stream(execution)
        return step.value
    finally:
-        execution.close()
+        if not handed_off:
+            execution.close()
+
+
+class Stream(AsyncIterator[object]):
+    """A streamed native call: each read resumes the execution until its next chunk."""
+
+    def __init__(self, execution: Execution) -> None:
+        self._execution: Final = execution
+        self._done = False
+
+    def __aiter__(self) -> Stream:
+        return self
+
+    async def __anext__(self) -> object:
+        if self._done:
+            raise StopAsyncIteration
+        try:
+            step: Final = await _settle(self._execution, self._execution.resume_value(None))
+        except BaseException:
+            self._finish()
+            raise
+        if isinstance(step, Yield):
+            return step.value
+        self._finish()
+        raise StopAsyncIteration
+
+    async def aclose(self) -> None:
+        if self._done:
+            return
+        try:
+            await _settle(self._execution, self._execution.resume_error(StreamClosed()))
+        finally:
+            self._finish()
+
+    def _finish(self) -> None:
+        self._done = True
+        self._execution.close()
+
+
+class SyncStream(Iterator[object]):
+    """The sync form of `Stream`; its execution never suspends on an awaitable."""
+
+    def __init__(self, execution: Execution) -> None:
+        self._execution: Final = execution
+        self._done = False
+
+    def __iter__(self) -> SyncStream:
+        return self
+
+    def __next__(self) -> object:
+        if self._done:
+            raise StopIteration
+        try:
+            step: Final = _settled(self._execution.resume_value(None))
+        except BaseException:
+            self._finish()
+            raise
+        if isinstance(step, Yield):
+            return step.value
+        self._finish()
+        raise StopIteration
+
+    def close(self) -> None:
+        if self._done:
+            return
+        try:
+            _settled(self._execution.resume_error(StreamClosed()))
+        finally:
+            self._finish()
+
+    def _finish(self) -> None:
+        self._done = True
+        self._execution.close()
--- a/litellm/rust_bridge/messages/entrypoints.py
+++ b/litellm/rust_bridge/messages/entrypoints.py
@ -1,6 +1,6 @@
 from __future__ import annotations

-from collections.abc import Awaitable, Mapping, Sequence
+from collections.abc import AsyncIterator, Awaitable, Iterator, Mapping, Sequence
 from dataclasses import dataclass
 from typing import Final, Protocol, cast  # noqa: TID251  # validates dynamically loaded native callables

@ -26,7 +26,7 @@ class NativeMessages(Protocol):
        request: LiteLLMMessagesRequest,
        args: tuple[object, ...],
        kwargs: Mapping[str, object],
-    ) -> AnthropicMessagesResponse: ...
+    ) -> AnthropicMessagesResponse | Iterator[bytes]: ...


 class NativeAmessages(Protocol):
@ -35,7 +35,7 @@ class NativeAmessages(Protocol):
        request: LiteLLMMessagesRequest,
        args: tuple[object, ...],
        kwargs: Mapping[str, object],
-    ) -> Awaitable[AnthropicMessagesResponse]: ...
+    ) -> Awaitable[AnthropicMessagesResponse | AsyncIterator[bytes]]: ...


 def _messages_binding(value: object) -> NativeMessages | None:
@ -50,5 +50,5 @@ def _amessages_binding(value: object) -> NativeAmessages | None:
    return cast("NativeAmessages", value)  # cast-ok: callable validated at the native binding boundary


-NATIVE_MESSAGES: Final = NativeBinding("anthropic_messages_handler", validate=_messages_binding)
-NATIVE_AMESSAGES: Final = NativeBinding("anthropic_messages", validate=_amessages_binding)
+NATIVE_MESSAGES: Final = NativeBinding("messages", validate=_messages_binding)
+NATIVE_AMESSAGES: Final = NativeBinding("amessages", validate=_amessages_binding)
--- a/litellm/rust_bridge/messages/route_host.py
+++ b/litellm/rust_bridge/messages/route_host.py
@ -20,4 +20,4 @@ def arguments(request: LiteLLMMessagesRequest) -> Mapping[str, object]:


 def map_failure(error: Exception, request: LiteLLMMessagesRequest, request_provider: str) -> Exception:
-    return failures.map_failure(error, request.model, request_provider, arguments(request))
+    return failures.map_native_failure(error, request.model, request_provider, arguments(request), request.api_base)
--- a/litellm/rust_bridge/ocr/route_host.py
+++ b/litellm/rust_bridge/ocr/route_host.py
@ -4,40 +4,17 @@ from collections.abc import Mapping
 from types import MappingProxyType
 from typing import Final

-import httpx
-import openai
-from pydantic import TypeAdapter, ValidationError
+from pydantic import TypeAdapter

 import litellm
 from litellm.llms.base_llm.ocr.transformation import PROVIDER_NATIVE_RESPONSE_KEY, OCRResponse
 from litellm.rust_bridge import failures
+from litellm.rust_bridge.failures import UpstreamFailure
 from litellm.rust_bridge.ocr.entrypoints import LiteLLMOcrRequest

+__all__ = ("UpstreamFailure", "arguments", "map_failure", "response")
+
 _RESPONSE_ADAPTER: Final = TypeAdapter(dict[str, object])
-_UPSTREAM_ARGS: Final = TypeAdapter(tuple[int, str])
-_UPSTREAM_HEADERS: Final = TypeAdapter(list[tuple[str, str]])
-
-
-class UpstreamFailure(Exception):
-    def __init__(self, response: httpx.Response, cause: Exception) -> None:
-        super().__init__(str(cause))
-        self.message: Final = str(cause)
-        self.response: Final = response
-        self.status_code: Final = response.status_code
-        self.__cause__ = cause
-
-
-def _upstream_failure(error: Exception, request: LiteLLMOcrRequest) -> Exception:
-    try:
-        status, body = _UPSTREAM_ARGS.validate_python(error.args)
-        headers: Final = _UPSTREAM_HEADERS.validate_python(getattr(error, "headers", None))
-    except ValidationError:
-        return error
-    http_request: Final = httpx.Request("POST", request.api_base or "https://docs.litellm.ai/docs")
-    return UpstreamFailure(
-        httpx.Response(status, content=body.encode(), headers=headers, request=http_request),
-        error,
-    )


 def response(value: Mapping[str, object]) -> OCRResponse:
@ -61,11 +38,4 @@ def map_failure(error: Exception, request: LiteLLMOcrRequest, request_provider:
            model=request.model.removeprefix(f"{request_provider}/"),
            llm_provider=request_provider,
        )
-    original: Final = _upstream_failure(error, request)
-    public_error: Final = failures.map_failure(original, request.model, request_provider, arguments(request))
-    if isinstance(original, UpstreamFailure) and public_error.__context__ is original:
-        public_error.__context__ = error
-        if isinstance(public_error, openai.APIStatusError):
-            public_error.response = original.response
-            public_error.status_code = original.status_code
-    return public_error
+    return failures.map_native_failure(error, request.model, request_provider, arguments(request), request.api_base)
--- a/tests/test_litellm/rust_bridge/native_route_wheel_test.py
+++ b/tests/test_litellm/rust_bridge/native_route_wheel_test.py
@ -73,7 +73,7 @@ def assert_native_request(
    headers: HTTPMessage,
    body: object,
 ) -> None:
-    if route not in {"transcription", "messages", "chat_completions"}:
+    if route not in {"transcription", "chat_completions"}:
        raise AssertionError(f"unexpected route marker: {route!r}")
    if outcome not in {"success", "429", "hang"}:
        raise AssertionError(f"unexpected outcome marker: {outcome!r}")
@ -89,10 +89,6 @@ def assert_native_request(
    assert path == "/v1/messages"
    assert headers.get("x-api-key") == "sk-native"
    assert body["model"] == "claude-sonnet-4-5"
-    if route == "messages":
-        assert body["max_tokens"] == 16
-        assert body["messages"][0]["content"] == "hello-from-messages"
-        return
    assert body["max_tokens"] == 17
    assert body["messages"][0]["content"] == [{"type": "text", "text": "hello-from-chat"}]

@ -132,17 +128,6 @@ def route_kwargs(route: str, api_base: str, outcome: str) -> dict[str, object]:
                "language": "en",
            },
        }
-    if route == "messages":
-        return common | {
-            "model": "claude-sonnet-4-5",
-            "body": {
-                "model": "claude-sonnet-4-5",
-                "max_tokens": 16,
-                "messages": [{"role": "user", "content": "hello-from-messages"}],
-            },
-            "api_key": "sk-native",
-            "custom_llm_provider": "anthropic",
-        }
    if route == "chat_completions":
        return common | {
            "model": "anthropic/claude-sonnet-4-5",
@ -165,8 +150,6 @@ def assert_success(route: str, response: object) -> None:
 def success_value(route: str, response: dict[object, object]) -> object:
    if route == "transcription":
        return response["text"]
-    if route == "messages":
-        return response["content"][0]["text"]
    return response["choices"][0]["message"]["content"]


@ -181,7 +164,7 @@ def assert_rate_limit(native: object, route: str, error: BaseException) -> None:


 def exercise_sync(native: object, api_base: str) -> None:
-    for route in ("transcription", "messages", "chat_completions"):
+    for route in ("transcription", "chat_completions"):
        function: Final = getattr(native, route)
        assert_success(route, function(**route_kwargs(route, api_base, "success")))
        try:
@ -193,7 +176,7 @@ def exercise_sync(native: object, api_base: str) -> None:


 async def exercise_async(native: object, api_base: str) -> None:
-    for route in ("transcription", "messages", "chat_completions"):
+    for route in ("transcription", "chat_completions"):
        function: Final = getattr(native, f"a{route}")
        assert_success(route, await function(**route_kwargs(route, api_base, "success")))
        try:
@ -206,11 +189,11 @@ async def exercise_async(native: object, api_base: str) -> None:

 async def exercise_async_concurrency(native: object, api_base: str) -> None:
    responses: Final = await asyncio.wait_for(
-        asyncio.gather(*(native.amessages(**route_kwargs("messages", api_base, "success")) for _ in range(32))),
+        asyncio.gather(*(native.achat_completions(**route_kwargs("chat_completions", api_base, "success")) for _ in range(32))),
        timeout=15,
    )
    for response in responses:
-        assert_success("messages", response)
+        assert_success("chat_completions", response)


 def exercise_routes(native_path: Path, api_base: str) -> object:
@ -223,8 +206,8 @@ def exercise_routes(native_path: Path, api_base: str) -> object:

 def exercise_signal(native: object, api_base: str) -> int:
    try:
-        native.messages(
-            **route_kwargs("messages", api_base, "hang"),
+        native.chat_completions(
+            **route_kwargs("chat_completions", api_base, "hang"),
        )
    except KeyboardInterrupt:
        sys.stdout.write("KeyboardInterrupt\n")
--- a/tests/test_litellm/rust_bridge/test_bindings.py
+++ b/tests/test_litellm/rust_bridge/test_bindings.py
@ -43,8 +43,8 @@ def test_binding_validates_native_attribute(
 ROUTE_BINDINGS: Final = (
    ("completion", chat_completions.NATIVE_COMPLETION),
    ("acompletion", chat_completions.NATIVE_ACOMPLETION),
-    ("anthropic_messages_handler", messages.NATIVE_MESSAGES),
-    ("anthropic_messages", messages.NATIVE_AMESSAGES),
+    ("messages", messages.NATIVE_MESSAGES),
+    ("amessages", messages.NATIVE_AMESSAGES),
    ("responses", responses.NATIVE_RESPONSES),
    ("aresponses", responses.NATIVE_ARESPONSES),
    ("ocr", ocr.NATIVE_OCR),
--- a/tests/test_litellm/rust_bridge/test_catalog.py
+++ b/tests/test_litellm/rust_bridge/test_catalog.py
@ -40,6 +40,10 @@ def test_shipped_decisions(
        enabled: Final = environment == "1" if environment is not None else process is not False
        assert catalog.rollout(context) is Rollout.RUST_OPT_OUT
        assert catalog.decision(context) is (Decision.RUST_WITH_FALLBACK if enabled else Decision.PYTHON)
+    elif route is Route.MESSAGES:
+        enabled: Final = environment == "1" if environment is not None else process is True
+        assert catalog.rollout(context) is Rollout.RUST_OPT_IN
+        assert catalog.decision(context) is (Decision.RUST_WITH_FALLBACK if enabled else Decision.PYTHON)
    elif route is Route.TRANSCRIPTION and provider == "bedrock":
        assert catalog.rollout(context) is Rollout.RUST_REQUIRED
        assert catalog.decision(context) is Decision.RUST_REQUIRED
--- a/tests/test_litellm/rust_bridge/test_runtime.py
+++ b/tests/test_litellm/rust_bridge/test_runtime.py
@ -157,7 +157,6 @@ def test_context_outside_rule_stays_on_python() -> None:
    (
        Context(Route.CHAT_COMPLETIONS, provider="anthropic"),
        Context(Route.CHAT_COMPLETIONS, provider="bedrock"),
-        Context(Route.MESSAGES, provider="anthropic"),
        Context(Route.RESPONSES, provider="openai"),
        Context(Route.TRANSCRIPTION, provider="openai"),
    ),
--- a/tests/test_litellm_rust/messages/init.py
+++ b/tests/test_litellm_rust/messages/init.py
--- a/tests/test_litellm_rust/messages/test_callbacks.py
+++ b/tests/test_litellm_rust/messages/test_callbacks.py
@ -0,0 +1,175 @@
+from collections.abc import AsyncIterator, Iterator
+from typing import Final
+
+import pytest
+
+import litellm
+from litellm.integrations.custom_logger import CustomLogger
+from tests.test_litellm_rust.support.callback_recorder import RecordingLogger, drain_logging
+from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec
+from tests.test_litellm_rust.support.requests import (
+    MESSAGES,
+    MESSAGES_EVENTS,
+    MESSAGES_MODEL,
+    MESSAGES_RESPONSE,
+    request_body,
+)
+
+pytestmark = pytest.mark.requires_rust_extension
+
+STREAM: Final = ResponseSpec(body=None, events=MESSAGES_EVENTS)
+
+
+@pytest.fixture
+def messages_server(recording_server: RecordingServer) -> RecordingServer:
+    recording_server.default_response = ResponseSpec(body=MESSAGES_RESPONSE)
+    return recording_server
+
+
+def arguments(server: RecordingServer, **kwargs: object) -> dict[str, object]:
+    return {
+        "model": MESSAGES_MODEL,
+        "messages": [dict(message) for message in MESSAGES],
+        "max_tokens": 64,
+        "api_key": "test-key",
+        "api_base": server.base_url,
+        **kwargs,
+    }
+
+
+def assert_served_natively(server: RecordingServer) -> None:
+    assert len(server.requests) == 1
+    assert not server.requests[0].headers.get("user-agent", "").startswith("python-httpx")
+
+
+@pytest.mark.asyncio
+async def test_native_messages_callbacks_see_the_provider_request_and_the_public_response(
+    messages_server: RecordingServer,
+) -> None:
+    recorder: Final = RecordingLogger()
+
+    response: Final = await litellm.anthropic.messages.acreate(
+        **arguments(messages_server, callbacks=[recorder], litellm_call_id="messages-success")
+    )
+
+    assert_served_natively(messages_server)
+    assert response["content"] == MESSAGES_RESPONSE["content"]
+    sent: Final = messages_server.requests[0]
+    assert sent.path == "/v1/messages"
+    assert sent.body == {"model": "claude-sonnet-5", "messages": list(MESSAGES), "max_tokens": 64, "stream": False}
+    pre_call: Final = recorder.wait_for("log_pre_api_call")
+    assert request_body(pre_call[0].kwargs) == sent.body
+    success: Final = await recorder.wait_for_async("async_log_success_event")
+    assert len(success) == 1
+    assert success[0].call_type == "anthropic_messages"
+    assert success[0].kwargs["litellm_call_id"] == "messages-success"
+    assert success[0].response.choices[0].message.content == "Hello from native Messages"
+
+
+@pytest.mark.asyncio
+async def test_native_messages_pre_call_body_edit_reaches_the_provider(messages_server: RecordingServer) -> None:
+    class Edit(CustomLogger):
+        def log_pre_api_call(self, model, messages, kwargs):
+            request_body(kwargs)["temperature"] = 0.25
+
+    await litellm.anthropic.messages.acreate(**arguments(messages_server, callbacks=[Edit()]))
+
+    assert messages_server.requests[0].body["temperature"] == 0.25
+
+
+@pytest.mark.asyncio
+async def test_native_messages_provider_error_reaches_caller_and_failure_callbacks_as_one_public_error(
+    messages_server: RecordingServer,
+) -> None:
+    messages_server.enqueue(
+        ResponseSpec(body={"type": "error", "error": {"type": "invalid_request_error", "message": "bad"}}, status=400)
+    )
+    observed: Final = []
+
+    class Observe(CustomLogger):
+        def log_failure_event(self, kwargs, response_obj, start_time, end_time):
+            observed.append(("sync", kwargs["exception"]))
+
+        async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
+            observed.append(("async", kwargs["exception"]))
+
+    with pytest.raises(litellm.BadRequestError) as raised:
+        await litellm.anthropic.messages.acreate(**arguments(messages_server, callbacks=[Observe()]))
+
+    assert_served_natively(messages_server)
+    assert [phase for phase, _ in observed] == ["sync", "async"]
+    assert all(error is raised.value for _, error in observed)
+
+
+def sse_payload() -> bytes:
+    return b"".join(STREAM.payloads())
+
+
+@pytest.mark.asyncio
+async def test_native_messages_stream_relays_provider_events_and_logs_success_once_after_the_last_chunk(
+    messages_server: RecordingServer,
+) -> None:
+    messages_server.enqueue(STREAM)
+    recorder: Final = RecordingLogger()
+
+    stream: Final = await litellm.anthropic.messages.acreate(
+        **arguments(messages_server, stream=True, callbacks=[recorder])
+    )
+    assert isinstance(stream, AsyncIterator)
+    first: Final = await anext(stream)
+    await drain_logging()
+    assert "async_log_success_event" not in recorder.names
+    rest: Final = [chunk async for chunk in stream]
+
+    assert first + b"".join(rest) == sse_payload()
+    assert_served_natively(messages_server)
+    assert messages_server.requests[0].body["stream"] is True
+    success: Final = await recorder.wait_for_async("async_log_success_event")
+    assert len(success) == 1
+    assert success[0].kwargs["stream"] is True
+    assert success[0].kwargs["completion_start_time"] is not None
+    assert "log_failure_event" not in recorder.names
+
+
+@pytest.mark.asyncio
+async def test_native_messages_stream_closed_early_logs_success_once_for_what_was_delivered(
+    messages_server: RecordingServer,
+) -> None:
+    messages_server.enqueue(STREAM)
+    recorder: Final = RecordingLogger()
+
+    stream: Final = await litellm.anthropic.messages.acreate(
+        **arguments(messages_server, stream=True, callbacks=[recorder])
+    )
+    assert isinstance(stream, AsyncIterator)
+    await anext(stream)
+    await stream.aclose()
+
+    success: Final = await recorder.wait_for_async("async_log_success_event")
+    assert len(success) == 1
+    with pytest.raises(StopAsyncIteration):
+        await anext(stream)
+
+
+def test_native_sync_messages_stream_relays_provider_events_and_logs_success_once(
+    messages_server: RecordingServer,
+) -> None:
+    messages_server.enqueue(STREAM)
+    recorder: Final = RecordingLogger()
+
+    stream: Final = litellm.anthropic.messages.create(**arguments(messages_server, stream=True, callbacks=[recorder]))
+    assert isinstance(stream, Iterator)
+
+    assert b"".join(stream) == sse_payload()
+    assert_served_natively(messages_server)
+    assert len(recorder.wait_for("async_log_success_event")) == 1
+
+
+def test_native_sync_messages_returns_the_provider_message(messages_server: RecordingServer) -> None:
+    recorder: Final = RecordingLogger()
+
+    response: Final = litellm.anthropic.messages.create(**arguments(messages_server, callbacks=[recorder]))
+
+    assert_served_natively(messages_server)
+    assert response["content"] == MESSAGES_RESPONSE["content"]
+    assert len(recorder.wait_for("log_success_event")) == 1
--- a/tests/test_litellm_rust/support/recording_server.py
+++ b/tests/test_litellm_rust/support/recording_server.py
@ -25,6 +25,12 @@ class ResponseSpec:
    status: int = 200
    headers: dict[str, str] = field(default_factory=dict)
    delay: float = 0
+    events: tuple[tuple[str, object], ...] = ()
+
+    def payloads(self) -> tuple[bytes, ...]:
+        if not self.events:
+            return (json.dumps(self.body).encode(),)
+        return tuple(f"event: {event}\ndata: {json.dumps(data)}\n\n".encode() for event, data in self.events)


@dataclass
@ -73,15 +79,17 @@ def recording_service() -> Iterator[RecordingServer]:
            response: Final = responses.pop(0) if responses else copy.deepcopy(recording_server.default_response)
            if response.delay:
                time.sleep(response.delay)
-            payload: Final = json.dumps(response.body).encode()
+            payloads: Final = response.payloads()
            self.send_response(response.status)
-            self.send_header("Content-Type", "application/json")
-            self.send_header("Content-Length", str(len(payload)))
+            self.send_header("Content-Type", "text/event-stream" if response.events else "application/json")
+            self.send_header("Content-Length", str(sum(len(payload) for payload in payloads)))
            for name, value in response.headers.items():
                self.send_header(name, value)
            self.end_headers()
            try:
-                self.wfile.write(payload)
+                for payload in payloads:
+                    self.wfile.write(payload)
+                    self.wfile.flush()
            except (BrokenPipeError, ConnectionResetError):
                pass

--- a/tests/test_litellm_rust/support/requests.py
+++ b/tests/test_litellm_rust/support/requests.py
@ -12,6 +12,41 @@ OCR_RESPONSE: Final = {
    "usage_info": {"pages_processed": 1, "doc_size_bytes": 3},
 }

+MESSAGES_MODEL: Final = "anthropic/claude-sonnet-5"
+MESSAGES: Final = ({"role": "user", "content": "Hello"},)
+MESSAGES_RESPONSE: Final = {
+    "id": "msg_native",
+    "type": "message",
+    "role": "assistant",
+    "model": "claude-sonnet-5",
+    "content": [{"type": "text", "text": "Hello from native Messages"}],
+    "stop_reason": "end_turn",
+    "stop_sequence": None,
+    "usage": {"input_tokens": 5, "output_tokens": 4},
+}
+MESSAGES_EVENTS: Final = (
+    ("message_start", {"type": "message_start", "message": {**MESSAGES_RESPONSE, "content": [], "stop_reason": None}}),
+    ("content_block_start", {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}),
+    (
+        "content_block_delta",
+        {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "text_delta", "text": "Hello from native Messages"},
+        },
+    ),
+    ("content_block_stop", {"type": "content_block_stop", "index": 0}),
+    (
+        "message_delta",
+        {
+            "type": "message_delta",
+            "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+            "usage": {"output_tokens": 4},
+        },
+    ),
+    ("message_stop", {"type": "message_stop"}),
+)
+

 def ocr_arguments(server: RecordingServer, **kwargs: object) -> dict[str, object]:
    return {