ccproxy.claude_sdk.client

ccproxy.claude_sdk.client

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

ClaudeSDKClient

ClaudeSDKClient(settings=None, session_manager=None)

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. Supports both stateless query() calls and pooled connection reuse for improved performance.

Parameters:

Name	Type	Description	Default
settings	Settings | None	Application settings for session pool configuration	None
session_manager	SessionManager | None	Optional SessionManager instance for dependency injection	None

Source code in ccproxy/claude_sdk/client.py

def __init__(
    self,
    settings: Settings | None = None,
    session_manager: SessionManager | None = None,
) -> None:
    """Initialize the Claude SDK client.

    Args:
        settings: Application settings for session pool configuration
        session_manager: Optional SessionManager instance for dependency injection
    """
    self._last_api_call_time_ms: float = 0.0
    self._settings = settings
    self._session_manager = session_manager

query_completion async

query_completion(
    message, options, request_id=None, session_id=None
)

Execute a query using the Claude Code SDK and return a StreamHandle.

Parameters:

Name	Type	Description	Default
message	SDKMessage	SDKMessage to send to Claude SDK	required
options	ClaudeCodeOptions	Claude Code options configuration	required
request_id	str | None	Optional request ID for correlation	None
session_id	str | None	Optional session ID for conversation continuity	None

Returns:

Type	Description
StreamHandle	StreamHandle that can create listeners for the stream

Raises:

Type	Description
ClaudeSDKError	If the query fails

Source code in ccproxy/claude_sdk/client.py

async def query_completion(
    self,
    message: SDKMessage,
    options: ClaudeCodeOptions,
    request_id: str | None = None,
    session_id: str | None = None,
) -> StreamHandle:
    """
    Execute a query using the Claude Code SDK and return a StreamHandle.

    Args:
        message: SDKMessage to send to Claude SDK
        options: Claude Code options configuration
        request_id: Optional request ID for correlation
        session_id: Optional session ID for conversation continuity

    Returns:
        StreamHandle that can create listeners for the stream

    Raises:
        ClaudeSDKError: If the query fails
    """
    # Determine routing strategy
    if self._should_use_session_pool(session_id):
        return await self._create_session_pool_stream_handle(
            message, options, request_id, session_id
        )
    else:
        return await self._create_direct_stream_handle(
            message, options, request_id, session_id
        )

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

interrupt_session async

interrupt_session(session_id)

Interrupt a specific session due to client disconnection.

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to interrupt	required

Returns:

Type	Description
bool	True if session was found and interrupted, False otherwise

Source code in ccproxy/claude_sdk/client.py

async def interrupt_session(self, session_id: str) -> bool:
    """Interrupt a specific session due to client disconnection.

    Args:
        session_id: The session ID to interrupt

    Returns:
        True if session was found and interrupted, False otherwise
    """
    logger.debug("sdk_client_interrupt_session_started", session_id=session_id)
    if self._session_manager:
        logger.info(
            "client_interrupt_session_requested",
            session_id=session_id,
            has_session_manager=True,
        )
        return await self._session_manager.interrupt_session(session_id)
    else:
        logger.warning(
            "client_interrupt_session_no_session_manager",
            session_id=session_id,
        )
        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