Skip to content

ccproxy.services.adapters.protocols

ccproxy.services.adapters.protocols

Protocol definitions for streaming adapters.

StreamAccumulatorProtocol

Bases: Protocol

Protocol for stream accumulation strategies.

Implementations should accumulate streaming chunks and return complete objects when they are ready for validation. This allows handling providers that send data in partial chunks (like tool calls across multiple chunks).

accumulate_chunk 

accumulate_chunk(chunk)

Accumulate a streaming chunk and return complete object if ready.

Parameters:

Name Type Description Default
chunk dict[str, Any]

The incoming stream chunk data

 required

Returns:

Type Description
dict[str, Any] | None

None if accumulation is ongoing, or the complete object when ready
dict[str, Any] | None

for validation. The complete object should have all required fields
dict[str, Any] | None

properly populated.
Source code in ccproxy/services/adapters/protocols.py 
16
17
18
19
20
21
22
23
24
25
26
27
def accumulate_chunk(self, chunk: dict[str, Any]) -> dict[str, Any] | None:
    """Accumulate a streaming chunk and return complete object if ready.

    Args:
        chunk: The incoming stream chunk data

    Returns:
        None if accumulation is ongoing, or the complete object when ready
        for validation. The complete object should have all required fields
        properly populated.
    """
    ...

reset 

reset()

Reset accumulator state for next message.

Called when starting a new message or when an error occurs to ensure clean state for the next accumulation cycle.

Source code in ccproxy/services/adapters/protocols.py 
29
30
31
32
33
34
35
def reset(self) -> None:
    """Reset accumulator state for next message.

    Called when starting a new message or when an error occurs to ensure
    clean state for the next accumulation cycle.
    """
    ...