ccproxy.plugins.claude_api.tasks

ccproxy.plugins.claude_api.tasks

Scheduled tasks for Claude API plugin.

ClaudeAPIDetectionRefreshTask

ClaudeAPIDetectionRefreshTask(
    name,
    interval_seconds,
    detection_service,
    enabled=True,
    skip_initial_run=True,
    **kwargs,
)

Bases: BaseScheduledTask

Task to periodically refresh Claude CLI detection headers.

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

def __init__(
    self,
    name: str,
    interval_seconds: float,
    detection_service: "ClaudeAPIDetectionService",
    enabled: bool = True,
    skip_initial_run: bool = True,
    **kwargs: Any,
):
    super().__init__(
        name=name,
        interval_seconds=interval_seconds,
        enabled=enabled,
        **kwargs,
    )
    self.detection_service = detection_service
    self.skip_initial_run = skip_initial_run
    self._first_run = True

run async

run()

Execute the detection refresh.

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

async def run(self) -> bool:
    """Execute the detection refresh."""
    if self._first_run and self.skip_initial_run:
        self._first_run = False
        logger.debug(
            "claude_api_detection_refresh_skipped_initial",
            task_name=self.name,
        )
        return True

    self._first_run = False

    try:
        logger.info(
            "claude_api_detection_refresh_starting",
            task_name=self.name,
        )
        detection_data = await self.detection_service.initialize_detection()

        logger.info(
            "claude_api_detection_refresh_completed",
            task_name=self.name,
            version=detection_data.claude_version if detection_data else "unknown",
        )
        return True

    except Exception as e:
        logger.error(
            "claude_api_detection_refresh_failed",
            task_name=self.name,
            error=str(e),
        )
        return False

setup async

setup()

Setup before task execution starts.

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

async def setup(self) -> None:
    """Setup before task execution starts."""
    logger.debug(
        "claude_api_detection_refresh_setup",
        task_name=self.name,
    )

cleanup async

cleanup()

Cleanup after task execution stops.

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

async def cleanup(self) -> None:
    """Cleanup after task execution stops."""
    logger.info(
        "claude_api_detection_refresh_cleanup",
        task_name=self.name,
    )