ccproxy.adapters

ccproxy.adapters

Adapter modules for API format conversion.

APIAdapter

Bases: ABC

Abstract base class for API format adapters.

Combines all transformation interfaces to provide a complete adapter for converting between different API formats.

adapt_request abstractmethod async

adapt_request(request)

Convert a request from one API format to another.

Parameters:

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

Returns:

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

Raises:

Type	Description
ValueError	If the request format is invalid or unsupported

Source code in ccproxy/adapters/base.py

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

    Args:
        request: The request data to convert

    Returns:
        The converted request data

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

adapt_response abstractmethod async

adapt_response(response)

Convert a response from one API format to another.

Parameters:

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

Returns:

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

Raises:

Type	Description
ValueError	If the response format is invalid or unsupported

Source code in ccproxy/adapters/base.py

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

    Args:
        response: The response data to convert

    Returns:
        The converted response data

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

adapt_stream abstractmethod async

adapt_stream(stream)

Convert a streaming response from one API format to another.

Parameters:

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

Yields:

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

Raises:

Type	Description
ValueError	If the stream format is invalid or unsupported

Source code in ccproxy/adapters/base.py

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

    Args:
        stream: The streaming response data to convert

    Yields:
        The converted streaming response chunks

    Raises:
        ValueError: If the stream format is invalid or unsupported
    """
    # This should be implemented as an async generator
    # async def adapt_stream(self, stream):
    #     async for item in stream:
    #         yield transformed_item
    raise NotImplementedError

BaseAPIAdapter

BaseAPIAdapter(name)

Bases: APIAdapter

Base implementation with common functionality.

Source code in ccproxy/adapters/base.py

def __init__(self, name: str):
    self.name = name

OpenAIAdapter

OpenAIAdapter(include_sdk_content_as_xml=False)

Bases: APIAdapter

OpenAI API adapter for converting between OpenAI and Anthropic formats.

Source code in ccproxy/adapters/openai/adapter.py

def __init__(self, include_sdk_content_as_xml: bool = False) -> None:
    """Initialize the OpenAI adapter."""
    self.include_sdk_content_as_xml = include_sdk_content_as_xml

adapt_request

adapt_request(request)

Convert OpenAI request format to Anthropic format.

Parameters:

Name	Type	Description	Default
request	dict[str, Any]	OpenAI format request	required

Returns:

Type	Description
dict[str, Any]	Anthropic format request

Raises:

Type	Description
ValueError	If the request format is invalid or unsupported

Source code in ccproxy/adapters/openai/adapter.py

def adapt_request(self, request: dict[str, Any]) -> dict[str, Any]:
    """Convert OpenAI request format to Anthropic format.

    Args:
        request: OpenAI format request

    Returns:
        Anthropic format request

    Raises:
        ValueError: If the request format is invalid or unsupported
    """
    try:
        # Parse OpenAI request
        openai_req = OpenAIChatCompletionRequest(**request)
    except ValidationError as e:
        raise ValueError(f"Invalid OpenAI request format: {e}") from e

    # Map OpenAI model to Claude model
    model = map_model_to_claude(openai_req.model)

    # Convert messages
    messages, system_prompt = self._convert_messages_to_anthropic(
        openai_req.messages
    )

    # Build base Anthropic request
    anthropic_request = {
        "model": model,
        "messages": messages,
        "max_tokens": openai_req.max_tokens or 4096,
    }

    # Add system prompt if present
    if system_prompt:
        anthropic_request["system"] = system_prompt

    # Add optional parameters
    self._handle_optional_parameters(openai_req, anthropic_request)

    # Handle metadata
    self._handle_metadata(openai_req, anthropic_request)

    # Handle response format
    anthropic_request = self._handle_response_format(openai_req, anthropic_request)

    # Handle thinking configuration
    anthropic_request = self._handle_thinking_parameters(
        openai_req, anthropic_request
    )

    # Log unsupported parameters
    self._log_unsupported_parameters(openai_req)

    # Handle tools and tool choice
    self._handle_tools(openai_req, anthropic_request)

    logger.debug(
        "format_conversion_completed",
        from_format="openai",
        to_format="anthropic",
        original_model=openai_req.model,
        anthropic_model=anthropic_request.get("model"),
        has_tools=bool(anthropic_request.get("tools")),
        has_system=bool(anthropic_request.get("system")),
        message_count=len(cast(list[Any], anthropic_request["messages"])),
        operation="adapt_request",
    )
    return anthropic_request

adapt_response

adapt_response(response)

Convert Anthropic response format to OpenAI format.

Parameters:

Name	Type	Description	Default
response	dict[str, Any]	Anthropic format response	required

Returns:

Type	Description
dict[str, Any]	OpenAI format response

Raises:

Type	Description
ValueError	If the response format is invalid or unsupported

Source code in ccproxy/adapters/openai/adapter.py

def adapt_response(self, response: dict[str, Any]) -> dict[str, Any]:
    """Convert Anthropic response format to OpenAI format.

    Args:
        response: Anthropic format response

    Returns:
        OpenAI format response

    Raises:
        ValueError: If the response format is invalid or unsupported
    """
    try:
        # Extract original model from response metadata if available
        original_model = response.get("model", "gpt-4")

        # Generate response ID
        request_id = generate_openai_response_id()

        # Convert content and extract tool calls
        content, tool_calls = self._convert_content_blocks(response)

        # Create OpenAI message
        message = self._create_openai_message(content, tool_calls)

        # Create choice with proper finish reason
        choice = self._create_openai_choice(message, response)

        # Create usage information
        usage = self._create_openai_usage(response)

        # Create final OpenAI response
        openai_response = OpenAIChatCompletionResponse(
            id=request_id,
            object="chat.completion",
            created=int(time.time()),
            model=original_model,
            choices=[choice],
            usage=usage,
            system_fingerprint=generate_openai_system_fingerprint(),
        )

        logger.debug(
            "format_conversion_completed",
            from_format="anthropic",
            to_format="openai",
            response_id=request_id,
            original_model=original_model,
            finish_reason=choice.finish_reason,
            content_length=len(content) if content else 0,
            tool_calls_count=len(tool_calls),
            input_tokens=usage.prompt_tokens,
            output_tokens=usage.completion_tokens,
            operation="adapt_response",
            choice=choice,
        )
        return openai_response.model_dump()

    except ValidationError as e:
        raise ValueError(f"Invalid Anthropic response format: {e}") from e

adapt_stream async

adapt_stream(stream)

Convert Anthropic streaming response to OpenAI streaming format.

Parameters:

Name	Type	Description	Default
stream	AsyncIterator[dict[str, Any]]	Anthropic streaming response	required

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	OpenAI format streaming chunks

Raises:

Type	Description
ValueError	If the stream format is invalid or unsupported

Source code in ccproxy/adapters/openai/adapter.py

async def adapt_stream(
    self, stream: AsyncIterator[dict[str, Any]]
) -> AsyncIterator[dict[str, Any]]:
    """Convert Anthropic streaming response to OpenAI streaming format.

    Args:
        stream: Anthropic streaming response

    Yields:
        OpenAI format streaming chunks

    Raises:
        ValueError: If the stream format is invalid or unsupported
    """
    # Create stream processor with dict output format
    processor = OpenAIStreamProcessor(
        enable_usage=True,
        enable_tool_calls=True,
        output_format="dict",  # Output dict objects instead of SSE strings
    )

    try:
        # Process the stream - now yields dict objects directly
        async for chunk in processor.process_stream(stream):
            yield chunk  # type: ignore[misc]  # chunk is guaranteed to be dict when output_format="dict"
    except Exception as e:
        logger.error(
            "streaming_conversion_failed",
            error=str(e),
            error_type=type(e).__name__,
            operation="adapt_stream",
            exc_info=True,
        )
        raise ValueError(f"Error processing streaming response: {e}") from e

adapt_error

adapt_error(error_body)

Convert Anthropic error format to OpenAI error format.

Parameters:

Name	Type	Description	Default
error_body	dict[str, Any]	Anthropic error response	required

Returns:

Type	Description
dict[str, Any]	OpenAI-formatted error response

Source code in ccproxy/adapters/openai/adapter.py

def adapt_error(self, error_body: dict[str, Any]) -> dict[str, Any]:
    """Convert Anthropic error format to OpenAI error format.

    Args:
        error_body: Anthropic error response

    Returns:
        OpenAI-formatted error response
    """
    # Extract error details from Anthropic format
    anthropic_error = error_body.get("error", {})
    error_type = anthropic_error.get("type", "internal_server_error")
    error_message = anthropic_error.get("message", "An error occurred")

    # Map Anthropic error types to OpenAI error types
    error_type_mapping = {
        "invalid_request_error": "invalid_request_error",
        "authentication_error": "invalid_request_error",
        "permission_error": "invalid_request_error",
        "not_found_error": "invalid_request_error",
        "rate_limit_error": "rate_limit_error",
        "internal_server_error": "internal_server_error",
        "overloaded_error": "server_error",
    }

    openai_error_type = error_type_mapping.get(error_type, "invalid_request_error")

    # Return OpenAI-formatted error
    return {
        "error": {
            "message": error_message,
            "type": openai_error_type,
            "code": error_type,  # Preserve original error type as code
        }
    }