Merge pull request #41531 from BerriAI/litellm_anthropic_stream_types

feat(rust): map Anthropic Messages transformations

litellm-rust/Cargo.lock

@@ -1953,6 +1953,8 @@ dependencies = [
 name = "litellm-core"
 version = "0.1.0"
 dependencies = [
+ "aws-smithy-eventstream",
+ "aws-smithy-types",
  "base64 0.22.1",
  "bytes",
  "data-url",
@@ -1961,6 +1963,7 @@ dependencies = [
  "litellm-auth-aws",
  "litellm-auth-azure",
  "litellm-auth-gcp",
+ "litellm-framing",
  "mime_guess",
  "moka",
  "rand 0.8.7",
@@ -1975,6 +1978,7 @@ dependencies = [
  "strum",
  "subtle",
  "thiserror 2.0.19",
+ "time",
  "tokio",
  "tokio-tungstenite",
  "url",

litellm-rust/Cargo.toml

@@ -11,6 +11,7 @@ repository = "https://github.com/BerriAI/litellm"
 [workspace.dependencies]
 bytes = "1"
 litellm-core = { path = "crates/core" }
+litellm-framing = { path = "crates/framer" }
 litellm-auth = { path = "crates/auth" }
 litellm-auth-aws = { path = "crates/auth-aws" }
 litellm-auth-azure = { path = "crates/auth-azure" }
@@ -39,6 +40,7 @@ base64 = "0.22"
 moka = { version = "0.12.16", features = ["future"] }
 strum = { version = "0.28.0", features = ["derive"] }
 url = "2.5.8"
+time = { version = "0.3.53", features = ["parsing"] }
 criterion = "0.8.2"
 veil = "0.3.0"
 

litellm-rust/crates/core/Cargo.toml

@@ -15,6 +15,7 @@ litellm-auth.workspace = true
 litellm-auth-aws.workspace = true
 litellm-auth-azure.workspace = true
 litellm-auth-gcp.workspace = true
+litellm-framing.workspace = true
 moka.workspace = true
 mime_guess = "2.0.5"
 rand.workspace = true
@@ -29,9 +30,12 @@ subtle.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 tokio-tungstenite.workspace = true
 thiserror.workspace = true
+time.workspace = true
 sha2.workspace = true
 url.workspace = true
 veil.workspace = true
 
 [dev-dependencies]
+aws-smithy-eventstream = "=0.61.1"
+aws-smithy-types = "1.6.1"
 rstest.workspace = true

litellm-rust/crates/core/src/chat_completions/mod.rs

@@ -14,6 +14,7 @@ pub mod conversation;
 pub(crate) mod handler;
 mod prepare;
 pub mod response_utils;
+pub mod streaming;
 pub mod transformation;
 pub mod types;
 

litellm-rust/crates/core/src/chat_completions/streaming.rs

@@ -0,0 +1,9 @@
+pub trait StreamTransformer {
+    type Input;
+    type Output;
+    type Error;
+
+    fn transform(&mut self, input: Self::Input) -> Result<Vec<Self::Output>, Self::Error>;
+
+    fn finish(&mut self) -> Result<Vec<Self::Output>, Self::Error>;
+}

litellm-rust/crates/core/src/chat_completions/types.rs

@@ -120,3 +120,83 @@ pub struct ChatCompletionsResponse {
     pub choices: Vec<ChatCompletionsChoice>,
     pub usage: ChatCompletionsUsage,
 }
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct ChatCompletionToolCallFunctionChunk {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub name: Option<String>,
+    pub arguments: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub provider_specific_fields: Option<Map<String, Value>>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct ChatCompletionToolCallChunk {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub id: Option<String>,
+    #[serde(rename = "type")]
+    pub tool_type: String,
+    pub function: ChatCompletionToolCallFunctionChunk,
+    pub index: i64,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum ChatCompletionThinkingBlock {
+    Thinking {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        thinking: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        signature: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        cache_control: Option<Value>,
+    },
+    RedactedThinking {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        data: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        cache_control: Option<Value>,
+    },
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
+pub struct ChatCompletionDelta {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub content: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub role: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub tool_calls: Option<Vec<ChatCompletionToolCallChunk>>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub reasoning_content: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub thinking_blocks: Option<Vec<ChatCompletionThinkingBlock>>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub provider_specific_fields: Option<Map<String, Value>>,
+    #[serde(flatten)]
+    pub extra: Map<String, Value>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct ChatCompletionStreamingChoice {
+    pub index: u64,
+    pub delta: ChatCompletionDelta,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub finish_reason: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub logprobs: Option<Value>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct ChatCompletionChunk {
+    pub id: String,
+    pub created: u64,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub model: Option<String>,
+    pub object: String,
+    pub choices: Vec<ChatCompletionStreamingChoice>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub usage: Option<ChatCompletionsUsage>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub provider_specific_fields: Option<Map<String, Value>>,
+}

litellm-rust/crates/core/src/messages/error.rs

@@ -2,16 +2,54 @@
 pub enum Error {
     #[error("invalid provider: {0}")]
     InvalidProvider(String),
+    #[error("missing required field: {0}")]
+    MissingField(&'static str),
     #[error("invalid request: {0}")]
     InvalidRequest(String),
     #[error("invalid response: {0}")]
     InvalidResponse(String),
-    #[error("routing error: {0}")]
-    Routing(String),
+    #[error("unsupported by the Rust messages route: {0}")]
+    Unsupported(&'static str),
     #[error(transparent)]
     Auth(#[from] litellm_auth::Error),
     #[error(transparent)]
     Transport(#[from] crate::transport::Error),
     #[error(transparent)]
     Headers(#[from] crate::http_utils::HeaderError),
+    #[error("stream framing failed: {0}")]
+    StreamFraming(String),
+    #[error("Anthropic SSE frame has no data")]
+    MissingStreamData,
+    #[error("Anthropic stream event is invalid: {0}")]
+    InvalidStreamEvent(String),
+    #[error("Bedrock event payload is invalid: {0}")]
+    InvalidBedrockPayload(String),
+    #[error("Bedrock event payload has invalid base64: {0}")]
+    InvalidBedrockBase64(String),
+}
+
+impl Error {
+    pub fn is_request(&self) -> bool {
+        match self {
+            Self::InvalidProvider(_)
+            | Self::MissingField(_)
+            | Self::InvalidRequest(_)
+            | Self::Unsupported(_)
+            | Self::Headers(_) => true,
+            Self::Auth(error) => !matches!(error, litellm_auth::Error::MissingApiKey { .. }),
+            _ => false,
+        }
+    }
+
+    pub fn is_response(&self) -> bool {
+        matches!(
+            self,
+            Self::InvalidResponse(_)
+                | Self::StreamFraming(_)
+                | Self::MissingStreamData
+                | Self::InvalidStreamEvent(_)
+                | Self::InvalidBedrockPayload(_)
+                | Self::InvalidBedrockBase64(_)
+        )
+    }
 }

litellm-rust/crates/core/src/messages/handler.rs

@@ -46,9 +46,7 @@ pub(super) async fn execute_messages_provider_stream(
 ) -> Result<reqwest::Response, Error> {
     let request = prepare_provider_request(request)?;
     if request.provider != ANTHROPIC_MESSAGES_PROVIDER {
-        return Err(Error::InvalidRequest(
-            "streaming messages is not supported for this provider".to_string(),
-        ));
+        return Err(Error::Unsupported("streaming messages for this provider"));
     }
 
     let mut request_builder = http_client().post(&request.url).json(&request.body);

litellm-rust/crates/core/src/providers/anthropic/chat_completions/mod.rs

@@ -1 +1,2 @@
+pub mod streaming;
 pub mod transformation;

litellm-rust/crates/core/src/providers/anthropic/chat_completions/streaming.rs

@@ -0,0 +1,164 @@
+use std::collections::HashMap;
+
+use serde_json::Value;
+
+use crate::chat_completions::Error;
+use crate::chat_completions::streaming::StreamTransformer;
+use crate::chat_completions::types::{
+    ChatCompletionChunk, ChatCompletionThinkingBlock, ChatCompletionToolCallChunk,
+    ChatCompletionsUsage,
+};
+use crate::providers::anthropic::messages::streaming::{
+    AnthropicContentBlock, AnthropicContentBlockDelta, AnthropicMessagesStreamEvent,
+    AnthropicStreamUsage,
+};
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum AnthropicJsonChunkType {
+    ValidJson,
+    AccumulatedJson,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum AnthropicContentBlockType {
+    Text,
+    ToolUse,
+    ServerToolUse,
+    Thinking,
+    RedactedThinking,
+    Compaction,
+    ToolResult(String),
+    Other(String),
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct AnthropicContentBlockDeltaEvent {
+    pub index: u64,
+    pub delta: AnthropicContentBlockDelta,
+}
+
+pub struct AnthropicChatCompletionsStreamTransformer {
+    pub content_blocks: Vec<AnthropicContentBlockDeltaEvent>,
+    pub tool_index: i64,
+    pub json_mode: bool,
+    pub speed: Option<String>,
+    pub tool_name_reverse_map: HashMap<String, String>,
+    pub response_id: String,
+    pub served_model: Option<String>,
+    pub is_response_format_tool: bool,
+    pub converted_response_format_tool: bool,
+    pub accumulated_json: String,
+    pub chunk_type: AnthropicJsonChunkType,
+    pub current_content_block_type: Option<AnthropicContentBlockType>,
+    pub web_search_results: Vec<Value>,
+    pub web_search_calls: HashMap<String, Value>,
+    pub compaction_blocks: Vec<Value>,
+    pub reasoning_content_chunks: Vec<String>,
+    pub server_tool_inputs: HashMap<String, Value>,
+    pub tool_results: Vec<Value>,
+    pub current_server_tool_id: Option<String>,
+    pub container_id: Option<String>,
+}
+
+impl AnthropicChatCompletionsStreamTransformer {
+    pub fn new(
+        _json_mode: bool,
+        _speed: Option<String>,
+        _tool_name_reverse_map: HashMap<String, String>,
+    ) -> Self {
+        todo!()
+    }
+
+    pub fn check_empty_tool_call_args(&self) -> bool {
+        todo!()
+    }
+
+    pub fn handle_usage(&mut self, _usage: AnthropicStreamUsage) -> ChatCompletionsUsage {
+        todo!()
+    }
+
+    pub fn handle_content_block_delta(
+        &mut self,
+        _index: u64,
+        _delta: AnthropicContentBlockDelta,
+    ) -> (
+        String,
+        Option<ChatCompletionToolCallChunk>,
+        Vec<ChatCompletionThinkingBlock>,
+        Option<Value>,
+        Option<String>,
+    ) {
+        todo!()
+    }
+
+    pub fn handle_content_block_start(
+        &mut self,
+        _index: u64,
+        _content_block: AnthropicContentBlock,
+    ) -> Result<ChatCompletionChunk, Error> {
+        todo!()
+    }
+
+    pub fn handle_json_mode_chunk(
+        &mut self,
+        _text: String,
+        _tool_use: Option<ChatCompletionToolCallChunk>,
+    ) -> (String, Option<ChatCompletionToolCallChunk>) {
+        todo!()
+    }
+
+    pub fn handle_accumulated_json_chunk(
+        &mut self,
+        _data: &str,
+        _is_final: bool,
+    ) -> Result<Option<ChatCompletionChunk>, Error> {
+        todo!()
+    }
+
+    pub fn handle_redacted_thinking_content(
+        &mut self,
+        _content_block: &AnthropicContentBlock,
+    ) -> Vec<ChatCompletionThinkingBlock> {
+        todo!()
+    }
+
+    pub fn web_search_call_snapshot(&self) -> HashMap<String, Value> {
+        todo!()
+    }
+
+    pub fn complete_web_search_call(&mut self, _result: Value) {
+        todo!()
+    }
+
+    pub fn build_code_interpreter_results(&self) -> Vec<Value> {
+        todo!()
+    }
+
+    pub fn handle_message_delta(
+        &mut self,
+        _event: AnthropicMessagesStreamEvent,
+    ) -> (Option<String>, Option<ChatCompletionsUsage>, Option<Value>) {
+        todo!()
+    }
+
+    pub fn chunk_parser(
+        &mut self,
+        _event: AnthropicMessagesStreamEvent,
+    ) -> Result<ChatCompletionChunk, Error> {
+        todo!()
+    }
+}
+
+impl StreamTransformer for AnthropicChatCompletionsStreamTransformer {
+    type Input = AnthropicMessagesStreamEvent;
+    type Output = ChatCompletionChunk;
+    type Error = Error;
+
+    fn transform(&mut self, _input: Self::Input) -> Result<Vec<Self::Output>, Self::Error> {
+        todo!()
+    }
+
+    fn finish(&mut self) -> Result<Vec<Self::Output>, Self::Error> {
+        todo!()
+    }
+}

litellm-rust/crates/core/src/providers/anthropic/messages/batches.rs

@@ -0,0 +1,338 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use time::OffsetDateTime;
+use url::Url;
+
+use crate::messages::Error;
+use crate::messages::types::AnthropicMessagesResponse;
+use crate::providers::anthropic::messages::transformation::resolve_anthropic_api_base;
+
+const BATCHES_PATH_SUFFIX: &str = "/v1/messages/batches";
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct AnthropicBatchRequestCounts {
+    #[serde(default)]
+    pub processing: u64,
+    #[serde(default)]
+    pub succeeded: u64,
+    #[serde(default)]
+    pub errored: u64,
+    #[serde(default)]
+    pub canceled: u64,
+    #[serde(default)]
+    pub expired: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct AnthropicMessageBatch {
+    #[serde(default)]
+    pub id: String,
+    #[serde(default = "default_processing_status")]
+    pub processing_status: String,
+    pub created_at: Option<String>,
+    pub ended_at: Option<String>,
+    pub expires_at: Option<String>,
+    pub cancel_initiated_at: Option<String>,
+    pub archived_at: Option<String>,
+    #[serde(default)]
+    pub request_counts: AnthropicBatchRequestCounts,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum BatchStatus {
+    InProgress,
+    Cancelling,
+    Completed,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BatchRequestCounts {
+    pub total: u64,
+    pub completed: u64,
+    pub failed: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LiteLlmMessageBatch {
+    pub id: String,
+    pub object: String,
+    pub endpoint: String,
+    pub input_file_id: String,
+    pub completion_window: String,
+    pub status: BatchStatus,
+    pub output_file_id: String,
+    pub created_at: i64,
+    pub in_progress_at: Option<i64>,
+    pub expires_at: Option<i64>,
+    pub completed_at: Option<i64>,
+    pub expired_at: Option<i64>,
+    pub cancelling_at: Option<i64>,
+    pub cancelled_at: Option<i64>,
+    pub request_counts: BatchRequestCounts,
+}
+
+pub trait AnthropicBatchesConfig {
+    fn create_batch_url(
+        &self,
+        api_base: Option<&str>,
+        env_lookup: &dyn Fn(&str) -> Option<String>,
+    ) -> Result<String, Error>;
+
+    fn transform_create_batch_request(&self) -> Result<Value, Error>;
+
+    fn transform_create_batch_response(
+        &self,
+        response: AnthropicMessageBatch,
+        now: i64,
+    ) -> Result<LiteLlmMessageBatch, Error>;
+
+    fn retrieve_batch_url(
+        &self,
+        api_base: Option<&str>,
+        batch_id: &str,
+        env_lookup: &dyn Fn(&str) -> Option<String>,
+    ) -> Result<String, Error>;
+
+    fn transform_retrieve_batch_request(&self) -> Value;
+
+    fn transform_retrieve_batch_response(
+        &self,
+        response: AnthropicMessageBatch,
+        now: i64,
+    ) -> LiteLlmMessageBatch;
+
+    fn transform_batch_results(&self, body: &str) -> Result<Vec<AnthropicMessagesResponse>, Error>;
+}
+
+pub struct AnthropicBatchesTransformation;
+
+pub const ANTHROPIC_BATCHES_TRANSFORMATION: AnthropicBatchesTransformation =
+    AnthropicBatchesTransformation;
+
+fn default_processing_status() -> String {
+    "in_progress".into()
+}
+
+fn timestamp(value: Option<&str>) -> Option<i64> {
+    value
+        .and_then(|value| {
+            OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).ok()
+        })
+        .map(OffsetDateTime::unix_timestamp)
+}
+
+fn batches_base_url(
+    api_base: Option<&str>,
+    env_lookup: &dyn Fn(&str) -> Option<String>,
+) -> Result<Url, Error> {
+    let api_base = resolve_anthropic_api_base(api_base, env_lookup);
+    let api_base = api_base.trim_end_matches('/');
+    let complete_url = if api_base.ends_with(BATCHES_PATH_SUFFIX) {
+        api_base.to_string()
+    } else if let Some(base) = api_base.strip_suffix("/v1/messages") {
+        format!("{base}{BATCHES_PATH_SUFFIX}")
+    } else {
+        format!("{api_base}{BATCHES_PATH_SUFFIX}")
+    };
+    Url::parse(&complete_url)
+        .map_err(|error| Error::InvalidRequest(format!("invalid Anthropic API base: {error}")))
+}
+
+impl AnthropicBatchesConfig for AnthropicBatchesTransformation {
+    fn create_batch_url(
+        &self,
+        api_base: Option<&str>,
+        env_lookup: &dyn Fn(&str) -> Option<String>,
+    ) -> Result<String, Error> {
+        Ok(batches_base_url(api_base, env_lookup)?.into())
+    }
+
+    fn transform_create_batch_request(&self) -> Result<Value, Error> {
+        Err(Error::Unsupported("Anthropic message batch creation"))
+    }
+
+    fn transform_create_batch_response(
+        &self,
+        _response: AnthropicMessageBatch,
+        _now: i64,
+    ) -> Result<LiteLlmMessageBatch, Error> {
+        Err(Error::Unsupported("Anthropic message batch creation"))
+    }
+
+    fn retrieve_batch_url(
+        &self,
+        api_base: Option<&str>,
+        batch_id: &str,
+        env_lookup: &dyn Fn(&str) -> Option<String>,
+    ) -> Result<String, Error> {
+        if batch_id.is_empty() {
+            return Err(Error::MissingField("batch_id"));
+        }
+        let mut url = batches_base_url(api_base, env_lookup)?;
+        url.path_segments_mut()
+            .map_err(|_| Error::InvalidRequest("Anthropic API base cannot be a base URL".into()))?
+            .push(batch_id);
+        Ok(url.into())
+    }
+
+    fn transform_retrieve_batch_request(&self) -> Value {
+        Value::Object(Default::default())
+    }
+
+    fn transform_retrieve_batch_response(
+        &self,
+        response: AnthropicMessageBatch,
+        now: i64,
+    ) -> LiteLlmMessageBatch {
+        let created_at = timestamp(response.created_at.as_deref());
+        let ended_at = timestamp(response.ended_at.as_deref());
+        let expires_at = timestamp(response.expires_at.as_deref());
+        let cancel_initiated_at = timestamp(response.cancel_initiated_at.as_deref());
+        let archived_at = timestamp(response.archived_at.as_deref());
+        let status = match response.processing_status.as_str() {
+            "canceling" => BatchStatus::Cancelling,
+            "ended" => BatchStatus::Completed,
+            _ => BatchStatus::InProgress,
+        };
+        let request_counts = BatchRequestCounts {
+            total: response.request_counts.processing
+                + response.request_counts.succeeded
+                + response.request_counts.errored
+                + response.request_counts.canceled
+                + response.request_counts.expired,
+            completed: response.request_counts.succeeded,
+            failed: response.request_counts.errored,
+        };
+
+        LiteLlmMessageBatch {
+            id: response.id.clone(),
+            object: "batch".into(),
+            endpoint: "/v1/messages".into(),
+            input_file_id: "None".into(),
+            completion_window: "24h".into(),
+            status,
+            output_file_id: response.id,
+            created_at: created_at.unwrap_or(now),
+            in_progress_at: (response.processing_status == "in_progress")
+                .then_some(created_at)
+                .flatten(),
+            expires_at,
+            completed_at: (response.processing_status == "ended")
+                .then_some(ended_at)
+                .flatten(),
+            expired_at: archived_at,
+            cancelling_at: (response.processing_status == "canceling")
+                .then_some(cancel_initiated_at)
+                .flatten(),
+            cancelled_at: (response.processing_status == "canceling")
+                .then_some(ended_at)
+                .flatten(),
+            request_counts,
+        }
+    }
+
+    fn transform_batch_results(&self, body: &str) -> Result<Vec<AnthropicMessagesResponse>, Error> {
+        body.lines()
+            .filter(|line| !line.trim().is_empty())
+            .filter_map(|line| serde_json::from_str::<Value>(line.trim()).ok())
+            .map(|record| {
+                serde_json::from_value(record["result"]["message"].clone()).map_err(|error| {
+                    Error::InvalidResponse(format!("invalid Anthropic batch result: {error}"))
+                })
+            })
+            .collect()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::json;
+
+    use super::*;
+
+    #[test]
+    fn builds_and_encodes_message_batch_urls() {
+        assert_eq!(
+            ANTHROPIC_BATCHES_TRANSFORMATION
+                .create_batch_url(None, &|_| None)
+                .unwrap(),
+            "https://api.anthropic.com/v1/messages/batches"
+        );
+        assert_eq!(
+            ANTHROPIC_BATCHES_TRANSFORMATION
+                .create_batch_url(Some("https://proxy.test/v1/messages/batches"), &|_| None)
+                .unwrap(),
+            "https://proxy.test/v1/messages/batches"
+        );
+        assert_eq!(
+            ANTHROPIC_BATCHES_TRANSFORMATION
+                .retrieve_batch_url(Some("https://proxy.test"), "batch/id ?", &|_| None)
+                .unwrap(),
+            "https://proxy.test/v1/messages/batches/batch%2Fid%20%3F"
+        );
+        assert_eq!(
+            ANTHROPIC_BATCHES_TRANSFORMATION.transform_retrieve_batch_request(),
+            json!({})
+        );
+    }
+
+    #[test]
+    fn maps_retrieved_batch_status_counts_and_timestamps_like_python() {
+        let response: AnthropicMessageBatch = serde_json::from_value(json!({
+            "id": "msgbatch_1",
+            "processing_status": "ended",
+            "created_at": "2025-01-01T00:00:00Z",
+            "ended_at": "2025-01-01T00:01:00Z",
+            "expires_at": "not-a-timestamp",
+            "request_counts": {
+                "processing": 1,
+                "succeeded": 2,
+                "errored": 3,
+                "canceled": 4,
+                "expired": 5
+            }
+        }))
+        .unwrap();
+
+        let batch = ANTHROPIC_BATCHES_TRANSFORMATION.transform_retrieve_batch_response(response, 7);
+        assert_eq!(batch.status, BatchStatus::Completed);
+        assert_eq!(batch.created_at, 1_735_689_600);
+        assert_eq!(batch.completed_at, Some(1_735_689_660));
+        assert_eq!(batch.expires_at, None);
+        assert_eq!(
+            batch.request_counts,
+            BatchRequestCounts {
+                total: 15,
+                completed: 2,
+                failed: 3
+            }
+        );
+    }
+
+    #[test]
+    fn extracts_message_responses_from_ndjson_and_skips_non_json_lines() {
+        let body = r#"not-json
+{"result":{"message":{"id":"msg_1","type":"message","role":"assistant","model":"claude-test","content":[],"stop_reason":"end_turn","stop_sequence":null}}}
+"#;
+        let messages = ANTHROPIC_BATCHES_TRANSFORMATION
+            .transform_batch_results(body)
+            .unwrap();
+
+        assert_eq!(messages.len(), 1);
+        assert_eq!(messages[0].id, "msg_1");
+    }
+
+    #[test]
+    fn preserves_python_placeholder_for_batch_creation() {
+        assert!(matches!(
+            ANTHROPIC_BATCHES_TRANSFORMATION.transform_create_batch_request(),
+            Err(Error::Unsupported("Anthropic message batch creation"))
+        ));
+        let response: AnthropicMessageBatch = serde_json::from_value(json!({})).unwrap();
+        assert!(matches!(
+            ANTHROPIC_BATCHES_TRANSFORMATION.transform_create_batch_response(response, 0),
+            Err(Error::Unsupported("Anthropic message batch creation"))
+        ));
+    }
+}

litellm-rust/crates/core/src/providers/anthropic/messages/count_tokens.rs

@@ -0,0 +1,168 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+use crate::constants::ANTHROPIC_OAUTH_TOKEN_PREFIX;
+use crate::messages::Error;
+use crate::messages::types::{AnthropicMessage, SystemPrompt};
+
+const COUNT_TOKENS_ENDPOINT: &str = "https://api.anthropic.com/v1/messages/count_tokens";
+const TOKEN_COUNTING_BETA: &str = "token-counting-2024-11-01";
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct AnthropicCountTokensRequest {
+    pub model: String,
+    pub messages: Vec<AnthropicMessage>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub tools: Option<Vec<Value>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub system: Option<SystemPrompt>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct AnthropicCountTokensResponse {
+    pub input_tokens: u64,
+}
+
+pub trait AnthropicCountTokensConfig {
+    fn endpoint(&self) -> &'static str;
+
+    fn validate_request(&self, model: &str, messages: &[AnthropicMessage]) -> Result<(), Error>;
+
+    fn transform_request(
+        &self,
+        model: &str,
+        messages: Vec<AnthropicMessage>,
+        tools: Option<Vec<Value>>,
+        system: Option<SystemPrompt>,
+    ) -> Result<AnthropicCountTokensRequest, Error>;
+
+    fn required_headers(&self, api_key: &str) -> Vec<(&'static str, String)>;
+}
+
+pub struct AnthropicCountTokensTransformation;
+
+pub const ANTHROPIC_COUNT_TOKENS_TRANSFORMATION: AnthropicCountTokensTransformation =
+    AnthropicCountTokensTransformation;
+
+impl AnthropicCountTokensConfig for AnthropicCountTokensTransformation {
+    fn endpoint(&self) -> &'static str {
+        COUNT_TOKENS_ENDPOINT
+    }
+
+    fn transform_request(
+        &self,
+        model: &str,
+        messages: Vec<AnthropicMessage>,
+        tools: Option<Vec<Value>>,
+        system: Option<SystemPrompt>,
+    ) -> Result<AnthropicCountTokensRequest, Error> {
+        self.validate_request(model, &messages)?;
+
+        Ok(AnthropicCountTokensRequest {
+            model: model.to_string(),
+            messages,
+            tools,
+            system,
+        })
+    }
+
+    fn validate_request(&self, model: &str, messages: &[AnthropicMessage]) -> Result<(), Error> {
+        if model.is_empty() {
+            return Err(Error::MissingField("model"));
+        }
+        if messages.is_empty() {
+            return Err(Error::MissingField("messages"));
+        }
+        Ok(())
+    }
+
+    fn required_headers(&self, api_key: &str) -> Vec<(&'static str, String)> {
+        let auth = if api_key.starts_with(ANTHROPIC_OAUTH_TOKEN_PREFIX) {
+            ("authorization", format!("Bearer {api_key}"))
+        } else {
+            ("x-api-key", api_key.to_string())
+        };
+        vec![
+            ("content-type", "application/json".to_string()),
+            auth,
+            ("anthropic-version", "2023-06-01".to_string()),
+            ("anthropic-beta", TOKEN_COUNTING_BETA.to_string()),
+        ]
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::{Map, json};
+
+    use super::*;
+    use crate::messages::types::MessageContent;
+
+    fn message() -> AnthropicMessage {
+        AnthropicMessage {
+            role: "user".into(),
+            content: MessageContent::Text("hello".into()),
+            extra: Map::new(),
+        }
+    }
+
+    #[test]
+    fn maps_the_python_count_tokens_contract() {
+        let request = ANTHROPIC_COUNT_TOKENS_TRANSFORMATION
+            .transform_request(
+                "claude-test",
+                vec![message()],
+                Some(vec![json!({"name": "lookup"})]),
+                Some(SystemPrompt::Text("system".into())),
+            )
+            .unwrap();
+
+        assert_eq!(
+            serde_json::to_value(request).unwrap(),
+            json!({
+                "model": "claude-test",
+                "messages": [{"role": "user", "content": "hello"}],
+                "tools": [{"name": "lookup"}],
+                "system": "system"
+            })
+        );
+        assert_eq!(
+            ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.endpoint(),
+            COUNT_TOKENS_ENDPOINT
+        );
+    }
+
+    #[test]
+    fn rejects_the_invalid_requests_python_rejects() {
+        assert!(matches!(
+            ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.transform_request(
+                "",
+                vec![message()],
+                None,
+                None
+            ),
+            Err(Error::MissingField("model"))
+        ));
+        assert!(matches!(
+            ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.transform_request(
+                "claude-test",
+                vec![],
+                None,
+                None
+            ),
+            Err(Error::MissingField("messages"))
+        ));
+    }
+
+    #[test]
+    fn uses_api_key_or_oauth_headers_without_combining_credentials() {
+        let api_key = ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.required_headers("sk-ant-api");
+        assert!(api_key.contains(&("x-api-key", "sk-ant-api".into())));
+        assert!(!api_key.iter().any(|(name, _)| *name == "authorization"));
+
+        let oauth = ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.required_headers("sk-ant-oat-test");
+        assert!(oauth.contains(&("authorization", "Bearer sk-ant-oat-test".into())));
+        assert!(!oauth.iter().any(|(name, _)| *name == "x-api-key"));
+        assert!(oauth.contains(&("anthropic-beta", TOKEN_COUNTING_BETA.into())));
+    }
+}

litellm-rust/crates/core/src/providers/anthropic/messages/mod.rs

@@ -1 +1,4 @@
+pub mod batches;
+pub mod count_tokens;
+pub mod streaming;
 pub mod transformation;

litellm-rust/crates/core/src/providers/anthropic/messages/streaming.rs

@@ -0,0 +1,282 @@
+use base64::Engine;
+use bytes::Buf;
+use futures_util::{Stream, StreamExt};
+use litellm_framing::Framer;
+use litellm_framing::aws_event_stream::{AwsEventStreamFrame, AwsEventStreamFramer};
+use litellm_framing::sse::{SseFrame, SseFramer};
+use serde::{Deserialize, Serialize};
+use serde_json::{Map, Value};
+
+use crate::messages::Error;
+
+#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
+pub struct AnthropicStreamUsage {
+    #[serde(default)]
+    pub input_tokens: u64,
+    #[serde(default)]
+    pub output_tokens: u64,
+    #[serde(default)]
+    pub cache_creation_input_tokens: u64,
+    #[serde(default)]
+    pub cache_read_input_tokens: u64,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub server_tool_use: Option<Value>,
+    #[serde(flatten)]
+    pub extra: Map<String, Value>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct AnthropicStreamMessage {
+    pub id: String,
+    #[serde(rename = "type")]
+    pub message_type: String,
+    pub role: String,
+    pub model: String,
+    pub content: Vec<Value>,
+    pub stop_reason: Option<String>,
+    pub stop_sequence: Option<String>,
+    pub usage: AnthropicStreamUsage,
+    #[serde(flatten)]
+    pub extra: Map<String, Value>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum AnthropicContentBlockDelta {
+    TextDelta {
+        text: String,
+    },
+    InputJsonDelta {
+        partial_json: String,
+    },
+    #[serde(rename = "citations_delta")]
+    Citations {
+        citation: Value,
+    },
+    ThinkingDelta {
+        thinking: String,
+    },
+    SignatureDelta {
+        signature: String,
+    },
+    CompactionDelta {
+        content: String,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct AnthropicContentBlock {
+    #[serde(rename = "type")]
+    pub block_type: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub name: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub text: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub input: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub thinking: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub signature: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub data: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub content: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub caller: Option<Value>,
+    #[serde(flatten)]
+    pub extra: Map<String, Value>,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
+pub struct AnthropicMessageDelta {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stop_reason: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stop_sequence: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stop_details: Option<Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub container: Option<Value>,
+    #[serde(flatten)]
+    pub extra: Map<String, Value>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct AnthropicStreamError {
+    #[serde(rename = "type")]
+    pub error_type: String,
+    pub message: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub details: Option<Value>,
+    #[serde(flatten)]
+    pub extra: Map<String, Value>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum AnthropicMessagesStreamEvent {
+    MessageStart {
+        message: AnthropicStreamMessage,
+    },
+    ContentBlockStart {
+        index: u64,
+        content_block: AnthropicContentBlock,
+    },
+    ContentBlockDelta {
+        index: u64,
+        delta: AnthropicContentBlockDelta,
+    },
+    ContentBlockStop {
+        index: u64,
+    },
+    MessageDelta {
+        delta: AnthropicMessageDelta,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        usage: Option<AnthropicStreamUsage>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        context_management: Option<Value>,
+    },
+    MessageStop,
+    Ping,
+    Error {
+        error: AnthropicStreamError,
+    },
+}
+
+#[derive(Deserialize)]
+struct BedrockChunkPayload {
+    bytes: String,
+}
+
+pub fn decode_anthropic_sse_frame(frame: SseFrame) -> Result<AnthropicMessagesStreamEvent, Error> {
+    let data = frame.data.ok_or(Error::MissingStreamData)?;
+    serde_json::from_str(&data).map_err(|error| Error::InvalidStreamEvent(error.to_string()))
+}
+
+pub fn decode_bedrock_anthropic_frame(
+    frame: AwsEventStreamFrame,
+) -> Result<AnthropicMessagesStreamEvent, Error> {
+    let payload: BedrockChunkPayload = serde_json::from_slice(&frame.payload)
+        .map_err(|error| Error::InvalidBedrockPayload(error.to_string()))?;
+    let event = base64::engine::general_purpose::STANDARD
+        .decode(payload.bytes)
+        .map_err(|error| Error::InvalidBedrockBase64(error.to_string()))?;
+    serde_json::from_slice(&event).map_err(|error| Error::InvalidStreamEvent(error.to_string()))
+}
+
+pub fn direct_anthropic_event_stream<S, B, E>(
+    input: S,
+) -> impl Stream<Item = Result<AnthropicMessagesStreamEvent, Error>> + Send
+where
+    S: Stream<Item = Result<B, E>> + Send,
+    B: Buf + Send,
+    E: std::error::Error + Send + Sync + 'static,
+{
+    SseFramer.frame(input).map(|frame| {
+        let frame = frame.map_err(|error| Error::StreamFraming(error.to_string()))?;
+        decode_anthropic_sse_frame(frame)
+    })
+}
+
+pub fn bedrock_anthropic_event_stream<S, B, E>(
+    input: S,
+) -> impl Stream<Item = Result<AnthropicMessagesStreamEvent, Error>> + Send
+where
+    S: Stream<Item = Result<B, E>> + Send,
+    B: Buf + Send,
+    E: std::error::Error + Send + Sync + 'static,
+{
+    AwsEventStreamFramer.frame(input).map(|frame| {
+        let frame = frame.map_err(|error| Error::StreamFraming(error.to_string()))?;
+        decode_bedrock_anthropic_frame(frame)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use std::io;
+
+    use aws_smithy_eventstream::frame::write_message_to;
+    use aws_smithy_types::event_stream::{Header, HeaderValue, Message};
+    use base64::engine::general_purpose::STANDARD;
+    use bytes::Bytes;
+    use futures_util::TryStreamExt;
+
+    use super::*;
+
+    const TEXT_DELTA: &str =
+        r#"{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello"}}"#;
+
+    #[tokio::test]
+    async fn direct_anthropic_sse_frames_into_typed_events() {
+        let wire = format!("event: content_block_delta\ndata: {TEXT_DELTA}\n\n");
+        let events = direct_anthropic_event_stream(futures_util::stream::iter(
+            wire.as_bytes().chunks(3).map(Ok::<_, io::Error>),
+        ))
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap();
+
+        assert_eq!(
+            events,
+            vec![AnthropicMessagesStreamEvent::ContentBlockDelta {
+                index: 0,
+                delta: AnthropicContentBlockDelta::TextDelta {
+                    text: "hello".into(),
+                },
+            }]
+        );
+    }
+
+    #[test]
+    fn decodes_citations_delta_events() {
+        let event = decode_anthropic_sse_frame(SseFrame {
+            event: Some("content_block_delta".into()),
+            data: Some(
+                r#"{"type":"content_block_delta","index":0,"delta":{"type":"citations_delta","citation":{"type":"char_location"}}}"#
+                    .into(),
+            ),
+            id: None,
+            retry: None,
+        })
+        .unwrap();
+
+        assert!(matches!(
+            event,
+            AnthropicMessagesStreamEvent::ContentBlockDelta {
+                delta: AnthropicContentBlockDelta::Citations { .. },
+                ..
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn bedrock_aws_frames_into_the_same_typed_events() {
+        let payload = serde_json::json!({"bytes": STANDARD.encode(TEXT_DELTA)});
+        let message = Message::new(Bytes::from(serde_json::to_vec(&payload).unwrap())).add_header(
+            Header::new(":event-type", HeaderValue::String("chunk".into())),
+        );
+        let mut wire = Vec::new();
+        write_message_to(&message, &mut wire).unwrap();
+
+        let events = bedrock_anthropic_event_stream(futures_util::stream::iter(
+            wire.chunks(3).map(Ok::<_, io::Error>),
+        ))
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap();
+
+        assert_eq!(
+            events,
+            vec![AnthropicMessagesStreamEvent::ContentBlockDelta {
+                index: 0,
+                delta: AnthropicContentBlockDelta::TextDelta {
+                    text: "hello".into(),
+                },
+            }]
+        );
+    }
+}

litellm-rust/crates/core/src/providers/anthropic/messages/transformation.rs

@@ -31,10 +31,7 @@ pub fn complete_anthropic_url(
     api_base: Option<&str>,
     env_lookup: &dyn Fn(&str) -> Option<String>,
 ) -> String {
-    let api_base = non_empty(api_base)
-        .map(str::to_string)
-        .or_else(|| env_lookup(ANTHROPIC_API_BASE_ENV).filter(|value| !value.trim().is_empty()))
-        .unwrap_or_else(|| DEFAULT_ANTHROPIC_API_BASE.to_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_SUFFIX) {
@@ -43,6 +40,16 @@ pub fn complete_anthropic_url(
     format!("{api_base}{MESSAGES_PATH_SUFFIX}")
 }
 
+pub fn resolve_anthropic_api_base(
+    api_base: Option<&str>,
+    env_lookup: &dyn Fn(&str) -> Option<String>,
+) -> String {
+    non_empty(api_base)
+        .map(str::to_string)
+        .or_else(|| env_lookup(ANTHROPIC_API_BASE_ENV).filter(|value| !value.trim().is_empty()))
+        .unwrap_or_else(|| DEFAULT_ANTHROPIC_API_BASE.to_string())
+}
+
 impl AnthropicMessagesProviderConfig for AnthropicMessagesConfig {
     fn complete_url(
         &self,

litellm-rust/crates/python-bridge/src/errors.rs

@@ -46,10 +46,7 @@ pub(crate) fn core_error_to_pyerr(error: Error) -> PyErr {
         ),
         Error::Messages(error) => match error {
             messages::Error::Auth(source) => auth_is_value_error(source),
-            messages::Error::InvalidProvider(_)
-            | messages::Error::InvalidRequest(_)
-            | messages::Error::Headers(_) => true,
-            _ => false,
+            _ => error.is_request(),
         },
         Error::AudioTranscription(error) => match error {
             audio_transcription::Error::Auth(source) => auth_is_value_error(source),