ccproxy.core.http_transformers

ccproxy.core.http_transformers

HTTP-level transformers for proxy service.

RequestData

Bases: TypedDict

Typed structure for transformed request data.

ResponseData

Bases: TypedDict

Typed structure for transformed response data.

HTTPRequestTransformer

HTTPRequestTransformer()

Bases: RequestTransformer

HTTP request transformer that implements the abstract RequestTransformer interface.

Source code in ccproxy/core/http_transformers.py

def __init__(self) -> None:
    """Initialize HTTP request transformer."""
    super().__init__()

transform_proxy_request async

transform_proxy_request(
    method,
    path,
    headers,
    body,
    query_params,
    access_token,
    target_base_url="https://api.anthropic.com",
    app_state=None,
    injection_mode="minimal",
)

Transform request using direct parameters from ProxyService.

This method provides the same functionality as ProxyService._transform_request() but is properly located in the transformer layer.

Parameters:

Name	Type	Description	Default
method	str	HTTP method	required
path	str	Request path	required
headers	dict[str, str]	Request headers	required
body	bytes | None	Request body	required
query_params	dict[str, str | list[str]] | None	Query parameters	required
access_token	str	OAuth access token	required
target_base_url	str	Base URL for the target API	'https://api.anthropic.com'
app_state	Any	Optional app state containing detection data	None
injection_mode	str	System prompt injection mode	'minimal'

Returns:

Type	Description
RequestData	Dictionary with transformed request data (method, url, headers, body)

Source code in ccproxy/core/http_transformers.py

async def transform_proxy_request(
    self,
    method: str,
    path: str,
    headers: dict[str, str],
    body: bytes | None,
    query_params: dict[str, str | list[str]] | None,
    access_token: str,
    target_base_url: str = "https://api.anthropic.com",
    app_state: Any = None,
    injection_mode: str = "minimal",
) -> RequestData:
    """Transform request using direct parameters from ProxyService.

    This method provides the same functionality as ProxyService._transform_request()
    but is properly located in the transformer layer.

    Args:
        method: HTTP method
        path: Request path
        headers: Request headers
        body: Request body
        query_params: Query parameters
        access_token: OAuth access token
        target_base_url: Base URL for the target API
        app_state: Optional app state containing detection data
        injection_mode: System prompt injection mode

    Returns:
        Dictionary with transformed request data (method, url, headers, body)
    """
    import urllib.parse

    # Transform path
    transformed_path = self.transform_path(path, self.proxy_mode)
    target_url = f"{target_base_url.rstrip('/')}{transformed_path}"

    # Add beta=true query parameter for /v1/messages requests if not already present
    if transformed_path == "/v1/messages":
        if query_params is None:
            query_params = {}
        elif "beta" not in query_params:
            query_params = dict(query_params)  # Make a copy

        if "beta" not in query_params:
            query_params["beta"] = "true"

    # Transform body first (as it might change size)
    proxy_body = None
    if body:
        proxy_body = self.transform_request_body(
            body, path, self.proxy_mode, app_state, injection_mode
        )

    # Transform headers (and update Content-Length if body changed)
    proxy_headers = self.create_proxy_headers(
        headers, access_token, self.proxy_mode, app_state
    )

    # Update Content-Length if body was transformed and size changed
    if proxy_body and body and len(proxy_body) != len(body):
        # Remove any existing content-length headers (case-insensitive)
        proxy_headers = {
            k: v for k, v in proxy_headers.items() if k.lower() != "content-length"
        }
        proxy_headers["Content-Length"] = str(len(proxy_body))
    elif proxy_body and not body:
        # New body was created where none existed
        proxy_headers["Content-Length"] = str(len(proxy_body))

    # Add query parameters to URL if present
    if query_params:
        query_string = urllib.parse.urlencode(query_params)
        target_url = f"{target_url}?{query_string}"

    return RequestData(
        method=method,
        url=target_url,
        headers=proxy_headers,
        body=proxy_body,
    )

transform_path

transform_path(path, proxy_mode='full')

Transform request path.

Source code in ccproxy/core/http_transformers.py

