ccproxy.claude_sdk

ccproxy.claude_sdk

Claude SDK integration module.

ClaudeSDKClient

ClaudeSDKClient()

Minimal Claude SDK client wrapper that handles core SDK interactions.

This class provides a clean interface to the Claude Code SDK while handling error translation and basic query execution.

Source code in ccproxy/claude_sdk/client.py

def __init__(self) -> None:
    """Initialize the Claude SDK client."""
    self._last_api_call_time_ms: float = 0.0

query_completion async

query_completion(prompt, options, request_id=None)

Execute a query using the Claude Code SDK.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt string to send to Claude	required
options	ClaudeCodeOptions	Claude Code options configuration	required
request_id	str | None	Optional request ID for correlation	None

Yields:

Type	Description
AsyncIterator[UserMessage | AssistantMessage | SystemMessage | ResultMessage]	Messages from the Claude Code SDK

Raises:

Type	Description
ClaudeSDKError	If the query fails

Source code in ccproxy/claude_sdk/client.py

async def query_completion(
    self, prompt: str, options: ClaudeCodeOptions, request_id: str | None = None
) -> AsyncIterator[UserMessage | AssistantMessage | SystemMessage | ResultMessage]:
    """
    Execute a query using the Claude Code SDK.

    Args:
        prompt: The prompt string to send to Claude
        options: Claude Code options configuration
        request_id: Optional request ID for correlation

    Yields:
        Messages from the Claude Code SDK

    Raises:
        ClaudeSDKError: If the query fails
    """
    async with timed_operation("claude_sdk_query", request_id) as op:
        try:
            logger.debug("claude_sdk_query_start", prompt_length=len(prompt))

            message_count = 0
            async for message in query(prompt=prompt, options=options):
                message_count += 1
                yield message

            # Store final metrics
            op["message_count"] = message_count
            self._last_api_call_time_ms = op.get("duration_ms", 0.0)

            logger.debug(
                "claude_sdk_query_completed",
                message_count=message_count,
                duration_ms=op.get("duration_ms"),
            )

        except (CLINotFoundError, CLIConnectionError) as e:
            logger.error(
                "claude_sdk_connection_failed",
                error=str(e),
                error_type=type(e).__name__,
            )
            raise ServiceUnavailableError(
                f"Claude CLI not available: {str(e)}"
            ) from e
        except (ProcessError, CLIJSONDecodeError) as e:
            logger.error(
                "claude_sdk_process_failed",
                error=str(e),
                error_type=type(e).__name__,
            )
            raise ClaudeProxyError(
                message=f"Claude process error: {str(e)}",
                error_type="service_unavailable_error",
                status_code=503,
            ) from e
        except Exception as e:
            logger.error(
                "claude_sdk_unexpected_error_occurred",
                error=str(e),
                error_type=type(e).__name__,
            )
            raise ClaudeProxyError(
                message=f"Unexpected error: {str(e)}",
                error_type="internal_server_error",
                status_code=500,
            ) from e

get_last_api_call_time_ms

get_last_api_call_time_ms()

Get the duration of the last Claude API call in milliseconds.

Returns:

Type	Description
float	Duration in milliseconds, or 0.0 if no call has been made yet

Source code in ccproxy/claude_sdk/client.py

def get_last_api_call_time_ms(self) -> float:
    """
    Get the duration of the last Claude API call in milliseconds.

    Returns:
        Duration in milliseconds, or 0.0 if no call has been made yet
    """
    return self._last_api_call_time_ms

validate_health async

validate_health()

Validate that the Claude SDK is healthy.

Returns:

Type	Description
bool	True if healthy, False otherwise

Source code in ccproxy/claude_sdk/client.py

async def validate_health(self) -> bool:
    """
    Validate that the Claude SDK is healthy.

    Returns:
        True if healthy, False otherwise
    """
    try:
        logger.debug("health_check_start", component="claude_sdk")

        # Simple health check - the SDK is available if we can import it
        # More sophisticated checks could be added here
        is_healthy = True

        logger.debug(
            "health_check_completed", component="claude_sdk", healthy=is_healthy
        )
        return is_healthy
    except Exception as e:
        logger.error(
            "health_check_failed",
            component="claude_sdk",
            error=str(e),
            error_type=type(e).__name__,
        )
        return False

