Skip to content

ccproxy.streaming.interfaces

ccproxy.streaming.interfaces

Streaming interfaces for provider implementations.

This module defines interfaces that providers can implement to extend streaming functionality without coupling core code to specific providers.

StreamingMetrics

Bases: TypedDict

Standard streaming metrics structure.

IStreamingMetricsCollector

Bases: Protocol

Interface for provider-specific streaming metrics collection.

Providers implement this interface to extract token usage and other metrics from their specific streaming response formats.

process_chunk 

process_chunk(chunk_str)

Process a streaming chunk to extract metrics.

Parameters:

Name Type Description Default
chunk_str str

Raw chunk string from streaming response

 required

Returns:

Type Description
bool

True if this was the final chunk with complete metrics, False otherwise
Source code in ccproxy/streaming/interfaces.py 
29
30
31
32
33
34
35
36
37
38
def process_chunk(self, chunk_str: str) -> bool:
    """Process a streaming chunk to extract metrics.

    Args:
        chunk_str: Raw chunk string from streaming response

    Returns:
        True if this was the final chunk with complete metrics, False otherwise
    """
    ...

process_raw_chunk 

process_raw_chunk(chunk_str)

Process a raw provider chunk before any format conversion.

This method is called with chunks in the provider's native format, before any OpenAI/Anthropic format conversion happens.

Parameters:

Name Type Description Default
chunk_str str

Raw chunk string in provider's native format

 required

Returns:

Type Description
bool

True if this was the final chunk with complete metrics, False otherwise
Source code in ccproxy/streaming/interfaces.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
def process_raw_chunk(self, chunk_str: str) -> bool:
    """Process a raw provider chunk before any format conversion.

    This method is called with chunks in the provider's native format,
    before any OpenAI/Anthropic format conversion happens.

    Args:
        chunk_str: Raw chunk string in provider's native format

    Returns:
        True if this was the final chunk with complete metrics, False otherwise
    """
    ...

process_converted_chunk 

process_converted_chunk(chunk_str)

Process a chunk after format conversion.

This method is called with chunks after they've been converted to a different format (e.g., OpenAI format).

Parameters:

Name Type Description Default
chunk_str str

Chunk string after format conversion

 required

Returns:

Type Description
bool

True if this was the final chunk with complete metrics, False otherwise
Source code in ccproxy/streaming/interfaces.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
def process_converted_chunk(self, chunk_str: str) -> bool:
    """Process a chunk after format conversion.

    This method is called with chunks after they've been converted
    to a different format (e.g., OpenAI format).

    Args:
        chunk_str: Chunk string after format conversion

    Returns:
        True if this was the final chunk with complete metrics, False otherwise
    """
    ...

get_metrics 

get_metrics()

Get the collected metrics.

Returns:

Type Description
StreamingMetrics

Dictionary with provider-specific metrics (tokens, costs, etc.)
Source code in ccproxy/streaming/interfaces.py 
68
69
70
71
72
73
74
def get_metrics(self) -> StreamingMetrics:
    """Get the collected metrics.

    Returns:
        Dictionary with provider-specific metrics (tokens, costs, etc.)
    """
    ...