Skip to content

ccproxy.plugins.claude_sdk.detection_service

ccproxy.plugins.claude_sdk.detection_service

Claude SDK CLI detection service using centralized detection.

ClaudeDetectionData

Bases: NamedTuple

Detection data for Claude CLI.

ClaudeSDKDetectionService

ClaudeSDKDetectionService(settings, cli_service=None)

Service for detecting Claude CLI availability.

This detection service checks if the Claude CLI exists either as a direct binary in PATH or via package manager execution (e.g., bunx). Unlike the Claude API plugin, this doesn't support fallback mode as the SDK requires the actual CLI to be present.

Parameters:

Name Type Description Default
settings Settings

Application settings

required
cli_service CLIDetectionService | None

Optional CLI detection service instance. If None, creates a new one.

None
Source code in ccproxy/plugins/claude_sdk/detection_service.py
def __init__(
    self, settings: Settings, cli_service: CLIDetectionService | None = None
) -> None:
    """Initialize the Claude SDK detection service.

    Args:
        settings: Application settings
        cli_service: Optional CLI detection service instance. If None, creates a new one.
    """
    self.settings = settings
    self._cli_service = cli_service or CLIDetectionService(settings)
    self._version: str | None = None
    self._cli_command: list[str] | None = None
    self._is_available = False
    self._cli_info: ClaudeCliInfoType | None = None

initialize_detection async

initialize_detection()

Initialize Claude CLI detection with caching.

Returns:

Type Description
ClaudeDetectionData

ClaudeDetectionData with detection results

Note

No fallback support - SDK requires actual CLI presence

Source code in ccproxy/plugins/claude_sdk/detection_service.py
@async_ttl_cache(maxsize=16, ttl=600.0)  # 10 minute cache for CLI detection
async def initialize_detection(self) -> ClaudeDetectionData:
    """Initialize Claude CLI detection with caching.

    Returns:
        ClaudeDetectionData with detection results

    Note:
        No fallback support - SDK requires actual CLI presence
    """
    logger.debug("detection_starting", category="plugin")

    # Use centralized CLI detection service
    # For SDK, we don't want fallback - require actual CLI
    original_fallback = self._cli_service.resolver.fallback_enabled
    self._cli_service.resolver.fallback_enabled = False

    try:
        result = await self._cli_service.detect_cli(
            binary_name="claude",
            package_name="@anthropic-ai/claude-code",
            version_flag="--version",
            fallback_data=None,  # No fallback for SDK
            cache_key="claude_sdk",
        )

        # Accept both direct binary and package manager execution
        if result.is_available:
            self._version = result.version
            self._cli_command = result.command
            self._is_available = True
            logger.debug(
                "cli_detection_completed",
                cli_command=self._cli_command,
                version=self._version,
                source=result.source,
                cached=hasattr(result, "cached") and result.cached,
                category="plugin",
            )
        else:
            self._is_available = False
            logger.error(
                "claude_sdk_detection_failed",
                message="Claude CLI not found - SDK plugin cannot function without CLI",
                category="plugin",
            )
    finally:
        # Restore original fallback setting
        self._cli_service.resolver.fallback_enabled = original_fallback

    return ClaudeDetectionData(
        claude_version=self._version,
        cli_command=self._cli_command,
        is_available=self._is_available,
    )

get_version

get_version()

Get the detected Claude CLI version.

Returns:

Type Description
str | None

Version string if available, None otherwise

Source code in ccproxy/plugins/claude_sdk/detection_service.py
def get_version(self) -> str | None:
    """Get the detected Claude CLI version.

    Returns:
        Version string if available, None otherwise
    """
    return self._version

get_cli_path

get_cli_path()

Get the detected Claude CLI command.

Returns:

Type Description
list[str] | None

CLI command list if available, None otherwise

Source code in ccproxy/plugins/claude_sdk/detection_service.py
def get_cli_path(self) -> list[str] | None:
    """Get the detected Claude CLI command.

    Returns:
        CLI command list if available, None otherwise
    """
    return self._cli_command

is_claude_available

is_claude_available()

Check if Claude CLI is available.

Returns:

Type Description
bool

True if Claude CLI was detected, False otherwise

Source code in ccproxy/plugins/claude_sdk/detection_service.py
def is_claude_available(self) -> bool:
    """Check if Claude CLI is available.

    Returns:
        True if Claude CLI was detected, False otherwise
    """
    return self._is_available

get_cli_health_info

get_cli_health_info()

Return CLI health info model using current detection state.

Returns:

Type Description
Any

ClaudeCliInfo with availability, version, and binary path

Source code in ccproxy/plugins/claude_sdk/detection_service.py
def get_cli_health_info(self) -> Any:
    """Return CLI health info model using current detection state.

    Returns:
        ClaudeCliInfo with availability, version, and binary path
    """
    from ..claude_api.models import ClaudeCliInfo, ClaudeCliStatus

    if self._cli_info is not None:
        return self._cli_info

    status = (
        ClaudeCliStatus.AVAILABLE
        if self._is_available
        else ClaudeCliStatus.NOT_INSTALLED
    )
    cli_info = ClaudeCliInfo(
        status=status,
        version=self._version,
        binary_path=self._cli_command[0] if self._cli_command else None,
    )
    self._cli_info = cli_info
    return cli_info

invalidate_cache

invalidate_cache()

Clear all cached detection data.

Source code in ccproxy/plugins/claude_sdk/detection_service.py
def invalidate_cache(self) -> None:
    """Clear all cached detection data."""
    # Clear the async cache for initialize_detection
    if hasattr(self.initialize_detection, "cache_clear"):
        self.initialize_detection.cache_clear()
    self._cli_info = None
    logger.debug("detection_cache_cleared", category="plugin")