ccproxy.plugins.claude_api.detection_service

ccproxy.plugins.claude_api.detection_service

Claude API plugin detection service using centralized detection.

ClaudeAPIDetectionService

ClaudeAPIDetectionService(
    settings, cli_service=None, redact_sensitive_cache=True
)

Claude API plugin detection service for automatically detecting Claude CLI headers.

Parameters:

Name	Type	Description	Default
settings	Settings	Application settings	required
cli_service	CLIDetectionService | None	Optional CLIDetectionService instance for dependency injection. If None, creates a new instance for backward compatibility.	None

Source code in ccproxy/plugins/claude_api/detection_service.py

def __init__(
    self,
    settings: Settings,
    cli_service: CLIDetectionService | None = None,
    redact_sensitive_cache: bool = True,
) -> None:
    """Initialize Claude detection service.

    Args:
        settings: Application settings
        cli_service: Optional CLIDetectionService instance for dependency injection.
                    If None, creates a new instance for backward compatibility.
    """
    self.settings = settings
    self.cache_dir = get_ccproxy_cache_dir()
    self.cache_dir.mkdir(parents=True, exist_ok=True)
    self._cached_data: ClaudeCacheData | None = None
    self._cli_service = cli_service or CLIDetectionService(settings)
    self._cli_info: ClaudeCliInfo | None = None
    self._redact_sensitive_cache = redact_sensitive_cache

initialize_detection async

initialize_detection()

Initialize Claude detection at startup.

Source code in ccproxy/plugins/claude_api/detection_service.py

async def initialize_detection(self) -> ClaudeCacheData:
    """Initialize Claude detection at startup."""
    try:
        # Get current Claude version
        current_version = await self._get_claude_version()

        # Try to load from cache first
        cached = False
        try:
            detected_data = self._load_from_cache(current_version)
            cached = detected_data is not None

        except Exception as e:
            logger.warning(
                "invalid_cache_file",
                error=str(e),
                category="plugin",
                exc_info=e,
            )

        if not cached:
            # No cache or version changed - detect fresh
            detected_data = await self._detect_claude_headers(current_version)
            # Cache the results
            self._save_to_cache(detected_data)

        self._cached_data = detected_data

        logger.trace(
            "detection_headers_completed",
            version=current_version,
            cached=cached,
        )

        if detected_data is None:
            raise ValueError("Claude detection failed")
        return detected_data

    except Exception as e:
        logger.warning(
            "detection_claude_headers_failed",
            fallback=True,
            error=e,
            category="plugin",
        )
        # Return fallback data
        fallback_data = self._get_fallback_data()
        self._cached_data = fallback_data
        return fallback_data

get_cached_data

get_cached_data()

Get currently cached detection data.

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_cached_data(self) -> ClaudeCacheData | None:
    """Get currently cached detection data."""
    return self._cached_data

get_detected_headers

get_detected_headers()

Return cached headers as structured data.

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_detected_headers(self) -> DetectedHeaders:
    """Return cached headers as structured data."""

    data = self.get_cached_data()
    if not data:
        return DetectedHeaders()
    return data.headers

get_detected_prompts

get_detected_prompts()

Return cached prompt metadata as structured data.

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_detected_prompts(self) -> DetectedPrompts:
    """Return cached prompt metadata as structured data."""

    data = self.get_cached_data()
    if not data:
        return DetectedPrompts()
    return data.prompts

get_ignored_headers

get_ignored_headers()

Headers that should be ignored when injecting CLI values.

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_ignored_headers(self) -> list[str]:
    """Headers that should be ignored when injecting CLI values."""

    return list(self.ignores_header)

get_redacted_headers

get_redacted_headers()

Headers that must never be forwarded from detection cache.

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_redacted_headers(self) -> list[str]:
    """Headers that must never be forwarded from detection cache."""

    return list(self.redact_headers)

get_cli_health_info

get_cli_health_info()

Get lightweight CLI health info using centralized detection, cached locally.

Returns:

Type	Description
ClaudeCliInfo	ClaudeCliInfo with availability, version, and binary path

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_cli_health_info(self) -> ClaudeCliInfo:
    """Get lightweight CLI health info using centralized detection, cached locally.

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

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

    info = self._cli_service.get_cli_info("claude")
    status = (
        ClaudeCliStatus.AVAILABLE
        if info["is_available"]
        else ClaudeCliStatus.NOT_INSTALLED
    )
    cli_info = ClaudeCliInfo(
        status=status,
        version=info.get("version"),
        binary_path=info.get("path"),
    )
    self._cli_info = cli_info
    return cli_info

get_version

get_version()

Get the detected Claude CLI version.

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_version(self) -> str | None:
    """Get the detected Claude CLI version."""
    if self._cached_data:
        return self._cached_data.claude_version
    return None

get_cli_path

get_cli_path()

Get the Claude CLI command with caching.

Returns:

Type	Description
list[str] | None	Command list to execute Claude CLI if found, None otherwise

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_cli_path(self) -> list[str] | None:
    """Get the Claude CLI command with caching.

    Returns:
        Command list to execute Claude CLI if found, None otherwise
    """
    info = self._cli_service.get_cli_info("claude")
    return info["command"] if info["is_available"] else None

get_binary_path

get_binary_path()

Alias for get_cli_path for consistency with Codex.

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_binary_path(self) -> list[str] | None:
    """Alias for get_cli_path for consistency with Codex."""
    return self.get_cli_path()

invalidate_cache

invalidate_cache()

Clear all cached detection data.

Source code in ccproxy/plugins/claude_api/detection_service.py

def invalidate_cache(self) -> None:
    """Clear all cached detection data."""
    # Clear the async cache for _get_claude_version
    if hasattr(self._get_claude_version, "cache_clear"):
        self._get_claude_version.cache_clear()
    # Clear CLI info cache
    self._cli_info = None
    logger.debug("detection_cache_cleared", category="plugin")

get_system_prompt

get_system_prompt(mode='minimal')

Return a system prompt dict for injection based on cached prompts.

mode: "none", "minimal", or "full"

Source code in ccproxy/plugins/claude_api/detection_service.py

def get_system_prompt(self, mode: str | None = "minimal") -> dict[str, Any]:
    """Return a system prompt dict for injection based on cached prompts.

    mode: "none", "minimal", or "full"
    """
    prompts = self.get_detected_prompts()
    mode_value = "full" if mode is None else mode
    return prompts.system_payload(mode=mode_value)