Add anthropic retrieve batches and retreive file content support

litellm/batches/batch_utils.py

@@ -9,12 +9,12 @@ from litellm._logging import verbose_logger
 from litellm._uuid import uuid
 from litellm.types.llms.openai import Batch
 from litellm.types.utils import CallTypes, ModelResponse, Usage
-from litellm.utils import token_counter
+from litellm.utils import token_counter, ProviderConfigManager
 
 
 async def calculate_batch_cost_and_usage(
     file_content_dictionary: List[dict],
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"],
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"],
     model_name: Optional[str] = None,
 ) -> Tuple[float, Usage, List[str]]:
     """
@@ -30,14 +30,16 @@ async def calculate_batch_cost_and_usage(
         custom_llm_provider=custom_llm_provider,
         model_name=model_name,
     )
-    batch_models = _get_batch_models_from_file_content(file_content_dictionary, model_name)
+    batch_models = _get_batch_models_from_file_content(
+        file_content_dictionary, model_name, custom_llm_provider=custom_llm_provider
+    )
 
     return batch_cost, batch_usage, batch_models
 
 
 async def _handle_completed_batch(
     batch: Batch,
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"],
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"],
     model_name: Optional[str] = None,
 ) -> Tuple[float, Usage, List[str]]:
     """Helper function to process a completed batch and handle logging"""
@@ -58,14 +60,136 @@ async def _handle_completed_batch(
         model_name=model_name,
     )
 
-    batch_models = _get_batch_models_from_file_content(file_content_dictionary, model_name)
+    batch_models = _get_batch_models_from_file_content(
+        file_content_dictionary, model_name, custom_llm_provider=custom_llm_provider
+    )
 
     return batch_cost, batch_usage, batch_models
 
 
+def transform_raw_provider_response_to_openai(
+    raw_response: dict,
+    custom_llm_provider: str,
+    model: Optional[str] = None,
+    messages: Optional[list] = None,
+) -> ModelResponse:
+    """
+    Unified method to transform any raw LLM provider response to OpenAI format.
+    
+    Args:
+        raw_response: Raw response dictionary from any provider (Anthropic, OpenAI, etc.)
+        custom_llm_provider: Provider name (e.g., "anthropic", "openai", "vertex_ai")
+        model: Model name (optional, will try to extract from raw_response if not provided)
+        messages: Original messages list (optional, defaults to empty list)
+    
+    Returns:
+        ModelResponse: OpenAI-compatible response object
+    """
+    # Lazy import to avoid circular dependency
+    from litellm.litellm_core_utils.litellm_logging import Logging
+    
+    # Extract model from response if not provided
+    if model is None:
+        model = raw_response.get("model", "unknown-model")
+    
+    # Default messages if not provided
+    if messages is None:
+        messages = []
+    
+    # Get provider config using ProviderConfigManager
+    provider_config = ProviderConfigManager.get_provider_chat_config(
+        model=model,
+        provider=litellm.LlmProviders(custom_llm_provider)
+    )
+    
+    if provider_config is None:
+        raise ValueError(f"Could not get config for provider: {custom_llm_provider}")
+    
+    # Create a mock httpx.Response from the dict
+    response_text = json.dumps(raw_response)
+    mock_httpx_response = httpx.Response(
+        status_code=200,
+        content=response_text.encode('utf-8'),
+        headers={"content-type": "application/json"}
+    )
+    
+    # Create a minimal logging object
+    logging_obj = Logging(
+        model=model,
+        messages=messages,
+        stream=False,
+        call_type=CallTypes.completion.value,
+        start_time=time.time(),
+        litellm_call_id=None,
+        function_id=None,
+    )
+    
+    # Create empty ModelResponse to be populated
+    model_response = ModelResponse()
+    
+    # Call transform_response on the provider config
+    transformed_response = provider_config.transform_response(
+        model=model,
+        raw_response=mock_httpx_response,
+        model_response=model_response,
+        logging_obj=logging_obj,
+        request_data={},
+        messages=messages,
+        optional_params={},
+        litellm_params={},
+        encoding=litellm.encoding,
+        api_key=None,
+        json_mode=None,
+    )
+    
+    return transformed_response
+
+
+def _extract_raw_response_from_batch_item(
+    batch_item: dict,
+    custom_llm_provider: str,
+) -> Optional[dict]:
+    """
+    Extract the raw provider response from a batch output file item.
+    
+    Handles different batch output formats:
+    - Anthropic: {"result": {"type": "succeeded", "message": {...}}}
+    - Vertex AI: {"status": "JOB_STATE_SUCCEEDED", "response": {...}}
+    - OpenAI/Azure: {"response": {"status_code": 200, "body": {...}}}
+    
+    Args:
+        batch_item: A single item from the batch output file
+        custom_llm_provider: Provider name (e.g., "anthropic", "openai", "vertex_ai")
+    
+    Returns:
+        Raw response dict or None if not successful
+    """
+    # Anthropic format: {"result": {"type": "succeeded", "message": {...}}}
+    if custom_llm_provider == "anthropic":
+        result = batch_item.get("result", {})
+        if result.get("type") == "succeeded":
+            return result.get("message", {})
+        return None
+    
+    # Vertex AI format: {"status": "JOB_STATE_SUCCEEDED", "response": {...}}
+    if custom_llm_provider == "vertex_ai":
+        if batch_item.get("status") == "JOB_STATE_SUCCEEDED":
+            return batch_item.get("response", {})
+        return None
+    
+    # OpenAI/Azure format: {"response": {"status_code": 200, "body": {...}}}
+    # Default to OpenAI format for openai, azure, hosted_vllm, etc.
+    response = batch_item.get("response", {})
+    if response.get("status_code") == 200:
+        return response.get("body", {})
+    
+    return None
+
+
 def _get_batch_models_from_file_content(
     file_content_dictionary: List[dict],
     model_name: Optional[str] = None,
+    custom_llm_provider: str = "openai",
 ) -> List[str]:
     """
     Get the models from the file content
