Skip to content

ccproxy.plugins.claude_sdk.health

ccproxy.plugins.claude_sdk.health

Health check implementation for Claude SDK plugin.

claude_sdk_health_check async 

claude_sdk_health_check(
    config, detection_service, *, version
)

Perform health check for Claude SDK plugin.

Parameters:

Name Type Description Default
config ClaudeSDKSettings | None

Claude SDK plugin configuration

 required
detection_service ClaudeSDKDetectionService | None

Claude CLI detection service

 required

Returns:

Type Description
HealthCheckResult

HealthCheckResult with plugin status
Source code in ccproxy/plugins/claude_sdk/health.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
async def claude_sdk_health_check(
    config: "ClaudeSDKSettings | None",
    detection_service: "ClaudeSDKDetectionService | None",
    *,
    version: str,
) -> HealthCheckResult:
    """Perform health check for Claude SDK plugin.

    Args:
        config: Claude SDK plugin configuration
        detection_service: Claude CLI detection service

    Returns:
        HealthCheckResult with plugin status
    """
    checks = []
    status: str = "pass"

    # Check if plugin is enabled
    if not config or not config.enabled:
        return HealthCheckResult(
            status="fail",
            componentId="plugin-claude_sdk",
            output="Plugin is disabled",
            version=version,
            details={"enabled": False},
        )

    # Check Claude CLI detection
    if detection_service:
        cli_version = detection_service.get_version()
        cli_path = detection_service.get_cli_path()
        is_available = detection_service.is_claude_available()
        cli_info = detection_service.get_cli_health_info()

        if is_available and cli_path:
            checks.append(f"CLI: {cli_version or 'detected'} at {cli_path}")
        else:
            checks.append("CLI: not found")
            status = "warn"  # CLI not found is a warning, not a failure
    else:
        checks.append("CLI: detection service not initialized")
        status = "warn"

    # Check configuration
    if config:
        checks.append(f"Models: {len(config.models_endpoint)} configured")
        checks.append(
            f"Session pool: {'enabled' if config.session_pool_enabled else 'disabled'}"
        )
        checks.append(
            f"Streaming: {'enabled' if config.supports_streaming else 'disabled'}"
        )
    else:
        checks.append("Config: not loaded")
        status = "fail"

    # Standardized details
    from ccproxy.core.plugins.models import (
        CLIHealth,
        ConfigHealth,
        ProviderHealthDetails,
    )

    cli_health = None
    if detection_service:
        path_list = detection_service.get_cli_path()
        cli_status = cli_info.status.value if cli_info else "unknown"
        cli_health = CLIHealth(
            available=bool(detection_service.is_claude_available()),
            status=cli_status,
            version=detection_service.get_version(),
            path=(path_list[0] if path_list else None),
        )

    details = ProviderHealthDetails(
        provider="claude_sdk",
        enabled=bool(config and config.enabled),
        base_url=None,
        cli=cli_health,
        auth=None,
        config=ConfigHealth(
            model_count=len(config.models_endpoint)
            if config and config.models_endpoint
            else 0,
            supports_openai_format=config.supports_streaming if config else None,
            extra={
                "session_pool_enabled": bool(config.session_pool_enabled)
                if config
                else None
            },
        ),
    ).model_dump()

    return HealthCheckResult(
        status=cast(Literal["pass", "warn", "fail"], status),
        componentId="plugin-claude_sdk",
        output="; ".join(checks),
        version=version,
        details=details,
    )