Merge pull request #16594 from Chesars/feat/anthropic-files-api

feat: Add Anthropic Files API support
This commit is contained in:
Cesar Garcia 2026-03-11 13:45:53 -03:00 • committed by GitHub
commit ebe75a1bc6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 863 additions and 28 deletions

View file

@ -326,4 +326,10 @@ print("file content=", content.text)
### [Bedrock](./providers/bedrock_batches#4-retrieve-batch-results)
### [Anthropic](./providers/anthropic#files-api)
:::note
Anthropic Files API has a different purpose than OpenAI's. It's **not** for Batches or Fine-tuning—it's for uploading files once and referencing them by `file_id` in multiple messages, avoiding re-uploads. File API operations are free — file content used in Messages requests is priced as input tokens.
:::
## [Swagger API Reference](https://litellm-api.up.railway.app/#/files)

View file

@ -1965,6 +1965,98 @@ curl -L -X POST 'http://0.0.0.0:4000/v1/chat/completions' \
</TabItem>
</Tabs>
## Files API
Upload files once and reference them by `file_id` in multiple requests—no need to re-upload content each time.
:::info
The `file_id` obtained from Anthropic only works with Anthropic Claude models. You cannot use it with other providers (OpenAI, Bedrock, etc.).
:::
- **Max file size:** 500 MB | **Total storage:** 100 GB per org
- **Pricing:** File API operations are free. File content used in Messages requests is priced as input tokens.
**Supported models by file type:**
- **Images:** All Claude 3+ models
- **PDFs:** All Claude 3.5+ models
- **Other file types** (for code execution): Claude 3.5 Haiku + all Claude 3.7+ models
### Quick Start
```python
import litellm
import os
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-..."
# 1. Upload a file once
file = litellm.create_file(
file=open("document.pdf", "rb"),
purpose="messages",
custom_llm_provider="anthropic",
)
# 2. Use file_id in messages (no re-upload needed)
response = litellm.completion(
model="anthropic/claude-sonnet-4-5-20250929",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": "Summarize this document"},
{"type": "file", "file": {"file_id": file.id, "format": "application/pdf"}}
]
}]
)
```
### File Operations
| Operation | Function |
|-----------|----------|
| Upload | `litellm.create_file(file, purpose="messages", custom_llm_provider="anthropic")` |
| List | `litellm.file_list(custom_llm_provider="anthropic")` |
| Retrieve | `litellm.file_retrieve(file_id, custom_llm_provider="anthropic")` |
| Delete | `litellm.file_delete(file_id, custom_llm_provider="anthropic")` |
| Download | `litellm.file_content(file_id, custom_llm_provider="anthropic")` |
:::note
Download only works for files created by the [code execution tool](https://docs.anthropic.com/en/docs/agents-and-tools/tool-use/code-execution-tool), not uploaded files.
:::
### Supported Formats
| File Type | Format Value |
|-----------|-------------|
| PDF | `application/pdf` |
| Plain text | `text/plain` |
| JPEG | `image/jpeg` |
| PNG | `image/png` |
| GIF | `image/gif` |
| WebP | `image/webp` |
### Using Images
```python
# Upload image
image = litellm.create_file(
file=open("photo.jpg", "rb"),
purpose="messages",
custom_llm_provider="anthropic",
)
# Use in message
response = litellm.completion(
model="anthropic/claude-sonnet-4-5-20250929",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": "What's in this image?"},
{"type": "file", "file": {"file_id": image.id, "format": "image/jpeg"}}
]
}]
)
```
## Usage - passing 'user_id' to Anthropic
LiteLLM translates the OpenAI `user` param to Anthropic's `metadata[user_id]` param.

View file

@ -18,7 +18,6 @@ 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.common_utils import get_azure_credentials
from litellm.llms.azure.files.handler import AzureOpenAIFilesAPI
from litellm.llms.bedrock.files.handler import BedrockFilesHandler
@ -54,16 +53,15 @@ openai_files_instance = OpenAIFilesAPI()
azure_files_instance = AzureOpenAIFilesAPI()
vertex_ai_files_instance = VertexAIFilesHandler()
bedrock_files_instance = BedrockFilesHandler()
anthropic_files_instance = AnthropicFilesHandler()
#################################################
@client
async def acreate_file(
file: FileTypes,
purpose: Literal["assistants", "batch", "fine-tune"],
purpose: Literal["assistants", "batch", "fine-tune", "messages"],
expires_after: Optional[FileExpiresAfter] = None,
custom_llm_provider: Literal["openai", "azure", "gemini", "vertex_ai", "bedrock", "hosted_vllm", "manus"] = "openai",
custom_llm_provider: Literal["openai", "azure", "gemini", "vertex_ai", "bedrock", "hosted_vllm", "manus", "anthropic"] = "openai",
extra_headers: Optional[Dict[str, str]] = None,
extra_body: Optional[Dict[str, str]] = None,
**kwargs,
@ -106,9 +104,9 @@ async def acreate_file(
@client
def create_file(
file: FileTypes,
purpose: Literal["assistants", "batch", "fine-tune"],
purpose: Literal["assistants", "batch", "fine-tune", "messages"],
expires_after: Optional[FileExpiresAfter] = None,
custom_llm_provider: Optional[Literal["openai", "azure", "gemini", "vertex_ai", "bedrock", "hosted_vllm", "manus"]] = None,
custom_llm_provider: Optional[Literal["openai", "azure", "gemini", "vertex_ai", "bedrock", "hosted_vllm", "manus", "anthropic"]] = None,
extra_headers: Optional[Dict[str, str]] = None,
extra_body: Optional[Dict[str, str]] = None,
**kwargs,
@ -218,7 +216,7 @@ def create_file(
)
else:
raise litellm.exceptions.BadRequestError(
message="LiteLLM doesn't support {} for 'create_file'. Only ['openai', 'azure', 'vertex_ai', 'manus'] are supported.".format(
message="LiteLLM doesn't support {} for 'create_file'. Only ['openai', 'azure', 'vertex_ai', 'manus', 'anthropic'] are supported.".format(
custom_llm_provider
),
model="n/a",
@ -237,7 +235,7 @@ def create_file(
@client
async def afile_retrieve(
file_id: str,
custom_llm_provider: Literal["openai", "azure", "gemini", "vertex_ai", "hosted_vllm", "manus"] = "openai",
custom_llm_provider: Literal["openai", "azure", "gemini", "vertex_ai", "hosted_vllm", "manus", "anthropic"] = "openai",
extra_headers: Optional[Dict[str, str]] = None,
extra_body: Optional[Dict[str, str]] = None,
**kwargs,
@ -278,7 +276,7 @@ async def afile_retrieve(
@client
def file_retrieve(
file_id: str,
custom_llm_provider: Literal["openai", "azure", "gemini", "vertex_ai", "hosted_vllm", "manus"] = "openai",
custom_llm_provider: Literal["openai", "azure", "gemini", "vertex_ai", "hosted_vllm", "manus", "anthropic"] = "openai",
extra_headers: Optional[Dict[str, str]] = None,
extra_body: Optional[Dict[str, str]] = None,
**kwargs,
@ -382,7 +380,7 @@ def file_retrieve(
)
else:
raise litellm.exceptions.BadRequestError(
message="LiteLLM doesn't support {} for 'file_retrieve'. Only 'openai', 'azure', and 'manus' are supported.".format(
message="LiteLLM doesn't support {} for 'file_retrieve'. Only 'openai', 'azure', 'manus', and 'anthropic' are supported.".format(
custom_llm_provider
),
model="n/a",
@ -403,7 +401,7 @@ def file_retrieve(
@client
async def afile_delete(
file_id: str,
custom_llm_provider: Literal["openai", "azure", "gemini", "manus"] = "openai",
custom_llm_provider: Literal["openai", "azure", "gemini", "manus", "anthropic"] = "openai",
extra_headers: Optional[Dict[str, str]] = None,
extra_body: Optional[Dict[str, str]] = None,
**kwargs,
@ -447,7 +445,7 @@ async def afile_delete(
def file_delete(
file_id: str,
model: Optional[str] = None,
custom_llm_provider: Union[Literal["openai", "azure", "gemini", "manus"], str] = "openai",
custom_llm_provider: Union[Literal["openai", "azure", "gemini", "manus", "anthropic"], str] = "openai",
extra_headers: Optional[Dict[str, str]] = None,
extra_body: Optional[Dict[str, str]] = None,
**kwargs,
@ -558,7 +556,7 @@ def file_delete(
)
else:
raise litellm.exceptions.BadRequestError(
message="LiteLLM doesn't support {} for 'file_delete'. Only 'openai', 'azure', 'gemini', and 'manus' are supported.".format(
message="LiteLLM doesn't support {} for 'file_delete'. Only 'openai', 'azure', 'gemini', 'manus', and 'anthropic' are supported.".format(
custom_llm_provider
),
model="n/a",
@ -577,7 +575,7 @@ def file_delete(
# List files
@client
async def afile_list(
custom_llm_provider: Literal["openai", "azure", "manus"] = "openai",
custom_llm_provider: Literal["openai", "azure", "manus", "anthropic"] = "openai",
purpose: Optional[str] = None,
extra_headers: Optional[Dict[str, str]] = None,
extra_body: Optional[Dict[str, str]] = None,
@ -618,7 +616,7 @@ async def afile_list(
@client
def file_list(
custom_llm_provider: Literal["openai", "azure", "manus"] = "openai",
custom_llm_provider: Literal["openai", "azure", "manus", "anthropic"] = "openai",
purpose: Optional[str] = None,
extra_headers: Optional[Dict[str, str]] = None,
extra_body: Optional[Dict[str, str]] = None,
@ -723,7 +721,7 @@ def file_list(
)
else:
raise litellm.exceptions.BadRequestError(
message="LiteLLM doesn't support {} for 'file_list'. Only 'openai', 'azure', and 'manus' are supported.".format(
message="LiteLLM doesn't support {} for 'file_list'. Only 'openai', 'azure', 'manus', and 'anthropic' are supported.".format(
custom_llm_provider
),
model="n/a",
@ -834,15 +832,41 @@ 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,
# Check if provider has a custom files config (e.g., Anthropic, Manus)
provider_config = ProviderConfigManager.get_provider_files_config(
model="",
provider=LlmProviders(custom_llm_provider),
)
if provider_config is not None:
litellm_params_dict["api_key"] = optional_params.api_key
litellm_params_dict["api_base"] = optional_params.api_base
logging_obj = kwargs.get("litellm_logging_obj")
if logging_obj is None:
logging_obj = LiteLLMLoggingObj(
model="",
messages=[],
stream=False,
call_type="afile_content" if _is_async else "file_content",
start_time=time.time(),
litellm_call_id=kwargs.get("litellm_call_id", str(uuid_module.uuid4())),
function_id=str(kwargs.get("id") or ""),
)
response = base_llm_http_handler.retrieve_file_content(
file_content_request=_file_content_request,
api_base=optional_params.api_base,
api_key=optional_params.api_key,
provider_config=provider_config,
litellm_params=litellm_params_dict,
headers=extra_headers or {},
logging_obj=logging_obj,
_is_async=_is_async,
client=(
client
if client is not None
and isinstance(client, (HTTPHandler, AsyncHTTPHandler))
else None
),
timeout=timeout,
max_retries=optional_params.max_retries,
)
return response
@ -915,7 +939,7 @@ def file_content(
)
else:
raise litellm.exceptions.BadRequestError(
message="LiteLLM doesn't support {} for 'file_content'. Supported providers are 'openai', 'azure', 'vertex_ai', 'bedrock', 'manus'.".format(
message="LiteLLM doesn't support {} for 'file_content'. Supported providers are 'openai', 'azure', 'vertex_ai', 'bedrock', 'manus', 'anthropic'.".format(
custom_llm_provider
),
model="n/a",

View file

@ -644,6 +644,10 @@ def extract_file_data(file_data: FileTypes) -> ExtractedFileData:
content = f.read()
elif isinstance(file_content, io.IOBase):
# If it's a file-like object
# Try to get filename from file handle if not already set
if not filename and hasattr(file_content, 'name'):
filename = Path(file_content.name).name
content = file_content.read()
if isinstance(content, str):

View file

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

View file

@ -0,0 +1,303 @@
"""
Anthropic Files API transformation config.
Implements BaseFilesConfig for Anthropic's Files API (beta).
Reference: https://docs.anthropic.com/en/docs/build-with-claude/files
Anthropic Files API endpoints:
- POST /v1/files - Upload a file
- GET /v1/files - List files
- GET /v1/files/{file_id} - Retrieve file metadata
- DELETE /v1/files/{file_id} - Delete a file
- GET /v1/files/{file_id}/content - Download file content
"""
import calendar
import time
from typing import Any, Dict, List, Optional, Union
import httpx
from openai.types.file_deleted import FileDeleted
from litellm.litellm_core_utils.prompt_templates.common_utils import extract_file_data
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.files.transformation import (
BaseFilesConfig,
LiteLLMLoggingObj,
)
from litellm.types.llms.openai import (
CreateFileRequest,
FileContentRequest,
HttpxBinaryResponseContent,
OpenAICreateFileRequestOptionalParams,
OpenAIFileObject,
)
from litellm.types.utils import LlmProviders
from ..common_utils import AnthropicError, AnthropicModelInfo
ANTHROPIC_FILES_API_BASE = "https://api.anthropic.com"
ANTHROPIC_FILES_BETA_HEADER = "files-api-2025-04-14"
class AnthropicFilesConfig(BaseFilesConfig):
"""
Transformation config for Anthropic Files API.
Anthropic uses:
- x-api-key header for authentication
- anthropic-beta: files-api-2025-04-14 header
- multipart/form-data for file uploads
- purpose="messages" (Anthropic-specific, not for batches/fine-tuning)
"""
def __init__(self):
pass
@property
def custom_llm_provider(self) -> LlmProviders:
return LlmProviders.ANTHROPIC
def get_complete_url(
self,
api_base: Optional[str],
api_key: Optional[str],
model: str,
optional_params: dict,
litellm_params: dict,
stream: Optional[bool] = None,
) -> str:
api_base = AnthropicModelInfo.get_api_base(api_base) or ANTHROPIC_FILES_API_BASE
return f"{api_base.rstrip('/')}/v1/files"
def get_error_class(
self,
error_message: str,
status_code: int,
headers: Union[dict, httpx.Headers],
) -> BaseLLMException:
return AnthropicError(
status_code=status_code,
message=error_message,
headers=headers,
)
def validate_environment(
self,
headers: dict,
model: str,
messages: list,
optional_params: dict,
litellm_params: dict,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
) -> dict:
api_key = AnthropicModelInfo.get_api_key(api_key)
if not api_key:
raise ValueError(
"Anthropic API key is required. Set ANTHROPIC_API_KEY environment variable or pass api_key parameter."
)
headers.update(
{
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
"anthropic-beta": ANTHROPIC_FILES_BETA_HEADER,
}
)
return headers
def get_supported_openai_params(
self, model: str
) -> List[OpenAICreateFileRequestOptionalParams]:
return ["purpose"]
def map_openai_params(
self,
non_default_params: dict,
optional_params: dict,
model: str,
drop_params: bool,
) -> dict:
return optional_params
def transform_create_file_request(
self,
model: str,
create_file_data: CreateFileRequest,
optional_params: dict,
litellm_params: dict,
) -> dict:
"""
Transform to multipart form data for Anthropic file upload.
Anthropic expects: POST /v1/files with multipart form-data
- file: the file content
- purpose: "messages" (defaults to "messages" if not provided)
"""
file_data = create_file_data.get("file")
if file_data is None:
raise ValueError("File data is required")
extracted = extract_file_data(file_data)
filename = extracted["filename"] or f"file_{int(time.time())}"
content = extracted["content"]
content_type = extracted.get("content_type", "application/octet-stream")
purpose = create_file_data.get("purpose", "messages")
return {
"file": (filename, content, content_type),
"purpose": (None, purpose),
}
def transform_create_file_response(
self,
model: Optional[str],
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
) -> OpenAIFileObject:
"""
Transform Anthropic file response to OpenAI format.
Anthropic response:
{
"id": "file-xxx",
"type": "file",
"filename": "document.pdf",
"mime_type": "application/pdf",
"size_bytes": 12345,
"created_at": "2025-01-01T00:00:00Z"
}
"""
response_json = raw_response.json()
return self._parse_anthropic_file(response_json)
def transform_retrieve_file_request(
self,
file_id: str,
optional_params: dict,
litellm_params: dict,
) -> tuple[str, dict]:
api_base = AnthropicModelInfo.get_api_base(
litellm_params.get("api_base")
) or ANTHROPIC_FILES_API_BASE
return f"{api_base.rstrip('/')}/v1/files/{file_id}", {}
def transform_retrieve_file_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
) -> OpenAIFileObject:
response_json = raw_response.json()
return self._parse_anthropic_file(response_json)
def transform_delete_file_request(
self,
file_id: str,
optional_params: dict,
litellm_params: dict,
) -> tuple[str, dict]:
api_base = AnthropicModelInfo.get_api_base(
litellm_params.get("api_base")
) or ANTHROPIC_FILES_API_BASE
return f"{api_base.rstrip('/')}/v1/files/{file_id}", {}
def transform_delete_file_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
) -> FileDeleted:
response_json = raw_response.json()
file_id = response_json.get("id", "")
return FileDeleted(
id=file_id,
deleted=True,
object="file",
)
def transform_list_files_request(
self,
purpose: Optional[str],
optional_params: dict,
litellm_params: dict,
) -> tuple[str, dict]:
api_base = AnthropicModelInfo.get_api_base(
litellm_params.get("api_base")
) or ANTHROPIC_FILES_API_BASE
url = f"{api_base.rstrip('/')}/v1/files"
params: Dict[str, Any] = {}
if purpose:
params["purpose"] = purpose
return url, params
def transform_list_files_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
) -> List[OpenAIFileObject]:
"""
Anthropic list response:
{
"data": [...],
"has_more": false,
"first_id": "...",
"last_id": "..."
}
"""
response_json = raw_response.json()
files_data = response_json.get("data", [])
return [self._parse_anthropic_file(f) for f in files_data]
def transform_file_content_request(
self,
file_content_request: FileContentRequest,
optional_params: dict,
litellm_params: dict,
) -> tuple[str, dict]:
file_id = file_content_request.get("file_id")
api_base = AnthropicModelInfo.get_api_base(
litellm_params.get("api_base")
) or ANTHROPIC_FILES_API_BASE
return f"{api_base.rstrip('/')}/v1/files/{file_id}/content", {}
def transform_file_content_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
) -> HttpxBinaryResponseContent:
return HttpxBinaryResponseContent(response=raw_response)
@staticmethod
def _parse_anthropic_file(file_data: dict) -> OpenAIFileObject:
"""Parse Anthropic file object into OpenAI format."""
created_at_str = file_data.get("created_at", "")
if created_at_str:
try:
created_at = int(
calendar.timegm(
time.strptime(
created_at_str.replace("Z", "+00:00")[:19],
"%Y-%m-%dT%H:%M:%S",
)
)
)
except (ValueError, TypeError):
created_at = int(time.time())
else:
created_at = int(time.time())
return OpenAIFileObject(
id=file_data.get("id", ""),
bytes=file_data.get("size_bytes", file_data.get("bytes", 0)),
created_at=created_at,
filename=file_data.get("filename", ""),
object="file",
purpose=file_data.get("purpose", "messages"),
status="uploaded",
status_details=None,
)

View file

@ -3012,6 +3012,15 @@ class BaseLLMHTTPHandler:
data=transformed_request,
timeout=timeout,
)
elif isinstance(transformed_request, dict) and "file" in transformed_request:
# Handle multipart form-data uploads (e.g., Anthropic Files API)
# The dict contains tuples suitable for httpx's `files` parameter
upload_response = sync_httpx_client.post(
url=api_base,
headers=headers,
files=transformed_request,
timeout=timeout,
)
else:
raise ValueError(f"Unsupported transformed_request type: {type(transformed_request)}")
@ -3140,6 +3149,15 @@ class BaseLLMHTTPHandler:
data=transformed_request,
timeout=timeout,
)
elif isinstance(transformed_request, dict) and "file" in transformed_request:
# Handle multipart form-data uploads (e.g., Anthropic Files API)
# The dict contains tuples suitable for httpx's `files` parameter
upload_response = await async_httpx_client.post(
url=api_base,
headers=headers,
files=transformed_request,
timeout=timeout,
)
else:
raise ValueError(f"Unsupported transformed_request type: {type(transformed_request)}")

View file

@ -287,6 +287,7 @@ OpenAIFilesPurpose = Literal[
"fine-tune-results",
"vision",
"user_data",
"messages",
]
@ -352,7 +353,7 @@ class OpenAIFileObject(BaseModel):
return self.dict()
CREATE_FILE_REQUESTS_PURPOSE = Literal["assistants", "batch", "fine-tune"]
CREATE_FILE_REQUESTS_PURPOSE = Literal["assistants", "batch", "fine-tune", "messages"]
# File expiration policy
@ -373,11 +374,11 @@ class FileExpiresAfter(TypedDict):
class CreateFileRequest(TypedDict, total=False):
"""
CreateFileRequest
Used by Assistants API, Batches API, and Fine-Tunes API
Used by Assistants API, Batches API, Fine-Tunes API, and Anthropic Files API
Required Params:
file: FileTypes
purpose: Literal['assistants', 'batch', 'fine-tune']
purpose: Literal['assistants', 'batch', 'fine-tune', 'messages']
Optional Params:
expires_after: Optional[FileExpiresAfter] - The expiration policy for a file

View file

@ -8529,6 +8529,12 @@ class ProviderConfigManager:
from litellm.llms.manus.files.transformation import ManusFilesConfig
return ManusFilesConfig()
elif LlmProviders.ANTHROPIC == provider:
from litellm.llms.anthropic.files.transformation import (
AnthropicFilesConfig,
)
return AnthropicFilesConfig()
return None
@staticmethod

View file

@ -17,3 +17,4 @@ exclude = ["litellm/types/*", "litellm/__init__.py", "litellm/proxy/example_conf
"litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py" = ["PLR0915"]
"litellm/proxy/guardrails/guardrail_hooks/guardrail_benchmarks/test_eval.py" = ["PLR0915"]
"litellm/responses/streaming_iterator.py" = ["PLR0915"]
"litellm/files/main.py" = ["PLR0915"]

View file

@ -0,0 +1,379 @@
"""
Test Anthropic Files API transformation functionality.
Tests the AnthropicFilesConfig class which transforms between
OpenAI-compatible file operations and Anthropic's Files API format.
"""
import io
import time
import httpx
import pytest
from unittest.mock import Mock, patch
from litellm.llms.anthropic.files.transformation import (
AnthropicFilesConfig,
ANTHROPIC_FILES_API_BASE,
ANTHROPIC_FILES_BETA_HEADER,
)
from litellm.types.llms.openai import OpenAIFileObject
from litellm.types.utils import LlmProviders
class TestAnthropicFilesConfig:
"""Test AnthropicFilesConfig transformation methods."""
def setup_method(self):
self.config = AnthropicFilesConfig()
def test_custom_llm_provider(self):
assert self.config.custom_llm_provider == LlmProviders.ANTHROPIC
def test_get_complete_url_default(self):
url = self.config.get_complete_url(
api_base=None,
api_key="test-key",
model="",
optional_params={},
litellm_params={},
)
assert url == f"{ANTHROPIC_FILES_API_BASE}/v1/files"
def test_get_complete_url_custom_base(self):
url = self.config.get_complete_url(
api_base="https://custom.api.com",
api_key="test-key",
model="",
optional_params={},
litellm_params={},
)
assert url == "https://custom.api.com/v1/files"
def test_get_complete_url_strips_trailing_slash(self):
url = self.config.get_complete_url(
api_base="https://custom.api.com/",
api_key="test-key",
model="",
optional_params={},
litellm_params={},
)
assert url == "https://custom.api.com/v1/files"
def test_validate_environment_sets_headers(self):
headers = {}
result = self.config.validate_environment(
headers=headers,
model="",
messages=[],
optional_params={},
litellm_params={},
api_key="sk-ant-test-key",
)
assert result["x-api-key"] == "sk-ant-test-key"
assert result["anthropic-version"] == "2023-06-01"
assert result["anthropic-beta"] == ANTHROPIC_FILES_BETA_HEADER
@patch.dict("os.environ", {}, clear=True)
@patch(
"litellm.llms.anthropic.common_utils.AnthropicModelInfo.get_api_key",
return_value=None,
)
def test_validate_environment_missing_api_key(self, mock_get_key):
with pytest.raises(ValueError, match="Anthropic API key is required"):
self.config.validate_environment(
headers={},
model="",
messages=[],
optional_params={},
litellm_params={},
api_key=None,
)
def test_get_supported_openai_params(self):
params = self.config.get_supported_openai_params(model="")
assert "purpose" in params
def test_transform_create_file_request(self):
file_content = b"test file content"
file_tuple = ("test.txt", file_content, "text/plain")
result = self.config.transform_create_file_request(
model="",
create_file_data={
"file": file_tuple,
"purpose": "messages",
},
optional_params={},
litellm_params={},
)
assert "file" in result
assert "purpose" in result
# file should be a tuple (filename, content, content_type)
assert result["file"][0] == "test.txt"
assert result["file"][1] == file_content
assert result["file"][2] == "text/plain"
# purpose should be (None, value) for multipart form field
assert result["purpose"] == (None, "messages")
def test_transform_create_file_request_missing_file(self):
with pytest.raises(ValueError, match="File data is required"):
self.config.transform_create_file_request(
model="",
create_file_data={"purpose": "messages"},
optional_params={},
litellm_params={},
)
def test_transform_create_file_request_default_purpose(self):
file_tuple = ("test.txt", b"content", "text/plain")
result = self.config.transform_create_file_request(
model="",
create_file_data={"file": file_tuple},
optional_params={},
litellm_params={},
)
assert result["purpose"] == (None, "messages")
def test_transform_create_file_response(self):
mock_response = Mock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "file-abc123",
"type": "file",
"filename": "document.pdf",
"mime_type": "application/pdf",
"size_bytes": 12345,
"created_at": "2025-01-15T10:30:00Z",
}
result = self.config.transform_create_file_response(
model=None,
raw_response=mock_response,
logging_obj=Mock(),
litellm_params={},
)
assert isinstance(result, OpenAIFileObject)
assert result.id == "file-abc123"
assert result.filename == "document.pdf"
assert result.bytes == 12345
assert result.object == "file"
assert result.purpose == "messages"
assert result.status == "uploaded"
def test_transform_retrieve_file_request(self):
url, params = self.config.transform_retrieve_file_request(
file_id="file-abc123",
optional_params={},
litellm_params={},
)
assert url == f"{ANTHROPIC_FILES_API_BASE}/v1/files/file-abc123"
assert params == {}
def test_transform_retrieve_file_request_custom_base(self):
url, params = self.config.transform_retrieve_file_request(
file_id="file-abc123",
optional_params={},
litellm_params={"api_base": "https://custom.api.com"},
)
assert url == "https://custom.api.com/v1/files/file-abc123"
assert params == {}
def test_transform_retrieve_file_response(self):
mock_response = Mock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "file-abc123",
"type": "file",
"filename": "document.pdf",
"mime_type": "application/pdf",
"size_bytes": 5000,
"created_at": "2025-06-01T12:00:00Z",
}
result = self.config.transform_retrieve_file_response(
raw_response=mock_response,
logging_obj=Mock(),
litellm_params={},
)
assert isinstance(result, OpenAIFileObject)
assert result.id == "file-abc123"
assert result.bytes == 5000
def test_transform_delete_file_request(self):
url, params = self.config.transform_delete_file_request(
file_id="file-abc123",
optional_params={},
litellm_params={},
)
assert url == f"{ANTHROPIC_FILES_API_BASE}/v1/files/file-abc123"
assert params == {}
def test_transform_delete_file_response(self):
mock_response = Mock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "file-abc123",
"type": "file_deleted",
}
result = self.config.transform_delete_file_response(
raw_response=mock_response,
logging_obj=Mock(),
litellm_params={},
)
assert result.id == "file-abc123"
assert result.deleted is True
assert result.object == "file"
def test_transform_list_files_request(self):
url, params = self.config.transform_list_files_request(
purpose=None,
optional_params={},
litellm_params={},
)
assert url == f"{ANTHROPIC_FILES_API_BASE}/v1/files"
assert params == {}
def test_transform_list_files_request_with_purpose(self):
url, params = self.config.transform_list_files_request(
purpose="messages",
optional_params={},
litellm_params={},
)
assert url == f"{ANTHROPIC_FILES_API_BASE}/v1/files"
assert params == {"purpose": "messages"}
def test_transform_list_files_response(self):
mock_response = Mock(spec=httpx.Response)
mock_response.json.return_value = {
"data": [
{
"id": "file-1",
"filename": "a.txt",
"size_bytes": 100,
"created_at": "2025-01-01T00:00:00Z",
},
{
"id": "file-2",
"filename": "b.txt",
"size_bytes": 200,
"created_at": "2025-01-02T00:00:00Z",
},
],
"has_more": False,
}
result = self.config.transform_list_files_response(
raw_response=mock_response,
logging_obj=Mock(),
litellm_params={},
)
assert len(result) == 2
assert result[0].id == "file-1"
assert result[0].filename == "a.txt"
assert result[1].id == "file-2"
def test_transform_list_files_response_empty(self):
mock_response = Mock(spec=httpx.Response)
mock_response.json.return_value = {"data": [], "has_more": False}
result = self.config.transform_list_files_response(
raw_response=mock_response,
logging_obj=Mock(),
litellm_params={},
)
assert result == []
def test_transform_file_content_request(self):
url, params = self.config.transform_file_content_request(
file_content_request={"file_id": "file-abc123"},
optional_params={},
litellm_params={},
)
assert url == f"{ANTHROPIC_FILES_API_BASE}/v1/files/file-abc123/content"
assert params == {}
def test_transform_file_content_response(self):
mock_response = Mock(spec=httpx.Response)
result = self.config.transform_file_content_response(
raw_response=mock_response,
logging_obj=Mock(),
litellm_params={},
)
assert result.response == mock_response
def test_parse_anthropic_file_with_size_bytes(self):
"""Test that size_bytes is correctly mapped to bytes field."""
result = AnthropicFilesConfig._parse_anthropic_file(
{
"id": "file-test",
"filename": "test.pdf",
"size_bytes": 9999,
"created_at": "2025-03-01T00:00:00Z",
}
)
assert result.bytes == 9999
def test_parse_anthropic_file_fallback_bytes_field(self):
"""Test fallback to 'bytes' field when 'size_bytes' is missing."""
result = AnthropicFilesConfig._parse_anthropic_file(
{
"id": "file-test",
"filename": "test.pdf",
"bytes": 7777,
"created_at": "2025-03-01T00:00:00Z",
}
)
assert result.bytes == 7777
def test_parse_anthropic_file_invalid_timestamp(self):
"""Test that invalid timestamps fall back to current time."""
result = AnthropicFilesConfig._parse_anthropic_file(
{
"id": "file-test",
"filename": "test.pdf",
"size_bytes": 100,
"created_at": "not-a-date",
}
)
# Should not raise, should use current time
assert isinstance(result.created_at, int)
assert result.created_at > 0
def test_parse_anthropic_file_missing_timestamp(self):
"""Test that missing timestamps fall back to current time."""
result = AnthropicFilesConfig._parse_anthropic_file(
{
"id": "file-test",
"filename": "test.pdf",
"size_bytes": 100,
}
)
assert isinstance(result.created_at, int)
assert result.created_at > 0
def test_get_error_class(self):
error = self.config.get_error_class(
error_message="Not found",
status_code=404,
headers={},
)
assert error.status_code == 404
assert error.message == "Not found"
class TestProviderConfigRegistration:
"""Test that AnthropicFilesConfig is properly registered."""
def test_provider_config_returns_anthropic_files_config(self):
from litellm.utils import ProviderConfigManager
config = ProviderConfigManager.get_provider_files_config(
model="",
provider=LlmProviders.ANTHROPIC,
)
assert config is not None
assert isinstance(config, AnthropicFilesConfig)