@@ -74,119 +198,79 @@ def _get_batch_models_from_file_content(
         return [model_name]
     batch_models = []
     for _item in file_content_dictionary:
-        if _batch_response_was_successful(_item):
-            _response_body = _get_response_from_batch_job_output_file(_item)
-            _model = _response_body.get("model")
-            if _model:
-                batch_models.append(_model)
+        if _batch_response_was_successful(_item, custom_llm_provider=custom_llm_provider):
+            # Extract raw response using generalized method
+            raw_response = _extract_raw_response_from_batch_item(
+                batch_item=_item,
+                custom_llm_provider=custom_llm_provider,
+            )
+            if raw_response:
+                _model = raw_response.get("model")
+                if _model:
+                    batch_models.append(_model)
     return batch_models
 
 
 def _batch_cost_calculator(
     file_content_dictionary: List[dict],
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"] = "openai",
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"] = "openai",
     model_name: Optional[str] = None,
 ) -> float:
     """
     Calculate the cost of a batch based on the output file id
     """
-    # Handle Vertex AI with specialized method
-    if custom_llm_provider == "vertex_ai" and model_name:
-        batch_cost, _ = calculate_vertex_ai_batch_cost_and_usage(file_content_dictionary, model_name)
-        verbose_logger.debug("vertex_ai_total_cost=%s", batch_cost)
-        return batch_cost
+    total_cost: float = 0.0
+    
+    for batch_item in file_content_dictionary:
+        if not _batch_response_was_successful(batch_item, custom_llm_provider=custom_llm_provider):
+            continue
+        
+        # Extract raw response from batch item
+        raw_response = _extract_raw_response_from_batch_item(
+            batch_item=batch_item,
+            custom_llm_provider=custom_llm_provider,
+        )
+        
+        if raw_response is None:
+            continue
+        
+        # Extract model from response if not provided
+        actual_model = model_name or raw_response.get("model")
+        if actual_model is None:
+            verbose_logger.warning("Could not determine model for batch item, skipping cost calculation")
+            continue
+        
+        try:
+            # Transform to OpenAI format using generalized method
+            openai_format_response = transform_raw_provider_response_to_openai(
+                raw_response=raw_response,
+                custom_llm_provider=custom_llm_provider,
+                model=actual_model,
+                messages=[],  # Messages not needed for cost calculation
+            )
+            
+            # Calculate cost using standard OpenAI cost calculation
+            cost = litellm.completion_cost(
+                completion_response=openai_format_response,
+                custom_llm_provider=custom_llm_provider,
+                call_type=CallTypes.aretrieve_batch.value,
+            )
+            total_cost += cost
+            verbose_logger.debug("item_cost=%s, total_cost=%s", cost, total_cost)
+        except Exception as e:
+            verbose_logger.warning(
+                f"Error calculating cost for batch item: {e}. Skipping this item."
+            )
+            continue
     
-    # For other providers, use the existing logic
-    total_cost = _get_batch_job_cost_from_file_content(
-        file_content_dictionary=file_content_dictionary,
-        custom_llm_provider=custom_llm_provider,
-    )
     verbose_logger.debug("total_cost=%s", total_cost)
     return total_cost
 
 
-def calculate_vertex_ai_batch_cost_and_usage(
-    vertex_ai_batch_responses: List[dict],
-    model_name: Optional[str] = None,
-) -> Tuple[float, Usage]:
-    """
-    Calculate both cost and usage from Vertex AI batch responses
-    """
-    from litellm.litellm_core_utils.litellm_logging import Logging
-    from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
-        VertexGeminiConfig,
-    )
-    total_cost = 0.0
-    total_tokens = 0
-    prompt_tokens = 0
-    completion_tokens = 0
-    
-    for response in vertex_ai_batch_responses:
-        if response.get("status") == "JOB_STATE_SUCCEEDED":  # Check if response was successful
-            # Transform Vertex AI response to OpenAI format if needed
-
-            # Create required arguments for the transformation method
-            model_response = ModelResponse()
-            
-            # Ensure model_name is not None
-            actual_model_name = model_name or "gemini-2.5-flash"
-            
-            # Create a real LiteLLM logging object
-            logging_obj = Logging(
-                model=actual_model_name,
-                messages=[{"role": "user", "content": "batch_request"}],
-                stream=False,
-                call_type=CallTypes.aretrieve_batch,
-                start_time=time.time(),
-                litellm_call_id="batch_" + str(uuid.uuid4()),
-                function_id="batch_processing",
-                litellm_trace_id=str(uuid.uuid4()),
-                kwargs={"optional_params": {}}
-            )
-            
-            # Add the optional_params attribute that the Vertex AI transformation expects
-            logging_obj.optional_params = {}
-            raw_response = httpx.Response(200)  # Mock response object
-            
-            openai_format_response = VertexGeminiConfig()._transform_google_generate_content_to_openai_model_response(
-                completion_response=response["response"],
-                model_response=model_response,
-                model=actual_model_name,
-                logging_obj=logging_obj,
-                raw_response=raw_response,
-            )
-            
-            # Calculate cost using existing function
-            cost = litellm.completion_cost(
-                completion_response=openai_format_response,
-                custom_llm_provider="vertex_ai",
-                call_type=CallTypes.aretrieve_batch.value,
-            )
-            total_cost += cost
-            
-            # Extract usage from the transformed response
-            usage_obj = getattr(openai_format_response, 'usage', None)
-            if usage_obj:
-                usage = usage_obj
-            else:
-                # Fallback: create usage from response dict
-                response_dict = openai_format_response.dict() if hasattr(openai_format_response, 'dict') else {}
-                usage = _get_batch_job_usage_from_response_body(response_dict)
-            
-            total_tokens += usage.total_tokens
-            prompt_tokens += usage.prompt_tokens
-            completion_tokens += usage.completion_tokens
-    
-    return total_cost, Usage(
-        total_tokens=total_tokens,
-        prompt_tokens=prompt_tokens,
-        completion_tokens=completion_tokens,
-    )
-
 
 async def _get_batch_output_file_content_as_dictionary(
     batch: Batch,
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"] = "openai",
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"] = "openai",
 ) -> List[dict]:
     """
     Get the batch output file content as a list of dictionaries
@@ -223,62 +307,72 @@ def _get_file_content_as_dictionary(file_content: bytes) -> List[dict]:
         raise e
 
 
-def _get_batch_job_cost_from_file_content(
-    file_content_dictionary: List[dict],
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"] = "openai",
-) -> float:
-    """
-    Get the cost of a batch job from the file content
-    """
-    try:
-        total_cost: float = 0.0
-        # parse the file content as json
-        verbose_logger.debug(
-            "file_content_dictionary=%s", json.dumps(file_content_dictionary, indent=4)
-        )
-        for _item in file_content_dictionary:
-            if _batch_response_was_successful(_item):
-                _response_body = _get_response_from_batch_job_output_file(_item)
-                total_cost += litellm.completion_cost(
-                    completion_response=_response_body,
-                    custom_llm_provider=custom_llm_provider,
-                    call_type=CallTypes.aretrieve_batch.value,
-                )
-                verbose_logger.debug("total_cost=%s", total_cost)
-        return total_cost
-    except Exception as e:
-        verbose_logger.error("error in _get_batch_job_cost_from_file_content", e)
-        raise e
-
-
 def _get_batch_job_total_usage_from_file_content(
     file_content_dictionary: List[dict],
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"] = "openai",
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"] = "openai",
     model_name: Optional[str] = None,
 ) -> Usage:
     """
     Get the tokens of a batch job from the file content
     """
-    # Handle Vertex AI with specialized method
-    if custom_llm_provider == "vertex_ai" and model_name:
-        _, batch_usage = calculate_vertex_ai_batch_cost_and_usage(file_content_dictionary, model_name)
-        return batch_usage
+    from litellm.cost_calculator import BaseTokenUsageProcessor
     
-    # For other providers, use the existing logic
-    total_tokens: int = 0
-    prompt_tokens: int = 0
-    completion_tokens: int = 0
-    for _item in file_content_dictionary:
-        if _batch_response_was_successful(_item):
-            _response_body = _get_response_from_batch_job_output_file(_item)
-            usage: Usage = _get_batch_job_usage_from_response_body(_response_body)
-            total_tokens += usage.total_tokens
-            prompt_tokens += usage.prompt_tokens
-            completion_tokens += usage.completion_tokens
+    all_usage: List[Usage] = []
+    
+    for batch_item in file_content_dictionary:
+        if not _batch_response_was_successful(batch_item, custom_llm_provider=custom_llm_provider):
+            continue
+        
+        # Extract raw response from batch item
+        raw_response = _extract_raw_response_from_batch_item(
+            batch_item=batch_item,
+            custom_llm_provider=custom_llm_provider,
+        )
+        
+        if raw_response is None:
+            continue
+        
+        # Extract model from response if not provided
+        actual_model = model_name or raw_response.get("model")
+        if actual_model is None:
+            verbose_logger.warning("Could not determine model for batch item, skipping usage calculation")
+            continue
+        
+        try:
+            # Transform to OpenAI format using generalized method
+            openai_format_response = transform_raw_provider_response_to_openai(
+                raw_response=raw_response,
+                custom_llm_provider=custom_llm_provider,
+                model=actual_model,
+                messages=[],  # Messages not needed for usage extraction
+            )
+            
+            # Extract usage from transformed response
+            usage_obj = getattr(openai_format_response, 'usage', None)
+            if usage_obj:
+                all_usage.append(usage_obj)
+            else:
+                # Fallback: try to extract from response dict
+                response_dict = openai_format_response.dict() if hasattr(openai_format_response, 'dict') else {}
+                usage = _get_batch_job_usage_from_response_body(response_dict)
+                if usage and usage.total_tokens > 0:
+                    all_usage.append(usage)
+        except Exception as e:
+            verbose_logger.warning(
+                f"Error extracting usage for batch item: {e}. Skipping this item."
+            )
+            continue
+    
+    # Combine all usage objects
+    if all_usage:
+        combined_usage = BaseTokenUsageProcessor.combine_usage_objects(all_usage)
+        return combined_usage
+    
+    # Return empty usage if no valid responses
     return Usage(
-        total_tokens=total_tokens,
-        prompt_tokens=prompt_tokens,
-        completion_tokens=completion_tokens,
+        total_tokens=0,
+        prompt_tokens=0,
+        completion_tokens=0,
     )
 
 def _get_batch_job_input_file_usage(
@@ -318,18 +412,30 @@ def _get_batch_job_usage_from_response_body(response_body: dict) -> Usage:
     return usage
 
 
-def _get_response_from_batch_job_output_file(batch_job_output_file: dict) -> Any:
+def _batch_response_was_successful(
+    batch_job_output_file: dict,
+    custom_llm_provider: str = "openai",
+) -> bool:
     """
