ccproxy.llms.formatters

ccproxy.llms.formatters

LLM format adapters with typed interfaces.

APIAdapter

Bases: ABC, Generic[RequestType, ResponseType, StreamEventType]

Abstract base class for API format adapters.

Provides strongly-typed interface for converting between different API formats with full type safety and validation.

adapt_request abstractmethod async

adapt_request(request)

Convert a request using strongly-typed Pydantic models.

Parameters:

Name	Type	Description	Default
request	RequestType	The typed request model to convert	required

Returns:

Type	Description
BaseModel	The converted typed request model

Raises:

Type	Description
ValueError	If the request format is invalid or unsupported

Source code in ccproxy/llms/formatters/base.py

@abstractmethod
async def adapt_request(self, request: RequestType) -> BaseModel:
    """Convert a request using strongly-typed Pydantic models.

    Args:
        request: The typed request model to convert

    Returns:
        The converted typed request model

    Raises:
        ValueError: If the request format is invalid or unsupported
    """
    pass

adapt_response abstractmethod async

adapt_response(response)

Convert a response using strongly-typed Pydantic models.

Parameters:

Name	Type	Description	Default
response	ResponseType	The typed response model to convert	required

Returns:

Type	Description
BaseModel	The converted typed response model

Raises:

Type	Description
ValueError	If the response format is invalid or unsupported

Source code in ccproxy/llms/formatters/base.py

@abstractmethod
async def adapt_response(self, response: ResponseType) -> BaseModel:
    """Convert a response using strongly-typed Pydantic models.

    Args:
        response: The typed response model to convert

    Returns:
        The converted typed response model

    Raises:
        ValueError: If the response format is invalid or unsupported
    """
    pass

adapt_stream abstractmethod

adapt_stream(stream)

Convert a streaming response using strongly-typed Pydantic models.

Parameters:

Name	Type	Description	Default
stream	AsyncIterator[StreamEventType]	The typed streaming response data to convert	required

Yields:

Type	Description
AsyncGenerator[BaseModel, None]	The converted typed streaming response chunks

Raises:

Type	Description
ValueError	If the stream format is invalid or unsupported

Source code in ccproxy/llms/formatters/base.py

@abstractmethod
def adapt_stream(
    self, stream: AsyncIterator[StreamEventType]
) -> AsyncGenerator[BaseModel, None]:
    """Convert a streaming response using strongly-typed Pydantic models.

    Args:
        stream: The typed streaming response data to convert

    Yields:
        The converted typed streaming response chunks

    Raises:
        ValueError: If the stream format is invalid or unsupported
    """
    # This should be implemented as an async generator
    # Subclasses must override this method
    ...

adapt_error abstractmethod async

adapt_error(error)

Convert an error response using strongly-typed Pydantic models.

Parameters:

Name	Type	Description	Default
error	BaseModel	The typed error response model to convert	required

Returns:

Type	Description
BaseModel	The converted typed error response model

Raises:

Type	Description
ValueError	If the error format is invalid or unsupported

Source code in ccproxy/llms/formatters/base.py

@abstractmethod
async def adapt_error(self, error: BaseModel) -> BaseModel:
    """Convert an error response using strongly-typed Pydantic models.

    Args:
        error: The typed error response model to convert

    Returns:
        The converted typed error response model

    Raises:
        ValueError: If the error format is invalid or unsupported
    """
    pass

BaseAPIAdapter

BaseAPIAdapter(name)

Bases: APIAdapter[RequestType, ResponseType, StreamEventType], StreamingConfigurable

Base implementation with common functionality.

Provides strongly-typed interface for API format conversion with better type safety and validation.

Source code in ccproxy/llms/formatters/base.py

def __init__(self, name: str):
    self.name = name
    # Optional streaming flags that subclasses may use
    self._openai_thinking_xml: bool | None = None

adapt_request abstractmethod async

adapt_request(request)

Convert a request using strongly-typed Pydantic models.

Source code in ccproxy/llms/formatters/base.py

@abstractmethod
async def adapt_request(self, request: RequestType) -> BaseModel:
    """Convert a request using strongly-typed Pydantic models."""
    pass

adapt_response abstractmethod async

adapt_response(response)

Convert a response using strongly-typed Pydantic models.

Source code in ccproxy/llms/formatters/base.py

@abstractmethod
async def adapt_response(self, response: ResponseType) -> BaseModel:
    """Convert a response using strongly-typed Pydantic models."""
    pass

adapt_stream abstractmethod

adapt_stream(stream)

Convert a streaming response using strongly-typed Pydantic models.

Source code in ccproxy/llms/formatters/base.py

@abstractmethod
def adapt_stream(
    self, stream: AsyncIterator[StreamEventType]
) -> AsyncGenerator[BaseModel, None]:
    """Convert a streaming response using strongly-typed Pydantic models."""
    # This should be implemented as an async generator
    # Subclasses must override this method
    ...

adapt_error abstractmethod async

adapt_error(error)

Convert an error response using strongly-typed Pydantic models.

Source code in ccproxy/llms/formatters/base.py

@abstractmethod
async def adapt_error(self, error: BaseModel) -> BaseModel:
    """Convert an error response using strongly-typed Pydantic models."""
    pass

LlmBaseModel

Bases: BaseModel

Base model for all LLM API models with proper JSON serialization.

Excludes None values and empty collections to match API conventions.

model_dump

model_dump(**kwargs)

Override to exclude empty collections as well as None values.

Source code in ccproxy/llms/formatters/base_model.py

def model_dump(self, **kwargs: Any) -> dict[str, Any]:
    """Override to exclude empty collections as well as None values."""
    # Extract exclude_none from kwargs, defaulting to True for our convention
    exclude_none = kwargs.pop("exclude_none", True)
    # First get the data with None values excluded
    data = super().model_dump(exclude_none=exclude_none, **kwargs)

    # Filter out empty collections (lists, dicts, sets)
    filtered_data = {}
    for key, value in data.items():
        if isinstance(value, list | dict | set) and len(value) == 0:
            # Skip empty collections
            continue
        filtered_data[key] = value

    return filtered_data