def transform_path(self, path: str, proxy_mode: str = "full") -> str:
    """Transform request path."""
    # Remove /api prefix if present (for new proxy endpoints)
    if path.startswith("/api"):
        path = path[4:]  # Remove "/api" prefix

    # Remove /openai prefix if present
    if path.startswith("/openai"):
        path = path[7:]  # Remove "/openai" prefix

    # Convert OpenAI chat completions to Anthropic messages
    if path == "/v1/chat/completions":
        return "/v1/messages"

    return path

create_proxy_headers

create_proxy_headers(
    headers, access_token, proxy_mode="full", app_state=None
)

Create proxy headers from original headers with Claude CLI identity.

Source code in ccproxy/core/http_transformers.py

def create_proxy_headers(
    self,
    headers: dict[str, str],
    access_token: str,
    proxy_mode: str = "full",
    app_state: Any = None,
) -> dict[str, str]:
    """Create proxy headers from original headers with Claude CLI identity."""
    proxy_headers = {}

    # Strip potentially problematic headers
    excluded_headers = {
        "host",
        "x-forwarded-for",
        "x-forwarded-proto",
        "x-forwarded-host",
        "forwarded",
        # Authentication headers to be replaced
        "authorization",
        "x-api-key",
        # Compression headers to avoid decompression issues
        "accept-encoding",
        "content-encoding",
        # CORS headers - should not be forwarded to upstream
        "origin",
        "access-control-request-method",
        "access-control-request-headers",
        "access-control-allow-origin",
        "access-control-allow-methods",
        "access-control-allow-headers",
        "access-control-allow-credentials",
        "access-control-max-age",
        "access-control-expose-headers",
    }

    # Copy important headers (excluding problematic ones)
    for key, value in headers.items():
        lower_key = key.lower()
        if lower_key not in excluded_headers:
            proxy_headers[key] = value

    # Set authentication with OAuth token
    if access_token:
        proxy_headers["Authorization"] = f"Bearer {access_token}"

    # Set defaults for essential headers
    if "content-type" not in [k.lower() for k in proxy_headers]:
        proxy_headers["Content-Type"] = "application/json"
    if "accept" not in [k.lower() for k in proxy_headers]:
        proxy_headers["Accept"] = "application/json"
    if "connection" not in [k.lower() for k in proxy_headers]:
        proxy_headers["Connection"] = "keep-alive"

    # Use detected Claude CLI headers when available
    if app_state and hasattr(app_state, "claude_detection_data"):
        claude_data = app_state.claude_detection_data
        detected_headers = claude_data.headers.to_headers_dict()
        proxy_headers.update(detected_headers)
        logger.debug("using_detected_headers", version=claude_data.claude_version)
    else:
        # Fallback to hardcoded Claude/Anthropic headers
        proxy_headers["anthropic-beta"] = (
            "claude-code-20250219,oauth-2025-04-20,"
            "interleaved-thinking-2025-05-14,fine-grained-tool-streaming-2025-05-14"
        )
        proxy_headers["anthropic-version"] = "2023-06-01"
        proxy_headers["anthropic-dangerous-direct-browser-access"] = "true"

        # Claude CLI identity headers
        proxy_headers["x-app"] = "cli"
        proxy_headers["User-Agent"] = "claude-cli/1.0.60 (external, cli)"

        # Stainless SDK compatibility headers
        proxy_headers["X-Stainless-Lang"] = "js"
        proxy_headers["X-Stainless-Retry-Count"] = "0"
        proxy_headers["X-Stainless-Timeout"] = "60"
        proxy_headers["X-Stainless-Package-Version"] = "0.55.1"
        proxy_headers["X-Stainless-OS"] = "Linux"
        proxy_headers["X-Stainless-Arch"] = "x64"
        proxy_headers["X-Stainless-Runtime"] = "node"
        proxy_headers["X-Stainless-Runtime-Version"] = "v24.3.0"
        logger.debug("using_fallback_headers")

    # Standard HTTP headers for proper API interaction
    proxy_headers["accept-language"] = "*"
    proxy_headers["sec-fetch-mode"] = "cors"
    # Note: accept-encoding removed to avoid compression issues
    # HTTPX handles compression automatically

    return proxy_headers

transform_request_body

transform_request_body(
    body,
    path,
    proxy_mode="full",
    app_state=None,
    injection_mode="minimal",
)

Transform request body.

Source code in ccproxy/core/http_transformers.py

