ccproxy.adapters.openai.adapter

ccproxy.adapters.openai.adapter

OpenAI API adapter implementation.

This module provides the OpenAI adapter that implements the APIAdapter interface for converting between OpenAI and Anthropic API formats.

OpenAIChatCompletionRequest

Bases: BaseModel

OpenAI-compatible chat completion request model.

validate_model classmethod

validate_model(v)

Validate model name - just return as-is like Anthropic endpoint.

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

@field_validator("model")
@classmethod
def validate_model(cls, v: str) -> str:
    """Validate model name - just return as-is like Anthropic endpoint."""
    return v

validate_messages classmethod

validate_messages(v)

Validate message structure.

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

@field_validator("messages")
@classmethod
def validate_messages(cls, v: list[OpenAIMessage]) -> list[OpenAIMessage]:
    """Validate message structure."""
    if not v:
        raise ValueError("At least one message is required")
    return v

validate_stop classmethod

validate_stop(v)

Validate stop sequences.

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

@field_validator("stop")
@classmethod
def validate_stop(cls, v: str | list[str] | None) -> str | list[str] | None:
    """Validate stop sequences."""
    if v is not None:
        if isinstance(v, str):
            return v
        elif isinstance(v, list):
            if len(v) > 4:
                raise ValueError("Maximum 4 stop sequences allowed")
            return v
    return v

validate_tools classmethod

validate_tools(v)

Validate tools array.

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

@field_validator("tools")
@classmethod
def validate_tools(cls, v: list[OpenAITool] | None) -> list[OpenAITool] | None:
    """Validate tools array."""
    if v is not None and len(v) > 128:
        raise ValueError("Maximum 128 tools allowed")
    return v

OpenAIChatCompletionResponse

Bases: BaseModel

OpenAI chat completion response.

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
        }
    }