Skip to content

ccproxy.plugins.claude_sdk.tasks

ccproxy.plugins.claude_sdk.tasks

Scheduled tasks for Claude SDK plugin.

ClaudeSDKDetectionRefreshTask 

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

Bases: BaseScheduledTask

Task to periodically refresh Claude CLI detection.

Parameters:

Name Type Description Default
name str

Task name

 required
interval_seconds float

How often to run the task

 required
detection_service ClaudeSDKDetectionService

Claude CLI detection service

 required
enabled bool

Whether the task is enabled

 True
skip_initial_run bool

Whether to skip the first run

 True
**kwargs Any

Additional task arguments

 {}
Source code in ccproxy/plugins/claude_sdk/tasks.py 
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
def __init__(
    self,
    name: str,
    interval_seconds: float,
    detection_service: "ClaudeSDKDetectionService",
    enabled: bool = True,
    skip_initial_run: bool = True,
    **kwargs: Any,
):
    """Initialize the detection refresh task.

    Args:
        name: Task name
        interval_seconds: How often to run the task
        detection_service: Claude CLI detection service
        enabled: Whether the task is enabled
        skip_initial_run: Whether to skip the first run
        **kwargs: Additional task arguments
    """
    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 Claude CLI detection refresh.

Returns:

Type Description
bool

True if successful, False otherwise
Source code in ccproxy/plugins/claude_sdk/tasks.py 
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
async def run(self) -> bool:
    """Execute the Claude CLI detection refresh.

    Returns:
        True if successful, False otherwise
    """
    if self._first_run and self.skip_initial_run:
        self._first_run = False
        logger.debug(
            "claude_sdk_detection_refresh_skipped_initial",
            task_name=self.name,
        )
        return True

    self._first_run = False

    try:
        logger.debug(
            "claude_sdk_detection_refresh_starting",
            task_name=self.name,
        )

        # Refresh Claude CLI detection
        detection_data = await self.detection_service.initialize_detection()

        logger.debug(
            "claude_sdk_detection_refresh_completed",
            task_name=self.name,
            version=detection_data.claude_version or "unknown",
            cli_command=detection_data.cli_command,
            is_available=detection_data.is_available,
        )
        return True

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

setup async 

setup()

Setup before task execution starts.

Source code in ccproxy/plugins/claude_sdk/tasks.py 
91
92
93
async def setup(self) -> None:
    """Setup before task execution starts."""
    pass

cleanup async 

cleanup()

Cleanup after task execution stops.

Source code in ccproxy/plugins/claude_sdk/tasks.py 
95
96
97
async def cleanup(self) -> None:
    """Cleanup after task execution stops."""
    pass