def transform_request_body(
    self,
    body: bytes,
    path: str,
    proxy_mode: str = "full",
    app_state: Any = None,
    injection_mode: str = "minimal",
) -> bytes:
    """Transform request body."""
    if not body:
        return body

    # Check if this is an OpenAI request and transform it
    if self._is_openai_request(path, body):
        # Transform OpenAI format to Anthropic format
        body = self._transform_openai_to_anthropic(body)

    # Apply system prompt transformation for Claude Code identity
    return self.transform_system_prompt(body, app_state, injection_mode)

transform_system_prompt

transform_system_prompt(
    body, app_state=None, injection_mode="minimal"
)

Transform system prompt based on injection mode.

Parameters:

Name	Type	Description	Default
body	bytes	Original request body as bytes	required
app_state	Any	Optional app state containing detection data	None
injection_mode	str	System prompt injection mode ('minimal' or 'full')	'minimal'

Returns:

Type	Description
bytes	Transformed request body as bytes with system prompt injection

Source code in ccproxy/core/http_transformers.py

def transform_system_prompt(
    self, body: bytes, app_state: Any = None, injection_mode: str = "minimal"
) -> bytes:
    """Transform system prompt based on injection mode.

    Args:
        body: Original request body as bytes
        app_state: Optional app state containing detection data
        injection_mode: System prompt injection mode ('minimal' or 'full')

    Returns:
        Transformed request body as bytes with system prompt injection
    """
    try:
        import json

        data = json.loads(body.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError) as e:
        # Return original if not valid JSON
        logger.warning(
            "http_transform_json_decode_failed",
            error=str(e),
            body_preview=body[:200].decode("utf-8", errors="replace")
            if body
            else None,
            body_length=len(body) if body else 0,
        )
        return body

    # Get the system field to inject
    detected_system = get_detected_system_field(app_state, injection_mode)
    if detected_system is None:
        # No detection data, use fallback
        detected_system = get_fallback_system_field()

    # Always inject the system prompt (detected or fallback)
    if "system" not in data:
        # No existing system prompt, inject the detected/fallback one
        data["system"] = detected_system
    else:
        # Request has existing system prompt, prepend the detected/fallback one
        existing_system = data["system"]

        if isinstance(detected_system, str):
            # Detected system is a string
            if isinstance(existing_system, str):
                # Both are strings, convert to list format
                data["system"] = [
                    {"type": "text", "text": detected_system},
                    {"type": "text", "text": existing_system},
                ]
            elif isinstance(existing_system, list):
                # Detected is string, existing is list
                data["system"] = [
                    {"type": "text", "text": detected_system}
                ] + existing_system
        elif isinstance(detected_system, list):
            # Detected system is a list
            if isinstance(existing_system, str):
                # Detected is list, existing is string
                data["system"] = detected_system + [
                    {"type": "text", "text": existing_system}
                ]
            elif isinstance(existing_system, list):
                # Both are lists, concatenate
                data["system"] = detected_system + existing_system

    # Limit cache_control blocks to comply with Anthropic's limit
    data = self._limit_cache_control_blocks(data)

    return json.dumps(data).encode("utf-8")

HTTPResponseTransformer

HTTPResponseTransformer()

Bases: ResponseTransformer

HTTP response transformer that implements the abstract ResponseTransformer interface.

Source code in ccproxy/core/http_transformers.py

def __init__(self) -> None:
    """Initialize HTTP response transformer."""
    super().__init__()

transform_proxy_response async

transform_proxy_response(
    status_code,
    headers,
    body,
    original_path,
    proxy_mode="full",
)

Transform response using direct parameters from ProxyService.

This method provides the same functionality as ProxyService._transform_response() but is properly located in the transformer layer.

Parameters:

Name	Type	Description	Default
status_code	int	HTTP status code	required
headers	dict[str, str]	Response headers	required
body	bytes	Response body	required
original_path	str	Original request path for context	required
proxy_mode	str	Proxy transformation mode	'full'

Returns:

Type	Description
ResponseData	Dictionary with transformed response data (status_code, headers, body)

Source code in ccproxy/core/http_transformers.py

