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, ContainerFileListResponse, ContainerFileObject, ContainerListOptionalRequestParams, ContainerListResponse, ContainerObject, DeleteContainerResult, ) from litellm.types.llms.openai import FileTypes from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import CallTypes from litellm.utils import ProviderConfigManager, client __all__ = [ "acreate_container", "adelete_container", "alist_container_files", "alist_containers", "aretrieve_container", "aupload_container_file", "create_container", "delete_container", "list_container_files", "list_containers", "retrieve_container", "upload_container_file", ] ##### 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 api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, 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 # Pass credential params explicitly since they're named args, not in kwargs litellm_params = GenericLiteLLMParams( api_key=api_key, api_base=api_base, api_version=api_version, **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 api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, 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 # Pass credential params explicitly since they're named args, not in kwargs litellm_params = GenericLiteLLMParams( api_key=api_key, api_base=api_base, api_version=api_version, **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 api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, 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 # Pass credential params explicitly since they're named args, not in kwargs litellm_params = GenericLiteLLMParams( api_key=api_key, api_base=api_base, api_version=api_version, **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 api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, 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 # Pass credential params explicitly since they're named args, not in kwargs litellm_params = GenericLiteLLMParams( api_key=api_key, api_base=api_base, api_version=api_version, **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, ) ##### Container Files List ####################### @client async def alist_container_files( container_id: str, after: Optional[str] = None, limit: Optional[int] = None, order: Optional[str] = None, timeout=600, # default to 10 minutes custom_llm_provider: Literal["openai"] = "openai", extra_headers: Optional[Dict[str, Any]] = None, extra_query: Optional[Dict[str, Any]] = None, extra_body: Optional[Dict[str, Any]] = None, **kwargs, ) -> ContainerFileListResponse: """Asynchronously list files in a container. Parameters: - `container_id` (str): The ID of the container - `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` (ContainerFileListResponse): The list of container files """ local_vars = locals() try: loop = asyncio.get_event_loop() kwargs["async_call"] = True func = partial( list_container_files, container_id=container_id, 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_container_files( container_id: str, after: Optional[str] = None, limit: Optional[int] = None, order: Optional[str] = None, timeout=600, api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, custom_llm_provider: Literal["openai"] = "openai", *, alist_container_files: Literal[True], **kwargs, ) -> Coroutine[Any, Any, ContainerFileListResponse]: ... @overload def list_container_files( container_id: str, after: Optional[str] = None, limit: Optional[int] = None, order: Optional[str] = None, timeout=600, api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, custom_llm_provider: Literal["openai"] = "openai", *, alist_container_files: Literal[False] = False, **kwargs, ) -> ContainerFileListResponse: ... # fmt: on @client def list_container_files( container_id: str, 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", extra_headers: Optional[Dict[str, Any]] = None, extra_query: Optional[Dict[str, Any]] = None, extra_body: Optional[Dict[str, Any]] = None, **kwargs, ) -> Union[ ContainerFileListResponse, Coroutine[Any, Any, ContainerFileListResponse], ]: """List files in 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 = ContainerFileListResponse(**mock_response) return response # get llm provider logic # Pass credential params explicitly since they're named args, not in kwargs litellm_params = GenericLiteLLMParams( api_key=api_key, api_base=api_base, api_version=api_version, **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={"container_id": container_id, "after": after, "limit": limit, "order": order}, 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.list_container_files.value return base_llm_http_handler.container_file_list_handler( container_id=container_id, 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 File Upload ####################### @client async def aupload_container_file( container_id: str, file: FileTypes, timeout=600, # default to 10 minutes custom_llm_provider: Literal["openai"] = "openai", extra_headers: Optional[Dict[str, Any]] = None, extra_query: Optional[Dict[str, Any]] = None, extra_body: Optional[Dict[str, Any]] = None, **kwargs, ) -> ContainerFileObject: """Asynchronously upload a file to a container. This endpoint allows uploading files directly to a container session, supporting various file types like CSV, Excel, Python scripts, etc. Parameters: - `container_id` (str): The ID of the container to upload the file to - `file` (FileTypes): The file to upload. Can be: - A tuple of (filename, content, content_type) - A tuple of (filename, content) - A file-like object with read() method - Bytes - A string path to a file - `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` (ContainerFileObject): The uploaded file object Example: ```python import litellm # Upload a CSV file response = await litellm.aupload_container_file( container_id="container_abc123", file=("data.csv", open("data.csv", "rb").read(), "text/csv"), custom_llm_provider="openai", ) print(response) ``` """ local_vars = locals() try: loop = asyncio.get_event_loop() kwargs["async_call"] = True func = partial( upload_container_file, container_id=container_id, file=file, 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 upload_container_file( container_id: str, file: FileTypes, timeout=600, api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, custom_llm_provider: Literal["openai"] = "openai", *, aupload_container_file: Literal[True], **kwargs, ) -> Coroutine[Any, Any, ContainerFileObject]: ... @overload def upload_container_file( container_id: str, file: FileTypes, timeout=600, api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, custom_llm_provider: Literal["openai"] = "openai", *, aupload_container_file: Literal[False] = False, **kwargs, ) -> ContainerFileObject: ... # fmt: on @client def upload_container_file( container_id: str, file: FileTypes, 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", extra_headers: Optional[Dict[str, Any]] = None, extra_query: Optional[Dict[str, Any]] = None, extra_body: Optional[Dict[str, Any]] = None, **kwargs, ) -> Union[ ContainerFileObject, Coroutine[Any, Any, ContainerFileObject], ]: """Upload a file to a container using the OpenAI Container API. This endpoint allows uploading files directly to a container session, supporting various file types like CSV, Excel, Python scripts, JSON, etc. This is useful when /chat/completions or /responses sends files to the container but the input file type is limited to PDF. This endpoint lets you work with other file types. Currently supports OpenAI Example: ```python import litellm # Upload a CSV file response = litellm.upload_container_file( container_id="container_abc123", file=("data.csv", open("data.csv", "rb").read(), "text/csv"), custom_llm_provider="openai", ) print(response) # Upload a Python script response = litellm.upload_container_file( container_id="container_abc123", file=("script.py", b"print('hello world')", "text/x-python"), custom_llm_provider="openai", ) print(response) ``` """ from litellm.llms.custom_httpx.container_handler import generic_container_handler 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 = ContainerFileObject(**mock_response) return response # get llm provider logic # Pass credential params explicitly since they're named args, not in kwargs litellm_params = GenericLiteLLMParams( api_key=api_key, api_base=api_base, api_version=api_version, **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={"container_id": container_id}, 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.upload_container_file.value return generic_container_handler.handle( endpoint_name="upload_container_file", 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, container_id=container_id, file=file, ) 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, )