ccproxy.adapters.openai

ccproxy.adapters.openai

OpenAI adapter module for API format conversion.

This module provides the OpenAI adapter implementation for converting between OpenAI and Anthropic API formats.

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

OpenAIChatCompletionResponse

Bases: BaseModel

OpenAI chat completion response.

OpenAIChoice

Bases: BaseModel

OpenAI choice in response.

OpenAIMessage

Bases: BaseModel

OpenAI message model.

OpenAIMessageContent

Bases: BaseModel

OpenAI message content block.

OpenAIResponseMessage

Bases: BaseModel

OpenAI response message.

OpenAIStreamingChatCompletionResponse

Bases: BaseModel

OpenAI streaming chat completion response.

OpenAIToolCall

Bases: BaseModel

OpenAI tool call.

OpenAIUsage

Bases: BaseModel

OpenAI usage information.

OpenAISSEFormatter

Formats streaming responses to match OpenAI's SSE format.

format_data_event staticmethod

format_data_event(data)

Format a data event for OpenAI-compatible Server-Sent Events.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Event data dictionary	required

Returns:

Type	Description
str	Formatted SSE string

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

@staticmethod
def format_data_event(data: dict[str, Any]) -> str:
    """Format a data event for OpenAI-compatible Server-Sent Events.

    Args:
        data: Event data dictionary

    Returns:
        Formatted SSE string
    """
    json_data = json.dumps(data, separators=(",", ":"))
    return f"data: {json_data}\n\n"

format_first_chunk staticmethod

format_first_chunk(
    message_id, model, created, role="assistant"
)

Format the first chunk with role and basic metadata.

Parameters:

Name	Type	Description	Default
message_id	str	Unique identifier for the completion	required
model	str	Model name being used	required
created	int	Unix timestamp when the completion was created	required
role	str	Role of the assistant	'assistant'

Returns:

Type	Description
str	Formatted SSE string

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

@staticmethod
def format_first_chunk(
    message_id: str, model: str, created: int, role: str = "assistant"
) -> str:
    """Format the first chunk with role and basic metadata.

    Args:
        message_id: Unique identifier for the completion
        model: Model name being used
        created: Unix timestamp when the completion was created
        role: Role of the assistant

    Returns:
        Formatted SSE string
    """
    data = {
        "id": message_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
        "choices": [
            {
                "index": 0,
                "delta": {"role": role},
                "logprobs": None,
                "finish_reason": None,
            }
        ],
    }
    return OpenAISSEFormatter.format_data_event(data)

format_content_chunk staticmethod

format_content_chunk(
    message_id, model, created, content, choice_index=0
)

Format a content chunk with text delta.

Parameters:

Name	Type	Description	Default
message_id	str	Unique identifier for the completion	required
model	str	Model name being used	required
created	int	Unix timestamp when the completion was created	required
content	str	Text content to include in the delta	required
choice_index	int	Index of the choice (usually 0)	0

Returns:

Type	Description
str	Formatted SSE string

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

@staticmethod
def format_content_chunk(
    message_id: str, model: str, created: int, content: str, choice_index: int = 0
) -> str:
    """Format a content chunk with text delta.

    Args:
        message_id: Unique identifier for the completion
        model: Model name being used
        created: Unix timestamp when the completion was created
        content: Text content to include in the delta
        choice_index: Index of the choice (usually 0)

    Returns:
        Formatted SSE string
    """
    data = {
        "id": message_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
        "choices": [
            {
                "index": choice_index,
                "delta": {"content": content},
                "logprobs": None,
                "finish_reason": None,
            }
        ],
    }
    return OpenAISSEFormatter.format_data_event(data)

format_tool_call_chunk staticmethod

format_tool_call_chunk(
    message_id,
    model,
    created,
    tool_call_id,
    function_name=None,
    function_arguments=None,
    tool_call_index=0,
    choice_index=0,
)

Format a tool call chunk.

Parameters:

Name	Type	Description	Default
message_id	str	Unique identifier for the completion	required
model	str	Model name being used	required
created	int	Unix timestamp when the completion was created	required
tool_call_id	str	ID of the tool call	required
function_name	str | None	Name of the function being called	None
function_arguments	str | None	Arguments for the function	None
tool_call_index	int	Index of the tool call	0
choice_index	int	Index of the choice (usually 0)	0

Returns:

Type	Description
str	Formatted SSE string

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