-    Get the response from the batch job output file
-    """
-    _response: dict = batch_job_output_file.get("response", None) or {}
-    _response_body = _response.get("body", None) or {}
-    return _response_body
-
-
-def _batch_response_was_successful(batch_job_output_file: dict) -> bool:
-    """
-    Check if the batch job response status == 200
+    Check if the batch job response was successful.
+    
+    Args:
+        batch_job_output_file: A single item from the batch output file
+        custom_llm_provider: Provider name (e.g., "anthropic", "openai", "vertex_ai")
+    
+    Returns:
+        True if the batch response was successful, False otherwise
     """
+    # Anthropic format: {"result": {"type": "succeeded", "message": {...}}}
+    if custom_llm_provider == "anthropic":
+        result = batch_job_output_file.get("result", {})
+        return result.get("type") == "succeeded"
+    
+    # Vertex AI format: {"status": "JOB_STATE_SUCCEEDED", "response": {...}}
+    if custom_llm_provider == "vertex_ai":
+        return batch_job_output_file.get("status") == "JOB_STATE_SUCCEEDED"
+    
+    # OpenAI/Azure format: {"response": {"status_code": 200, "body": {...}}}
+    # Default to OpenAI format for openai, azure, hosted_vllm, etc.
     _response: dict = batch_job_output_file.get("response", None) or {}
     return _response.get("status_code", None) == 200

litellm/batches/main.py

@@ -22,6 +22,7 @@ from openai.types.batch import BatchRequestCounts
 import litellm
 from litellm._logging import verbose_logger
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.anthropic.batches.handler import AnthropicBatchesHandler
 from litellm.llms.azure.batches.handler import AzureBatchesAPI
 from litellm.llms.bedrock.batches.handler import BedrockBatchesHandler
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
@@ -53,6 +54,7 @@ from litellm.utils import (
 openai_batches_instance = OpenAIBatchesAPI()
 azure_batches_instance = AzureBatchesAPI()
 vertex_ai_batches_instance = VertexAIBatchPrediction(gcs_bucket_name="")
+anthropic_batches_instance = AnthropicBatchesHandler()
 base_llm_http_handler = BaseLLMHTTPHandler()
 #################################################
 
@@ -355,7 +357,7 @@ def create_batch(
 @client
 async def aretrieve_batch(
     batch_id: str,
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"] = "openai",
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"] = "openai",
     metadata: Optional[Dict[str, str]] = None,
     extra_headers: Optional[Dict[str, str]] = None,
     extra_body: Optional[Dict[str, str]] = None,
@@ -401,7 +403,7 @@ def _handle_retrieve_batch_providers_without_provider_config(
     litellm_params: dict,
     _retrieve_batch_request: RetrieveBatchRequest,
     _is_async: bool,
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"] = "openai",
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"] = "openai",
 ):
     api_base: Optional[str] = None
     if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS:
@@ -498,6 +500,27 @@ def _handle_retrieve_batch_providers_without_provider_config(
             timeout=timeout,
             max_retries=optional_params.max_retries,
         )
+    elif custom_llm_provider == "anthropic":
+        api_base = (
+            optional_params.api_base
+            or litellm.api_base
+            or get_secret_str("ANTHROPIC_API_BASE")
+        )
+        api_key = (
+            optional_params.api_key
+            or litellm.api_key
+            or litellm.azure_key
+            or get_secret_str("ANTHROPIC_API_KEY")
+        )
+
+        response = anthropic_batches_instance.retrieve_batch(
+            _is_async=_is_async,
+            batch_id=batch_id,
+            api_base=api_base,
+            api_key=api_key,
+            timeout=timeout,
+            max_retries=optional_params.max_retries,
+        )
     else:
         raise litellm.exceptions.BadRequestError(
             message="LiteLLM doesn't support {} for 'create_batch'. Only 'openai' is supported.".format(
@@ -517,7 +540,7 @@ def _handle_retrieve_batch_providers_without_provider_config(
 @client
 def retrieve_batch(
     batch_id: str,
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"] = "openai",
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"] = "openai",
     metadata: Optional[Dict[str, str]] = None,
     extra_headers: Optional[Dict[str, str]] = None,
     extra_body: Optional[Dict[str, str]] = None,
@@ -608,7 +631,7 @@ def retrieve_batch(
                 api_key=optional_params.api_key,
                 logging_obj=litellm_logging_obj
                 or LiteLLMLoggingObj(
-                    model=model or "bedrock/unknown",
+                    model=model or f"{custom_llm_provider}/unknown",
                     messages=[],
                     stream=False,
                     call_type="batch_retrieve",

litellm/files/main.py

@@ -17,6 +17,7 @@ import litellm
 from litellm import get_secret_str
 from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.anthropic.files.handler import AnthropicFilesHandler
 from litellm.llms.azure.files.handler import AzureOpenAIFilesAPI
 from litellm.llms.bedrock.files.handler import BedrockFilesHandler
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
@@ -49,6 +50,7 @@ openai_files_instance = OpenAIFilesAPI()
 azure_files_instance = AzureOpenAIFilesAPI()
 vertex_ai_files_instance = VertexAIFilesHandler()
 bedrock_files_instance = BedrockFilesHandler()
+anthropic_files_instance = AnthropicFilesHandler()
 #################################################
 
 
@@ -757,7 +759,7 @@ def file_list(
 @client
 async def afile_content(
     file_id: str,
-    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"] = "openai",
+    custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"] = "openai",
     extra_headers: Optional[Dict[str, str]] = None,
     extra_body: Optional[Dict[str, str]] = None,
     **kwargs,
@@ -802,7 +804,7 @@ def file_content(
     file_id: str,
     model: Optional[str] = None,
     custom_llm_provider: Optional[
-        Union[Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"], str]
+        Union[Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"], str]
     ] = None,
     extra_headers: Optional[Dict[str, str]] = None,
     extra_body: Optional[Dict[str, str]] = None,
@@ -849,6 +851,18 @@ def file_content(
 
         _is_async = kwargs.pop("afile_content", False) is True
 
+        # Check if this is an Anthropic batch results request
+        if custom_llm_provider == "anthropic":
+            response = anthropic_files_instance.file_content(
+                _is_async=_is_async,
+                file_content_request=_file_content_request,
+                api_base=optional_params.api_base,
+                api_key=optional_params.api_key,
+                timeout=timeout,
+                max_retries=optional_params.max_retries,
+            )
+            return response
+
         if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS:
             # for deepinfra/perplexity/anyscale/groq we check in get_llm_provider and pass in the api base from there
             api_base = (

litellm/llms/anthropic/batches/__init__.py

@@ -0,0 +1,5 @@
+from .handler import AnthropicBatchesHandler
+from .transformation import AnthropicBatchesConfig
+
+__all__ = ["AnthropicBatchesHandler", "AnthropicBatchesConfig"]
+

litellm/llms/anthropic/batches/handler.py

@@ -0,0 +1,169 @@
+"""
+Anthropic Batches API Handler
+"""
+
+import asyncio
+from typing import TYPE_CHECKING, Any, Coroutine, Optional, Union
+
+import httpx
+
+from litellm.llms.custom_httpx.http_handler import (
+    get_async_httpx_client,
+)
+from litellm.types.llms.openai import RetrieveBatchRequest
+from litellm.types.utils import LiteLLMBatch
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+else:
+    LiteLLMLoggingObj = Any
+
+from ..common_utils import AnthropicModelInfo
+from .transformation import AnthropicBatchesConfig
+
+
+class AnthropicBatchesHandler:
+    """
+    Handler for Anthropic Message Batches API.
+    
+    Supports:
+    - retrieve_batch() - Retrieve batch status and information
+    """
+
+    def __init__(self):
+        self.anthropic_model_info = AnthropicModelInfo()
+        self.provider_config = AnthropicBatchesConfig()
+
+    async def aretrieve_batch(
+        self,
+        batch_id: str,
+        api_base: Optional[str],
+        api_key: Optional[str],
+        timeout: Union[float, httpx.Timeout],
+        max_retries: Optional[int],
+        logging_obj: Optional[LiteLLMLoggingObj] = None,
+    ) -> LiteLLMBatch:
+        """
+        Async: Retrieve a batch from Anthropic.
+        
+        Args:
+            batch_id: The batch ID to retrieve
+            api_base: Anthropic API base URL
+            api_key: Anthropic API key
+            timeout: Request timeout
+            max_retries: Max retry attempts (unused for now)
+            logging_obj: Optional logging object
+            
+        Returns:
+            LiteLLMBatch: Batch information in OpenAI format
+        """
+        # Resolve API credentials
+        api_base = api_base or self.anthropic_model_info.get_api_base(api_base)
+        api_key = api_key or self.anthropic_model_info.get_api_key()
+        
+        if not api_key:
+            raise ValueError("Missing Anthropic API Key")
+        
+        # Create a minimal logging object if not provided
+        if logging_obj is None:
+            from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObjClass
+            logging_obj = LiteLLMLoggingObjClass(
+                model="anthropic/unknown",
+                messages=[],
+                stream=False,
+                call_type="batch_retrieve",
+                start_time=None,
+                litellm_call_id=f"batch_retrieve_{batch_id}",
+                function_id="batch_retrieve",
+            )
+        
+        # Get the complete URL for batch retrieval
+        retrieve_url = self.provider_config.get_retrieve_batch_url(
+            api_base=api_base,
+            batch_id=batch_id,
+            optional_params={},
+            litellm_params={},
+        )
+        
+        # Validate environment and get headers
+        headers = self.provider_config.validate_environment(
+            headers={},
+            model="",
+            messages=[],
+            optional_params={},
+            litellm_params={},
+            api_key=api_key,
+            api_base=api_base,
+        )
+
+        logging_obj.pre_call(
+            input=batch_id,
+            api_key=api_key,
+            additional_args={
+                "api_base": retrieve_url,
+                "headers": headers,
+                "complete_input_dict": {},
+            },
+        )
+        # Make the request
+        async_client = get_async_httpx_client(llm_provider="anthropic")
+        response = await async_client.get(
+            url=retrieve_url,
+            headers=headers
+        )
+        response.raise_for_status()
+        
+        # Transform response to LiteLLM format
+        return self.provider_config.transform_retrieve_batch_response(
+            model=None,
+            raw_response=response,
+            logging_obj=logging_obj,
+            litellm_params={},
+        )
+
+    def retrieve_batch(
+        self,
+        _is_async: bool,
+        batch_id: str,
+        api_base: Optional[str],
+        api_key: Optional[str],
+        timeout: Union[float, httpx.Timeout],
+        max_retries: Optional[int],
+        logging_obj: Optional[LiteLLMLoggingObj] = None,
+    ) -> Union[LiteLLMBatch, Coroutine[Any, Any, LiteLLMBatch]]:
+        """
+        Retrieve a batch from Anthropic.
+        
+        Args:
+            _is_async: Whether to run asynchronously
+            batch_id: The batch ID to retrieve
+            api_base: Anthropic API base URL
+            api_key: Anthropic API key
+            timeout: Request timeout
+            max_retries: Max retry attempts (unused for now)
+            logging_obj: Optional logging object
+            
+        Returns:
+            LiteLLMBatch or Coroutine: Batch information in OpenAI format
+        """
+        if _is_async:
+            return self.aretrieve_batch(
+                batch_id=batch_id,
+                api_base=api_base,
+                api_key=api_key,
+                timeout=timeout,
+                max_retries=max_retries,
+                logging_obj=logging_obj,
+            )
+        else:
+            return asyncio.run(
+                self.aretrieve_batch(
+                    batch_id=batch_id,
+                    api_base=api_base,
+                    api_key=api_key,
+                    timeout=timeout,
+                    max_retries=max_retries,
+                    logging_obj=logging_obj,
+                )
+            )
+

litellm/llms/anthropic/batches/transformation.py

@@ -1,10 +1,14 @@
 import json
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
+import time
+from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union, cast
 
-from httpx import Response
+import httpx
+from httpx import Headers, Response
 
-from litellm.types.llms.openai import AllMessageValues
-from litellm.types.utils import ModelResponse
+from litellm.llms.base_llm.batches.transformation import BaseBatchesConfig
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
+from litellm.types.llms.openai import AllMessageValues, CreateBatchRequest
+from litellm.types.utils import LiteLLMBatch, LlmProviders, ModelResponse
 
 if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@@ -14,11 +18,225 @@ else:
     LoggingClass = Any
 
 
-class AnthropicBatchesConfig:
+class AnthropicBatchesConfig(BaseBatchesConfig):
     def __init__(self):
         from ..chat.transformation import AnthropicConfig
+        from ..common_utils import AnthropicError, AnthropicModelInfo
 
         self.anthropic_chat_config = AnthropicConfig()  # initialize once
+        self.anthropic_model_info = AnthropicModelInfo()
+
+    @property
+    def custom_llm_provider(self) -> LlmProviders:
+        """Return the LLM provider type for this configuration."""
+        return LlmProviders.ANTHROPIC
+
+    def validate_environment(
+        self,
+        headers: dict,
+        model: str,
+        messages: List[AllMessageValues],
+        optional_params: dict,
+        litellm_params: dict,
+        api_key: Optional[str] = None,
+        api_base: Optional[str] = None,
+    ) -> dict:
+        """Validate and prepare environment-specific headers and parameters."""
+        # Resolve api_key from environment if not provided
+        api_key = api_key or self.anthropic_model_info.get_api_key()
+        if api_key is None:
+            raise ValueError(
+                "Missing Anthropic API Key - A call is being made to anthropic but no key is set either in the environment variables or via params"
+            )
+        _headers = {
+            "accept": "application/json",
+            "anthropic-version": "2023-06-01",
+            "content-type": "application/json",
+            "x-api-key": api_key,
+        }
+        # Add beta header for message batches
+        if "anthropic-beta" not in headers:
+            headers["anthropic-beta"] = "message-batches-2024-09-24"
+        headers.update(_headers)
+        return headers
+
+    def get_complete_batch_url(
+        self,
+        api_base: Optional[str],
+        api_key: Optional[str],
+        model: str,
+        optional_params: Dict,
+        litellm_params: Dict,
+        data: CreateBatchRequest,
+    ) -> str:
+        """Get the complete URL for batch creation request."""
+        api_base = api_base or self.anthropic_model_info.get_api_base(api_base)
+        if not api_base.endswith("/v1/messages/batches"):
+            api_base = f"{api_base.rstrip('/')}/v1/messages/batches"
+        return api_base
+
+    def transform_create_batch_request(
+        self,
+        model: str,
+        create_batch_data: CreateBatchRequest,
+        optional_params: dict,
+        litellm_params: dict,
+    ) -> Union[bytes, str, Dict[str, Any]]:
+        """
+        Transform the batch creation request to Anthropic format.
+        
+        Not currently implemented - placeholder to satisfy abstract base class.
+        """
+        raise NotImplementedError("Batch creation not yet implemented for Anthropic")
+
+    def transform_create_batch_response(
+        self,
+        model: Optional[str],
+        raw_response: httpx.Response,
+        logging_obj: LoggingClass,
+        litellm_params: dict,
+    ) -> LiteLLMBatch:
+        """
+        Transform Anthropic MessageBatch creation response to LiteLLM format.
+        
+        Not currently implemented - placeholder to satisfy abstract base class.
+        """
+        raise NotImplementedError("Batch creation not yet implemented for Anthropic")
+
+    def get_retrieve_batch_url(
+        self,
+        api_base: Optional[str],
+        batch_id: str,
+        optional_params: Dict,
+        litellm_params: Dict,
+    ) -> str:
+        """
+        Get the complete URL for batch retrieval request.
+        
+        Args:
+            api_base: Base API URL (optional, will use default if not provided)
+            batch_id: Batch ID to retrieve
+            optional_params: Optional parameters
+            litellm_params: LiteLLM parameters
+            
+        Returns:
+            Complete URL for Anthropic batch retrieval: {api_base}/v1/messages/batches/{batch_id}
+        """
+        api_base = api_base or self.anthropic_model_info.get_api_base(api_base)
+        return f"{api_base.rstrip('/')}/v1/messages/batches/{batch_id}"
+
+    def transform_retrieve_batch_request(
+        self,
+        batch_id: str,
+        optional_params: dict,
+        litellm_params: dict,
+    ) -> Union[bytes, str, Dict[str, Any]]:
+        """
+        Transform batch retrieval request for Anthropic.
+        
+        For Anthropic, the URL is constructed by get_retrieve_batch_url(),
+        so this method returns an empty dict (no additional request params needed).
+        """
+        # No additional request params needed - URL is handled by get_retrieve_batch_url
+        return {}
+
+    def transform_retrieve_batch_response(
+        self,
+        model: Optional[str],
+        raw_response: httpx.Response,
+        logging_obj: LoggingClass,
+        litellm_params: dict,
+    ) -> LiteLLMBatch:
+        """Transform Anthropic MessageBatch retrieval response to LiteLLM format."""
+        try:
+            response_data = raw_response.json()
+        except Exception as e:
+            raise ValueError(f"Failed to parse Anthropic batch response: {e}")
+
+        # Map Anthropic MessageBatch to OpenAI Batch format
+        batch_id = response_data.get("id", "")
+        processing_status = response_data.get("processing_status", "in_progress")
+        
+        # Map Anthropic processing_status to OpenAI status
+        status_mapping: Dict[str, Literal["validating", "failed", "in_progress", "finalizing", "completed", "expired", "cancelling", "cancelled"]] = {
+            "in_progress": "in_progress",
+            "canceling": "cancelling",
+            "ended": "completed",
+        }
+        openai_status = status_mapping.get(processing_status, "in_progress")
+
+        # Parse timestamps
+        def parse_timestamp(ts_str: Optional[str]) -> Optional[int]:
+            if not ts_str:
+                return None
+            try:
+                from datetime import datetime
+                dt = datetime.fromisoformat(ts_str.replace('Z', '+00:00'))
+                return int(dt.timestamp())
+            except Exception:
+                return None
+
+        created_at = parse_timestamp(response_data.get("created_at"))
+        ended_at = parse_timestamp(response_data.get("ended_at"))
+        expires_at = parse_timestamp(response_data.get("expires_at"))
+        cancel_initiated_at = parse_timestamp(response_data.get("cancel_initiated_at"))
+        archived_at = parse_timestamp(response_data.get("archived_at"))
+
+        # Extract request counts
+        request_counts_data = response_data.get("request_counts", {})
+        from openai.types.batch import BatchRequestCounts
+        request_counts = BatchRequestCounts(
+            total=sum([
+                request_counts_data.get("processing", 0),
+                request_counts_data.get("succeeded", 0),
+                request_counts_data.get("errored", 0),
+                request_counts_data.get("canceled", 0),
+                request_counts_data.get("expired", 0),
+            ]),
+            completed=request_counts_data.get("succeeded", 0),
+            failed=request_counts_data.get("errored", 0),
+        )
+
+        # Extract results_url - this will be used for file content retrieval
+        results_url = response_data.get("results_url")
+        # Store results_url in output_file_id for later retrieval
+        # We'll encode it in a way that we can detect it's an Anthropic results URL
+        output_file_id = None
+        if results_url:
+            # Encode the batch_id and results_url so we can retrieve it later
+            # Format: anthropic_batch_results:{batch_id}
+            output_file_id = f"anthropic_batch_results:{batch_id}"
+
+        return LiteLLMBatch(
+            id=batch_id,
+            object="batch",
+            endpoint="/v1/messages",
+            errors=None,
+            input_file_id=None,
+            completion_window="24h",
+            status=openai_status,
+            output_file_id=output_file_id,
+            error_file_id=None,
+            created_at=created_at or int(time.time()),
+            in_progress_at=created_at if processing_status == "in_progress" else None,
+            expires_at=expires_at,
+            finalizing_at=None,
+            completed_at=ended_at if processing_status == "ended" else None,
+            failed_at=None,
+            expired_at=archived_at if archived_at else None,
+            cancelling_at=cancel_initiated_at if processing_status == "canceling" else None,
+            cancelled_at=ended_at if processing_status == "canceling" and ended_at else None,
+            request_counts=request_counts,
+            metadata={},
+        )
+
+    def get_error_class(
+        self, error_message: str, status_code: int, headers: Union[Dict, Headers]
+    ) -> "BaseLLMException":
+        """Get the appropriate error class for Anthropic."""
+        from ..common_utils import AnthropicError
+
+        return AnthropicError(status_code=status_code, message=error_message, headers=headers)
 
     def transform_response(
         self,

litellm/llms/anthropic/files/__init__.py

@@ -0,0 +1,4 @@
+from .handler import AnthropicFilesHandler
+
+__all__ = ["AnthropicFilesHandler"]
+

litellm/llms/anthropic/files/handler.py

@@ -0,0 +1,142 @@
+import asyncio
+from typing import Any, Coroutine, Optional, Union
+
+import httpx
+
+from litellm.llms.custom_httpx.http_handler import (
+    AsyncHTTPHandler,
+    HTTPHandler,
+    _get_httpx_client,
+    get_async_httpx_client,
+)
+from litellm.types.llms.openai import (
+    FileContentRequest,
+    HttpxBinaryResponseContent,
+)
+
+from ..common_utils import AnthropicModelInfo
+
+
+class AnthropicFilesHandler:
+    """
+    Handles Anthropic Files API operations.
+    
+    Currently supports:
+    - file_content() for retrieving Anthropic Message Batch results
+    """
+
+    def __init__(self):
+        self.anthropic_model_info = AnthropicModelInfo()
+
+    async def afile_content(
+        self,
+        file_content_request: FileContentRequest,
+        api_base: Optional[str] = None,
+        api_key: Optional[str] = None,
+        timeout: Union[float, httpx.Timeout] = 600.0,
+        max_retries: Optional[int] = None,
+    ) -> HttpxBinaryResponseContent:
+        """
+        Async: Retrieve file content from Anthropic.
+        
+        For batch results, the file_id should be the batch_id.
+        This will call Anthropic's /v1/messages/batches/{batch_id}/results endpoint.
+        
+        Args:
+            file_content_request: Contains file_id (batch_id for batch results)
+            api_base: Anthropic API base URL
+            api_key: Anthropic API key
+            timeout: Request timeout
+            max_retries: Max retry attempts (unused for now)
+            
+        Returns:
+            HttpxBinaryResponseContent: Binary content wrapped in compatible response format
+        """
+        file_id = file_content_request.get("file_id")
+        if not file_id:
+            raise ValueError("file_id is required in file_content_request")
+
+        # Extract batch_id from file_id
+        # Handle both formats: "anthropic_batch_results:{batch_id}" or just "{batch_id}"
+        if file_id.startswith("anthropic_batch_results:"):
+            batch_id = file_id.replace("anthropic_batch_results:", "", 1)
+        else:
+            batch_id = file_id
+
+        # Get Anthropic API credentials
+        api_base = self.anthropic_model_info.get_api_base(api_base)
+        api_key = api_key or self.anthropic_model_info.get_api_key()
+
+        if not api_key:
+            raise ValueError("Missing Anthropic API Key")
+
+        # Construct the Anthropic batch results URL
+        results_url = f"{api_base.rstrip('/')}/v1/messages/batches/{batch_id}/results"
+
+        # Prepare headers
+        headers = {
+            "accept": "application/json",
+            "anthropic-version": "2023-06-01",
+            "x-api-key": api_key,
+        }
+
+        # Make the request to Anthropic
+        async_client = get_async_httpx_client(llm_provider="anthropic")
+        try:
+            anthropic_response = await async_client.get(
+                url=results_url,
+                headers=headers
+            )
+            anthropic_response.raise_for_status()
+
+            # Return the response content
+            return HttpxBinaryResponseContent(response=anthropic_response)
+        finally:
+            await async_client.aclose()
+
+    def file_content(
+        self,
+        _is_async: bool,
+        file_content_request: FileContentRequest,
+        api_base: Optional[str] = None,
+        api_key: Optional[str] = None,
+        timeout: Union[float, httpx.Timeout] = 600.0,
+        max_retries: Optional[int] = None,
+    ) -> Union[
+        HttpxBinaryResponseContent, Coroutine[Any, Any, HttpxBinaryResponseContent]
+    ]:
+        """
+        Retrieve file content from Anthropic.
+        
+        For batch results, the file_id should be the batch_id.
+        This will call Anthropic's /v1/messages/batches/{batch_id}/results endpoint.
+        
+        Args:
+            _is_async: Whether to run asynchronously
+            file_content_request: Contains file_id (batch_id for batch results)
+            api_base: Anthropic API base URL
+            api_key: Anthropic API key
+            timeout: Request timeout
+            max_retries: Max retry attempts (unused for now)
+            
+        Returns:
+            HttpxBinaryResponseContent or Coroutine: Binary content wrapped in compatible response format
+        """
+        if _is_async:
+            return self.afile_content(
+                file_content_request=file_content_request,
+                api_base=api_base,
+                api_key=api_key,
+                max_retries=max_retries,
+            )
+        else:
+            return asyncio.run(
+                self.afile_content(
+                    file_content_request=file_content_request,
+                    api_base=api_base,
+                    api_key=api_key,
+                    timeout=timeout,
+                    max_retries=max_retries,
+                )
+            )
+