close async

close()

Close the client and cleanup resources.

Source code in ccproxy/claude_sdk/client.py

async def close(self) -> None:
    """Close the client and cleanup resources."""
    # Claude Code SDK doesn't require explicit cleanup
    pass

ClaudeSDKConnectionError

Bases: ClaudeSDKError

Raised when unable to connect to Claude Code.

ClaudeSDKError

Bases: Exception

Base exception for Claude SDK errors.

ClaudeSDKProcessError

Bases: ClaudeSDKError

Raised when Claude Code process fails.

MessageConverter

Handles conversion between Anthropic API format and Claude SDK format.

format_messages_to_prompt staticmethod

format_messages_to_prompt(messages)

Convert Anthropic messages format to a single prompt string.

Parameters:

Name	Type	Description	Default
messages	list[dict[str, Any]]	List of messages in Anthropic format	required

Returns:

Type	Description
str	Single prompt string formatted for Claude SDK

Source code in ccproxy/claude_sdk/converter.py

@staticmethod
def format_messages_to_prompt(messages: list[dict[str, Any]]) -> str:
    """
    Convert Anthropic messages format to a single prompt string.

    Args:
        messages: List of messages in Anthropic format

    Returns:
        Single prompt string formatted for Claude SDK
    """
    prompt_parts = []

    for message in messages:
        role = message.get("role", "")
        content = message.get("content", "")

        if isinstance(content, list):
            # Handle content blocks
            text_parts = []
            for block in content:
                if block.get("type") == "text":
                    text_parts.append(block.get("text", ""))
            content = " ".join(text_parts)

        if role == "user":
            prompt_parts.append(f"Human: {content}")
        elif role == "assistant":
            prompt_parts.append(f"Assistant: {content}")
        elif role == "system":
            # System messages are handled via options
            continue

    return "\n\n".join(prompt_parts)

extract_text_from_content staticmethod

extract_text_from_content(content)

Extract text content from Claude SDK content blocks.

Parameters:

Name	Type	Description	Default
content	TextBlock | ToolUseBlock | ToolResultBlock	List of content blocks from Claude SDK	required

Returns:

Type	Description
str	Extracted text content

Source code in ccproxy/claude_sdk/converter.py

@staticmethod
def extract_text_from_content(
    content: TextBlock | ToolUseBlock | ToolResultBlock,
) -> str:
    """
    Extract text content from Claude SDK content blocks.

    Args:
        content: List of content blocks from Claude SDK

    Returns:
        Extracted text content
    """
    if isinstance(content, TextBlock):
        return content.text
    elif isinstance(content, ToolUseBlock):
        # Return full XML representation of ToolUseBlock
        tool_id = escape(str(getattr(content, "id", f"tool_{id(content)}")))
        tool_name = escape(content.name)
        tool_input = getattr(content, "input", {}) or {}
        # Convert input dict to JSON string and escape for XML
        input_json = escape(json.dumps(tool_input, ensure_ascii=False))
        return f'<tooluseblock id="{tool_id}" name="{tool_name}">{input_json}</tooluseblock>'
    elif isinstance(content, ToolResultBlock):
        # Return full XML representation of ToolResultBlock
        tool_use_id = escape(str(getattr(content, "tool_use_id", "")))
        result_content = content.content if isinstance(content.content, str) else ""
        escaped_content = escape(result_content)
        return f'<toolresultblock tool_use_id="{tool_use_id}">{escaped_content}</toolresultblock>'

extract_contents staticmethod

extract_contents(contents)

Extract content from Claude SDK blocks, preserving custom blocks.

Parameters:

Name	Type	Description	Default
content		List of content blocks from Claude SDK	required

Returns:

Type	Description
str	Content with thinking blocks preserved

Source code in ccproxy/claude_sdk/converter.py

@staticmethod
def extract_contents(
    contents: list[TextBlock | ToolUseBlock | ToolResultBlock],
) -> str:
    """
    Extract content from Claude SDK blocks, preserving custom blocks.

    Args:
        content: List of content blocks from Claude SDK

    Returns:
        Content with thinking blocks preserved
    """
    text_parts = []

    for block in contents:
        text_parts.append(MessageConverter.extract_text_from_content(block))

    return "\n".join(text_parts)

