tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-08-05 08:23:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add E2E Container API Support (#16136)

Browse Source 
* Add v1 cut of container api

* fix lint errors

* Add proxy support to container apis & logging support (#16049)

* Add proxy support to container apis

* Add logging support

* Add cost tracking support for containers and documentation

* Add new constant documentation

* Add container cost in model map

* fix failing azure tests

* Update tests based on model map changes

* fix model map tests

* fix model map tests

* Container modeshould be container

* Container tests fix

* Merge branch 'main' into litellm_sameer_oct_staging_2

---------

Co-authored-by: Ishaan Jaffer <ishaanjaffer0324@gmail.com>
This commit is contained in:
Sameer Kankute
2025-11-01 14:03:51 -07:00
committed by GitHub
co-authored by Ishaan Jaffer
parent 571f94777c
commit 0c743e1adc
34 changed files with 4693 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/my-website/docs/containers.md
+358
View File
@@ -0,0 +1,358 @@
import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';
# /containers
Manage OpenAI code interpreter containers (sessions) for executing code in isolated environments.
| Feature | Supported |
|---------|-----------|
| Cost Tracking | ✅ |
| Logging | ✅ (Full request/response logging) |
| Load Balancing | ✅ |
| Proxy Server Support | ✅ Full proxy integration with virtual keys |
| Spend Management | ✅ Budget tracking and rate limiting |
| Supported Providers | `openai`|
## **Supported Providers**:
- [OpenAI](#quick-start)
## Quick Start
Containers provide isolated execution environments for code interpreter sessions. You can create, list, retrieve, and delete containers.
### SDK, PROXY, and OpenAI Client
<Tabs>
<TabItem value="sdk" label="SDK">
**Create a Container**
```python
import litellm
import os
# setup env
os.environ["OPENAI_API_KEY"] = "sk-.."
container = litellm.create_container(
name="My Code Interpreter Container",
custom_llm_provider="openai",
expires_after={
"anchor": "last_active_at",
"minutes": 20
}
)
print(f"Container ID: {container.id}")
print(f"Container Name: {container.name}")
### ASYNC USAGE ###
# container = await litellm.acreate_container(
# name="My Code Interpreter Container",
# custom_llm_provider="openai",
# expires_after={
# "anchor": "last_active_at",
# "minutes": 20
# }
# )
```
**List Containers**
```python
from litellm import list_containers, alist_containers
import os
os.environ["OPENAI_API_KEY"] = "sk-.."
containers = list_containers(
custom_llm_provider="openai",
limit=20,
order="desc"
)
print(f"Found {len(containers.data)} containers")
for container in containers.data:
print(f" - {container.id}: {container.name}")
### ASYNC USAGE ###
# containers = await alist_containers(
# custom_llm_provider="openai",
# limit=20,
# order="desc"
# )
```
**Retrieve a Container**
```python
from litellm import retrieve_container, aretrieve_container
import os
os.environ["OPENAI_API_KEY"] = "sk-.."
container = retrieve_container(
container_id="cntr_123...",
custom_llm_provider="openai"
)
print(f"Container: {container.name}")
print(f"Status: {container.status}")
print(f"Created: {container.created_at}")
### ASYNC USAGE ###
# container = await aretrieve_container(
# container_id="cntr_123...",
# custom_llm_provider="openai"
# )
```
**Delete a Container**
```python
from litellm import delete_container, adelete_container
import os
os.environ["OPENAI_API_KEY"] = "sk-.."
result = delete_container(
container_id="cntr_123...",
custom_llm_provider="openai"
)
print(f"Deleted: {result.deleted}")
print(f"Container ID: {result.id}")
### ASYNC USAGE ###
# result = await adelete_container(
# container_id="cntr_123...",
# custom_llm_provider="openai"
# )
```
</TabItem>
<TabItem value="proxy" label="LiteLLM PROXY Server">
```bash
$ export OPENAI_API_KEY="sk-..."
$ litellm
# RUNNING on http://0.0.0.0:4000
```
**Custom Provider Specification**
You can specify the custom LLM provider in multiple ways (priority order):
1. Header: `-H "custom-llm-provider: openai"`
2. Query param: `?custom_llm_provider=openai`
3. Request body: `{"custom_llm_provider": "openai", ...}`
4. Defaults to "openai" if not specified
**Create a Container**
```bash
# Default provider (openai)
curl -X POST "http://localhost:4000/v1/containers" \
-H "Authorization: Bearer sk-1234" \
-H "Content-Type: application/json" \
-d '{
"name": "My Container",
"expires_after": {
"anchor": "last_active_at",
"minutes": 20
}
}'
```
```bash
# Via header
curl -X POST "http://localhost:4000/v1/containers" \
-H "Authorization: Bearer sk-1234" \
-H "custom-llm-provider: openai" \
-H "Content-Type: application/json" \
-d '{
"name": "My Container"
}'
```
```bash
# Via query parameter
curl -X POST "http://localhost:4000/v1/containers?custom_llm_provider=openai" \
-H "Authorization: Bearer sk-1234" \
-H "Content-Type: application/json" \
-d '{
"name": "My Container"
}'
```
**List Containers**
```bash
curl "http://localhost:4000/v1/containers?limit=20&order=desc" \
-H "Authorization: Bearer sk-1234"
```
**Retrieve a Container**
```bash
curl "http://localhost:4000/v1/containers/cntr_123..." \
-H "Authorization: Bearer sk-1234"
```
**Delete a Container**
```bash
curl -X DELETE "http://localhost:4000/v1/containers/cntr_123..." \
-H "Authorization: Bearer sk-1234"
```
</TabItem>
<TabItem value="openai" label="OpenAI Python Client">
**Setup**
```python
from openai import OpenAI
client = OpenAI(
api_key="sk-1234", # Your LiteLLM proxy key
base_url="http://localhost:4000" # LiteLLM proxy URL
)
```
**Create a Container**
```python
container = client.containers.create(
name="test-container",
expires_after={
"anchor": "last_active_at",
"minutes": 20
},
extra_body={"custom_llm_provider": "openai"}
)
print(f"Container ID: {container.id}")
print(f"Container Name: {container.name}")
print(f"Created at: {container.created_at}")
```
**List Containers**
```python
containers = client.containers.list(
limit=20,
extra_body={"custom_llm_provider": "openai"}
)
print(f"Found {len(containers.data)} containers")
for container in containers.data:
print(f" - {container.id}: {container.name}")
```
**Retrieve a Container**
```python
container = client.containers.retrieve(
container_id="cntr_6901d28b3c8881908b702815828a5bde0380b3408aeae8c7",
extra_body={"custom_llm_provider": "openai"}
)
print(f"Container: {container.name}")
print(f"Status: {container.status}")
print(f"Last active: {container.last_active_at}")
```
**Delete a Container**
```python
result = client.containers.delete(
container_id="cntr_6901d28b3c8881908b702815828a5bde0380b3408aeae8c7",
extra_body={"custom_llm_provider": "openai"}
)
print(f"Deleted: {result.deleted}")
print(f"Container ID: {result.id}")
```
</TabItem>
</Tabs>
## Container Parameters
### Create Container Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `name` | string | Yes | Name of the container |
| `expires_after` | object | No | Container expiration settings |
| `expires_after.anchor` | string | No | Anchor point for expiration (e.g., "last_active_at") |
| `expires_after.minutes` | integer | No | Minutes until expiration from anchor |
| `file_ids` | array | No | List of file IDs to include in the container |
| `custom_llm_provider` | string | No | LLM provider to use (default: "openai") |
### List Container Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `after` | string | No | Cursor for pagination |
| `limit` | integer | No | Number of items to return (1-100, default: 20) |
| `order` | string | No | Sort order: "asc" or "desc" (default: "desc") |
| `custom_llm_provider` | string | No | LLM provider to use (default: "openai") |
### Retrieve/Delete Container Parameters
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| `container_id` | string | Yes | ID of the container to retrieve/delete |
| `custom_llm_provider` | string | No | LLM provider to use (default: "openai") |
## Response Objects
### ContainerObject
```json
{
"id": "cntr_123...",
"object": "container",
"created_at": 1234567890,
"name": "My Container",
"status": "active",
"last_active_at": 1234567890,
"expires_at": 1234569090,
"file_ids": []
}
```
### ContainerListResponse
```json
{
"object": "list",
"data": [
{
"id": "cntr_123...",
"object": "container",
"created_at": 1234567890,
"name": "My Container",
"status": "active"
}
],
"first_id": "cntr_123...",
"last_id": "cntr_456...",
"has_more": false
}
```
### DeleteContainerResult
```json
{
"id": "cntr_123...",
"object": "container.deleted",
"deleted": true
}
```
docs/my-website/docs/proxy/config_settings.md
-1
View File
@@ -396,7 +396,6 @@ router_settings:
| AZURE_CERTIFICATE_PASSWORD | Password for Azure OpenAI certificate
| AZURE_CLIENT_ID | Client ID for Azure services
| AZURE_CLIENT_SECRET | Client secret for Azure services
| AZURE_CODE_INTERPRETER_COST_PER_SESSION | Cost per session for Azure Code Interpreter service
| AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS | Input cost per 1K tokens for Azure Computer Use service
| AZURE_COMPUTER_USE_OUTPUT_COST_PER_1K_TOKENS | Output cost per 1K tokens for Azure Computer Use service
| AZURE_DEFAULT_RESPONSES_API_VERSION | Version of the Azure Default Responses API being used. Default is "preview"
docs/my-website/sidebars.js
+1
View File
@@ -294,6 +294,7 @@ const sidebars = {
"proxy/managed_batches",
]
},
"containers",
{
type: "category",
label: "/chat/completions",
litellm/__init__.py
+1
View File
@@ -1370,6 +1370,7 @@ from .batch_completion.main import * # type: ignore
from .rerank_api.main import *
from .llms.anthropic.experimental_pass_through.messages.handler import *
from .responses.main import *
from .containers.main import *
from .ocr.main import *
from .search.main import *
from .realtime_api.main import _arealtime
litellm/constants.py
-5
View File
@@ -175,11 +175,6 @@ OPENAI_FILE_SEARCH_COST_PER_1K_CALLS = float(
AZURE_FILE_SEARCH_COST_PER_GB_PER_DAY = float(
os.getenv("AZURE_FILE_SEARCH_COST_PER_GB_PER_DAY", 0.1) # $0.1 USD per 1 GB/Day
)
AZURE_CODE_INTERPRETER_COST_PER_SESSION = float(
os.getenv(
"AZURE_CODE_INTERPRETER_COST_PER_SESSION", 0.03
) # $0.03 USD per 1 Session
)
AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS = float(
os.getenv(
"AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS", 3.0
litellm/containers/__init__.py
+24
View File
@@ -0,0 +1,24 @@
"""Container management functions for LiteLLM."""
from .main import (
acreate_container,
adelete_container,
alist_containers,
aretrieve_container,
create_container,
delete_container,
list_containers,
retrieve_container,
)
__all__ = [
"acreate_container",
"adelete_container",
"alist_containers",
"aretrieve_container",
"create_container",
"delete_container",
"list_containers",
"retrieve_container",
]
litellm/containers/main.py
+801
View File
@@ -0,0 +1,801 @@
import asyncio
import contextvars
import json
from functools import partial
from typing import Any, Coroutine, Dict, List, Literal, Optional, Union, overload
import litellm
from litellm.constants import request_timeout as DEFAULT_REQUEST_TIMEOUT
from litellm.containers.utils import ContainerRequestUtils
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.containers.transformation import BaseContainerConfig
from litellm.main import base_llm_http_handler
from litellm.types.containers.main import (
ContainerCreateOptionalRequestParams,
ContainerListOptionalRequestParams,
ContainerListResponse,
ContainerObject,
DeleteContainerResult,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import CallTypes
from litellm.utils import ProviderConfigManager, client
__all__ = [
"acreate_container",
"adelete_container",
"alist_containers",
"aretrieve_container",
"create_container",
"delete_container",
"list_containers",
"retrieve_container",
]
##### Container Create #######################
@client
async def acreate_container(
name: str,
expires_after: Optional[Dict[str, Any]] = None,
file_ids: Optional[List[str]] = None,
timeout=600, # default to 10 minutes
# LiteLLM specific params,
custom_llm_provider: Literal["openai"] = "openai",
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
**kwargs,
) -> ContainerObject:
"""Asynchronously calls the `create_container` function with the given arguments and keyword arguments.
Parameters:
- `name` (str): Name of the container to create
- `expires_after` (Optional[Dict[str, Any]]): Container expiration time settings
- `file_ids` (Optional[List[str]]): IDs of files to copy to the container
- `timeout` (int): Request timeout in seconds
- `custom_llm_provider` (Optional[Literal["openai"]]): The LLM provider to use
- `extra_headers` (Optional[Dict[str, Any]]): Additional headers
- `extra_query` (Optional[Dict[str, Any]]): Additional query parameters
- `extra_body` (Optional[Dict[str, Any]]): Additional body parameters
- `kwargs` (dict): Additional keyword arguments
Returns:
- `response` (ContainerObject): The created container object
"""
local_vars = locals()
try:
loop = asyncio.get_event_loop()
kwargs["async_call"] = True
func = partial(
create_container,
name=name,
expires_after=expires_after,
file_ids=file_ids,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
**kwargs,
)
ctx = contextvars.copy_context()
func_with_context = partial(ctx.run, func)
init_response = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
except Exception as e:
raise litellm.exception_type(
model="",
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
# fmt: off
# Overload for when acreate_container=True (returns Coroutine)
@overload
def create_container(
name: str,
expires_after: Optional[Dict[str, Any]] = None,
file_ids: Optional[List[str]] = None,
timeout=600, # default to 10 minutes
api_key: Optional[str] = None,
api_base: Optional[str] = None,
api_version: Optional[str] = None,
custom_llm_provider: Literal["openai"] = "openai",
*,
acreate_container: Literal[True],
**kwargs,
) -> Coroutine[Any, Any, ContainerObject]:
...
@overload
def create_container(
name: str,
expires_after: Optional[Dict[str, Any]] = None,
file_ids: Optional[List[str]] = None,
timeout=600, # default to 10 minutes
api_key: Optional[str] = None,
api_base: Optional[str] = None,
api_version: Optional[str] = None,
custom_llm_provider: Literal["openai"] = "openai",
*,
acreate_container: Literal[False] = False,
**kwargs,
) -> ContainerObject:
...
# fmt: on
@client
def create_container(
name: str,
expires_after: Optional[Dict[str, Any]] = None,
file_ids: Optional[List[str]] = None,
timeout=600, # default to 10 minutes
custom_llm_provider: Literal["openai"] = "openai",
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Union[
ContainerObject,
Coroutine[Any, Any, ContainerObject],
]:
"""Create a container using the OpenAI Container API.
Currently supports OpenAI
Example:
```python
import litellm
response = litellm.create_container(
name="My Container",
custom_llm_provider="openai",
)
print(response)
```
"""
local_vars = locals()
try:
litellm_logging_obj: LiteLLMLoggingObj = kwargs.pop("litellm_logging_obj") # type: ignore
litellm_call_id: Optional[str] = kwargs.get("litellm_call_id")
_is_async = kwargs.pop("async_call", False) is True
# Check for mock response first
mock_response = kwargs.get("mock_response")
if mock_response is not None:
if isinstance(mock_response, str):
mock_response = json.loads(mock_response)
response = ContainerObject(**mock_response)
return response
# get llm provider logic
litellm_params = GenericLiteLLMParams(**kwargs)
# get provider config
container_provider_config: Optional[BaseContainerConfig] = (
ProviderConfigManager.get_provider_container_config(
provider=litellm.LlmProviders(custom_llm_provider),
)
)
if container_provider_config is None:
raise ValueError(f"container operations are not supported for {custom_llm_provider}")
local_vars.update(kwargs)
# Get ContainerCreateOptionalRequestParams with only valid parameters
container_create_optional_params: ContainerCreateOptionalRequestParams = (
ContainerRequestUtils.get_requested_container_create_optional_param(local_vars)
)
# Get optional parameters for the container API
container_create_request_params: Dict = (
ContainerRequestUtils.get_optional_params_container_create(
container_provider_config=container_provider_config,
container_create_optional_params=container_create_optional_params,
)
)
# Pre Call logging
litellm_logging_obj.update_environment_variables(
model="",
optional_params=dict(container_create_request_params),
litellm_params={
"litellm_call_id": litellm_call_id,
**container_create_request_params,
},
custom_llm_provider=custom_llm_provider,
)
# Set the correct call type for container creation
litellm_logging_obj.call_type = CallTypes.create_container.value
return base_llm_http_handler.container_create_handler(
name=name,
container_create_request_params=container_create_request_params,
container_provider_config=container_provider_config,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
extra_headers=extra_headers,
timeout=timeout or DEFAULT_REQUEST_TIMEOUT,
_is_async=_is_async,
)
except Exception as e:
raise litellm.exception_type(
model="",
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
##### Container List #######################
@client
async def alist_containers(
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
timeout=600, # default to 10 minutes
custom_llm_provider: Literal["openai"] = "openai",
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
**kwargs,
) -> ContainerListResponse:
"""Asynchronously list containers.
Parameters:
- `after` (Optional[str]): A cursor for pagination
- `limit` (Optional[int]): Number of items to return (1-100, default 20)
- `order` (Optional[str]): Sort order ('asc' or 'desc', default 'desc')
- `timeout` (int): Request timeout in seconds
- `custom_llm_provider` (Literal["openai"]): The LLM provider to use
- `extra_headers` (Optional[Dict[str, Any]]): Additional headers
- `extra_query` (Optional[Dict[str, Any]]): Additional query parameters
- `extra_body` (Optional[Dict[str, Any]]): Additional body parameters
- `kwargs` (dict): Additional keyword arguments
Returns:
- `response` (ContainerListResponse): The list of containers
"""
local_vars = locals()
try:
loop = asyncio.get_event_loop()
kwargs["async_call"] = True
func = partial(
list_containers,
after=after,
limit=limit,
order=order,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
**kwargs,
)
ctx = contextvars.copy_context()
func_with_context = partial(ctx.run, func)
init_response = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
except Exception as e:
raise litellm.exception_type(
model="",
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
# fmt: off
@overload
def list_containers(
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
timeout=600, # default to 10 minutes
api_key: Optional[str] = None,
api_base: Optional[str] = None,
api_version: Optional[str] = None,
custom_llm_provider: Literal["openai"] = "openai",
*,
alist_containers: Literal[True],
**kwargs,
) -> Coroutine[Any, Any, ContainerListResponse]:
...
@overload
def list_containers(
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
timeout=600, # default to 10 minutes
api_key: Optional[str] = None,
api_base: Optional[str] = None,
api_version: Optional[str] = None,
custom_llm_provider: Literal["openai"] = "openai",
*,
alist_containers: Literal[False] = False,
**kwargs,
) -> ContainerListResponse:
...
# fmt: on
@client
def list_containers(
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
timeout=600, # default to 10 minutes
custom_llm_provider: Literal["openai"] = "openai",
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Union[
ContainerListResponse,
Coroutine[Any, Any, ContainerListResponse],
]:
"""List containers using the OpenAI Container API.
Currently supports OpenAI
"""
local_vars = locals()
try:
litellm_logging_obj: LiteLLMLoggingObj = kwargs.pop("litellm_logging_obj") # type: ignore
litellm_call_id: Optional[str] = kwargs.get("litellm_call_id")
_is_async = kwargs.pop("async_call", False) is True
# Check for mock response first
mock_response = kwargs.get("mock_response")
if mock_response is not None:
if isinstance(mock_response, str):
mock_response = json.loads(mock_response)
response = ContainerListResponse(**mock_response)
return response
# get llm provider logic
litellm_params = GenericLiteLLMParams(**kwargs)
# get provider config
container_provider_config: Optional[BaseContainerConfig] = (
ProviderConfigManager.get_provider_container_config(
provider=litellm.LlmProviders(custom_llm_provider),
)
)
if container_provider_config is None:
raise ValueError(f"Container provider config not found for provider: {custom_llm_provider}")
# Get container list request parameters
container_list_optional_params: ContainerListOptionalRequestParams = (
ContainerRequestUtils.get_requested_container_list_optional_param(local_vars)
)
# Pre Call logging
litellm_logging_obj.update_environment_variables(
model="",
optional_params=dict(container_list_optional_params),
litellm_params={
"litellm_call_id": litellm_call_id,
**container_list_optional_params,
},
custom_llm_provider=custom_llm_provider,
)
# Set the correct call type
litellm_logging_obj.call_type = CallTypes.list_containers.value
return base_llm_http_handler.container_list_handler(
container_provider_config=container_provider_config,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
after=after,
limit=limit,
order=order,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout or DEFAULT_REQUEST_TIMEOUT,
_is_async=_is_async,
)
except Exception as e:
raise litellm.exception_type(
model="",
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
##### Container Retrieve #######################
@client
async def aretrieve_container(
container_id: str,
timeout=600, # default to 10 minutes
custom_llm_provider: Literal["openai"] = "openai",
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
**kwargs,
) -> ContainerObject:
"""Asynchronously retrieve a container.
Parameters:
- `container_id` (str): The ID of the container to retrieve
- `timeout` (int): Request timeout in seconds
- `custom_llm_provider` (Literal["openai"]): The LLM provider to use
- `extra_headers` (Optional[Dict[str, Any]]): Additional headers
- `extra_query` (Optional[Dict[str, Any]]): Additional query parameters
- `extra_body` (Optional[Dict[str, Any]]): Additional body parameters
- `kwargs` (dict): Additional keyword arguments
Returns:
- `response` (ContainerObject): The container object
"""
local_vars = locals()
try:
loop = asyncio.get_event_loop()
kwargs["async_call"] = True
func = partial(
retrieve_container,
container_id=container_id,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
**kwargs,
)
ctx = contextvars.copy_context()
func_with_context = partial(ctx.run, func)
init_response = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
except Exception as e:
raise litellm.exception_type(
model="",
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
# fmt: off
@overload
def retrieve_container(
container_id: str,
timeout=600, # default to 10 minutes
api_key: Optional[str] = None,
api_base: Optional[str] = None,
api_version: Optional[str] = None,
custom_llm_provider: Literal["openai"] = "openai",
*,
aretrieve_container: Literal[True],
**kwargs,
) -> Coroutine[Any, Any, ContainerObject]:
...
@overload
def retrieve_container(
container_id: str,
timeout=600, # default to 10 minutes
api_key: Optional[str] = None,
api_base: Optional[str] = None,
api_version: Optional[str] = None,
custom_llm_provider: Literal["openai"] = "openai",
*,
aretrieve_container: Literal[False] = False,
**kwargs,
) -> ContainerObject:
...
# fmt: on
@client
def retrieve_container(
container_id: str,
timeout=600, # default to 10 minutes
custom_llm_provider: Literal["openai"] = "openai",
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Union[
ContainerObject,
Coroutine[Any, Any, ContainerObject],
]:
"""Retrieve a container using the OpenAI Container API.
Currently supports OpenAI
"""
local_vars = locals()
try:
litellm_logging_obj: LiteLLMLoggingObj = kwargs.pop("litellm_logging_obj") # type: ignore
litellm_call_id: Optional[str] = kwargs.get("litellm_call_id")
_is_async = kwargs.pop("async_call", False) is True
# Check for mock response first
mock_response = kwargs.get("mock_response")
if mock_response is not None:
if isinstance(mock_response, str):
mock_response = json.loads(mock_response)
response = ContainerObject(**mock_response)
return response
# get llm provider logic
litellm_params = GenericLiteLLMParams(**kwargs)
# get provider config
container_provider_config: Optional[BaseContainerConfig] = (
ProviderConfigManager.get_provider_container_config(
provider=litellm.LlmProviders(custom_llm_provider),
)
)
if container_provider_config is None:
raise ValueError(f"Container provider config not found for provider: {custom_llm_provider}")
# Pre Call logging
litellm_logging_obj.update_environment_variables(
model="",
optional_params={},
litellm_params={
"litellm_call_id": litellm_call_id,
},
custom_llm_provider=custom_llm_provider,
)
# Set the correct call type
litellm_logging_obj.call_type = CallTypes.retrieve_container.value
return base_llm_http_handler.container_retrieve_handler(
container_id=container_id,
container_provider_config=container_provider_config,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout or DEFAULT_REQUEST_TIMEOUT,
_is_async=_is_async,
)
except Exception as e:
raise litellm.exception_type(
model="",
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
##### Container Delete #######################
@client
async def adelete_container(
container_id: str,
timeout=600, # default to 10 minutes
custom_llm_provider: Literal["openai"] = "openai",
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
**kwargs,
) -> DeleteContainerResult:
"""Asynchronously delete a container.
Parameters:
- `container_id` (str): The ID of the container to delete
- `timeout` (int): Request timeout in seconds
- `custom_llm_provider` (Literal["openai"]): The LLM provider to use
- `extra_headers` (Optional[Dict[str, Any]]): Additional headers
- `extra_query` (Optional[Dict[str, Any]]): Additional query parameters
- `extra_body` (Optional[Dict[str, Any]]): Additional body parameters
- `kwargs` (dict): Additional keyword arguments
Returns:
- `response` (DeleteContainerResult): The deletion result
"""
local_vars = locals()
try:
loop = asyncio.get_event_loop()
kwargs["async_call"] = True
func = partial(
delete_container,
container_id=container_id,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
**kwargs,
)
ctx = contextvars.copy_context()
func_with_context = partial(ctx.run, func)
init_response = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
except Exception as e:
raise litellm.exception_type(
model="",
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
# fmt: off
@overload
def delete_container(
container_id: str,
timeout=600, # default to 10 minutes
api_key: Optional[str] = None,
api_base: Optional[str] = None,
api_version: Optional[str] = None,
custom_llm_provider: Literal["openai"] = "openai",
*,
adelete_container: Literal[True],
**kwargs,
) -> Coroutine[Any, Any, DeleteContainerResult]:
...
@overload
def delete_container(
container_id: str,
timeout=600, # default to 10 minutes
api_key: Optional[str] = None,
api_base: Optional[str] = None,
api_version: Optional[str] = None,
custom_llm_provider: Literal["openai"] = "openai",
*,
adelete_container: Literal[False] = False,
**kwargs,
) -> DeleteContainerResult:
...
# fmt: on
@client
def delete_container(
container_id: str,
timeout=600, # default to 10 minutes
custom_llm_provider: Literal["openai"] = "openai",
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
extra_body: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Union[
DeleteContainerResult,
Coroutine[Any, Any, DeleteContainerResult],
]:
"""Delete a container using the OpenAI Container API.
Currently supports OpenAI
"""
local_vars = locals()
try:
litellm_logging_obj: LiteLLMLoggingObj = kwargs.pop("litellm_logging_obj") # type: ignore
litellm_call_id: Optional[str] = kwargs.get("litellm_call_id")
_is_async = kwargs.pop("async_call", False) is True
# Check for mock response first
mock_response = kwargs.get("mock_response")
if mock_response is not None:
if isinstance(mock_response, str):
mock_response = json.loads(mock_response)
response = DeleteContainerResult(**mock_response)
return response
# get llm provider logic
litellm_params = GenericLiteLLMParams(**kwargs)
# get provider config
container_provider_config: Optional[BaseContainerConfig] = (
ProviderConfigManager.get_provider_container_config(
provider=litellm.LlmProviders(custom_llm_provider),
)
)
if container_provider_config is None:
raise ValueError(f"Container provider config not found for provider: {custom_llm_provider}")
# Pre Call logging
litellm_logging_obj.update_environment_variables(
model="",
optional_params={},
litellm_params={
"litellm_call_id": litellm_call_id,
},
custom_llm_provider=custom_llm_provider,
)
# Set the correct call type
litellm_logging_obj.call_type = CallTypes.delete_container.value
return base_llm_http_handler.container_delete_handler(
container_id=container_id,
container_provider_config=container_provider_config,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout or DEFAULT_REQUEST_TIMEOUT,
_is_async=_is_async,
)
except Exception as e:
raise litellm.exception_type(
model="",
custom_llm_provider=custom_llm_provider,
original_exception=e,
completion_kwargs=local_vars,
extra_kwargs=kwargs,
)
litellm/containers/utils.py
+67
View File
@@ -0,0 +1,67 @@
from typing import Dict
from litellm.llms.base_llm.containers.transformation import BaseContainerConfig
from litellm.types.containers.main import ContainerCreateOptionalRequestParams, ContainerListOptionalRequestParams
class ContainerRequestUtils:
@staticmethod
def get_requested_container_create_optional_param(
passed_params: dict,
) -> ContainerCreateOptionalRequestParams:
"""Extract only valid container creation parameters from the passed parameters."""
container_create_optional_params = ContainerCreateOptionalRequestParams()
valid_params = [
"expires_after",
"file_ids",
"extra_headers",
"extra_body",
]
for param in valid_params:
if param in passed_params and passed_params[param] is not None:
container_create_optional_params[param] = passed_params[param] # type: ignore
return container_create_optional_params
@staticmethod
def get_optional_params_container_create(
container_provider_config: BaseContainerConfig,
container_create_optional_params: ContainerCreateOptionalRequestParams,
) -> Dict:
"""Get the optional parameters for container creation."""
supported_params = container_provider_config.get_supported_openai_params()
# Filter out unsupported parameters
filtered_params = {
k: v
for k, v in container_create_optional_params.items()
if k in supported_params
}
return container_provider_config.map_openai_params(
container_create_optional_params=filtered_params, # type: ignore
drop_params=False,
)
@staticmethod
def get_requested_container_list_optional_param(
passed_params: dict,
) -> ContainerListOptionalRequestParams:
"""Extract only valid container list parameters from the passed parameters."""
container_list_optional_params = ContainerListOptionalRequestParams()
valid_params = [
"after",
"limit",
"order",
"extra_headers",
"extra_query",
]
for param in valid_params:
if param in passed_params and passed_params[param] is not None:
container_list_optional_params[param] = passed_params[param] # type: ignore
return container_list_optional_params
litellm/litellm_core_utils/litellm_logging.py
+2
View File
@@ -116,6 +116,7 @@ from litellm.types.utils import (
Usage,
)
from litellm.types.videos.main import VideoObject
from litellm.types.containers.main import ContainerObject
from litellm.utils import _get_base_model_from_metadata, executor, print_verbose
from litellm.llms.base_llm.ocr.transformation import OCRResponse
@@ -1624,6 +1625,7 @@ class Logging(LiteLLMLoggingBaseClass):
or isinstance(logging_result, dict)
and logging_result.get("object") == "vector_store.search_results.page"
or isinstance(logging_result, VideoObject)
or isinstance(logging_result, ContainerObject)
or (self.call_type == CallTypes.call_mcp_tool.value)
):
return True
litellm/litellm_core_utils/llm_cost_calc/tool_call_cost_tracking.py
+35 -7
View File
@@ -595,6 +595,31 @@ class StandardBuiltInToolCostTracking:
# OpenAI doesn't charge separately for computer use yet
return 0.0
@staticmethod
def _get_code_interpreter_cost_from_model_map(
provider: str,
) -> Optional[float]:
"""
Get code interpreter cost per session from model cost map.
"""
import litellm
try:
container_model = f"{provider}/container"
model_info = litellm.get_model_info(
model=container_model,
custom_llm_provider=provider
)
model_key = model_info.get("key") if isinstance(model_info, dict) else getattr(model_info, "key", None)
if model_key and model_key in litellm.model_cost:
return litellm.model_cost[model_key].get("code_interpreter_cost_per_session")
except Exception:
pass
return None
@staticmethod
def get_cost_for_code_interpreter(
sessions: Optional[int] = None,
@@ -604,7 +629,8 @@ class StandardBuiltInToolCostTracking:
"""
Calculate cost for code interpreter feature.
Azure: $0.03 USD per session
Azure: $0.03 USD per session (from model cost map)
OpenAI: $0.03 USD per session (from model cost map)
"""
if sessions is None or sessions == 0:
return 0.0
@@ -613,13 +639,15 @@ class StandardBuiltInToolCostTracking:
if model_info and "code_interpreter_cost_per_session" in model_info:
return sessions * model_info["code_interpreter_cost_per_session"]
# Azure pricing for code interpreter
if provider == "azure":
from litellm.constants import AZURE_CODE_INTERPRETER_COST_PER_SESSION
# Try to get cost from model cost map for any provider
if provider:
cost_per_session = StandardBuiltInToolCostTracking._get_code_interpreter_cost_from_model_map(
provider=provider
)
if cost_per_session is not None:
return sessions * cost_per_session
return sessions * AZURE_CODE_INTERPRETER_COST_PER_SESSION
# OpenAI doesn't charge separately for code interpreter yet
return 0.0
@staticmethod
litellm/llms/base_llm/containers/transformation.py
+209
View File
@@ -0,0 +1,209 @@
from __future__ import annotations
import types
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any
import httpx
from litellm.types.containers.main import ContainerCreateOptionalRequestParams
from litellm.types.router import GenericLiteLLMParams
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
from litellm.types.containers.main import (
ContainerListResponse as _ContainerListResponse,
)
from litellm.types.containers.main import (
ContainerObject as _ContainerObject,
)
from litellm.types.containers.main import (
DeleteContainerResult as _DeleteContainerResult,
)
from ..chat.transformation import BaseLLMException as _BaseLLMException
LiteLLMLoggingObj = _LiteLLMLoggingObj
BaseLLMException = _BaseLLMException
ContainerObject = _ContainerObject
DeleteContainerResult = _DeleteContainerResult
ContainerListResponse = _ContainerListResponse
else:
LiteLLMLoggingObj = Any
BaseLLMException = Any
ContainerObject = Any
DeleteContainerResult = Any
ContainerListResponse = Any
class BaseContainerConfig(ABC):
def __init__(self):
pass
@classmethod
def get_config(cls):
return {
k: v
for k, v in cls.__dict__.items()
if not k.startswith("__")
and not k.startswith("_abc")
and not isinstance(
v,
(
types.FunctionType,
types.BuiltinFunctionType,
classmethod,
staticmethod,
),
)
and v is not None
}
@abstractmethod
def get_supported_openai_params(self) -> list:
pass
@abstractmethod
def map_openai_params(
self,
container_create_optional_params: ContainerCreateOptionalRequestParams,
drop_params: bool,
) -> dict:
pass
@abstractmethod
def validate_environment(
self,
headers: dict,
api_key: str | None = None,
) -> dict:
return {}
@abstractmethod
def get_complete_url(
self,
api_base: str | None,
litellm_params: dict,
) -> str:
"""Get the complete url for the request.
OPTIONAL - Some providers need `model` in `api_base`.
"""
if api_base is None:
msg = "api_base is required"
raise ValueError(msg)
return api_base
@abstractmethod
def transform_container_create_request(
self,
name: str,
container_create_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> dict:
"""Transform the container creation request.
Returns:
dict: Request data for container creation.
"""
...
@abstractmethod
def transform_container_create_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ContainerObject:
"""Transform the container creation response."""
...
@abstractmethod
def transform_container_list_request(
self,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: dict[str, Any] | None = None,
) -> tuple[str, dict]:
"""Transform the container list request into a URL and params.
Returns:
tuple[str, dict]: (url, params) for the container list request.
"""
...
@abstractmethod
def transform_container_list_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ContainerListResponse:
"""Transform the container list response."""
...
@abstractmethod
def transform_container_retrieve_request(
self,
container_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[str, dict]:
"""Transform the container retrieve request into a URL and data/params.
Returns:
tuple[str, dict]: (url, params) for the container retrieve request.
"""
...
@abstractmethod
def transform_container_retrieve_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ContainerObject:
"""Transform the container retrieve response."""
...
@abstractmethod
def transform_container_delete_request(
self,
container_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[str, dict]:
"""Transform the container delete request into a URL and data.
Returns:
tuple[str, dict]: (url, data) for the container delete request.
"""
...
@abstractmethod
def transform_container_delete_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> DeleteContainerResult:
"""Transform the container delete response."""
...
def get_error_class(
self,
error_message: str,
status_code: int,
headers: dict | httpx.Headers,
) -> BaseLLMException:
from ..chat.transformation import BaseLLMException
raise BaseLLMException(
status_code=status_code,
message=error_message,
headers=headers,
)
litellm/llms/custom_httpx/llm_http_handler.py
+676
View File
@@ -31,6 +31,7 @@ from litellm.llms.base_llm.audio_transcription.transformation import (
from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
from litellm.llms.base_llm.batches.transformation import BaseBatchesConfig
from litellm.llms.base_llm.chat.transformation import BaseConfig
from litellm.llms.base_llm.containers.transformation import BaseContainerConfig
from litellm.llms.base_llm.embedding.transformation import BaseEmbeddingConfig
from litellm.llms.base_llm.files.transformation import BaseFilesConfig
from litellm.llms.base_llm.google_genai.transformation import (
@@ -60,6 +61,11 @@ from litellm.responses.streaming_iterator import (
ResponsesAPIStreamingIterator,
SyncResponsesAPIStreamingIterator,
)
from litellm.types.containers.main import (
ContainerListResponse,
ContainerObject,
DeleteContainerResult,
)
from litellm.types.llms.anthropic_messages.anthropic_response import (
AnthropicMessagesResponse,
)
@@ -3485,6 +3491,7 @@ class BaseLLMHTTPHandler:
BaseSearchConfig,
BaseTextToSpeechConfig,
"BasePassthroughConfig",
"BaseContainerConfig",
],
):
status_code = getattr(e, "status_code", 500)
@@ -4902,6 +4909,675 @@ class BaseLLMHTTPHandler:
e=e,
provider_config=video_status_provider_config,
)
###### CONTAINER HANDLER ######
def container_create_handler(
self,
name: str,
container_create_request_params: Dict,
container_provider_config: "BaseContainerConfig",
litellm_params: GenericLiteLLMParams,
logging_obj: "LiteLLMLoggingObj",
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Union[float, httpx.Timeout] = 600,
_is_async: bool = False,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
) -> Union["ContainerObject", Coroutine[Any, Any, "ContainerObject"]]:
if _is_async:
# Return the async coroutine if called with _is_async=True
return self.async_container_create_handler(
name=name,
container_create_request_params=container_create_request_params,
container_provider_config=container_provider_config,
litellm_params=litellm_params,
logging_obj=logging_obj,
extra_headers=extra_headers,
timeout=timeout,
client=client,
)
# For sync calls, use sync HTTP client
if client is None or not isinstance(client, HTTPHandler):
sync_httpx_client = _get_httpx_client(
params={"ssl_verify": litellm_params.get("ssl_verify", None)}
)
else:
sync_httpx_client = client
# Validate environment and get headers
headers = container_provider_config.validate_environment(
headers=extra_headers or {},
api_key=litellm_params.get("api_key", None),
)
# Add Content-Type header for JSON requests
headers["Content-Type"] = "application/json"
if extra_headers:
headers.update(extra_headers)
# Get the complete URL for the request
api_base = container_provider_config.get_complete_url(
api_base=litellm_params.get("api_base", None),
litellm_params=dict(litellm_params),
)
# Transform the request using the provider config
data = container_provider_config.transform_container_create_request(
name=name,
container_create_optional_request_params=container_create_request_params,
litellm_params=litellm_params,
headers=headers,
)
## LOGGING
logging_obj.pre_call(
input=name,
api_key="",
additional_args={
"complete_input_dict": data,
"api_base": api_base,
"headers": headers,
},
)
try:
response = sync_httpx_client.post(
url=api_base,
headers=headers,
json=data,
timeout=timeout,
)
return container_provider_config.transform_container_create_response(
raw_response=response,
logging_obj=logging_obj,
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=container_provider_config,
)
async def async_container_create_handler(
self,
name: str,
container_create_request_params: Dict,
container_provider_config: "BaseContainerConfig",
litellm_params: GenericLiteLLMParams,
logging_obj: "LiteLLMLoggingObj",
extra_headers: Optional[Dict[str, Any]] = None,
timeout: Union[float, httpx.Timeout] = 600,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
) -> "ContainerObject":
# For async calls, use async HTTP client
if client is None or not isinstance(client, AsyncHTTPHandler):
async_httpx_client = get_async_httpx_client(
llm_provider=litellm.LlmProviders.OPENAI,
params={"ssl_verify": litellm_params.get("ssl_verify", None)},
)
else:
async_httpx_client = client
# Validate environment and get headers
headers = container_provider_config.validate_environment(
headers=extra_headers or {},
api_key=litellm_params.get("api_key", None),
)
# Add Content-Type header for JSON requests
headers["Content-Type"] = "application/json"
if extra_headers:
headers.update(extra_headers)
# Get the complete URL for the request
api_base = container_provider_config.get_complete_url(
api_base=litellm_params.get("api_base", None),
litellm_params=dict(litellm_params),
)
# Transform the request using the provider config
data = container_provider_config.transform_container_create_request(
name=name,
container_create_optional_request_params=container_create_request_params,
litellm_params=litellm_params,
headers=headers,
)
## LOGGING
logging_obj.pre_call(
input=name,
api_key="",
additional_args={
"complete_input_dict": data,
"api_base": api_base,
"headers": headers,
},
)
try:
response = await async_httpx_client.post(
url=api_base,
headers=headers,
json=data,
timeout=timeout,
)
return container_provider_config.transform_container_create_response(
raw_response=response,
logging_obj=logging_obj,
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=container_provider_config,
)
def container_list_handler(
self,
container_provider_config: "BaseContainerConfig",
litellm_params: GenericLiteLLMParams,
logging_obj: "LiteLLMLoggingObj",
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Union[float, httpx.Timeout] = 600,
_is_async: bool = False,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
) -> Union["ContainerListResponse", Coroutine[Any, Any, "ContainerListResponse"]]:
if _is_async:
# Return the async coroutine if called with _is_async=True
return self.async_container_list_handler(
container_provider_config=container_provider_config,
litellm_params=litellm_params,
logging_obj=logging_obj,
after=after,
limit=limit,
order=order,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout,
client=client,
)
# For sync calls, use sync HTTP client
if client is None or not isinstance(client, HTTPHandler):
sync_httpx_client = _get_httpx_client(
params={"ssl_verify": litellm_params.get("ssl_verify", None)}
)
else:
sync_httpx_client = client
# Validate environment and get headers
headers = container_provider_config.validate_environment(
headers=extra_headers or {},
api_key=litellm_params.get("api_key", None),
)
if extra_headers:
headers.update(extra_headers)
# Get the complete URL for the request
api_base = container_provider_config.get_complete_url(
api_base=litellm_params.get("api_base", None),
litellm_params=dict(litellm_params),
)
# Transform the request using the provider config
url, params = container_provider_config.transform_container_list_request(
api_base=api_base,
litellm_params=litellm_params,
headers=headers,
after=after,
limit=limit,
order=order,
extra_query=extra_query,
)
## LOGGING
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
"params": params,
},
)
try:
response = sync_httpx_client.get(
url=url,
headers=headers,
params=params,
)
return container_provider_config.transform_container_list_response(
raw_response=response,
logging_obj=logging_obj,
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=container_provider_config,
)
async def async_container_list_handler(
self,
container_provider_config: "BaseContainerConfig",
litellm_params: GenericLiteLLMParams,
logging_obj: "LiteLLMLoggingObj",
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Union[float, httpx.Timeout] = 600,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
) -> "ContainerListResponse":
# For async calls, use async HTTP client
if client is None or not isinstance(client, AsyncHTTPHandler):
async_httpx_client = get_async_httpx_client(
llm_provider=litellm.LlmProviders.OPENAI,
params={"ssl_verify": litellm_params.get("ssl_verify", None)},
)
else:
async_httpx_client = client
# Validate environment and get headers
headers = container_provider_config.validate_environment(
headers=extra_headers or {},
api_key=litellm_params.get("api_key", None),
)
if extra_headers:
headers.update(extra_headers)
# Get the complete URL for the request
api_base = container_provider_config.get_complete_url(
api_base=litellm_params.get("api_base", None),
litellm_params=dict(litellm_params),
)
# Transform the request using the provider config
url, params = container_provider_config.transform_container_list_request(
api_base=api_base,
litellm_params=litellm_params,
headers=headers,
after=after,
limit=limit,
order=order,
extra_query=extra_query,
)
## LOGGING
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
"params": params,
},
)
try:
response = await async_httpx_client.get(
url=url,
headers=headers,
params=params,
)
return container_provider_config.transform_container_list_response(
raw_response=response,
logging_obj=logging_obj,
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=container_provider_config,
)
def container_retrieve_handler(
self,
container_id: str,
container_provider_config: "BaseContainerConfig",
litellm_params: GenericLiteLLMParams,
logging_obj: "LiteLLMLoggingObj",
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Union[float, httpx.Timeout] = 600,
_is_async: bool = False,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
) -> Union["ContainerObject", Coroutine[Any, Any, "ContainerObject"]]:
if _is_async:
# Return the async coroutine if called with _is_async=True
return self.async_container_retrieve_handler(
container_id=container_id,
container_provider_config=container_provider_config,
litellm_params=litellm_params,
logging_obj=logging_obj,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout,
client=client,
)
# For sync calls, use sync HTTP client
if client is None or not isinstance(client, HTTPHandler):
sync_httpx_client = _get_httpx_client(
params={"ssl_verify": litellm_params.get("ssl_verify", None)}
)
else:
sync_httpx_client = client
# Validate environment and get headers
headers = container_provider_config.validate_environment(
headers=extra_headers or {},
api_key=litellm_params.get("api_key", None),
)
if extra_headers:
headers.update(extra_headers)
# Get the complete URL for the request
api_base = container_provider_config.get_complete_url(
api_base=litellm_params.get("api_base", None),
litellm_params=dict(litellm_params),
)
# Transform the request using the provider config
url, params = container_provider_config.transform_container_retrieve_request(
container_id=container_id,
api_base=api_base,
litellm_params=litellm_params,
headers=headers,
)
# Add any extra query parameters
if extra_query:
params.update(extra_query)
## LOGGING
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
"params": params,
"container_id": container_id,
},
)
try:
response = sync_httpx_client.get(
url=url,
headers=headers,
params=params,
)
return container_provider_config.transform_container_retrieve_response(
raw_response=response,
logging_obj=logging_obj,
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=container_provider_config,
)
async def async_container_retrieve_handler(
self,
container_id: str,
container_provider_config: "BaseContainerConfig",
litellm_params: GenericLiteLLMParams,
logging_obj: "LiteLLMLoggingObj",
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Union[float, httpx.Timeout] = 600,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
) -> "ContainerObject":
# For async calls, use async HTTP client
if client is None or not isinstance(client, AsyncHTTPHandler):
async_httpx_client = get_async_httpx_client(
llm_provider=litellm.LlmProviders.OPENAI,
params={"ssl_verify": litellm_params.get("ssl_verify", None)},
)
else:
async_httpx_client = client
# Validate environment and get headers
headers = container_provider_config.validate_environment(
headers=extra_headers or {},
api_key=litellm_params.get("api_key", None),
)
if extra_headers:
headers.update(extra_headers)
# Get the complete URL for the request
api_base = container_provider_config.get_complete_url(
api_base=litellm_params.get("api_base", None),
litellm_params=dict(litellm_params),
)
# Transform the request using the provider config
url, params = container_provider_config.transform_container_retrieve_request(
container_id=container_id,
api_base=api_base,
litellm_params=litellm_params,
headers=headers,
)
# Add any extra query parameters
if extra_query:
params.update(extra_query)
## LOGGING
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
"params": params,
"container_id": container_id,
},
)
try:
response = await async_httpx_client.get(
url=url,
headers=headers,
params=params,
)
return container_provider_config.transform_container_retrieve_response(
raw_response=response,
logging_obj=logging_obj,
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=container_provider_config,
)
def container_delete_handler(
self,
container_id: str,
container_provider_config: "BaseContainerConfig",
litellm_params: GenericLiteLLMParams,
logging_obj: "LiteLLMLoggingObj",
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Union[float, httpx.Timeout] = 600,
_is_async: bool = False,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
) -> Union["DeleteContainerResult", Coroutine[Any, Any, "DeleteContainerResult"]]:
if _is_async:
# Return the async coroutine if called with _is_async=True
return self.async_container_delete_handler(
container_id=container_id,
container_provider_config=container_provider_config,
litellm_params=litellm_params,
logging_obj=logging_obj,
extra_headers=extra_headers,
extra_query=extra_query,
timeout=timeout,
client=client,
)
# For sync calls, use sync HTTP client
if client is None or not isinstance(client, HTTPHandler):
sync_httpx_client = _get_httpx_client(
params={"ssl_verify": litellm_params.get("ssl_verify", None)}
)
else:
sync_httpx_client = client
# Validate environment and get headers
headers = container_provider_config.validate_environment(
headers=extra_headers or {},
api_key=litellm_params.get("api_key", None),
)
if extra_headers:
headers.update(extra_headers)
# Get the complete URL for the request
api_base = container_provider_config.get_complete_url(
api_base=litellm_params.get("api_base", None),
litellm_params=dict(litellm_params),
)
# Transform the request using the provider config
url, params = container_provider_config.transform_container_delete_request(
container_id=container_id,
api_base=api_base,
litellm_params=litellm_params,
headers=headers,
)
# Add any extra query parameters
if extra_query:
params.update(extra_query)
## LOGGING
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
"params": params,
"container_id": container_id,
},
)
try:
response = sync_httpx_client.delete(
url=url,
headers=headers,
params=params,
)
return container_provider_config.transform_container_delete_response(
raw_response=response,
logging_obj=logging_obj,
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=container_provider_config,
)
async def async_container_delete_handler(
self,
container_id: str,
container_provider_config: "BaseContainerConfig",
litellm_params: GenericLiteLLMParams,
logging_obj: "LiteLLMLoggingObj",
extra_headers: Optional[Dict[str, Any]] = None,
extra_query: Optional[Dict[str, Any]] = None,
timeout: Union[float, httpx.Timeout] = 600,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
) -> "DeleteContainerResult":
# For async calls, use async HTTP client
if client is None or not isinstance(client, AsyncHTTPHandler):
async_httpx_client = get_async_httpx_client(
llm_provider=litellm.LlmProviders.OPENAI,
params={"ssl_verify": litellm_params.get("ssl_verify", None)},
)
else:
async_httpx_client = client
# Validate environment and get headers
headers = container_provider_config.validate_environment(
headers=extra_headers or {},
api_key=litellm_params.get("api_key", None),
)
if extra_headers:
headers.update(extra_headers)
# Get the complete URL for the request
api_base = container_provider_config.get_complete_url(
api_base=litellm_params.get("api_base", None),
litellm_params=dict(litellm_params),
)
# Transform the request using the provider config
url, params = container_provider_config.transform_container_delete_request(
container_id=container_id,
api_base=api_base,
litellm_params=litellm_params,
headers=headers,
)
# Add any extra query parameters
if extra_query:
params.update(extra_query)
## LOGGING
logging_obj.pre_call(
input="",
api_key="",
additional_args={
"api_base": url,
"headers": headers,
"params": params,
"container_id": container_id,
},
)
try:
response = await async_httpx_client.delete(
url=url,
headers=headers,
params=params,
)
return container_provider_config.transform_container_delete_response(
raw_response=response,
logging_obj=logging_obj,
)
except Exception as e:
raise self._handle_error(
e=e,
provider_config=container_provider_config,
)
###### VECTOR STORE HANDLER ######
async def async_vector_store_search_handler(
litellm/llms/openai/containers/transformation.py
+260
View File
@@ -0,0 +1,260 @@
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
import httpx
import litellm
from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
StandardBuiltInToolCostTracking,
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.containers.main import (
ContainerCreateOptionalRequestParams,
ContainerListResponse,
ContainerObject,
DeleteContainerResult,
)
from litellm.types.router import GenericLiteLLMParams
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
from ...base_llm.chat.transformation import BaseLLMException as _BaseLLMException
from ...base_llm.containers.transformation import BaseContainerConfig as _BaseContainerConfig
LiteLLMLoggingObj = _LiteLLMLoggingObj
BaseContainerConfig = _BaseContainerConfig
BaseLLMException = _BaseLLMException
else:
LiteLLMLoggingObj = Any
BaseContainerConfig = Any
BaseLLMException = Any
class OpenAIContainerConfig(BaseContainerConfig):
"""Configuration class for OpenAI container API.
"""
def __init__(self):
super().__init__()
def get_supported_openai_params(self) -> list:
"""Get the list of supported OpenAI parameters for container API.
"""
return [
"name",
"expires_after",
"file_ids",
"extra_headers",
]
def map_openai_params(
self,
container_create_optional_params: ContainerCreateOptionalRequestParams,
drop_params: bool,
) -> Dict:
"""No mapping applied since inputs are in OpenAI spec already"""
return dict(container_create_optional_params)
def validate_environment(
self,
headers: dict,
api_key: Optional[str] = None,
) -> dict:
api_key = (
api_key
or litellm.api_key
or litellm.openai_key
or get_secret_str("OPENAI_API_KEY")
)
headers.update(
{
"Authorization": f"Bearer {api_key}",
},
)
return headers
def get_complete_url(
self,
api_base: Optional[str],
litellm_params: dict,
) -> str:
"""Get the complete URL for OpenAI container API.
"""
if api_base is None:
api_base = "https://api.openai.com/v1"
return f"{api_base.rstrip('/')}/containers"
def transform_container_create_request(
self,
name: str,
container_create_optional_request_params: Dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Dict:
"""Transform the container creation request for OpenAI API.
"""
# Remove extra_headers from optional params as they're handled separately
container_create_optional_request_params = {
k: v for k, v in container_create_optional_request_params.items()
if k not in ["extra_headers"]
}
# Create the request data
request_dict = {
"name": name,
**container_create_optional_request_params,
}
return request_dict
def transform_container_create_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ContainerObject:
"""Transform the OpenAI container creation response.
"""
response_data = raw_response.json()
# Transform the response data
container_obj = ContainerObject(**response_data) # type: ignore[arg-type]
# Add cost for container creation (OpenAI containers are code interpreter sessions)
# https://platform.openai.com/docs/pricing
# Each container creation is 1 code interpreter session
container_cost = StandardBuiltInToolCostTracking.get_cost_for_code_interpreter(
sessions=1,
provider="openai",
)
if not hasattr(container_obj, "_hidden_params") or container_obj._hidden_params is None:
container_obj._hidden_params = {}
if "additional_headers" not in container_obj._hidden_params:
container_obj._hidden_params["additional_headers"] = {}
container_obj._hidden_params["additional_headers"]["llm_provider-x-litellm-response-cost"] = container_cost
return container_obj
def transform_container_list_request(
self,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
extra_query: Optional[Dict[str, Any]] = None,
) -> Tuple[str, Dict]:
"""Transform the container list request for OpenAI API.
OpenAI API expects the following request:
- GET /v1/containers
"""
# Use the api_base directly for container list
url = api_base
# Prepare query parameters
params = {}
if after is not None:
params["after"] = after
if limit is not None:
params["limit"] = str(limit)
if order is not None:
params["order"] = order
# Add any extra query parameters
if extra_query:
params.update(extra_query)
return url, params
def transform_container_list_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ContainerListResponse:
"""Transform the OpenAI container list response.
"""
response_data = raw_response.json()
# Transform the response data
container_list = ContainerListResponse(**response_data) # type: ignore[arg-type]
return container_list
def transform_container_retrieve_request(
self,
container_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
"""Transform the OpenAI container retrieve request.
"""
# For container retrieve, we just need to construct the URL
url = f"{api_base.rstrip('/')}/{container_id}"
# No additional data needed for GET request
data: Dict[str, Any] = {}
return url, data
def transform_container_retrieve_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ContainerObject:
"""Transform the OpenAI container retrieve response.
"""
response_data = raw_response.json()
# Transform the response data
container_obj = ContainerObject(**response_data) # type: ignore[arg-type]
return container_obj
def transform_container_delete_request(
self,
container_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
"""Transform the container delete request for OpenAI API.
OpenAI API expects the following request:
- DELETE /v1/containers/{container_id}
"""
# Construct the URL for container delete
url = f"{api_base.rstrip('/')}/{container_id}"
# No data needed for DELETE request
data: Dict[str, Any] = {}
return url, data
def transform_container_delete_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> DeleteContainerResult:
"""Transform the OpenAI container delete response.
"""
response_data = raw_response.json()
# Transform the response data
delete_result = DeleteContainerResult(**response_data) # type: ignore[arg-type]
return delete_result
def get_error_class(
self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers],
) -> BaseLLMException:
from ...base_llm.chat.transformation import BaseLLMException
raise BaseLLMException(
status_code=status_code,
message=error_message,
headers=headers,
)
litellm/model_prices_and_context_window_backup.json
+10
View File
@@ -24072,6 +24072,16 @@
"1280x720"
]
},
"openai/container": {
"code_interpreter_cost_per_session": 0.03,
"litellm_provider": "openai",
"mode": "container"
},
"azure/container": {
"code_interpreter_cost_per_session": 0.03,
"litellm_provider": "azure",
"mode": "container"
},
"azure/sora-2": {
"litellm_provider": "azure",
"mode": "video_generation",
litellm/proxy/common_request_processing.py
+8
View File
@@ -321,6 +321,10 @@ class ProxyBaseLLMRequestProcessing:
"avideo_status",
"avideo_content",
"avideo_remix",
"acreate_container",
"alist_containers",
"aretrieve_container",
"adelete_container",
],
version: Optional[str] = None,
user_model: Optional[str] = None,
@@ -419,6 +423,10 @@ class ProxyBaseLLMRequestProcessing:
"avideo_status",
"avideo_content",
"avideo_remix",
"acreate_container",
"alist_containers",
"aretrieve_container",
"adelete_container",
],
proxy_logging_obj: ProxyLogging,
general_settings: dict,
litellm/proxy/container_endpoints/__init__.py
View File
litellm/proxy/container_endpoints/endpoints.py
+406
View File
@@ -0,0 +1,406 @@
#### Container Endpoints #####
from typing import Any, Dict
from fastapi import APIRouter, Depends, Request, Response
from fastapi.responses import ORJSONResponse
from litellm.proxy._types import *
from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth, user_api_key_auth
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
from litellm.proxy.common_utils.openai_endpoint_utils import (
get_custom_llm_provider_from_request_headers,
get_custom_llm_provider_from_request_query,
get_custom_llm_provider_from_request_body,
)
router = APIRouter()
@router.post(
"/v1/containers",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse,
tags=["containers"],
)
@router.post(
"/containers",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse,
tags=["containers"],
)
async def create_container(
request: Request,
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Container creation endpoint for creating new containers.
Follows the OpenAI Containers API spec:
https://platform.openai.com/docs/api-reference/containers
Example:
```bash
curl -X POST "http://localhost:4000/v1/containers" \
-H "Authorization: Bearer sk-1234" \
-H "Content-Type: application/json" \
-d '{
"name": "My Container",
"expires_after": {
"anchor": "last_active_at",
"minutes": 20
}
}'
```
Or specify provider via header:
```bash
curl -X POST "http://localhost:4000/v1/containers" \
-H "Authorization: Bearer sk-1234" \
-H "custom-llm-provider: azure" \
-H "Content-Type: application/json" \
-d '{
"name": "My Container"
}'
```
"""
from litellm.proxy.proxy_server import (
general_settings,
llm_router,
proxy_config,
proxy_logging_obj,
select_data_generator,
user_api_base,
user_max_tokens,
user_model,
user_request_timeout,
user_temperature,
version,
)
# Read request body
data = await _read_request_body(request=request)
# Extract custom_llm_provider using priority chain
# Priority: headers > query params > request body > default
custom_llm_provider = (
get_custom_llm_provider_from_request_headers(request=request)
or get_custom_llm_provider_from_request_query(request=request)
or await get_custom_llm_provider_from_request_body(request=request)
or "openai"
)
# Add custom_llm_provider to data
data["custom_llm_provider"] = custom_llm_provider
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="acreate_container",
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
proxy_config=proxy_config,
select_data_generator=select_data_generator,
model=None,
user_model=user_model,
user_temperature=user_temperature,
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
version=version,
)
except Exception as e:
raise await processor._handle_llm_api_exception(
e=e,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
@router.get(
"/v1/containers",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse,
tags=["containers"],
)
@router.get(
"/containers",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse,
tags=["containers"],
)
async def list_containers(
request: Request,
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Container list endpoint for retrieving a list of containers.
Follows the OpenAI Containers API spec:
https://platform.openai.com/docs/api-reference/containers
Example:
```bash
curl -X GET "http://localhost:4000/v1/containers?limit=20&order=desc" \
-H "Authorization: Bearer sk-1234"
```
Or specify provider via header or query param:
```bash
curl -X GET "http://localhost:4000/v1/containers?custom_llm_provider=azure" \
-H "Authorization: Bearer sk-1234"
```
"""
from litellm.proxy.proxy_server import (
general_settings,
llm_router,
proxy_config,
proxy_logging_obj,
select_data_generator,
user_api_base,
user_max_tokens,
user_model,
user_request_timeout,
user_temperature,
version,
)
# Read query parameters
query_params = dict(request.query_params)
data: Dict[str, Any] = {"query_params": query_params}
# Extract custom_llm_provider using priority chain
custom_llm_provider = (
get_custom_llm_provider_from_request_headers(request=request)
or get_custom_llm_provider_from_request_query(request=request)
or "openai"
)
# Add custom_llm_provider to data
data["custom_llm_provider"] = custom_llm_provider
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="alist_containers",
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
proxy_config=proxy_config,
select_data_generator=select_data_generator,
model=None,
user_model=user_model,
user_temperature=user_temperature,
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
version=version,
)
except Exception as e:
raise await processor._handle_llm_api_exception(
e=e,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
@router.get(
"/v1/containers/{container_id}",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse,
tags=["containers"],
)
@router.get(
"/containers/{container_id}",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse,
tags=["containers"],
)
async def retrieve_container(
request: Request,
container_id: str,
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Container retrieve endpoint for getting details of a specific container.
Follows the OpenAI Containers API spec:
https://platform.openai.com/docs/api-reference/containers
Example:
```bash
curl -X GET "http://localhost:4000/v1/containers/cntr_123" \
-H "Authorization: Bearer sk-1234"
```
Or specify provider via header:
```bash
curl -X GET "http://localhost:4000/v1/containers/cntr_123" \
-H "Authorization: Bearer sk-1234" \
-H "custom-llm-provider: azure"
```
"""
from litellm.proxy.proxy_server import (
general_settings,
llm_router,
proxy_config,
proxy_logging_obj,
select_data_generator,
user_api_base,
user_max_tokens,
user_model,
user_request_timeout,
user_temperature,
version,
)
# Include container_id in request data
data: Dict[str, Any] = {"container_id": container_id}
# Extract custom_llm_provider using priority chain
custom_llm_provider = (
get_custom_llm_provider_from_request_headers(request=request)
or get_custom_llm_provider_from_request_query(request=request)
or "openai"
)
# Add custom_llm_provider to data
data["custom_llm_provider"] = custom_llm_provider
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="aretrieve_container",
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
proxy_config=proxy_config,
select_data_generator=select_data_generator,
model=None,
user_model=user_model,
user_temperature=user_temperature,
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
version=version,
)
except Exception as e:
raise await processor._handle_llm_api_exception(
e=e,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
@router.delete(
"/v1/containers/{container_id}",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse,
tags=["containers"],
)
@router.delete(
"/containers/{container_id}",
dependencies=[Depends(user_api_key_auth)],
response_class=ORJSONResponse,
tags=["containers"],
)
async def delete_container(
request: Request,
container_id: str,
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Container delete endpoint for deleting a specific container.
Follows the OpenAI Containers API spec:
https://platform.openai.com/docs/api-reference/containers
Example:
```bash
curl -X DELETE "http://localhost:4000/v1/containers/cntr_123" \
-H "Authorization: Bearer sk-1234"
```
Or specify provider via header:
```bash
curl -X DELETE "http://localhost:4000/v1/containers/cntr_123" \
-H "Authorization: Bearer sk-1234" \
-H "custom-llm-provider: azure"
```
"""
from litellm.proxy.proxy_server import (
general_settings,
llm_router,
proxy_config,
proxy_logging_obj,
select_data_generator,
user_api_base,
user_max_tokens,
user_model,
user_request_timeout,
user_temperature,
version,
)
# Include container_id in request data
data: Dict[str, Any] = {"container_id": container_id}
# Extract custom_llm_provider using priority chain
custom_llm_provider = (
get_custom_llm_provider_from_request_headers(request=request)
or get_custom_llm_provider_from_request_query(request=request)
or "openai"
)
# Add custom_llm_provider to data
data["custom_llm_provider"] = custom_llm_provider
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="adelete_container",
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
proxy_config=proxy_config,
select_data_generator=select_data_generator,
model=None,
user_model=user_model,
user_temperature=user_temperature,
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
version=version,
)
except Exception as e:
raise await processor._handle_llm_api_exception(
e=e,
user_api_key_dict=user_api_key_dict,
proxy_logging_obj=proxy_logging_obj,
version=version,
)
litellm/proxy/proxy_server.py
+11 -3
View File
@@ -227,6 +227,7 @@ from litellm.proxy.common_utils.openai_endpoint_utils import (
from litellm.proxy.common_utils.proxy_state import ProxyState
from litellm.proxy.common_utils.reset_budget_job import ResetBudgetJob
from litellm.proxy.common_utils.swagger_utils import ERROR_RESPONSES
from litellm.proxy.container_endpoints.endpoints import router as container_router
from litellm.proxy.credential_endpoints.endpoints import router as credential_router
from litellm.proxy.db.db_transaction_queue.spend_log_cleanup import SpendLogCleanup
from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
@@ -269,7 +270,9 @@ from litellm.proxy.management_endpoints.customer_endpoints import (
from litellm.proxy.management_endpoints.internal_user_endpoints import (
router as internal_user_router,
)
from litellm.proxy.management_endpoints.internal_user_endpoints import user_update
from litellm.proxy.management_endpoints.internal_user_endpoints import (
user_update,
)
from litellm.proxy.management_endpoints.key_management_endpoints import (
delete_verification_tokens,
duration_in_seconds,
@@ -320,7 +323,9 @@ from litellm.proxy.ocr_endpoints.endpoints import router as ocr_router
from litellm.proxy.openai_files_endpoints.files_endpoints import (
router as openai_files_router,
)
from litellm.proxy.openai_files_endpoints.files_endpoints import set_files_config
from litellm.proxy.openai_files_endpoints.files_endpoints import (
set_files_config,
)
from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
passthrough_endpoint_router,
)
@@ -407,7 +412,9 @@ from litellm.types.proxy.management_endpoints.ui_sso import (
LiteLLM_UpperboundKeyGenerateParams,
)
from litellm.types.realtime import RealtimeQueryParams
from litellm.types.router import DeploymentTypedDict
from litellm.types.router import (
DeploymentTypedDict,
)
from litellm.types.router import ModelInfo as RouterModelInfo
from litellm.types.router import (
RouterGeneralSettings,
@@ -10102,6 +10109,7 @@ app.include_router(public_endpoints_router)
app.include_router(rerank_router)
app.include_router(ocr_router)
app.include_router(video_router)
app.include_router(container_router)
app.include_router(search_router)
app.include_router(image_router)
app.include_router(fine_tuning_router)
litellm/proxy/route_llm_request.py
+16 -1
View File
@@ -32,6 +32,10 @@ ROUTE_ENDPOINT_MAPPING = {
"avideo_status": "/videos/{video_id}",
"avideo_content": "/videos/{video_id}/content",
"avideo_remix": "/videos/{video_id}/remix",
"acreate_container": "/containers",
"alist_containers": "/containers",
"aretrieve_container": "/containers/{container_id}",
"adelete_container": "/containers/{container_id}",
}
@@ -112,6 +116,10 @@ async def route_request(
"avideo_status",
"avideo_content",
"avideo_remix",
"acreate_container",
"alist_containers",
"aretrieve_container",
"adelete_container",
],
):
"""
@@ -152,6 +160,9 @@ async def route_request(
models = [model.strip() for model in data.pop("model").split(",")]
return llm_router.abatch_completion(models=models, **data)
elif llm_router is not None:
# Skip model-based routing for container operations
if route_type in ["acreate_container", "alist_containers", "aretrieve_container", "adelete_container"]:
return getattr(llm_router, f"{route_type}")(**data)
if route_type in [
"avideo_list",
"avideo_status",
@@ -203,7 +214,11 @@ async def route_request(
"alist_input_items",
"avector_store_create",
"avector_store_search",
"asearch"
"asearch",
"acreate_container",
"alist_containers",
"aretrieve_container",
"adelete_container",
]:
# moderation endpoint does not require `model` parameter
return getattr(llm_router, f"{route_type}")(**data)
litellm/router.py
+39 -1
View File
@@ -919,6 +919,28 @@ class Router:
self.avideo_remix = self.factory_function(avideo_remix, call_type="avideo_remix")
self.video_remix = self.factory_function(video_remix, call_type="video_remix")
# Container routes
#########################################################
from litellm.containers import (
acreate_container,
create_container,
alist_containers,
list_containers,
aretrieve_container,
retrieve_container,
adelete_container,
delete_container,
)
self.acreate_container = self.factory_function(acreate_container, call_type="acreate_container")
self.create_container = self.factory_function(create_container, call_type="create_container")
self.alist_containers = self.factory_function(alist_containers, call_type="alist_containers")
self.list_containers = self.factory_function(list_containers, call_type="list_containers")
self.aretrieve_container = self.factory_function(aretrieve_container, call_type="aretrieve_container")
self.retrieve_container = self.factory_function(retrieve_container, call_type="retrieve_container")
self.adelete_container = self.factory_function(adelete_container, call_type="adelete_container")
self.delete_container = self.factory_function(delete_container, call_type="delete_container")
def validate_fallbacks(self, fallback_param: Optional[List]):
"""
Validate the fallbacks parameter.
@@ -3663,7 +3685,15 @@ class Router:
"avideo_content",
"video_content",
"avideo_remix",
"video_remix"
"video_remix",
"acreate_container",
"create_container",
"alist_containers",
"list_containers",
"aretrieve_container",
"retrieve_container",
"adelete_container",
"delete_container"
] = "assistants",
):
"""
@@ -3687,6 +3717,10 @@ class Router:
"video_status",
"video_content",
"video_remix",
"create_container",
"list_containers",
"retrieve_container",
"delete_container",
):
def sync_wrapper(
@@ -3741,6 +3775,10 @@ class Router:
"avideo_status",
"avideo_content",
"avideo_remix",
"acreate_container",
"alist_containers",
"aretrieve_container",
"adelete_container",
):
return await self._ageneric_api_call_with_fallbacks(
original_function=original_function,
litellm/types/containers/main.py
+122
View File
@@ -0,0 +1,122 @@
from typing import Any, Dict, List, Literal, Optional
from typing_extensions import TypedDict
from pydantic import BaseModel
class ExpiresAfter(BaseModel):
"""Container expiration settings."""
anchor: Literal["last_active_at"]
minutes: int
class ContainerObject(BaseModel):
"""Represents a container object."""
id: str
object: Literal["container"]
created_at: int
status: str
expires_after: Optional[ExpiresAfter] = None
last_active_at: Optional[int] = None
name: Optional[str] = None
_hidden_params: Dict[str, Any] = {}
def __contains__(self, key):
# Define custom behavior for the 'in' operator
return hasattr(self, key)
def get(self, key, default=None):
# Custom .get() method to access attributes with a default value if the attribute doesn't exist
return getattr(self, key, default)
def __getitem__(self, key):
# Allow dictionary-style access to attributes
return getattr(self, key)
def json(self, **kwargs): # type: ignore
try:
return self.model_dump(**kwargs)
except Exception:
# if using pydantic v1
return self.dict()
class DeleteContainerResult(BaseModel):
"""Result of a delete container request."""
id: str
object: Literal["container.deleted"]
deleted: bool
def __contains__(self, key):
return hasattr(self, key)
def get(self, key, default=None):
return getattr(self, key, default)
def __getitem__(self, key):
return getattr(self, key)
def json(self, **kwargs): # type: ignore
try:
return self.model_dump(**kwargs)
except Exception:
return self.dict()
class ContainerListResponse(BaseModel):
"""Response object for list containers request."""
object: Literal["list"]
data: List[ContainerObject]
first_id: Optional[str] = None
last_id: Optional[str] = None
has_more: bool
def __contains__(self, key):
return hasattr(self, key)
def get(self, key, default=None):
return getattr(self, key, default)
def __getitem__(self, key):
return getattr(self, key)
def json(self, **kwargs): # type: ignore
try:
return self.model_dump(**kwargs)
except Exception:
return self.dict()
class ContainerCreateOptionalRequestParams(TypedDict, total=False):
"""
TypedDict for Optional parameters supported by OpenAI's container creation API.
Params here: https://platform.openai.com/docs/api-reference/containers/create
"""
expires_after: Optional[Dict[str, Any]] # ExpiresAfter object
file_ids: Optional[List[str]]
extra_headers: Optional[Dict[str, str]]
extra_body: Optional[Dict[str, str]]
class ContainerCreateRequestParams(ContainerCreateOptionalRequestParams, total=False):
"""
TypedDict for request parameters supported by OpenAI's container creation API.
Params here: https://platform.openai.com/docs/api-reference/containers/create
"""
name: str
class ContainerListOptionalRequestParams(TypedDict, total=False):
"""
TypedDict for Optional parameters supported by OpenAI's container list API.
Params here: https://platform.openai.com/docs/api-reference/containers/list
"""
after: Optional[str]
limit: Optional[int]
order: Optional[str]
extra_headers: Optional[Dict[str, str]]
extra_query: Optional[Dict[str, str]]
litellm/types/utils.py
+13
View File
@@ -298,6 +298,19 @@ class CallTypes(str, Enum):
avideo_retrieve_job = "avideo_retrieve_job"
video_delete = "video_delete"
avideo_delete = "avideo_delete"
#########################################################
# Container Call Types
#########################################################
create_container = "create_container"
acreate_container = "acreate_container"
list_containers = "list_containers"
alist_containers = "alist_containers"
retrieve_container = "retrieve_container"
aretrieve_container = "aretrieve_container"
delete_container = "delete_container"
adelete_container = "adelete_container"
acancel_fine_tuning_job = "acancel_fine_tuning_job"
cancel_fine_tuning_job = "cancel_fine_tuning_job"
alist_fine_tuning_jobs = "alist_fine_tuning_jobs"
litellm/utils.py
+13
View File
@@ -258,6 +258,7 @@ from litellm.llms.base_llm.base_utils import (
from litellm.llms.base_llm.batches.transformation import BaseBatchesConfig
from litellm.llms.base_llm.chat.transformation import BaseConfig
from litellm.llms.base_llm.completion.transformation import BaseTextCompletionConfig
from litellm.llms.base_llm.containers.transformation import BaseContainerConfig
from litellm.llms.base_llm.embedding.transformation import BaseEmbeddingConfig
from litellm.llms.base_llm.files.transformation import BaseFilesConfig
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
@@ -7676,6 +7677,18 @@ class ProviderConfigManager:
return AzureVideoConfig()
return None
@staticmethod
def get_provider_container_config(
provider: LlmProviders,
) -> Optional[BaseContainerConfig]:
if LlmProviders.OPENAI == provider:
from litellm.llms.openai.containers.transformation import (
OpenAIContainerConfig,
)
return OpenAIContainerConfig()
return None
@staticmethod
def get_provider_realtime_config(
model: str,
model_prices_and_context_window.json
+10
View File
@@ -24072,6 +24072,16 @@
"1280x720"
]
},
"openai/container": {
"code_interpreter_cost_per_session": 0.03,
"litellm_provider": "openai",
"mode": "container"
},
"azure/container": {
"code_interpreter_cost_per_session": 0.03,
"litellm_provider": "azure",
"mode": "container"
},
"azure/sora-2": {
"litellm_provider": "azure",
"mode": "video_generation",
tests/test_litellm/containers/__init__.py
View File
tests/test_litellm/containers/test_container_api.py
+363
View File
@@ -0,0 +1,363 @@
import json
import os
import sys
from unittest.mock import MagicMock, patch
import asyncio
import pytest
import httpx
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system path
import litellm
from litellm.types.containers.main import ContainerObject, ContainerListResponse, DeleteContainerResult
from litellm.containers.main import (
create_container, acreate_container,
list_containers, alist_containers,
retrieve_container, aretrieve_container,
delete_container, adelete_container
)
from litellm.llms.openai.containers.transformation import OpenAIContainerConfig
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
class TestContainerAPI:
"""Test suite for container API functionality."""
def test_create_container_basic(self):
"""Test basic container creation functionality."""
# Mock the container creation response
mock_response = ContainerObject(
id="cntr_123456",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Test Container"
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_create_handler.return_value = mock_response
response = create_container(
name="Test Container",
custom_llm_provider="openai"
)
assert isinstance(response, ContainerObject)
assert response.id == "cntr_123456"
assert response.name == "Test Container"
assert response.status == "running"
assert response.object == "container"
def test_create_container_with_expires_after(self):
"""Test container creation with expires_after parameter."""
mock_response = ContainerObject(
id="cntr_789",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 30},
last_active_at=1747857508,
name="Expiring Container"
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_create_handler.return_value = mock_response
response = create_container(
name="Expiring Container",
expires_after={"anchor": "last_active_at", "minutes": 30},
custom_llm_provider="openai"
)
assert response.expires_after.minutes == 30
assert response.expires_after.anchor == "last_active_at"
def test_create_container_with_file_ids(self):
"""Test container creation with file_ids parameter."""
mock_response = ContainerObject(
id="cntr_file_test",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Container with Files"
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_create_handler.return_value = mock_response
response = create_container(
name="Container with Files",
file_ids=["file_123", "file_456"],
custom_llm_provider="openai"
)
assert response.name == "Container with Files"
@pytest.mark.asyncio
async def test_acreate_container_basic(self):
"""Test basic async container creation functionality."""
mock_response = ContainerObject(
id="cntr_async_123",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Async Test Container"
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_create_handler.return_value = mock_response
response = await acreate_container(
name="Async Test Container",
custom_llm_provider="openai"
)
assert isinstance(response, ContainerObject)
assert response.id == "cntr_async_123"
assert response.name == "Async Test Container"
def test_list_containers_basic(self):
"""Test basic container listing functionality."""
mock_response = ContainerListResponse(
object="list",
data=[
ContainerObject(
id="cntr_1",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Container 1"
),
ContainerObject(
id="cntr_2",
object="container",
created_at=1747857600,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 15},
last_active_at=1747857600,
name="Container 2"
)
],
first_id="cntr_1",
last_id="cntr_2",
has_more=False
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_list_handler.return_value = mock_response
response = list_containers(
custom_llm_provider="openai"
)
assert isinstance(response, ContainerListResponse)
assert len(response.data) == 2
assert response.data[0].id == "cntr_1"
assert response.data[1].id == "cntr_2"
assert response.has_more == False
def test_list_containers_with_params(self):
"""Test container listing with parameters."""
mock_response = ContainerListResponse(
object="list",
data=[
ContainerObject(
id="cntr_limited",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Limited Container"
)
],
first_id="cntr_limited",
last_id="cntr_limited",
has_more=True
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_list_handler.return_value = mock_response
response = list_containers(
limit=1,
order="desc",
after="cntr_prev",
custom_llm_provider="openai"
)
assert len(response.data) == 1
assert response.has_more == True
@pytest.mark.asyncio
async def test_alist_containers_basic(self):
"""Test basic async container listing functionality."""
mock_response = ContainerListResponse(
object="list",
data=[
ContainerObject(
id="cntr_async_list",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Async List Container"
)
],
first_id="cntr_async_list",
last_id="cntr_async_list",
has_more=False
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_list_handler.return_value = mock_response
response = await alist_containers(
custom_llm_provider="openai"
)
assert isinstance(response, ContainerListResponse)
assert len(response.data) == 1
def test_retrieve_container_basic(self):
"""Test basic container retrieval functionality."""
container_id = "cntr_retrieve_test"
mock_response = ContainerObject(
id=container_id,
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Retrieved Container"
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_retrieve_handler.return_value = mock_response
response = retrieve_container(
container_id=container_id,
custom_llm_provider="openai"
)
assert isinstance(response, ContainerObject)
assert response.id == container_id
assert response.name == "Retrieved Container"
@pytest.mark.asyncio
async def test_aretrieve_container_basic(self):
"""Test basic async container retrieval functionality."""
container_id = "cntr_async_retrieve"
mock_response = ContainerObject(
id=container_id,
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Async Retrieved Container"
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_retrieve_handler.return_value = mock_response
response = await aretrieve_container(
container_id=container_id,
custom_llm_provider="openai"
)
assert isinstance(response, ContainerObject)
assert response.id == container_id
def test_delete_container_basic(self):
"""Test basic container deletion functionality."""
container_id = "cntr_delete_test"
mock_response = DeleteContainerResult(
id=container_id,
object="container.deleted",
deleted=True
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_delete_handler.return_value = mock_response
response = delete_container(
container_id=container_id,
custom_llm_provider="openai"
)
assert isinstance(response, DeleteContainerResult)
assert response.id == container_id
assert response.deleted == True
assert response.object == "container.deleted"
@pytest.mark.asyncio
async def test_adelete_container_basic(self):
"""Test basic async container deletion functionality."""
container_id = "cntr_async_delete"
mock_response = DeleteContainerResult(
id=container_id,
object="container.deleted",
deleted=True
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_delete_handler.return_value = mock_response
response = await adelete_container(
container_id=container_id,
custom_llm_provider="openai"
)
assert isinstance(response, DeleteContainerResult)
assert response.id == container_id
assert response.deleted == True
def test_create_container_error_handling(self):
"""Test error handling in container creation."""
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_create_handler.side_effect = Exception("API Error")
with pytest.raises(Exception):
create_container(
name="Error Test Container",
custom_llm_provider="openai"
)
def test_container_provider_config_retrieval(self):
"""Test that provider config is retrieved correctly."""
with patch('litellm.containers.main.ProviderConfigManager') as mock_config_manager:
mock_config_manager.get_provider_container_config.return_value = OpenAIContainerConfig()
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_response = ContainerObject(
id="cntr_config_test",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Config Test"
)
mock_handler.container_create_handler.return_value = mock_response
response = create_container(
name="Config Test",
custom_llm_provider="openai"
)
# Verify provider config was requested
mock_config_manager.get_provider_container_config.assert_called_once()
assert response.name == "Config Test"
tests/test_litellm/containers/test_container_integration.py
+396
View File
@@ -0,0 +1,396 @@
import json
import os
import sys
from unittest.mock import MagicMock, patch
import pytest
import httpx
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system path
import litellm
from litellm.types.containers.main import ContainerObject, ContainerListResponse, DeleteContainerResult
from litellm.containers.main import (
create_container, acreate_container,
list_containers, alist_containers,
retrieve_container, aretrieve_container,
delete_container, adelete_container
)
class TestContainerIntegration:
"""Integration tests for the complete container API flow."""
def setup_method(self):
"""Set up test fixtures."""
# Mock environment variable for API key
os.environ["OPENAI_API_KEY"] = "sk-test123"
def teardown_method(self):
"""Clean up after tests."""
if "OPENAI_API_KEY" in os.environ:
del os.environ["OPENAI_API_KEY"]
@patch('litellm.llms.custom_httpx.llm_http_handler.HTTPHandler')
def test_container_create_full_flow(self, mock_http_handler):
"""Test the complete container creation flow with mocked HTTP."""
# Setup mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "cntr_integration_test",
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"last_active_at": 1747857508,
"name": "Integration Test Container"
}
mock_response.status_code = 200
# Mock the HTTP handler
mock_client = MagicMock()
mock_client.post.return_value = mock_response
mock_http_handler.return_value = mock_client
with patch('litellm.llms.custom_httpx.llm_http_handler._get_httpx_client') as mock_get_client:
mock_get_client.return_value = mock_client
# Execute
response = create_container(
name="Integration Test Container",
expires_after={"anchor": "last_active_at", "minutes": 20},
custom_llm_provider="openai"
)
# Verify
assert isinstance(response, ContainerObject)
assert response.id == "cntr_integration_test"
assert response.name == "Integration Test Container"
assert response.status == "running"
@patch('litellm.llms.custom_httpx.llm_http_handler.HTTPHandler')
def test_container_list_full_flow(self, mock_http_handler):
"""Test the complete container listing flow with mocked HTTP."""
# Setup mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"object": "list",
"data": [
{
"id": "cntr_list_1",
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"last_active_at": 1747857508,
"name": "List Container 1"
},
{
"id": "cntr_list_2",
"object": "container",
"created_at": 1747857600,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 15},
"last_active_at": 1747857600,
"name": "List Container 2"
}
],
"first_id": "cntr_list_1",
"last_id": "cntr_list_2",
"has_more": False
}
mock_response.status_code = 200
# Mock the HTTP handler
mock_client = MagicMock()
mock_client.get.return_value = mock_response
mock_http_handler.return_value = mock_client
with patch('litellm.llms.custom_httpx.llm_http_handler._get_httpx_client') as mock_get_client:
mock_get_client.return_value = mock_client
# Execute
response = list_containers(
limit=10,
order="desc",
custom_llm_provider="openai"
)
# Verify
assert isinstance(response, ContainerListResponse)
assert len(response.data) == 2
assert response.data[0].id == "cntr_list_1"
assert response.data[1].id == "cntr_list_2"
assert response.has_more == False
@patch('litellm.llms.custom_httpx.llm_http_handler.HTTPHandler')
def test_container_retrieve_full_flow(self, mock_http_handler):
"""Test the complete container retrieval flow with mocked HTTP."""
container_id = "cntr_retrieve_integration"
# Setup mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"id": container_id,
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"last_active_at": 1747857508,
"name": "Retrieved Integration Container"
}
mock_response.status_code = 200
# Mock the HTTP handler
mock_client = MagicMock()
mock_client.get.return_value = mock_response
mock_http_handler.return_value = mock_client
with patch('litellm.llms.custom_httpx.llm_http_handler._get_httpx_client') as mock_get_client:
mock_get_client.return_value = mock_client
# Execute
response = retrieve_container(
container_id=container_id,
custom_llm_provider="openai"
)
# Verify
assert isinstance(response, ContainerObject)
assert response.id == container_id
assert response.name == "Retrieved Integration Container"
@patch('litellm.llms.custom_httpx.llm_http_handler.HTTPHandler')
def test_container_delete_full_flow(self, mock_http_handler):
"""Test the complete container deletion flow with mocked HTTP."""
container_id = "cntr_delete_integration"
# Setup mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"id": container_id,
"object": "container.deleted",
"deleted": True
}
mock_response.status_code = 200
# Mock the HTTP handler
mock_client = MagicMock()
mock_client.delete.return_value = mock_response
mock_http_handler.return_value = mock_client
with patch('litellm.llms.custom_httpx.llm_http_handler._get_httpx_client') as mock_get_client:
mock_get_client.return_value = mock_client
# Execute
response = delete_container(
container_id=container_id,
custom_llm_provider="openai"
)
# Verify
assert isinstance(response, DeleteContainerResult)
assert response.id == container_id
assert response.deleted == True
assert response.object == "container.deleted"
@pytest.mark.asyncio
@patch('litellm.llms.custom_httpx.llm_http_handler.AsyncHTTPHandler')
async def test_async_container_create_full_flow(self, mock_async_http_handler):
"""Test the complete async container creation flow with mocked HTTP."""
# Setup mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "cntr_async_integration",
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 30},
"last_active_at": 1747857508,
"name": "Async Integration Container"
}
mock_response.status_code = 200
# Mock the async HTTP handler
mock_client = MagicMock()
async def mock_post(*args, **kwargs):
return mock_response
mock_client.post = mock_post
mock_async_http_handler.return_value = mock_client
with patch('litellm.llms.custom_httpx.llm_http_handler.get_async_httpx_client') as mock_get_async_client:
mock_get_async_client.return_value = mock_client
# Execute
response = await acreate_container(
name="Async Integration Container",
expires_after={"anchor": "last_active_at", "minutes": 30},
custom_llm_provider="openai"
)
# Verify
assert isinstance(response, ContainerObject)
assert response.id == "cntr_async_integration"
assert response.name == "Async Integration Container"
@pytest.mark.asyncio
@patch('litellm.llms.custom_httpx.llm_http_handler.AsyncHTTPHandler')
async def test_async_container_list_full_flow(self, mock_async_http_handler):
"""Test the complete async container listing flow with mocked HTTP."""
# Setup mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"object": "list",
"data": [
{
"id": "cntr_async_list",
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 25},
"last_active_at": 1747857508,
"name": "Async List Container"
}
],
"first_id": "cntr_async_list",
"last_id": "cntr_async_list",
"has_more": False
}
mock_response.status_code = 200
# Mock the async HTTP handler
mock_client = MagicMock()
async def mock_get(*args, **kwargs):
return mock_response
mock_client.get = mock_get
mock_async_http_handler.return_value = mock_client
with patch('litellm.llms.custom_httpx.llm_http_handler.get_async_httpx_client') as mock_get_async_client:
mock_get_async_client.return_value = mock_client
# Execute
response = await alist_containers(
limit=5,
custom_llm_provider="openai"
)
# Verify
assert isinstance(response, ContainerListResponse)
assert len(response.data) == 1
assert response.data[0].id == "cntr_async_list"
def test_container_workflow_simulation(self):
"""Test a complete workflow: create -> list -> retrieve -> delete."""
container_id = "cntr_workflow_test"
# Mock all HTTP responses
create_response = MagicMock(spec=httpx.Response)
create_response.json.return_value = {
"id": container_id,
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"last_active_at": 1747857508,
"name": "Workflow Test Container"
}
list_response = MagicMock(spec=httpx.Response)
list_response.json.return_value = {
"object": "list",
"data": [create_response.json.return_value],
"first_id": container_id,
"last_id": container_id,
"has_more": False
}
retrieve_response = create_response # Same as create
delete_response = MagicMock(spec=httpx.Response)
delete_response.json.return_value = {
"id": container_id,
"object": "container.deleted",
"deleted": True
}
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
# Setup different responses for different operations
mock_handler.container_create_handler.return_value = ContainerObject(**create_response.json.return_value)
mock_handler.container_list_handler.return_value = ContainerListResponse(**list_response.json.return_value)
mock_handler.container_retrieve_handler.return_value = ContainerObject(**retrieve_response.json.return_value)
mock_handler.container_delete_handler.return_value = DeleteContainerResult(**delete_response.json.return_value)
# Execute workflow
# 1. Create container
created = create_container(
name="Workflow Test Container",
custom_llm_provider="openai"
)
assert created.id == container_id
# 2. List containers (should include our created one)
containers = list_containers(custom_llm_provider="openai")
assert len(containers.data) == 1
assert containers.data[0].id == container_id
# 3. Retrieve specific container
retrieved = retrieve_container(
container_id=container_id,
custom_llm_provider="openai"
)
assert retrieved.id == container_id
assert retrieved.name == "Workflow Test Container"
# 4. Delete container
deleted = delete_container(
container_id=container_id,
custom_llm_provider="openai"
)
assert deleted.id == container_id
assert deleted.deleted == True
def test_error_handling_integration(self):
"""Test error handling in the integration flow."""
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
# Simulate an API error
mock_handler.container_create_handler.side_effect = litellm.APIError(
status_code=400,
message="API Error occurred",
llm_provider="openai",
model=""
)
with pytest.raises(litellm.APIError):
create_container(
name="Error Test Container",
custom_llm_provider="openai"
)
@pytest.mark.parametrize("provider", ["openai"])
def test_provider_support(self, provider):
"""Test that the container API works with supported providers."""
mock_response = ContainerObject(
id="cntr_provider_test",
object="container",
created_at=1747857508,
status="running",
expires_after={"anchor": "last_active_at", "minutes": 20},
last_active_at=1747857508,
name="Provider Test Container"
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_create_handler.return_value = mock_response
response = create_container(
name="Provider Test Container",
custom_llm_provider=provider
)
assert response.name == "Provider Test Container"
tests/test_litellm/containers/test_container_transformation.py
+370
View File
@@ -0,0 +1,370 @@
import json
import os
import sys
from unittest.mock import MagicMock, patch
import httpx
import pytest
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system path
import litellm
from litellm.llms.openai.containers.transformation import OpenAIContainerConfig
from litellm.llms.base_llm.containers.transformation import BaseContainerConfig
from litellm.types.containers.main import ContainerObject, ContainerListResponse, DeleteContainerResult
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
class TestOpenAIContainerTransformation:
"""Test suite for OpenAI container transformation functionality."""
def setup_method(self):
"""Set up test fixtures."""
self.config = OpenAIContainerConfig()
self.logging_obj = LiteLLMLogging(
model="",
messages=[],
stream=False,
call_type="create_container",
start_time=None,
litellm_call_id="test_call_id",
function_id="test_function_id"
)
def test_get_supported_openai_params(self):
"""Test that supported OpenAI parameters are returned correctly."""
supported_params = self.config.get_supported_openai_params()
# Check that essential container parameters are supported
assert "name" in supported_params
assert "expires_after" in supported_params
assert "file_ids" in supported_params
def test_map_openai_params_basic(self):
"""Test basic parameter mapping for OpenAI."""
from litellm.types.containers.main import ContainerCreateOptionalRequestParams
optional_params = ContainerCreateOptionalRequestParams({
"expires_after": {"anchor": "last_active_at", "minutes": 30},
"file_ids": ["file_1", "file_2"]
})
mapped_params = self.config.map_openai_params(optional_params, drop_params=False)
assert mapped_params["expires_after"]["minutes"] == 30
assert mapped_params["file_ids"] == ["file_1", "file_2"]
def test_validate_environment(self):
"""Test environment validation adds proper headers."""
headers = {}
api_key = "sk-test123"
validated_headers = self.config.validate_environment(
headers=headers,
api_key=api_key
)
assert "Authorization" in validated_headers
assert validated_headers["Authorization"] == f"Bearer {api_key}"
# Note: Content-Type is not added by validate_environment method
def test_get_complete_url(self):
"""Test complete URL generation."""
api_base = "https://api.openai.com/v1"
litellm_params = {}
url = self.config.get_complete_url(
api_base=api_base,
litellm_params=litellm_params
)
assert url == "https://api.openai.com/v1/containers"
def test_get_complete_url_with_custom_base(self):
"""Test complete URL generation with custom API base."""
api_base = "https://custom.openai.com/v1"
litellm_params = {}
url = self.config.get_complete_url(
api_base=api_base,
litellm_params=litellm_params
)
assert url == "https://custom.openai.com/v1/containers"
def test_transform_container_create_request(self):
"""Test container create request transformation."""
from litellm.types.router import GenericLiteLLMParams
litellm_params = GenericLiteLLMParams()
headers = {"Authorization": "Bearer sk-test123"}
name = "Test Container"
container_create_optional_request_params = {
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"file_ids": ["file_123"]
}
data = self.config.transform_container_create_request(
name=name,
container_create_optional_request_params=container_create_optional_request_params,
litellm_params=litellm_params,
headers=headers
)
assert data["name"] == name
assert data["expires_after"] == container_create_optional_request_params["expires_after"]
assert data["file_ids"] == container_create_optional_request_params["file_ids"]
def test_transform_container_create_response(self):
"""Test container create response transformation."""
# Mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "cntr_123456",
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"last_active_at": 1747857508,
"name": "Test Container"
}
container = self.config.transform_container_create_response(
raw_response=mock_response,
logging_obj=self.logging_obj
)
assert isinstance(container, ContainerObject)
assert container.id == "cntr_123456"
assert container.name == "Test Container"
assert container.status == "running"
assert container.object == "container"
def test_transform_container_list_request(self):
"""Test container list request transformation."""
api_base = "https://api.openai.com/v1/containers"
litellm_params = {}
headers = {"Authorization": "Bearer sk-test123"}
after = "cntr_123"
limit = 10
order = "desc"
url, params = self.config.transform_container_list_request(
api_base=api_base,
litellm_params=litellm_params,
headers=headers,
after=after,
limit=limit,
order=order
)
assert url == api_base
assert params["after"] == after
assert params["limit"] == str(limit) # Should be string for query params
assert params["order"] == order
def test_transform_container_list_response(self):
"""Test container list response transformation."""
# Mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"object": "list",
"data": [
{
"id": "cntr_1",
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"last_active_at": 1747857508,
"name": "Container 1"
},
{
"id": "cntr_2",
"object": "container",
"created_at": 1747857600,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 15},
"last_active_at": 1747857600,
"name": "Container 2"
}
],
"first_id": "cntr_1",
"last_id": "cntr_2",
"has_more": False
}
container_list = self.config.transform_container_list_response(
raw_response=mock_response,
logging_obj=self.logging_obj
)
assert isinstance(container_list, ContainerListResponse)
assert len(container_list.data) == 2
assert container_list.first_id == "cntr_1"
assert container_list.last_id == "cntr_2"
assert container_list.has_more == False
def test_transform_container_retrieve_request(self):
"""Test container retrieve request transformation."""
container_id = "cntr_test123"
api_base = "https://api.openai.com/v1/containers"
litellm_params = {}
headers = {"Authorization": "Bearer sk-test123"}
url, params = self.config.transform_container_retrieve_request(
container_id=container_id,
api_base=api_base,
litellm_params=litellm_params,
headers=headers
)
assert url == f"{api_base}/{container_id}"
assert params == {} # No query params for retrieve
def test_transform_container_retrieve_response(self):
"""Test container retrieve response transformation."""
# Mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "cntr_retrieve_123",
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"last_active_at": 1747857508,
"name": "Retrieved Container"
}
container = self.config.transform_container_retrieve_response(
raw_response=mock_response,
logging_obj=self.logging_obj
)
assert isinstance(container, ContainerObject)
assert container.id == "cntr_retrieve_123"
assert container.name == "Retrieved Container"
def test_transform_container_delete_request(self):
"""Test container delete request transformation."""
container_id = "cntr_delete_123"
api_base = "https://api.openai.com/v1/containers"
litellm_params = {}
headers = {"Authorization": "Bearer sk-test123"}
url, params = self.config.transform_container_delete_request(
container_id=container_id,
api_base=api_base,
litellm_params=litellm_params,
headers=headers
)
assert url == f"{api_base}/{container_id}"
assert params == {} # No query params for delete
def test_transform_container_delete_response(self):
"""Test container delete response transformation."""
# Mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "cntr_delete_123",
"object": "container.deleted",
"deleted": True
}
delete_result = self.config.transform_container_delete_response(
raw_response=mock_response,
logging_obj=self.logging_obj
)
assert isinstance(delete_result, DeleteContainerResult)
assert delete_result.id == "cntr_delete_123"
assert delete_result.object == "container.deleted"
assert delete_result.deleted == True
def test_get_error_class(self):
"""Test error class handling."""
import httpx
from litellm.llms.base_llm.chat.transformation import BaseLLMException
with pytest.raises(BaseLLMException) as exc_info:
self.config.get_error_class(
error_message="Test error",
status_code=400,
headers={}
)
assert "Test error" in str(exc_info.value)
def test_transform_with_none_optional_params(self):
"""Test transformation handles None optional parameters correctly."""
from litellm.types.router import GenericLiteLLMParams
litellm_params = GenericLiteLLMParams()
headers = {"Authorization": "Bearer sk-test123"}
name = "Test Container"
container_create_optional_request_params = {
"expires_after": None,
"file_ids": None
}
data = self.config.transform_container_create_request(
name=name,
container_create_optional_request_params=container_create_optional_request_params,
litellm_params=litellm_params,
headers=headers
)
assert data["name"] == name
# None values should be included as None
assert data["expires_after"] is None
assert data["file_ids"] is None
def test_container_create_response_includes_cost(self):
"""Test that container create response includes code interpreter cost calculation."""
# Force use of local model cost map for CI/CD consistency
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import StandardBuiltInToolCostTracking
# Mock HTTP response
mock_response = MagicMock(spec=httpx.Response)
mock_response.json.return_value = {
"id": "cntr_cost_test",
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"last_active_at": 1747857508,
"name": "Cost Test Container"
}
# Transform the response
container = self.config.transform_container_create_response(
raw_response=mock_response,
logging_obj=self.logging_obj
)
# Verify the container object is created
assert isinstance(container, ContainerObject)
assert container.id == "cntr_cost_test"
# Verify that _hidden_params contains cost information
assert hasattr(container, "_hidden_params")
assert container._hidden_params is not None
assert "additional_headers" in container._hidden_params
assert "llm_provider-x-litellm-response-cost" in container._hidden_params["additional_headers"]
# Verify the cost matches expected value for OpenAI code interpreter (1 session)
# OpenAI charges $0.03 per code interpreter session
expected_cost = StandardBuiltInToolCostTracking.get_cost_for_code_interpreter(
sessions=1,
provider="openai"
)
actual_cost = container._hidden_params["additional_headers"]["llm_provider-x-litellm-response-cost"]
assert actual_cost == expected_cost
assert actual_cost == 0.03 # OpenAI code interpreter costs $0.03 per session
tests/test_litellm/containers/test_container_utils.py
+230
View File
@@ -0,0 +1,230 @@
import os
import sys
import pytest
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system path
import litellm
from litellm.containers.utils import ContainerRequestUtils
from litellm.llms.openai.containers.transformation import OpenAIContainerConfig
from litellm.types.containers.main import (
ContainerCreateOptionalRequestParams,
ContainerListOptionalRequestParams
)
class TestContainerRequestUtils:
"""Test suite for container request utilities."""
def test_get_optional_params_container_create_basic(self):
"""Test that optional parameters are correctly processed for container creation."""
# Setup
config = OpenAIContainerConfig()
optional_params = ContainerCreateOptionalRequestParams(
{
"expires_after": {"anchor": "last_active_at", "minutes": 30},
"file_ids": ["file_123", "file_456"]
}
)
# Execute
result = ContainerRequestUtils.get_optional_params_container_create(
container_provider_config=config,
container_create_optional_params=optional_params,
)
# Assert
assert result == optional_params
assert "expires_after" in result
assert result["expires_after"]["minutes"] == 30
assert "file_ids" in result
assert result["file_ids"] == ["file_123", "file_456"]
def test_get_optional_params_container_create_unsupported_param(self):
"""Test that unsupported parameters are filtered out by ContainerCreateOptionalRequestParams."""
# Setup
config = OpenAIContainerConfig()
# ContainerCreateOptionalRequestParams will only accept valid parameters
# so this test verifies the type validation works correctly
valid_params = ContainerCreateOptionalRequestParams(
{
"expires_after": {"anchor": "last_active_at", "minutes": 30},
"file_ids": ["file_123"]
}
)
# Execute - should work fine with valid parameters
result = ContainerRequestUtils.get_optional_params_container_create(
container_provider_config=config,
container_create_optional_params=valid_params,
)
assert result["expires_after"]["minutes"] == 30
assert result["file_ids"] == ["file_123"]
def test_get_requested_container_create_optional_param(self):
"""Test filtering parameters to only include those in ContainerCreateOptionalRequestParams."""
# Setup
params = {
"name": "Test Container", # This should be excluded as it's required
"expires_after": {"anchor": "last_active_at", "minutes": 30},
"file_ids": ["file_123"],
"invalid_param": "value",
"custom_llm_provider": "openai", # This should be excluded
}
# Execute
result = ContainerRequestUtils.get_requested_container_create_optional_param(
params
)
# Assert
assert "expires_after" in result
assert "file_ids" in result
assert "invalid_param" not in result
assert "name" not in result
assert "custom_llm_provider" not in result
assert result["expires_after"]["minutes"] == 30
assert result["file_ids"] == ["file_123"]
def test_get_requested_container_list_optional_param(self):
"""Test filtering parameters for container list requests."""
# Setup
params = {
"after": "cntr_123",
"limit": 10,
"order": "desc",
"invalid_param": "value",
"custom_llm_provider": "openai", # This should be excluded
}
# Execute
result = ContainerRequestUtils.get_requested_container_list_optional_param(
params
)
# Assert
assert "after" in result
assert "limit" in result
assert "order" in result
assert "invalid_param" not in result
assert "custom_llm_provider" not in result
assert result["after"] == "cntr_123"
assert result["limit"] == 10
assert result["order"] == "desc"
def test_get_optional_params_container_create_empty_params(self):
"""Test handling of empty optional parameters."""
# Setup
config = OpenAIContainerConfig()
optional_params = ContainerCreateOptionalRequestParams({})
# Execute
result = ContainerRequestUtils.get_optional_params_container_create(
container_provider_config=config,
container_create_optional_params=optional_params,
)
# Assert
assert result == optional_params
assert len(result) == 0
def test_get_optional_params_container_create_with_none_values(self):
"""Test handling of None values in optional parameters."""
# Setup
config = OpenAIContainerConfig()
optional_params = ContainerCreateOptionalRequestParams(
{
"expires_after": None,
"file_ids": None
}
)
# Execute
result = ContainerRequestUtils.get_optional_params_container_create(
container_provider_config=config,
container_create_optional_params=optional_params,
)
# Assert
assert result == optional_params
assert "expires_after" in result
assert "file_ids" in result
assert result["expires_after"] is None
assert result["file_ids"] is None
def test_get_requested_container_list_optional_param_partial(self):
"""Test filtering with only some list parameters present."""
# Setup
params = {
"limit": 5,
"custom_llm_provider": "openai", # Should be excluded
"timeout": 600, # Should be excluded
}
# Execute
result = ContainerRequestUtils.get_requested_container_list_optional_param(
params
)
# Assert
assert "limit" in result
assert "custom_llm_provider" not in result
assert "timeout" not in result
assert "after" not in result # Not present in input
assert "order" not in result # Not present in input
assert result["limit"] == 5
def test_container_create_optional_params_type_validation(self):
"""Test that ContainerCreateOptionalRequestParams validates types correctly."""
# Test with valid expires_after
valid_params = ContainerCreateOptionalRequestParams(
{
"expires_after": {"anchor": "last_active_at", "minutes": 20},
"file_ids": ["file_1", "file_2"]
}
)
assert valid_params["expires_after"]["anchor"] == "last_active_at"
assert valid_params["expires_after"]["minutes"] == 20
assert valid_params["file_ids"] == ["file_1", "file_2"]
def test_container_list_optional_params_type_validation(self):
"""Test that ContainerListOptionalRequestParams validates types correctly."""
# Test with valid parameters
valid_params = ContainerListOptionalRequestParams(
{
"after": "cntr_123",
"limit": 10,
"order": "desc"
}
)
assert valid_params["after"] == "cntr_123"
assert valid_params["limit"] == 10
assert valid_params["order"] == "desc"
def test_get_optional_params_with_supported_params_check(self):
"""Test that only supported parameters are accepted."""
# Setup
config = OpenAIContainerConfig()
# Get supported params to understand what should be allowed
supported_params = config.get_supported_openai_params()
# Create params with only valid parameters
test_params = {"expires_after": {"anchor": "last_active_at", "minutes": 15}}
optional_params = ContainerCreateOptionalRequestParams(test_params)
# Execute - should work fine with supported params
result = ContainerRequestUtils.get_optional_params_container_create(
container_provider_config=config,
container_create_optional_params=optional_params,
)
assert result["expires_after"]["minutes"] == 15
tests/test_litellm/litellm_core_utils/llm_cost_calc/test_azure_assistant_cost_tracking.py
+25 -5
View File
@@ -7,21 +7,34 @@ Tests cost calculation for Azure's new assistant features:
- Computer Use (token-based pricing)
- Vector Store (storage-based pricing)
"""
import os
import pytest
from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
StandardBuiltInToolCostTracking,
)
from litellm.constants import (
AZURE_FILE_SEARCH_COST_PER_GB_PER_DAY,
AZURE_CODE_INTERPRETER_COST_PER_SESSION,
AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS,
AZURE_COMPUTER_USE_OUTPUT_COST_PER_1K_TOKENS,
AZURE_VECTOR_STORE_COST_PER_GB_PER_DAY,
)
import litellm
class TestAzureAssistantCostTracking:
"""Test suite for Azure assistant features cost tracking."""
@pytest.fixture(autouse=True)
def setup_method(self):
"""Set up test environment to use local model cost map."""
# Force use of local model cost map for CI/CD consistency
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
yield
# Cleanup not strictly necessary but good practice
# Don't delete env var as other tests might need it
def test_azure_file_search_cost_calculation(self):
"""Test Azure file search cost calculation with storage-based pricing."""
@@ -60,7 +73,10 @@ class TestAzureAssistantCostTracking:
sessions=5,
provider="azure",
)
expected_cost = 5 * AZURE_CODE_INTERPRETER_COST_PER_SESSION # $0.15
# Read expected cost from model cost map (azure/container)
azure_container_info = litellm.model_cost.get("azure/container", {})
cost_per_session = azure_container_info.get("code_interpreter_cost_per_session", 0.03)
expected_cost = 5 * cost_per_session # $0.15
assert cost == expected_cost, f"Expected {expected_cost}, got {cost}"
def test_azure_code_interpreter_zero_sessions(self):
@@ -72,12 +88,12 @@ class TestAzureAssistantCostTracking:
assert cost == 0.0, "Should return 0 for zero sessions"
def test_openai_code_interpreter_free(self):
"""Test OpenAI code interpreter has no separate charges."""
"""Test OpenAI code interpreter cost from model cost map."""
cost = StandardBuiltInToolCostTracking.get_cost_for_code_interpreter(
sessions=5,
provider="openai",
)
assert cost == 0.0, "OpenAI should not charge separately for code interpreter"
assert cost == 0.15, "OpenAI code interpreter should return 0.15 based on current implementation"
@pytest.mark.parametrize("input_tokens,output_tokens,expected_cost", [
(1000, 500, 1000/1000 * AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS + 500/1000 * AZURE_COMPUTER_USE_OUTPUT_COST_PER_1K_TOKENS), # $0.009
@@ -180,7 +196,11 @@ class TestAzureAssistantCostTracking:
def test_constants_loaded_correctly(self):
"""Test that Azure pricing constants are loaded with expected values."""
assert AZURE_FILE_SEARCH_COST_PER_GB_PER_DAY == 0.1
assert AZURE_CODE_INTERPRETER_COST_PER_SESSION == 0.03
# Code interpreter cost is now in model cost map
azure_container_info = litellm.model_cost.get("azure/container", {})
assert azure_container_info.get("code_interpreter_cost_per_session") == 0.03
assert AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS == 3.0
assert AZURE_COMPUTER_USE_OUTPUT_COST_PER_1K_TOKENS == 12.0
assert AZURE_VECTOR_STORE_COST_PER_GB_PER_DAY == 0.1
tests/test_litellm/litellm_core_utils/llm_cost_calc/test_tool_call_cost_tracking.py
+4
View File
@@ -225,6 +225,10 @@ def test_azure_assistant_features_integrated_cost_tracking():
"""
Test integrated cost tracking for Azure assistant features.
"""
# Force use of local model cost map for CI/CD consistency
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "azure/gpt-4o"
# Test with multiple Azure assistant features
tests/test_litellm/llms/azure/test_azure_common_utils.py
+8
View File
@@ -562,6 +562,14 @@ async def test_ensure_initialize_azure_sdk_client_always_used(call_type):
):
# Skip video call types as they don't use Azure SDK client initialization
pytest.skip(f"Skipping {call_type.value} because Azure video calls don't use initialize_azure_sdk_client")
elif (
call_type == CallTypes.alist_containers
or call_type == CallTypes.aretrieve_container
or call_type == CallTypes.acreate_container
or call_type == CallTypes.adelete_container
):
# Skip container call types as they're not supported for Azure (only OpenAI)
pytest.skip(f"Skipping {call_type.value} because Azure doesn't support container operations")
# Mock the initialize_azure_sdk_client function
with patch(patch_target) as mock_init_azure:
tests/test_litellm/test_container_router.py
+213
View File
@@ -0,0 +1,213 @@
"""
Test suite for Container API router functionality.
Tests that the router method gets called correctly for container operations.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
import litellm
class TestContainerRouter:
"""Test suite for Container API router functionality"""
def setup_method(self):
"""Setup test fixtures"""
self.container_name = "Test Container"
self.container_id = "cntr_123456789"
@patch("litellm.containers.main.base_llm_http_handler")
def test_create_container_router_call_mock(self, mock_handler):
"""Test that create_container calls the router method with mock response"""
# Setup mock response
mock_response = {
"id": self.container_id,
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {
"anchor": "last_active_at",
"minutes": 20
},
"last_active_at": 1747857508,
"name": self.container_name
}
# Configure the mock handler
mock_handler.container_create_handler.return_value = mock_response
# Call the create_container function with mock response
result = litellm.create_container(
name=self.container_name,
custom_llm_provider="openai",
mock_response=mock_response
)
# Verify the result is a ContainerObject with the expected data
assert result.id == mock_response["id"]
assert result.object == mock_response["object"]
assert result.name == mock_response["name"]
assert result.status == mock_response["status"]
assert result.created_at == mock_response["created_at"]
@patch("litellm.containers.main.base_llm_http_handler")
def test_list_containers_router_call_mock(self, mock_handler):
"""Test that list_containers calls the router method with mock response"""
# Setup mock response
mock_response = {
"object": "list",
"data": [
{
"id": "cntr_123",
"object": "container",
"created_at": 1747857508,
"status": "running",
"name": "Container 1"
},
{
"id": "cntr_456",
"object": "container",
"created_at": 1747857509,
"status": "running",
"name": "Container 2"
}
],
"has_more": False
}
# Configure the mock handler
mock_handler.container_list_handler.return_value = mock_response
# Call the list_containers function with mock response
result = litellm.list_containers(
custom_llm_provider="openai",
mock_response=mock_response
)
# Verify the result is a ContainerListResponse with the expected data
assert result.object == "list"
assert len(result.data) == 2
assert result.data[0].id == "cntr_123"
assert result.data[1].id == "cntr_456"
assert result.has_more is False
@patch("litellm.containers.main.base_llm_http_handler")
def test_retrieve_container_router_call_mock(self, mock_handler):
"""Test that retrieve_container calls the router method with mock response"""
# Setup mock response
mock_response = {
"id": self.container_id,
"object": "container",
"created_at": 1747857508,
"status": "running",
"expires_after": {
"anchor": "last_active_at",
"minutes": 20
},
"last_active_at": 1747857508,
"name": self.container_name
}
# Configure the mock handler
mock_handler.container_retrieve_handler.return_value = mock_response
# Call the retrieve_container function with mock response
result = litellm.retrieve_container(
container_id=self.container_id,
custom_llm_provider="openai",
mock_response=mock_response
)
# Verify the result is a ContainerObject with the expected data
assert result.id == mock_response["id"]
assert result.object == mock_response["object"]
assert result.name == mock_response["name"]
assert result.status == mock_response["status"]
@patch("litellm.containers.main.base_llm_http_handler")
def test_delete_container_router_call_mock(self, mock_handler):
"""Test that delete_container calls the router method with mock response"""
# Setup mock response
mock_response = {
"id": self.container_id,
"object": "container.deleted",
"deleted": True
}
# Configure the mock handler
mock_handler.container_delete_handler.return_value = mock_response
# Call the delete_container function with mock response
result = litellm.delete_container(
container_id=self.container_id,
custom_llm_provider="openai",
mock_response=mock_response
)
# Verify the result is a DeleteContainerResult with the expected data
assert result.id == mock_response["id"]
assert result.object == mock_response["object"]
assert result.deleted is True
@pytest.mark.asyncio
@patch("litellm.containers.main.base_llm_http_handler")
async def test_acreate_container_router_call_mock(self, mock_handler):
"""Test that acreate_container (async) calls the router method with mock response"""
# Setup mock response
mock_response = {
"id": self.container_id,
"object": "container",
"created_at": 1747857508,
"status": "running",
"name": self.container_name
}
# Configure the mock handler
mock_handler.container_create_handler.return_value = mock_response
# Call the async create_container function with mock response
result = await litellm.acreate_container(
name=self.container_name,
custom_llm_provider="openai",
mock_response=mock_response
)
# Verify the result is a ContainerObject with the expected data
assert result.id == mock_response["id"]
assert result.object == mock_response["object"]
assert result.name == mock_response["name"]
assert result.status == mock_response["status"]
@pytest.mark.asyncio
@patch("litellm.containers.main.base_llm_http_handler")
async def test_alist_containers_router_call_mock(self, mock_handler):
"""Test that alist_containers (async) calls the router method with mock response"""
# Setup mock response
mock_response = {
"object": "list",
"data": [
{
"id": "cntr_123",
"object": "container",
"created_at": 1747857508,
"status": "running",
"name": "Container 1"
}
],
"has_more": False
}
# Configure the mock handler
mock_handler.container_list_handler.return_value = mock_response
# Call the async list_containers function with mock response
result = await litellm.alist_containers(
custom_llm_provider="openai",
mock_response=mock_response
)
# Verify the result is a ContainerListResponse with the expected data
assert result.object == "list"
assert len(result.data) == 1
assert result.data[0].id == "cntr_123"
tests/test_litellm/test_utils.py
+2
View File
@@ -548,6 +548,7 @@ def test_aaamodel_prices_and_context_window_json_is_valid():
"input_dbu_cost_per_token": {"type": "number"},
"annotation_cost_per_page": {"type": "number"},
"ocr_cost_per_page": {"type": "number"},
"code_interpreter_cost_per_session": {"type": "number"},
"litellm_provider": {"type": "string"},
"max_audio_length_hours": {"type": "number"},
"max_audio_per_prompt": {"type": "number"},
@@ -569,6 +570,7 @@ def test_aaamodel_prices_and_context_window_json_is_valid():
"audio_transcription",
"chat",
"completion",
"container",
"embedding",
"image_generation",
"video_generation",