ccproxy.llms.formatters.base

ccproxy.llms.formatters.base

Base adapter interface for API format conversion.

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