convert_to_anthropic_response staticmethod

convert_to_anthropic_response(
    assistant_message, result_message, model
)

Convert Claude SDK messages to Anthropic API response format.

Parameters:

Name	Type	Description	Default
assistant_message	AssistantMessage	The assistant message from Claude SDK	required
result_message	ResultMessage	The result message from Claude SDK	required
model	str	The model name used	required

Returns:

Type	Description
dict[str, Any]	Response in Anthropic API format

Source code in ccproxy/claude_sdk/converter.py

@staticmethod
def convert_to_anthropic_response(
    assistant_message: AssistantMessage,
    result_message: ResultMessage,
    model: str,
) -> dict[str, Any]:
    """
    Convert Claude SDK messages to Anthropic API response format.

    Args:
        assistant_message: The assistant message from Claude SDK
        result_message: The result message from Claude SDK
        model: The model name used

    Returns:
        Response in Anthropic API format
    """
    # Extract token usage from result message
    # First try to get usage from the usage field (preferred method)
    usage = getattr(result_message, "usage", {})
    if usage:
        input_tokens = usage.get("input_tokens", 0)
        output_tokens = usage.get("output_tokens", 0)
        cache_read_tokens = usage.get("cache_read_input_tokens", 0)
        cache_write_tokens = usage.get("cache_creation_input_tokens", 0)
    else:
        # Fallback to direct attributes
        input_tokens = getattr(result_message, "input_tokens", 0)
        output_tokens = getattr(result_message, "output_tokens", 0)
        cache_read_tokens = getattr(result_message, "cache_read_tokens", 0)
        cache_write_tokens = getattr(result_message, "cache_write_tokens", 0)

    # Log token extraction for debugging
    from structlog import get_logger

    logger = get_logger(__name__)

    logger.debug(
        "assistant_message_content",
        content_blocks=[
            type(block).__name__ for block in assistant_message.content
        ],
        content_count=len(assistant_message.content),
        first_block_text=(
            assistant_message.content[0].text[:100]
            if assistant_message.content
            and hasattr(assistant_message.content[0], "text")
            else None
        ),
    )

    logger.debug(
        "token_usage_extracted",
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cache_read_tokens=cache_read_tokens,
        cache_write_tokens=cache_write_tokens,
        source="claude_sdk",
    )

    # Calculate total tokens
    total_tokens = input_tokens + output_tokens

    # Build usage information
    usage_info = {
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "cache_read_tokens": cache_read_tokens,
        "cache_write_tokens": cache_write_tokens,
        "total_tokens": total_tokens,
    }

    # Add cost information if available
    total_cost_usd = getattr(result_message, "total_cost_usd", None)
    if total_cost_usd is not None:
        usage_info["cost_usd"] = total_cost_usd

    # Convert content blocks to Anthropic format, preserving thinking blocks
    content_blocks = []

    for block in assistant_message.content:
        if isinstance(block, TextBlock):
            # Parse text content for thinking blocks
            text = block.text

            # Check if the text contains thinking blocks
            import re

            thinking_pattern = r'<thinking signature="([^"]*)">(.*?)</thinking>'

            # Split the text by thinking blocks
            last_end = 0
            for match in re.finditer(thinking_pattern, text, re.DOTALL):
                # Add any text before the thinking block
                before_text = text[last_end : match.start()].strip()
                if before_text:
                    content_blocks.append({"type": "text", "text": before_text})

                # Add the thinking block
                signature, thinking_text = match.groups()
                content_blocks.append(
                    {
                        "type": "thinking",
                        "text": thinking_text,
                        "signature": signature,
                    }
                )

                last_end = match.end()

            # Add any remaining text after the last thinking block
            remaining_text = text[last_end:].strip()
            if remaining_text:
                content_blocks.append({"type": "text", "text": remaining_text})

            # If no thinking blocks were found, add the entire text as a text block
            if last_end == 0 and text:
                content_blocks.append({"type": "text", "text": text})

        elif isinstance(block, ToolUseBlock):
            tool_input = getattr(block, "input", {}) or {}
            content_blocks.append(
                cast(
                    dict[str, Any],
                    {
                        "type": "tool_use",
                        "id": getattr(block, "id", f"tool_{id(block)}"),
                        "name": block.name,
                        "input": tool_input,
                    },
                )
            )
        elif isinstance(block, ToolResultBlock):
            content_blocks.append(
                {
                    "type": "tool_result",
                    "tool_use_id": getattr(block, "tool_use_id", ""),
                    "content": block.content
                    if isinstance(block.content, str)
                    else "",
                }
            )

    return {
        "id": f"msg_{result_message.session_id}",
        "type": "message",
        "role": "assistant",
        "content": content_blocks,
        "model": model,
        "stop_reason": getattr(result_message, "stop_reason", "end_turn"),
        "stop_sequence": None,
        "usage": usage_info,
    }

