ccproxy.api.routes.health

ccproxy.api.routes.health

Health check endpoints for CCProxy API Server.

Implements modern health check patterns following 2024 best practices: - /health/live: Liveness probe for Kubernetes (minimal, fast) - /health/ready: Readiness probe for Kubernetes (critical dependencies) - /health: Detailed diagnostics (comprehensive status)

Follows IETF Health Check Response Format draft standard. TODO: health endpoint Content-Type header to only return application/health+json per IETF spec

liveness_probe async

liveness_probe(response)

Liveness probe for Kubernetes.

Minimal health check that only verifies the application process is running. Used by Kubernetes to determine if the pod should be restarted.

Returns:

Type	Description
dict[str, Any]	Simple health status following IETF health check format

Source code in ccproxy/api/routes/health.py

@router.get("/health/live")
async def liveness_probe(response: Response) -> dict[str, Any]:
    """Liveness probe for Kubernetes.

    Minimal health check that only verifies the application process is running.
    Used by Kubernetes to determine if the pod should be restarted.

    Returns:
        Simple health status following IETF health check format
    """
    # Add cache control headers as per best practices
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    response.headers["Content-Type"] = "application/health+json"

    logger.debug("Liveness probe request")

    return {
        "status": "pass",
        "version": __version__,
        "output": "Application process is running",
    }

readiness_probe async

readiness_probe(response)

Readiness probe for Kubernetes.

Checks critical dependencies to determine if the service is ready to accept traffic. Used by Kubernetes to determine if the pod should receive traffic.

Returns:

Type	Description
dict[str, Any]	Readiness status with critical dependency checks

Source code in ccproxy/api/routes/health.py

@router.get("/health/ready")
async def readiness_probe(response: Response) -> dict[str, Any]:
    """Readiness probe for Kubernetes.

    Checks critical dependencies to determine if the service is ready to accept traffic.
    Used by Kubernetes to determine if the pod should receive traffic.

    Returns:
        Readiness status with critical dependency checks
    """
    # Add cache control headers
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    response.headers["Content-Type"] = "application/health+json"

    logger.debug("Readiness probe request")

    # Check OAuth credentials, CLI, and SDK separately
    oauth_status, oauth_details = await _check_oauth2_credentials()
    cli_status, cli_details = await _check_claude_code()
    sdk_status, sdk_details = await _check_claude_sdk()

    # Service is ready if no check returns "fail"
    # "warn" statuses (missing credentials/CLI/SDK) don't prevent readiness
    if oauth_status == "fail" or cli_status == "fail" or sdk_status == "fail":
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        failed_components = []

        if oauth_status == "fail":
            failed_components.append("oauth2_credentials")
        if cli_status == "fail":
            failed_components.append("claude_cli")
        if sdk_status == "fail":
            failed_components.append("claude_sdk")

        return {
            "status": "fail",
            "version": __version__,
            "output": f"Critical dependency error: {', '.join(failed_components)}",
            "checks": {
                "oauth2_credentials": [
                    {
                        "status": oauth_status,
                        "output": oauth_details.get("error", "OAuth credentials error"),
                    }
                ],
                "claude_cli": [
                    {
                        "status": cli_status,
                        "output": cli_details.get("error", "Claude CLI error"),
                    }
                ],
                "claude_sdk": [
                    {
                        "status": sdk_status,
                        "output": sdk_details.get("error", "Claude SDK error"),
                    }
                ],
            },
        }

    return {
        "status": "pass",
        "version": __version__,
        "output": "Service is ready to accept traffic",
        "checks": {
            "oauth2_credentials": [
                {
                    "status": oauth_status,
                    "output": f"OAuth credentials: {oauth_details.get('auth_status', 'unknown')}",
                }
            ],
            "claude_cli": [
                {
                    "status": cli_status,
                    "output": f"Claude CLI: {cli_details.get('cli_status', 'unknown')}",
                }
            ],
            "claude_sdk": [
                {
                    "status": sdk_status,
                    "output": f"Claude SDK: {sdk_details.get('sdk_status', 'unknown')}",
                }
            ],
        },
    }

detailed_health_check async

detailed_health_check(response)

Comprehensive health check for diagnostics and monitoring.

Provides detailed status of all services and dependencies. Used by monitoring dashboards, debugging, and operations teams.

Returns:

Type	Description
dict[str, Any]	Detailed health status following IETF health check format

Source code in ccproxy/api/routes/health.py

@router.get("/health")
async def detailed_health_check(response: Response) -> dict[str, Any]:
    """Comprehensive health check for diagnostics and monitoring.

    Provides detailed status of all services and dependencies.
    Used by monitoring dashboards, debugging, and operations teams.

    Returns:
        Detailed health status following IETF health check format
    """
    # Add cache control headers
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    response.headers["Content-Type"] = "application/health+json"

    logger.debug("Detailed health check request")

    # Perform all health checks
    oauth_status, oauth_details = await _check_oauth2_credentials()
    cli_status, cli_details = await _check_claude_code()
    sdk_status, sdk_details = await _check_claude_sdk()

    # Determine overall status - prioritize failures, then warnings
    overall_status = "pass"
    if oauth_status == "fail" or cli_status == "fail" or sdk_status == "fail":
        overall_status = "fail"
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
    elif oauth_status == "warn" or cli_status == "warn" or sdk_status == "warn":
        overall_status = "warn"
        response.status_code = status.HTTP_200_OK

    current_time = datetime.now(UTC).isoformat()

    return {
        "status": overall_status,
        "version": __version__,
        "serviceId": "claude-code-proxy",
        "description": "CCProxy API Server",
        "time": current_time,
        "checks": {
            "oauth2_credentials": [
                {
                    "componentId": "oauth2-credentials",
                    "componentType": "authentication",
                    "status": oauth_status,
                    "time": current_time,
                    "output": f"OAuth2 credentials: {oauth_details.get('auth_status', 'unknown')}",
                    **oauth_details,
                }
            ],
            "claude_cli": [
                {
                    "componentId": "claude-cli",
                    "componentType": "external_dependency",
                    "status": cli_status,
                    "time": current_time,
                    "output": f"Claude CLI: {cli_details.get('cli_status', 'unknown')}",
                    **cli_details,
                }
            ],
            "claude_sdk": [
                {
                    "componentId": "claude-sdk",
                    "componentType": "python_package",
                    "status": sdk_status,
                    "time": current_time,
                    "output": f"Claude SDK: {sdk_details.get('sdk_status', 'unknown')}",
                    **sdk_details,
                }
            ],
            "proxy_service": [
                {
                    "componentId": "proxy-service",
                    "componentType": "service",
                    "status": "pass",
                    "time": current_time,
                    "output": "Proxy service operational",
                    "version": __version__,
                }
            ],
        },
    }