Skip to content

ccproxy.testing.response_handlers

ccproxy.testing.response_handlers

Response processing utilities for testing.

ResponseHandler

Handle responses from both Anthropic and OpenAI formats.

process_response 

process_response(response, scenario)

Process response based on format and streaming.

Source code in ccproxy/testing/response_handlers.py 
14
15
16
17
18
19
20
21
22
def process_response(
    self, response: httpx.Response, scenario: RequestScenario
) -> dict[str, Any]:
    """Process response based on format and streaming."""

    if scenario.streaming:
        return self._process_streaming_response(response, scenario)
    else:
        return self._process_standard_response(response, scenario)

MetricsExtractor

Extract metrics from API responses.

extract_token_metrics staticmethod 

extract_token_metrics(response_data, api_format)

Extract token usage from response data.

Source code in ccproxy/testing/response_handlers.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@staticmethod
def extract_token_metrics(
    response_data: dict[str, Any], api_format: str
) -> dict[str, int | None]:
    """Extract token usage from response data."""
    if api_format == "openai":
        usage = response_data.get("usage", {})
        return {
            "input_tokens": usage.get("prompt_tokens"),
            "output_tokens": usage.get("completion_tokens"),
            "cache_read_tokens": None,  # OpenAI doesn't expose cache metrics
            "cache_write_tokens": None,
        }
    else:  # anthropic
        usage = response_data.get("usage", {})
        return {
            "input_tokens": usage.get("input_tokens"),
            "output_tokens": usage.get("output_tokens"),
            "cache_read_tokens": usage.get("cache_read_input_tokens"),
            "cache_write_tokens": usage.get("cache_creation_input_tokens"),
        }

extract_content staticmethod 

extract_content(response_data, api_format)

Extract text content from response data.

Source code in ccproxy/testing/response_handlers.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
@staticmethod
def extract_content(response_data: dict[str, Any], api_format: str) -> str:
    """Extract text content from response data."""
    if api_format == "openai":
        content = (
            response_data.get("choices", [{}])[0]
            .get("message", {})
            .get("content", "")
        )
        return content if isinstance(content, str) else ""
    else:  # anthropic
        content = ""
        for block in response_data.get("content", []):
            if block.get("type") == "text":
                text = block.get("text", "")
                content += text if isinstance(text, str) else ""
        return content