create_streaming_start_chunk staticmethod

create_streaming_start_chunk(message_id, model)

Create the initial streaming chunk for Anthropic API format.

Parameters:

Name	Type	Description	Default
message_id	str	The message ID	required
model	str	The model name	required

Returns:

Type	Description
dict[str, Any]	Initial streaming chunk

Source code in ccproxy/claude_sdk/converter.py

@staticmethod
def create_streaming_start_chunk(message_id: str, model: str) -> dict[str, Any]:
    """
    Create the initial streaming chunk for Anthropic API format.

    Args:
        message_id: The message ID
        model: The model name

    Returns:
        Initial streaming chunk
    """
    return {
        "id": message_id,
        "type": "message_start",
        "message": {
            "id": message_id,
            "type": "message",
            "role": "assistant",
            "content": [],
            "model": model,
            "stop_reason": None,
            "stop_sequence": None,
            "usage": {
                "input_tokens": 0,
                "output_tokens": 0,
                "total_tokens": 0,
            },
        },
    }

create_streaming_delta_chunk staticmethod

create_streaming_delta_chunk(text)

Create a streaming delta chunk for Anthropic API format.

Parameters:

Name	Type	Description	Default
text	str	The text content to include	required

Returns:

Type	Description
dict[str, Any]	Delta chunk

Source code in ccproxy/claude_sdk/converter.py

@staticmethod
def create_streaming_delta_chunk(text: str) -> dict[str, Any]:
    """
    Create a streaming delta chunk for Anthropic API format.

    Args:
        text: The text content to include

    Returns:
        Delta chunk
    """
    return {
        "type": "content_block_delta",
        "index": 0,
        "delta": {"type": "text_delta", "text": text},
    }

create_streaming_end_chunk staticmethod

create_streaming_end_chunk(stop_reason='end_turn')

Create the final streaming chunk for Anthropic API format.

Parameters:

Name	Type	Description	Default
stop_reason	str	The reason for stopping	'end_turn'

Returns:

Type	Description
dict[str, Any]	Final streaming chunk

Source code in ccproxy/claude_sdk/converter.py

@staticmethod
def create_streaming_end_chunk(stop_reason: str = "end_turn") -> dict[str, Any]:
    """
    Create the final streaming chunk for Anthropic API format.

    Args:
        stop_reason: The reason for stopping

    Returns:
        Final streaming chunk
    """
    return {
        "type": "message_delta",
        "delta": {"stop_reason": stop_reason},
        "usage": {"output_tokens": 0},
    }

OptionsHandler

OptionsHandler(settings=None)

Handles creation and management of Claude SDK options.

Parameters:

Name	Type	Description	Default
settings	Settings | None	Application settings containing default Claude options	None

Source code in ccproxy/claude_sdk/options.py

def __init__(self, settings: Settings | None = None) -> None:
    """
    Initialize options handler.

    Args:
        settings: Application settings containing default Claude options
    """
    self.settings = settings

create_options

create_options(
    model,
    temperature=None,
    max_tokens=None,
    system_message=None,
    **kwargs,
)

Create Claude SDK options from API parameters.

Parameters:

Name	Type	Description	Default
model	str	The model name	required
temperature	float | None	Temperature for response generation	None
max_tokens	int | None	Maximum tokens in response	None
system_message	str | None	System message to include	None
**kwargs	Any	Additional options	{}

