ccproxy.core.interfaces

ccproxy.core.interfaces

Core interfaces and abstract base classes for the CCProxy API.

This module consolidates all abstract interfaces used throughout the application, providing a single location for defining contracts and protocols.

RequestTransformer

Bases: ABC

Abstract interface for request transformers.

transform_request abstractmethod async

transform_request(request)

Transform a request from one format to another.

Parameters:

Name	Type	Description	Default
request	dict[str, Any]	The request data to transform	required

Returns:

Type	Description
dict[str, Any]	The transformed request data

Raises:

Type	Description
ValueError	If the request format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def transform_request(self, request: dict[str, Any]) -> dict[str, Any]:
    """Transform a request from one format to another.

    Args:
        request: The request data to transform

    Returns:
        The transformed request data

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

ResponseTransformer

Bases: ABC

Abstract interface for response transformers.

transform_response abstractmethod async

transform_response(response)

Transform a response from one format to another.

Parameters:

Name	Type	Description	Default
response	dict[str, Any]	The response data to transform	required

Returns:

Type	Description
dict[str, Any]	The transformed response data

Raises:

Type	Description
ValueError	If the response format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def transform_response(self, response: dict[str, Any]) -> dict[str, Any]:
    """Transform a response from one format to another.

    Args:
        response: The response data to transform

    Returns:
        The transformed response data

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

StreamTransformer

Bases: ABC

Abstract interface for stream transformers.

transform_stream abstractmethod async

transform_stream(stream)

Transform a streaming response from one format to another.

Parameters:

Name	Type	Description	Default
stream	AsyncIterator[dict[str, Any]]	The streaming response data to transform	required

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	The transformed streaming response chunks

Raises:

Type	Description
ValueError	If the stream format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def transform_stream(
    self, stream: AsyncIterator[dict[str, Any]]
) -> AsyncIterator[dict[str, Any]]:
    """Transform a streaming response from one format to another.

    Args:
        stream: The streaming response data to transform

    Yields:
        The transformed streaming response chunks

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

TransformerProtocol

Bases: Protocol[T, R]

Protocol defining the transformer interface.

transform async

transform(data, context=None)

Transform the input data.

Source code in ccproxy/core/interfaces.py

async def transform(self, data: T, context: TransformContext | None = None) -> R:
    """Transform the input data."""
    ...

MetricExporter

Bases: ABC

Abstract interface for exporting metrics to external systems.

export_metrics abstractmethod async

export_metrics(metrics)

Export metrics to the target system.

Parameters:

Name	Type	Description	Default
metrics	dict[str, Any]	Dictionary of metrics to export	required

Returns:

Type	Description
bool	True if export was successful, False otherwise

Raises:

Type	Description
ConnectionError	If unable to connect to the metrics backend
ValueError	If metrics format is invalid

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def export_metrics(self, metrics: dict[str, Any]) -> bool:
    """Export metrics to the target system.

    Args:
        metrics: Dictionary of metrics to export

    Returns:
        True if export was successful, False otherwise

    Raises:
        ConnectionError: If unable to connect to the metrics backend
        ValueError: If metrics format is invalid
    """
    pass

StreamingConfigurable

Bases: Protocol

Protocol for adapters that accept streaming-related configuration.

Implementers can use this to receive DI-injected toggles such as whether to serialize thinking content as XML in OpenAI streams.

configure_streaming

configure_streaming(*, openai_thinking_xml=None)

Apply streaming flags.

Parameters:

Name	Type	Description	Default
openai_thinking_xml	bool | None	Enable/disable thinking-as-XML in OpenAI streams	None

Source code in ccproxy/core/interfaces.py

def configure_streaming(self, *, openai_thinking_xml: bool | None = None) -> None:
    """Apply streaming flags.

    Args:
        openai_thinking_xml: Enable/disable thinking-as-XML in OpenAI streams
    """
    ...