@staticmethod
def format_tool_call_chunk(
    message_id: str,
    model: str,
    created: int,
    tool_call_id: str,
    function_name: str | None = None,
    function_arguments: str | None = None,
    tool_call_index: int = 0,
    choice_index: int = 0,
) -> str:
    """Format a tool call chunk.

    Args:
        message_id: Unique identifier for the completion
        model: Model name being used
        created: Unix timestamp when the completion was created
        tool_call_id: ID of the tool call
        function_name: Name of the function being called
        function_arguments: Arguments for the function
        tool_call_index: Index of the tool call
        choice_index: Index of the choice (usually 0)

    Returns:
        Formatted SSE string
    """
    tool_call: dict[str, Any] = {
        "index": tool_call_index,
        "id": tool_call_id,
        "type": "function",
        "function": {},
    }

    if function_name is not None:
        tool_call["function"]["name"] = function_name

    if function_arguments is not None:
        tool_call["function"]["arguments"] = function_arguments

    data = {
        "id": message_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
        "choices": [
            {
                "index": choice_index,
                "delta": {"tool_calls": [tool_call]},
                "logprobs": None,
                "finish_reason": None,
            }
        ],
    }
    return OpenAISSEFormatter.format_data_event(data)

format_final_chunk staticmethod

format_final_chunk(
    message_id,
    model,
    created,
    finish_reason="stop",
    choice_index=0,
    usage=None,
)

Format the final chunk with finish_reason.

Parameters:

Name	Type	Description	Default
message_id	str	Unique identifier for the completion	required
model	str	Model name being used	required
created	int	Unix timestamp when the completion was created	required
finish_reason	str	Reason for completion (stop, length, tool_calls, etc.)	'stop'
choice_index	int	Index of the choice (usually 0)	0
usage	dict[str, int] | None	Optional usage information to include	None

Returns:

Type	Description
str	Formatted SSE string

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

@staticmethod
def format_final_chunk(
    message_id: str,
    model: str,
    created: int,
    finish_reason: str = "stop",
    choice_index: int = 0,
    usage: dict[str, int] | None = None,
) -> str:
    """Format the final chunk with finish_reason.

    Args:
        message_id: Unique identifier for the completion
        model: Model name being used
        created: Unix timestamp when the completion was created
        finish_reason: Reason for completion (stop, length, tool_calls, etc.)
        choice_index: Index of the choice (usually 0)
        usage: Optional usage information to include

    Returns:
        Formatted SSE string
    """
    data = {
        "id": message_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
        "choices": [
            {
                "index": choice_index,
                "delta": {},
                "logprobs": None,
                "finish_reason": finish_reason,
            }
        ],
    }

    # Add usage if provided
    if usage:
        data["usage"] = usage

    return OpenAISSEFormatter.format_data_event(data)

format_error_chunk staticmethod

format_error_chunk(
    message_id, model, created, error_type, error_message
)

Format an error chunk.

Parameters:

Name	Type	Description	Default
message_id	str	Unique identifier for the completion	required
model	str	Model name being used	required
created	int	Unix timestamp when the completion was created	required
error_type	str	Type of error	required
error_message	str	Error message	required

Returns:

Type	Description
str	Formatted SSE string

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

@staticmethod
def format_error_chunk(
    message_id: str, model: str, created: int, error_type: str, error_message: str
) -> str:
    """Format an error chunk.

    Args:
        message_id: Unique identifier for the completion
        model: Model name being used
        created: Unix timestamp when the completion was created
        error_type: Type of error
        error_message: Error message

    Returns:
        Formatted SSE string
    """
    data = {
        "id": message_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
        "choices": [
            {"index": 0, "delta": {}, "logprobs": None, "finish_reason": "error"}
        ],
        "error": {"type": error_type, "message": error_message},
    }
    return OpenAISSEFormatter.format_data_event(data)

format_done staticmethod

format_done()

Format the final DONE event.

Returns:

Type	Description
str	Formatted SSE termination string

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

@staticmethod
def format_done() -> str:
    """Format the final DONE event.

    Returns:
        Formatted SSE termination string
    """
    return "data: [DONE]\n\n"

OpenAIStreamProcessor

OpenAIStreamProcessor(
    message_id=None,
    model="claude-3-5-sonnet-20241022",
    created=None,
    enable_usage=True,
    enable_tool_calls=True,
    output_format="sse",
)

Processes Anthropic/Claude streaming responses into OpenAI format.

Parameters:

Name	Type	Description	Default
message_id	str | None	Response ID, generated if not provided	None
model	str	Model name for responses	'claude-3-5-sonnet-20241022'
created	int | None	Creation timestamp, current time if not provided	None
enable_usage	bool	Whether to include usage information	True
enable_tool_calls	bool	Whether to process tool calls	True
output_format	Literal['sse', 'dict']	Output format - "sse" for Server-Sent Events strings, "dict" for dict objects	'sse'

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

