Implementation:BerriAI Litellm Endpoint Factory

Attribute	Value
Sources	litellm/containers/endpoint_factory.py
Domains	Containers, SDK Generation, Endpoint Factory, Code Generation
Last Updated	2026-02-15 16:00 GMT

Overview

The endpoint_factory module dynamically generates synchronous and asynchronous SDK functions for container file operations from a JSON configuration file, using the generic container handler pattern.

Description

This module reads an endpoints.json configuration file and dynamically generates SDK functions for container file management operations. The factory pattern produces both sync and async variants of each endpoint:

• create_sync_endpoint_function - Generates a synchronous SDK function from an endpoint config dict. The generated function is decorated with @client, handles mock responses, resolves provider configuration via ProviderConfigManager, sets up logging, and delegates to generic_container_handler.handle.

• create_async_endpoint_function - Wraps a sync function in an async version that uses asyncio.get_event_loop().run_in_executor with context variable propagation via contextvars.copy_context.

• generate_container_endpoints - Orchestrates generation of all endpoints from the JSON config, returning a dictionary mapping function names to implementations.

• get_all_endpoint_names / get_async_endpoint_names - Utility functions for retrieving endpoint names for registration.

The module auto-generates and exports 10 functions on import: list_container_files, upload_container_file, retrieve_container_file, delete_container_file, retrieve_container_file_content, and their async counterparts (prefixed with a).

Usage

Import the generated container endpoint functions to interact with container file storage. These functions follow the same patterns as other LiteLLM SDK functions with support for custom providers, logging, and error handling.

Code Reference

Source Location

litellm/containers/endpoint_factory.py

Signature

def _load_endpoints_config() -> Dict
def create_sync_endpoint_function(endpoint_config: Dict) -> Callable
def create_async_endpoint_function(sync_func: Callable, endpoint_config: Dict) -> Callable
def generate_container_endpoints() -> Dict[str, Callable]
def get_all_endpoint_names() -> List[str]
def get_async_endpoint_names() -> List[str]

# Auto-generated exports:
list_container_files = _generated_endpoints.get("list_container_files")
alist_container_files = _generated_endpoints.get("alist_container_files")
upload_container_file = _generated_endpoints.get("upload_container_file")
aupload_container_file = _generated_endpoints.get("aupload_container_file")
retrieve_container_file = _generated_endpoints.get("retrieve_container_file")
aretrieve_container_file = _generated_endpoints.get("aretrieve_container_file")
delete_container_file = _generated_endpoints.get("delete_container_file")
adelete_container_file = _generated_endpoints.get("adelete_container_file")
retrieve_container_file_content = _generated_endpoints.get("retrieve_container_file_content")
aretrieve_container_file_content = _generated_endpoints.get("aretrieve_container_file_content")

Import

from litellm.containers.endpoint_factory import (
    list_container_files,
    alist_container_files,
    upload_container_file,
    aupload_container_file,
    retrieve_container_file,
    aretrieve_container_file,
    delete_container_file,
    adelete_container_file,
    retrieve_container_file_content,
    aretrieve_container_file_content,
)

I/O Contract

Inputs (Generated Functions)

Parameter	Type	Description
timeout	int	Request timeout in seconds. Defaults to 600.
custom_llm_provider	Literal["openai"]	The container provider. Defaults to "openai".
extra_headers	Optional[Dict[str, Any]]	Additional HTTP headers.
extra_query	Optional[Dict[str, Any]]	Additional query parameters.
extra_body	Optional[Dict[str, Any]]	Additional request body fields.
mock_response (kwarg)	Optional[str]	A JSON string to return as mock response for testing.
**kwargs	various	Path parameters and additional arguments specific to the endpoint.

Outputs

Function	Return Type	Description
list_container_files	ContainerFileListResponse	List of container files.
upload_container_file	ContainerFileObject	Uploaded file metadata.
retrieve_container_file	ContainerFileObject	Retrieved file metadata.
delete_container_file	DeleteContainerFileResponse	Deletion confirmation.
retrieve_container_file_content	varies	The file content.

Usage Examples

from litellm.containers.endpoint_factory import list_container_files, alist_container_files

# Sync: list container files
files = list_container_files(custom_llm_provider="openai", timeout=30)

# Async: list container files
import asyncio
async def main():
    files = await alist_container_files(custom_llm_provider="openai", timeout=30)
    return files

result = asyncio.run(main())

Related Pages

• BerriAI_Litellm_LLM_Base_Class - Base LLM class pattern used by container providers
• BerriAI_Litellm_Custom_LLM_Handler - Custom LLM handler for provider implementations