ccproxy.plugins.claude_sdk.plugin

ccproxy.plugins.claude_sdk.plugin

Claude SDK plugin v2 implementation.

ClaudeSDKRuntime

ClaudeSDKRuntime(manifest)

Bases: ProviderPluginRuntime

Runtime for Claude SDK plugin.

Source code in ccproxy/plugins/claude_sdk/plugin.py

def __init__(self, manifest: PluginManifest):
    """Initialize runtime."""
    super().__init__(manifest)
    self.session_manager: Any | None = None

ClaudeSDKFactory

ClaudeSDKFactory()

Bases: BaseProviderPluginFactory

Factory for Claude SDK plugin.

Source code in ccproxy/core/plugins/factories.py

def __init__(self) -> None:
    """Initialize factory with manifest built from class attributes."""
    # Validate required class attributes
    self._validate_class_attributes()

    # Validate runtime class is a proper subclass
    # Import locally to avoid circular import during module import
    from .runtime import ProviderPluginRuntime

    if not issubclass(self.runtime_class, ProviderPluginRuntime):
        raise TypeError(
            f"runtime_class {self.runtime_class.__name__} must be a subclass of ProviderPluginRuntime"
        )

    # Build routes from routers list
    routes = []
    for router_spec in self.routers:
        # Handle both router instances and router factory functions
        router_instance = router_spec.router
        if callable(router_spec.router) and not isinstance(
            router_spec.router, APIRouter
        ):
            # Router is a factory function, call it to get the actual router
            router_instance = router_spec.router()

        routes.append(
            RouteSpec(
                router=cast(APIRouter, router_instance),
                prefix=router_spec.prefix,
                tags=router_spec.tags or [],
                dependencies=router_spec.dependencies,
            )
        )

    # Create manifest from class attributes
    manifest = PluginManifest(
        name=self.plugin_name,
        version=self.plugin_version,
        description=self.plugin_description,
        is_provider=True,
        config_class=self.config_class,
        tool_accumulator_class=self.tool_accumulator_class,
        dependencies=self.dependencies.copy(),
        optional_requires=self.optional_requires.copy(),
        routes=routes,
        tasks=self.tasks.copy(),
        format_adapters=self.format_adapters.copy(),
        requires_format_adapters=self.requires_format_adapters.copy(),
        cli_commands=self.cli_commands.copy(),
        cli_arguments=self.cli_arguments.copy(),
    )

    # Format adapter specification validation is deferred to runtime
    # when settings are available via dependency injection

    # Store the manifest and runtime class directly
    # We don't call parent __init__ because ProviderPluginFactory
    # would override our runtime_class with ProviderPluginRuntime
    self.manifest = manifest
    self.runtime_class = self.__class__.runtime_class

create_adapter async

create_adapter(context)

Create the Claude SDK adapter.

This method overrides the base implementation because Claude SDK has different dependencies than HTTP-based adapters.

Parameters:

Name	Type	Description	Default
context	PluginContext	Plugin context	required

Returns:

Type	Description
BaseAdapter	ClaudeSDKAdapter instance

Source code in ccproxy/plugins/claude_sdk/plugin.py

async def create_adapter(self, context: PluginContext) -> BaseAdapter:
    """Create the Claude SDK adapter.

    This method overrides the base implementation because Claude SDK
    has different dependencies than HTTP-based adapters.

    Args:
        context: Plugin context

    Returns:
        ClaudeSDKAdapter instance
    """
    config = context.get("config")
    if not isinstance(config, ClaudeSDKSettings):
        raise RuntimeError("No configuration provided for Claude SDK adapter")

    # Get optional dependencies
    metrics = context.get("metrics")

    # Try to get hook_manager from context (provided by core services)
    hook_manager = context.get("hook_manager")
    if not hook_manager:
        # Try to get from app state as fallback
        app = context.get("app")
        if app and hasattr(app, "state") and hasattr(app.state, "hook_manager"):
            hook_manager = app.state.hook_manager

    if hook_manager:
        logger.debug("claude_sdk_hook_manager_found", source="context_or_app")

    # Create adapter with config and optional dependencies
    # Note: ClaudeSDKAdapter doesn't use an HTTP client, but it still
    #       needs access to the shared format registry so it can
    #       compose request/response converters declared by the core.
    format_registry = None
    service_container = context.get("service_container")
    if service_container:
        try:
            format_registry = service_container.get_format_registry()
        except Exception as exc:  # pragma: no cover - defensive logging
            logger.warning(
                "claude_sdk_format_registry_unavailable",
                error=str(exc),
                category="format",
            )

    adapter = ClaudeSDKAdapter(
        config=config,
        metrics=metrics,
        hook_manager=hook_manager,
        format_registry=format_registry,
        context=context,
    )

    return adapter

create_detection_service

create_detection_service(context)

Create the Claude SDK detection service with validation.

Parameters:

Name	Type	Description	Default
context	PluginContext	Plugin context	required

Returns:

Type	Description
DetectionServiceProtocol	ClaudeSDKDetectionService instance

Source code in ccproxy/plugins/claude_sdk/plugin.py

def create_detection_service(
    self, context: PluginContext
) -> DetectionServiceProtocol:
    """Create the Claude SDK detection service with validation.

    Args:
        context: Plugin context

    Returns:
        ClaudeSDKDetectionService instance
    """
    settings = context.get("settings")
    if not settings:
        raise RuntimeError("No settings provided for Claude SDK detection service")

    cli_service = context.get("cli_detection_service")
    service = ClaudeSDKDetectionService(settings, cli_service)
    return cast(DetectionServiceProtocol, service)

create_credentials_manager async

create_credentials_manager(context)

Create the credentials manager for Claude SDK.

Parameters:

Name	Type	Description	Default
context	PluginContext	Plugin context	required

Returns:

Type	Description
None	None - Claude SDK uses its own authentication mechanism

Source code in ccproxy/plugins/claude_sdk/plugin.py

async def create_credentials_manager(self, context: PluginContext) -> None:
    """Create the credentials manager for Claude SDK.

    Args:
        context: Plugin context

    Returns:
        None - Claude SDK uses its own authentication mechanism
    """
    # Claude SDK doesn't use a traditional credentials manager
    # It uses the built-in CLI authentication
    return None

create_context

create_context(core_services)

Create context and set up detection service in tasks.

Source code in ccproxy/plugins/claude_sdk/plugin.py

def create_context(self, core_services: Any) -> PluginContext:
    """Create context and set up detection service in tasks."""
    # Get base context
    context = super().create_context(core_services)

    # Create detection service early so it can be passed to tasks
    detection_service = self.create_detection_service(context)

    # Update task kwargs with detection service
    for task_spec in self.manifest.tasks:
        if task_spec.task_name == "claude_sdk_detection_refresh":
            task_spec.kwargs["detection_service"] = detection_service

    return context