def __init__(
    self,
    message_id: str | None = None,
    model: str = "claude-3-5-sonnet-20241022",
    created: int | None = None,
    enable_usage: bool = True,
    enable_tool_calls: bool = True,
    output_format: Literal["sse", "dict"] = "sse",
):
    """Initialize the stream processor.

    Args:
        message_id: Response ID, generated if not provided
        model: Model name for responses
        created: Creation timestamp, current time if not provided
        enable_usage: Whether to include usage information
        enable_tool_calls: Whether to process tool calls
        output_format: Output format - "sse" for Server-Sent Events strings, "dict" for dict objects
    """
    self.message_id = message_id or generate_openai_response_id()
    self.model = model
    self.created = created or int(time.time())
    self.enable_usage = enable_usage
    self.enable_tool_calls = enable_tool_calls
    self.output_format = output_format
    self.formatter = OpenAISSEFormatter()

    # State tracking
    self.role_sent = False
    self.accumulated_content = ""
    self.tool_calls: dict[str, dict[str, Any]] = {}
    self.usage_info: dict[str, int] | None = None
    # Thinking block tracking
    self.current_thinking_text = ""
    self.current_thinking_signature: str | None = None
    self.thinking_block_active = False

process_stream async

process_stream(claude_stream)

Process a Claude/Anthropic stream into OpenAI format.

Parameters:

Name	Type	Description	Default
claude_stream	AsyncIterator[dict[str, Any]]	Async iterator of Claude response chunks	required

Yields:

Type	Description
AsyncIterator[str | dict[str, Any]]	OpenAI-formatted SSE strings or dict objects based on output_format

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

async def process_stream(
    self, claude_stream: AsyncIterator[dict[str, Any]]
) -> AsyncIterator[str | dict[str, Any]]:
    """Process a Claude/Anthropic stream into OpenAI format.

    Args:
        claude_stream: Async iterator of Claude response chunks

    Yields:
        OpenAI-formatted SSE strings or dict objects based on output_format
    """
    try:
        chunk_count = 0
        processed_count = 0
        async for chunk in claude_stream:
            chunk_count += 1
            logger.debug(
                "openai_stream_chunk_received",
                chunk_count=chunk_count,
                chunk_type=chunk.get("type"),
                chunk=chunk,
            )
            async for sse_chunk in self._process_chunk(chunk):
                processed_count += 1
                logger.debug(
                    "openai_stream_chunk_processed",
                    processed_count=processed_count,
                    sse_chunk=sse_chunk,
                )
                yield sse_chunk

        logger.debug(
            "openai_stream_complete",
            total_chunks=chunk_count,
            processed_chunks=processed_count,
            usage_info=self.usage_info,
        )

        # Send final chunk
        if self.usage_info and self.enable_usage:
            yield self._format_chunk_output(
                finish_reason="stop",
                usage=self.usage_info,
            )
        else:
            yield self._format_chunk_output(finish_reason="stop")

        # Send DONE event (only for SSE format)
        if self.output_format == "sse":
            yield self.formatter.format_done()

    except Exception as e:
        # Send error chunk
        if self.output_format == "sse":
            yield self.formatter.format_error_chunk(
                self.message_id, self.model, self.created, "error", str(e)
            )
            yield self.formatter.format_done()
        else:
            # Dict format error
            yield self._create_chunk_dict(finish_reason="error")

format_openai_tool_call

format_openai_tool_call(tool_use)

Convert Anthropic tool use to OpenAI tool call format.

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

def format_openai_tool_call(tool_use: dict[str, Any]) -> OpenAIToolCall:
    """Convert Anthropic tool use to OpenAI tool call format."""
    tool_input = tool_use.get("input", {})
    if isinstance(tool_input, dict):
        arguments_str = json.dumps(tool_input)
    else:
        arguments_str = str(tool_input)

    return OpenAIToolCall(
        id=tool_use.get("id", ""),
        type="function",
        function=OpenAIFunctionCall(
            name=tool_use.get("name", ""),
            arguments=arguments_str,
        ),
    )

generate_openai_response_id

generate_openai_response_id()

Generate an OpenAI-compatible response ID.

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

def generate_openai_response_id() -> str:
    """Generate an OpenAI-compatible response ID."""
    return f"chatcmpl-{uuid.uuid4().hex[:29]}"

generate_openai_system_fingerprint

generate_openai_system_fingerprint()

Generate an OpenAI-compatible system fingerprint.

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

def generate_openai_system_fingerprint() -> str:
    """Generate an OpenAI-compatible system fingerprint."""
    return f"fp_{uuid.uuid4().hex[:8]}"