async def transform_proxy_response(
    self,
    status_code: int,
    headers: dict[str, str],
    body: bytes,
    original_path: str,
    proxy_mode: str = "full",
) -> ResponseData:
    """Transform response using direct parameters from ProxyService.

    This method provides the same functionality as ProxyService._transform_response()
    but is properly located in the transformer layer.

    Args:
        status_code: HTTP status code
        headers: Response headers
        body: Response body
        original_path: Original request path for context
        proxy_mode: Proxy transformation mode

    Returns:
        Dictionary with transformed response data (status_code, headers, body)
    """
    # For error responses, handle OpenAI transformation if needed
    if status_code >= 400:
        transformed_error_body = body
        if self._is_openai_request(original_path):
            try:
                import json

                from ccproxy.adapters.openai.adapter import OpenAIAdapter

                error_data = json.loads(body.decode("utf-8"))
                openai_adapter = OpenAIAdapter()
                openai_error = openai_adapter.adapt_error(error_data)
                transformed_error_body = json.dumps(openai_error).encode("utf-8")
            except (json.JSONDecodeError, UnicodeDecodeError):
                # Keep original error if parsing fails
                pass

        return ResponseData(
            status_code=status_code,
            headers=headers,
            body=transformed_error_body,
        )

    # For successful responses, transform normally
    transformed_body = self.transform_response_body(body, original_path, proxy_mode)

    transformed_headers = self.transform_response_headers(
        headers, original_path, len(transformed_body), proxy_mode
    )

    return ResponseData(
        status_code=status_code,
        headers=transformed_headers,
        body=transformed_body,
    )

transform_response_body

transform_response_body(body, path, proxy_mode='full')

Transform response body.

Source code in ccproxy/core/http_transformers.py

def transform_response_body(
    self, body: bytes, path: str, proxy_mode: str = "full"
) -> bytes:
    """Transform response body."""
    # Basic body transformation - pass through for now
    return body

transform_response_headers

transform_response_headers(
    headers, path, content_length, proxy_mode="full"
)

Transform response headers.

Source code in ccproxy/core/http_transformers.py

def transform_response_headers(
    self,
    headers: dict[str, str],
    path: str,
    content_length: int,
    proxy_mode: str = "full",
) -> dict[str, str]:
    """Transform response headers."""
    transformed_headers = {}

    # Copy important headers
    for key, value in headers.items():
        lower_key = key.lower()
        if lower_key not in [
            "content-length",
            "transfer-encoding",
            "content-encoding",
            "date",  # Remove upstream date header to avoid conflicts
        ]:
            transformed_headers[key] = value

    # Set content length
    transformed_headers["Content-Length"] = str(content_length)

    # Add CORS headers
    transformed_headers["Access-Control-Allow-Origin"] = "*"
    transformed_headers["Access-Control-Allow-Headers"] = "*"
    transformed_headers["Access-Control-Allow-Methods"] = "*"

    return transformed_headers

get_detected_system_field

get_detected_system_field(
    app_state=None, injection_mode="minimal"
)

Get the detected system field for injection.

Parameters:

Name	Type	Description	Default
app_state	Any	App state containing detection data	None
injection_mode	str	'minimal' or 'full' mode	'minimal'

Returns:

Type	Description
Any	The system field to inject (preserving exact Claude CLI structure), or None if no detection data available

Source code in ccproxy/core/http_transformers.py

def get_detected_system_field(
    app_state: Any = None, injection_mode: str = "minimal"
) -> Any:
    """Get the detected system field for injection.

    Args:
        app_state: App state containing detection data
        injection_mode: 'minimal' or 'full' mode

    Returns:
        The system field to inject (preserving exact Claude CLI structure), or None if no detection data available
    """
    if not app_state or not hasattr(app_state, "claude_detection_data"):
        return None

    claude_data = app_state.claude_detection_data
    detected_system = claude_data.system_prompt.system_field

    if injection_mode == "full":
        # Return the complete detected system field exactly as Claude CLI sent it
        return detected_system
    else:
        # Minimal mode: extract just the first system message, preserving its structure
        if isinstance(detected_system, str):
            return detected_system
        elif isinstance(detected_system, list) and detected_system:
            # Return only the first message object with its complete structure (type, text, cache_control)
            return [detected_system[0]]

    return None

get_fallback_system_field

get_fallback_system_field()

Get fallback system field when no detection data is available.

Source code in ccproxy/core/http_transformers.py

def get_fallback_system_field() -> list[dict[str, Any]]:
    """Get fallback system field when no detection data is available."""
    return [
        {
            "type": "text",
            "text": claude_code_prompt,
            "cache_control": {"type": "ephemeral"},
        }
    ]