Add batch passthrough endpoint cost tracking for anthropic

docs/my-website/docs/pass_through/anthropic_completion.md

@@ -7,7 +7,7 @@ Pass-through endpoints for Anthropic - call provider-specific endpoint, in nativ
 
 | Feature | Supported | Notes | 
 |-------|-------|-------|
-| Cost Tracking | ✅ | supports all models on `/messages` endpoint |
+| Cost Tracking | ✅ | supports all models on `/messages`, `/v1/messages/batches` endpoint |
 | Logging | ✅ | works across all integrations |
 | End-user Tracking | ✅ | disable prometheus tracking via `litellm.disable_end_user_cost_tracking_prometheus_only`|
 | Streaming | ✅ | |
@@ -263,6 +263,19 @@ curl https://api.anthropic.com/v1/messages/batches \
 }'
 ```
 
+:::note Configuration Required for Batch Cost Tracking
+For batch passthrough cost tracking to work properly, you need to define the Anthropic model in your `proxy_config.yaml`:
+
+```yaml
+model_list:
+  - model_name: claude-sonnet-4-5-20250929  # or any alias
+    litellm_params:
+      model: anthropic/claude-sonnet-4-5-20250929
+      api_key: os.environ/ANTHROPIC_API_KEY
+```
+
+This ensures the polling mechanism can correctly identify the provider and retrieve batch status for cost calculation.
+:::
 
 ## Advanced
 

litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py

@@ -16,7 +16,7 @@ from litellm.proxy.auth.auth_utils import get_end_user_id_from_request_body
 from litellm.types.passthrough_endpoints.pass_through_endpoints import (
     PassthroughStandardLoggingPayload,
 )
-from litellm.types.utils import ModelResponse, TextCompletionResponse
+from litellm.types.utils import LiteLLMBatch, ModelResponse, TextCompletionResponse
 
 if TYPE_CHECKING:
     from ..success_handler import PassThroughEndpointLogging
@@ -37,11 +37,28 @@ class AnthropicPassthroughLoggingHandler:
         start_time: datetime,
         end_time: datetime,
         cache_hit: bool,
+        request_body: Optional[dict] = None,
         **kwargs,
     ) -> PassThroughEndpointLoggingTypedDict:
         """
         Transforms Anthropic response to OpenAI response, generates a standard logging object so downstream logging can be handled
         """
+        # Check if this is a batch creation request
+        if "/v1/messages/batches" in url_route and httpx_response.status_code == 200:
+            # Get request body from parameter or kwargs
+            request_body = request_body or kwargs.get("request_body", {})
+            return AnthropicPassthroughLoggingHandler.batch_creation_handler(
+                httpx_response=httpx_response,
+                logging_obj=logging_obj,
+                url_route=url_route,
+                result=result,
+                start_time=start_time,
+                end_time=end_time,
+                cache_hit=cache_hit,
+                request_body=request_body,
+                **kwargs,
+            )
+        
         model = response_body.get("model", "")
         anthropic_config = get_anthropic_config(url_route)
         litellm_model_response: ModelResponse = anthropic_config().transform_response(
@@ -238,3 +255,288 @@ class AnthropicPassthroughLoggingHandler:
             logging_obj=litellm_logging_obj,
         )
         return complete_streaming_response
+
+    @staticmethod
+    def batch_creation_handler(  # noqa: PLR0915
+        httpx_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+        url_route: str,
+        result: str,
+        start_time: datetime,
+        end_time: datetime,
+        cache_hit: bool,
+        request_body: Optional[dict] = None,
+        **kwargs,
+    ) -> PassThroughEndpointLoggingTypedDict:
+        """
+        Handle Anthropic batch creation passthrough logging.
+        Creates a managed object for cost tracking when batch job is successfully created.
+        """
+        import base64
+
+        from litellm._uuid import uuid
+        from litellm.llms.anthropic.batches.transformation import (
+            AnthropicBatchesConfig,
+        )
+        from litellm.types.utils import Choices, SpecialEnums
+
+        try:
+            _json_response = httpx_response.json()
+            
+            
+            # Only handle successful batch job creation (POST requests with 201 status)
+            if httpx_response.status_code == 200 and "id" in _json_response:
+                # Transform Anthropic response to LiteLLM batch format
+                anthropic_batches_config = AnthropicBatchesConfig()
+                litellm_batch_response = anthropic_batches_config.transform_retrieve_batch_response(
+                    model=None,
+                    raw_response=httpx_response,
+                    logging_obj=logging_obj,
+                    litellm_params={},
+                )
+                # Set status to "validating" for newly created batches so polling mechanism picks them up
+                # The polling mechanism only looks for status="validating" jobs
+                litellm_batch_response.status = "validating"
+                
+                # Extract batch ID from the response
+                batch_id = _json_response.get("id", "")
+                
+                # Get model from request body (batch response doesn't include model)
+                request_body = request_body or {}
+                # Try to extract model from the batch request body, supporting Anthropic's nested structure
+                model_name: str = "unknown"
+                if isinstance(request_body, dict):
+                    # Standard: {"model": ...}
+                    model_name = request_body.get("model") or "unknown"
+                    if model_name == "unknown":
+                        # Anthropic batches: look under requests[0].params.model
+                        requests_list = request_body.get("requests", [])
+                        if isinstance(requests_list, list) and len(requests_list) > 0:
+                            first_req = requests_list[0]
+                            if isinstance(first_req, dict):
+                                params = first_req.get("params", {})
+                                if isinstance(params, dict):
+                                    extracted_model = params.get("model")
+                                    if extracted_model:
+                                        model_name = extracted_model
+                
+                
+                # Create unified object ID for tracking
+                # Format: base64(litellm_proxy;model_id:{};llm_batch_id:{})
+                # For Anthropic passthrough, prefix model with "anthropic/" so router can determine provider
+                actual_model_id = AnthropicPassthroughLoggingHandler.get_actual_model_id_from_router(model_name)
+                
+                # If model not in router, use "anthropic/{model_name}" format so router can determine provider
+                if actual_model_id == model_name and not actual_model_id.startswith("anthropic/"):
+                    actual_model_id = f"anthropic/{model_name}"
+
+                unified_id_string = SpecialEnums.LITELLM_MANAGED_BATCH_COMPLETE_STR.value.format(actual_model_id, batch_id)
+                unified_object_id = base64.urlsafe_b64encode(unified_id_string.encode()).decode().rstrip("=")
+                
+                # Store the managed object for cost tracking
+                # This will be picked up by check_batch_cost polling mechanism
+                AnthropicPassthroughLoggingHandler._store_batch_managed_object(
+                    unified_object_id=unified_object_id,
+                    batch_object=litellm_batch_response,
+                    model_object_id=batch_id,
+                    logging_obj=logging_obj,
+                    **kwargs,
+                )
+                
+                # Create a batch job response for logging
+                litellm_model_response = ModelResponse()
+                litellm_model_response.id = str(uuid.uuid4())
+                litellm_model_response.model = model_name
+                litellm_model_response.object = "batch"
+                litellm_model_response.created = int(start_time.timestamp())
+                
+                # Add batch-specific metadata to indicate this is a pending batch job
+                litellm_model_response.choices = [Choices(
+                    finish_reason="batch_pending",
+                    index=0,
+                    message={
+                        "role": "assistant",
+                        "content": f"Batch job {batch_id} created and is pending. Status will be updated when the batch completes.",
+                        "tool_calls": None,
+                        "function_call": None,
+                        "provider_specific_fields": {
+                            "batch_job_id": batch_id,
+                            "batch_job_state": "in_progress",
+                            "unified_object_id": unified_object_id
+                        }
+                    }
+                )]
+                
+                # Set response cost to 0 initially (will be updated when batch completes)
+                response_cost = 0.0
+                kwargs["response_cost"] = response_cost
+                kwargs["model"] = model_name
+                kwargs["batch_id"] = batch_id
+                kwargs["unified_object_id"] = unified_object_id
+                kwargs["batch_job_state"] = "in_progress"
+                
+                logging_obj.model = model_name
+                logging_obj.model_call_details["model"] = logging_obj.model
+                logging_obj.model_call_details["response_cost"] = response_cost
+                logging_obj.model_call_details["batch_id"] = batch_id
+                
+                return {
+                    "result": litellm_model_response,
+                    "kwargs": kwargs,
+                }
+            else:
+                # Handle non-successful responses
+                litellm_model_response = ModelResponse()
+                litellm_model_response.id = str(uuid.uuid4())
+                litellm_model_response.model = "anthropic_batch"
+                litellm_model_response.object = "batch"
+                litellm_model_response.created = int(start_time.timestamp())
+                
+                # Add error-specific metadata
+                litellm_model_response.choices = [Choices(
+                    finish_reason="batch_error",
+                    index=0,
+                    message={
+                        "role": "assistant",
+                        "content": f"Batch job creation failed. Status: {httpx_response.status_code}",
+                        "tool_calls": None,
+                        "function_call": None,
+                        "provider_specific_fields": {
+                            "batch_job_state": "failed",
+                            "status_code": httpx_response.status_code
+                        }
+                    }
+                )]
+                
+                kwargs["response_cost"] = 0.0
+                kwargs["model"] = "anthropic_batch"
+                kwargs["batch_job_state"] = "failed"
+                
+                return {
+                    "result": litellm_model_response,
+                    "kwargs": kwargs,
+                }
+                
+        except Exception as e:
+            verbose_proxy_logger.error(f"Error in batch_creation_handler: {e}")
+            # Return basic response on error
+            litellm_model_response = ModelResponse()
+            litellm_model_response.id = str(uuid.uuid4())
+            litellm_model_response.model = "anthropic_batch"
+            litellm_model_response.object = "batch"
+            litellm_model_response.created = int(start_time.timestamp())
+            
+            # Add error-specific metadata
+            litellm_model_response.choices = [Choices(
+                finish_reason="batch_error",
+                index=0,
+                message={
+                    "role": "assistant",
+                    "content": f"Error creating batch job: {str(e)}",
+                    "tool_calls": None,
+                    "function_call": None,
+                    "provider_specific_fields": {
+                        "batch_job_state": "failed",
+                        "error": str(e)
+                    }
+                }
+            )]
+            
+            kwargs["response_cost"] = 0.0
+            kwargs["model"] = "anthropic_batch"
+            kwargs["batch_job_state"] = "failed"
+            
+            return {
+                "result": litellm_model_response,
+                "kwargs": kwargs,
+            }
+
+    @staticmethod
+    def _store_batch_managed_object(
+        unified_object_id: str,
+        batch_object: LiteLLMBatch,
+        model_object_id: str,
+        logging_obj: LiteLLMLoggingObj,
+        **kwargs,
+    ) -> None:
+        """
+        Store batch managed object for cost tracking.
+        This will be picked up by the check_batch_cost polling mechanism.
+        """
+        try:
+            
+            # Get the managed files hook from the logging object
+            # This is a bit of a hack, but we need access to the proxy logging system
+            from litellm.proxy.proxy_server import proxy_logging_obj
+            
+            managed_files_hook = proxy_logging_obj.get_proxy_hook("managed_files")
+            if managed_files_hook is not None and hasattr(managed_files_hook, 'store_unified_object_id'):
+                # Create a mock user API key dict for the managed object storage
+                from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+                user_api_key_dict = UserAPIKeyAuth(
+                    user_id=kwargs.get("user_id", "default-user"),
+                    api_key="",
+                    team_id=None,
+                    team_alias=None,
+                    user_role=LitellmUserRoles.CUSTOMER,  # Use proper enum value
+                    user_email=None,
+                    max_budget=None,
+                    spend=0.0,  # Set to 0.0 instead of None
+                    models=[],  # Set to empty list instead of None
+                    tpm_limit=None,
+                    rpm_limit=None,
+                    budget_duration=None,
+                    budget_reset_at=None,
+                    max_parallel_requests=None,
+                    allowed_model_region=None,
+                    metadata={},  # Set to empty dict instead of None
+                    key_alias=None,
+                    permissions={},  # Set to empty dict instead of None
+                    model_max_budget={},  # Set to empty dict instead of None
+                    model_spend={},  # Set to empty dict instead of None
+                )
+                
+                # Store the unified object for batch cost tracking
+                import asyncio
+                asyncio.create_task(
+                    managed_files_hook.store_unified_object_id(  # type: ignore
+                        unified_object_id=unified_object_id,
+                        file_object=batch_object,
+                        litellm_parent_otel_span=None,
+                        model_object_id=model_object_id,
+                        file_purpose="batch",
+                        user_api_key_dict=user_api_key_dict,
+                    )
+                )
+                
+                verbose_proxy_logger.info(
+                    f"Stored Anthropic batch managed object with unified_object_id={unified_object_id}, batch_id={model_object_id}"
+                )
+            else:
+                verbose_proxy_logger.warning("Managed files hook not available, cannot store batch object for cost tracking")
+                
+        except Exception as e:
+            verbose_proxy_logger.error(f"Error storing Anthropic batch managed object: {e}")
+
+    @staticmethod
+    def get_actual_model_id_from_router(model_name: str) -> str:
+        from litellm.proxy.proxy_server import llm_router
+        
+        if llm_router is not None:
+            # Try to find the model in the router by the model name
+            # Use the existing get_model_ids method from router
+            model_ids = llm_router.get_model_ids(model_name=model_name)
+            if model_ids and len(model_ids) > 0:
+                # Use the first model ID found
+                actual_model_id = model_ids[0]
+                verbose_proxy_logger.info(f"Found model ID in router: {actual_model_id}")
+                return actual_model_id
+            else:
+                # Fallback to model name
+                actual_model_id = model_name
+                verbose_proxy_logger.warning(f"Model not found in router, using model name: {actual_model_id}")
+                return actual_model_id
+        else:
+            # Fallback if router is not available
+            verbose_proxy_logger.warning(f"Router not available, using model name: {model_name}")
+            return model_name

litellm/proxy/pass_through_endpoints/success_handler.py

@@ -46,7 +46,7 @@ class PassThroughEndpointLogging:
         ]
 
         # Anthropic
-        self.TRACKED_ANTHROPIC_ROUTES = ["/messages"]
+        self.TRACKED_ANTHROPIC_ROUTES = ["/messages", "/v1/messages/batches"]
 
         # Cohere
         self.TRACKED_COHERE_ROUTES = ["/v2/chat", "/v1/embed"]
@@ -169,6 +169,7 @@ class PassThroughEndpointLogging:
                     start_time=start_time,
                     end_time=end_time,
                     cache_hit=cache_hit,
+                    request_body=request_body,
                     **kwargs,
                 )
             )

tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py

@@ -3,7 +3,7 @@ import os
 import sys
 from datetime import datetime
 from typing import Any, Dict, List
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -279,4 +279,303 @@ class TestAzureAnthropicCostCalculation:
         mock_completion_cost.assert_called_once()
         call_kwargs = mock_completion_cost.call_args[1]
         assert call_kwargs["model"] == "azure_ai/claude-sonnet-4-5_gb_20250929"
-        assert call_kwargs["custom_llm_provider"] == "azure_ai" 
+        assert call_kwargs["custom_llm_provider"] == "azure_ai"
+
+
+class TestAnthropicBatchPassthroughCostTracking:
+    """Test cases for Anthropic batch passthrough cost tracking functionality"""
+
+    @pytest.fixture
+    def mock_httpx_response(self):
+        """Mock httpx response for batch job creation"""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "id": "msgbatch_01Wj7gkQk7gn4MpAKR8ZEDU2",
+            "archived_at": None,
+            "cancel_initiated_at": None,
+            "created_at": "2024-08-20T18:37:24.100435Z",
+            "ended_at": None,
+            "expires_at": "2024-08-21T18:37:24.100435Z",
+            "processing_status": "in_progress",
+            "request_counts": {
+                "canceled": 0,
+                "errored": 0,
+                "expired": 0,
+                "processing": 1,
+                "succeeded": 0
+            },
+            "results_url": "https://api.anthropic.com/v1/messages/batches/msgbatch_01Wj7gkQk7gn4MpAKR8ZEDU2/results",
+            "type": "message_batch"
+        }
+        return mock_response
+
+    @pytest.fixture
+    def mock_logging_obj(self):
+        """Mock logging object"""
+        mock = MagicMock()
+        mock.litellm_call_id = "test-call-id-123"
+        mock.model_call_details = {}
+        mock.model = None
+        return mock
+
+    @pytest.fixture
+    def mock_request_body(self):
+        """Mock request body for batch creation"""
+        return {
+            "requests": [
+                {
+                    "custom_id": "my-custom-id-1",
+                    "params": {
+                        "max_tokens": 1024,
+                        "messages": [
+                            {
+                                "content": "Hello, world",
+                                "role": "user"
+                            }
+                        ],
+                        "model": "claude-sonnet-4-5-20250929"
+                    }
+                }
+            ]
+        }
+
+    @patch('litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler.AnthropicPassthroughLoggingHandler._store_batch_managed_object')
+    @patch('litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler.AnthropicPassthroughLoggingHandler.get_actual_model_id_from_router')
+    @patch('litellm.llms.anthropic.batches.transformation.AnthropicBatchesConfig')
+    def test_batch_creation_handler_success(
+        self,
+        mock_batches_config,
+        mock_get_model_id,
+        mock_store_batch,
+        mock_httpx_response,
+        mock_logging_obj,
+        mock_request_body
+    ):
+        """Test successful batch creation and managed object storage"""
+        from litellm.types.utils import LiteLLMBatch
+        
+        # Setup mocks
+        mock_get_model_id.return_value = "claude-sonnet-4-5-20250929"
+        
+        mock_batch_response = LiteLLMBatch(
+            id="msgbatch_01Wj7gkQk7gn4MpAKR8ZEDU2",
+            object="batch",
+            endpoint="/v1/messages",
+            errors=None,
+            input_file_id="None",
+            completion_window="24h",
+            status="validating",
+            output_file_id="msgbatch_01Wj7gkQk7gn4MpAKR8ZEDU2",
+            error_file_id=None,
+            created_at=1704067200,
+            in_progress_at=1704067200,
+            expires_at=1704153600,
+            finalizing_at=None,
+            completed_at=None,
+            failed_at=None,
+            expired_at=None,
+            cancelling_at=None,
+            cancelled_at=None,
+            request_counts={"total": 1, "completed": 0, "failed": 0},
+            metadata={},
+        )
+        
+        mock_batches_config_instance = MagicMock()
+        mock_batches_config_instance.transform_retrieve_batch_response.return_value = mock_batch_response
+        mock_batches_config.return_value = mock_batches_config_instance
+        
+        # Test the handler
+        result = AnthropicPassthroughLoggingHandler.batch_creation_handler(
+            httpx_response=mock_httpx_response,
+            logging_obj=mock_logging_obj,
+            url_route="https://api.anthropic.com/v1/messages/batches",
+            result="success",
+            start_time=datetime.now(),
+            end_time=datetime.now(),
+            cache_hit=False,
+            request_body=mock_request_body,
+        )
+        
+        # Verify the result
+        assert result is not None
+        assert "result" in result
+        assert "kwargs" in result
+        # Model should be extracted from request body
+        assert result["kwargs"]["model"] == "claude-sonnet-4-5-20250929"
+        assert result["kwargs"]["batch_id"] == "msgbatch_01Wj7gkQk7gn4MpAKR8ZEDU2"
+        assert result["kwargs"]["batch_job_state"] == "in_progress"
+        assert "unified_object_id" in result["kwargs"]
+        
+        # Verify batch was stored
+        mock_store_batch.assert_called_once()
+        call_kwargs = mock_store_batch.call_args[1]
+        assert call_kwargs["model_object_id"] == "msgbatch_01Wj7gkQk7gn4MpAKR8ZEDU2"
+        assert call_kwargs["batch_object"].status == "validating"
+        
+        # Verify the response object
+        assert result["result"].model == "claude-sonnet-4-5-20250929"
+        assert result["result"].object == "batch"
+
+    @patch('litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler.AnthropicPassthroughLoggingHandler._store_batch_managed_object')
+    @patch('litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler.AnthropicPassthroughLoggingHandler.get_actual_model_id_from_router')
+    def test_batch_creation_handler_model_extraction_from_nested_request(
+        self,
+        mock_get_model_id,
+        mock_store_batch,
+        mock_httpx_response,
+        mock_logging_obj
+    ):
+        """Test that model is correctly extracted from nested request structure"""
+        from litellm.llms.anthropic.batches.transformation import AnthropicBatchesConfig
+        from litellm.types.utils import LiteLLMBatch
+        
+        # Setup mocks
+        mock_get_model_id.return_value = "claude-sonnet-4-5-20250929"
+        
+        mock_batch_response = LiteLLMBatch(
+            id="msgbatch_123",
+            object="batch",
+            endpoint="/v1/messages",
+            input_file_id="None",
+            completion_window="24h",
+            status="validating",
+            created_at=1704067200,
+            request_counts={"total": 1, "completed": 0, "failed": 0},
+        )
+        
+        with patch.object(AnthropicBatchesConfig, 'transform_retrieve_batch_response', return_value=mock_batch_response):
+            # Request body with nested model in requests[0].params.model
+            request_body = {
+                "requests": [
+                    {
+                        "custom_id": "test-1",
+                        "params": {
+                            "model": "claude-sonnet-4-5-20250929",
+                            "messages": [{"role": "user", "content": "test"}]
+                        }
+                    }
+                ]
+            }
+            
+            result = AnthropicPassthroughLoggingHandler.batch_creation_handler(
+                httpx_response=mock_httpx_response,
+                logging_obj=mock_logging_obj,
+                url_route="https://api.anthropic.com/v1/messages/batches",
+                result="success",
+                start_time=datetime.now(),
+                end_time=datetime.now(),
+                cache_hit=False,
+                request_body=request_body,
+            )
+            
+            # Verify model was extracted correctly
+            assert result["kwargs"]["model"] == "claude-sonnet-4-5-20250929"
+
+    @patch('litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler.AnthropicPassthroughLoggingHandler.get_actual_model_id_from_router')
+    def test_batch_creation_handler_model_prefix_when_not_in_router(
+        self,
+        mock_get_model_id,
+        mock_httpx_response,
+        mock_logging_obj,
+        mock_request_body
+    ):
+        """Test that model gets 'anthropic/' prefix when not found in router"""
+        from litellm.llms.anthropic.batches.transformation import AnthropicBatchesConfig
+        from litellm.types.utils import LiteLLMBatch
+        import base64
+        
+        # Model not in router - returns same model name
+        mock_get_model_id.return_value = "claude-sonnet-4-5-20250929"
+        
+        mock_batch_response = LiteLLMBatch(
+            id="msgbatch_123",
+            object="batch",
+            endpoint="/v1/messages",
+            input_file_id="None",
+            completion_window="24h",
+            status="validating",
+            created_at=1704067200,
+            request_counts={"total": 1, "completed": 0, "failed": 0},
+        )
+        
+        with patch.object(AnthropicBatchesConfig, 'transform_retrieve_batch_response', return_value=mock_batch_response):
+            with patch.object(AnthropicPassthroughLoggingHandler, '_store_batch_managed_object'):
+                result = AnthropicPassthroughLoggingHandler.batch_creation_handler(
+                    httpx_response=mock_httpx_response,
+                    logging_obj=mock_logging_obj,
+                    url_route="https://api.anthropic.com/v1/messages/batches",
+                    result="success",
+                    start_time=datetime.now(),
+                    end_time=datetime.now(),
+                    cache_hit=False,
+                    request_body=mock_request_body,
+                )
+                
+                # Verify unified_object_id contains anthropic/ prefix
+                unified_object_id = result["kwargs"]["unified_object_id"]
+                decoded = base64.urlsafe_b64decode(unified_object_id + "==").decode()
+                assert "anthropic/claude-sonnet-4-5-20250929" in decoded or "claude-sonnet-4-5-20250929" in decoded
+
+    def test_batch_creation_handler_failure_status_code(
+        self,
+        mock_logging_obj,
+        mock_request_body
+    ):
+        """Test batch creation handler with non-200 status code"""
+        mock_response = MagicMock()
+        mock_response.status_code = 400
+        mock_response.json.return_value = {"error": "Bad request"}
+        
+        result = AnthropicPassthroughLoggingHandler.batch_creation_handler(
+            httpx_response=mock_response,
+            logging_obj=mock_logging_obj,
+            url_route="https://api.anthropic.com/v1/messages/batches",
+            result="error",
+            start_time=datetime.now(),
+            end_time=datetime.now(),
+            cache_hit=False,
+            request_body=mock_request_body,
+        )
+        
+        # Verify error response
+        assert result is not None
+        assert result["kwargs"]["batch_job_state"] == "failed"
+        assert result["kwargs"]["response_cost"] == 0.0
+
+    @patch('litellm.proxy.proxy_server.proxy_logging_obj')
+    def test_store_batch_managed_object_success(
+        self,
+        mock_proxy_logging_obj,
+        mock_logging_obj
+    ):
+        """Test storing batch managed object"""
+        from litellm.types.utils import LiteLLMBatch
+        
+        # Setup mocks
+        mock_managed_files_hook = MagicMock()
+        mock_managed_files_hook.store_unified_object_id = AsyncMock()
+        mock_proxy_logging_obj.get_proxy_hook.return_value = mock_managed_files_hook
+        
+        batch_object = LiteLLMBatch(
+            id="msgbatch_123",
+            object="batch",
+            endpoint="/v1/messages",
+            input_file_id="None",
+            completion_window="24h",
+            status="validating",
+            created_at=1704067200,
+            request_counts={"total": 1, "completed": 0, "failed": 0},
+        )
+        
+        with patch('asyncio.create_task'):
+            AnthropicPassthroughLoggingHandler._store_batch_managed_object(
+                unified_object_id="test-unified-id",
+                batch_object=batch_object,
+                model_object_id="msgbatch_123",
+                logging_obj=mock_logging_obj,
+                user_id="test-user"
+            )
+            
+            # Verify managed files hook was called
+            mock_proxy_logging_obj.get_proxy_hook.assert_called_once_with("managed_files")