ccproxy.claude_sdk.client

ccproxy.claude_sdk.client

Claude SDK client wrapper for handling core Claude Code SDK interactions.

ClaudeSDKError

Bases: Exception

Base exception for Claude SDK errors.

ClaudeSDKConnectionError

Bases: ClaudeSDKError

Raised when unable to connect to Claude Code.

ClaudeSDKProcessError

Bases: ClaudeSDKError

Raised when Claude Code process fails.

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