Returns:

Type	Description
ClaudeCodeOptions	Configured ClaudeCodeOptions instance

Source code in ccproxy/claude_sdk/options.py

def create_options(
    self,
    model: str,
    temperature: float | None = None,
    max_tokens: int | None = None,
    system_message: str | None = None,
    **kwargs: Any,
) -> ClaudeCodeOptions:
    """
    Create Claude SDK options from API parameters.

    Args:
        model: The model name
        temperature: Temperature for response generation
        max_tokens: Maximum tokens in response
        system_message: System message to include
        **kwargs: Additional options

    Returns:
        Configured ClaudeCodeOptions instance
    """
    options = ClaudeCodeOptions(model=model)

    # First apply settings from configuration if available
    if self.settings and self.settings.claude.code_options:
        code_opts = self.settings.claude.code_options

        # Apply settings from configuration
        for attr_name in dir(code_opts):
            if not attr_name.startswith("_"):
                value = getattr(code_opts, attr_name, None)
                if value is not None and hasattr(options, attr_name):
                    setattr(options, attr_name, value)

    # Then apply API parameters (these override settings)
    if temperature is not None:
        options.temperature = temperature  # type: ignore[attr-defined]

    if max_tokens is not None:
        options.max_tokens = max_tokens  # type: ignore[attr-defined]

    if system_message is not None:
        options.system_prompt = system_message

    # Handle other options as needed
    for key, value in kwargs.items():
        if hasattr(options, key):
            setattr(options, key, value)

    return options

extract_system_message staticmethod

extract_system_message(messages)

Extract system message from Anthropic messages format.

Parameters:

Name	Type	Description	Default
messages	list[dict[str, Any]]	List of messages in Anthropic format	required

Returns:

Type	Description
str | None	System message content if found, None otherwise

Source code in ccproxy/claude_sdk/options.py

@staticmethod
def extract_system_message(messages: list[dict[str, Any]]) -> str | None:
    """
    Extract system message from Anthropic messages format.

    Args:
        messages: List of messages in Anthropic format

    Returns:
        System message content if found, None otherwise
    """
    for message in messages:
        if message.get("role") == "system":
            content = message.get("content", "")
            if isinstance(content, list):
                # Handle content blocks
                text_parts = []
                for block in content:
                    if block.get("type") == "text":
                        text_parts.append(block.get("text", ""))
                return " ".join(text_parts)
            return str(content)
    return None

get_supported_models staticmethod

get_supported_models()

Get list of supported Claude models.

Returns:

Type	Description
list[str]	List of supported model names

Source code in ccproxy/claude_sdk/options.py

@staticmethod
def get_supported_models() -> list[str]:
    """
    Get list of supported Claude models.

    Returns:
        List of supported model names
    """
    # Import here to avoid circular imports
    from ccproxy.adapters.openai.adapter import OPENAI_TO_CLAUDE_MODEL_MAPPING

    # Extract unique Claude models from OpenAI mapping
    claude_models = list(set(OPENAI_TO_CLAUDE_MODEL_MAPPING.values()))
    return sorted(claude_models)

validate_model staticmethod

validate_model(model)

Validate if a model is supported.

Parameters:

Name	Type	Description	Default
model	str	The model name to validate	required

Returns:

Type	Description
bool	True if supported, False otherwise

Source code in ccproxy/claude_sdk/options.py

@staticmethod
def validate_model(model: str) -> bool:
    """
    Validate if a model is supported.

    Args:
        model: The model name to validate

    Returns:
        True if supported, False otherwise
    """
    return model in OptionsHandler.get_supported_models()

get_default_options staticmethod

get_default_options()

Get default options for Claude SDK.

Returns:

Type	Description
dict[str, Any]	Dictionary of default options

Source code in ccproxy/claude_sdk/options.py

@staticmethod
def get_default_options() -> dict[str, Any]:
    """
    Get default options for Claude SDK.

    Returns:
        Dictionary of default options
    """
    return {
        "model": "claude-3-5-sonnet-20241022",
        "temperature": 0.7,
        "max_tokens": 4000,
    }