feat(rust): map anthropic messages transforms

litellm-rust/Cargo.lock

@@ -1978,6 +1978,7 @@ dependencies = [
  "strum",
  "subtle",
  "thiserror 2.0.19",
+ "time",
  "tokio",
  "tokio-tungstenite",
  "url",

litellm-rust/Cargo.toml

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

@@ -30,6 +30,7 @@ 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

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

@@ -0,0 +1,344 @@
+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::InvalidRequest(
+            "Batch creation not yet implemented for Anthropic".into(),
+        ))
+    }
+
+    fn transform_create_batch_response(
+        &self,
+        _response: AnthropicMessageBatch,
+        _now: i64,
+    ) -> Result<LiteLlmMessageBatch, Error> {
+        Err(Error::InvalidResponse(
+            "Batch creation not yet implemented for Anthropic".into(),
+        ))
+    }
+
+    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::InvalidRequest("batch_id is required".into()));
+        }
+        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::InvalidRequest(message))
+                if message == "Batch creation not yet implemented for Anthropic"
+        ));
+        let response: AnthropicMessageBatch = serde_json::from_value(json!({})).unwrap();
+        assert!(matches!(
+            ANTHROPIC_BATCHES_TRANSFORMATION.transform_create_batch_response(response, 0),
+            Err(Error::InvalidResponse(message))
+                if message == "Batch creation not yet implemented for Anthropic"
+        ));
+    }
+}

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

@@ -0,0 +1,170 @@
+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::InvalidRequest("model parameter is required".into()));
+        }
+        if messages.is_empty() {
+            return Err(Error::InvalidRequest(
+                "messages parameter is required".into(),
+            ));
+        }
+        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::InvalidRequest(message)) if message == "model parameter is required"
+        ));
+        assert!(matches!(
+            ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.transform_request(
+                "claude-test",
+                vec![],
+                None,
+                None
+            ),
+            Err(Error::InvalidRequest(message)) if message == "messages parameter is required"
+        ));
+    }
+
+    #[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,2 +1,4 @@
+pub mod batches;
+pub mod count_tokens;
 pub mod streaming;
 pub mod transformation;

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,