ccproxy.core.plugins.interfaces

ccproxy.core.plugins.interfaces

Abstract interfaces for the plugin system.

This module contains all abstract base classes and protocols to avoid circular dependencies between factory and runtime modules.

DetectionServiceProtocol

Bases: Protocol

Common capabilities shared by detection services.

TokenManagerProtocol

Bases: Protocol

Minimal async token manager contract used by adapters.

ProfiledTokenManagerProtocol

Bases: TokenManagerProtocol, Protocol

Token manager that can return a lightweight profile snapshot.

PluginFactory

Bases: ABC

Abstract factory for creating plugin runtime instances.

Each plugin must provide a factory that knows how to create its runtime instance from its manifest.

get_manifest abstractmethod

get_manifest()

Get the plugin manifest with static declarations.

Returns:

Type	Description
PluginManifest	Plugin manifest

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

@abstractmethod
def get_manifest(self) -> PluginManifest:
    """Get the plugin manifest with static declarations.

    Returns:
        Plugin manifest
    """
    ...

create_runtime abstractmethod

create_runtime()

Create a runtime instance for this plugin.

Returns:

Type	Description
Any	Plugin runtime instance

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

@abstractmethod
def create_runtime(self) -> Any:
    """Create a runtime instance for this plugin.

    Returns:
        Plugin runtime instance
    """
    ...

create_context abstractmethod

create_context(core_services)

Create the context for plugin initialization.

Parameters:

Name	Type	Description	Default
core_services	ServiceContainer	Core services container	required

Returns:

Type	Description
PluginContext	Plugin context with required services

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

@abstractmethod
def create_context(self, core_services: "ServiceContainer") -> PluginContext:
    """Create the context for plugin initialization.

    Args:
        core_services: Core services container

    Returns:
        Plugin context with required services
    """
    ...

BasePluginFactory

BasePluginFactory(manifest, runtime_class)

Bases: PluginFactory

Base implementation of plugin factory.

This class provides common functionality for creating plugin runtime instances from manifests.

Parameters:

Name	Type	Description	Default
manifest	PluginManifest	Plugin manifest	required
runtime_class	type[Any]	Runtime class to instantiate	required

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

def __init__(self, manifest: PluginManifest, runtime_class: type[Any]):
    """Initialize factory with manifest and runtime class.

    Args:
        manifest: Plugin manifest
        runtime_class: Runtime class to instantiate
    """
    self.manifest = manifest
    self.runtime_class = runtime_class

get_manifest

get_manifest()

Get the plugin manifest.

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

def get_manifest(self) -> PluginManifest:
    """Get the plugin manifest."""
    return self.manifest

create_runtime

create_runtime()

Create a runtime instance.

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

def create_runtime(self) -> Any:
    """Create a runtime instance."""
    return self.runtime_class(self.manifest)

create_context

create_context(service_container)

Create base context for plugin initialization.

Parameters:

Name	Type	Description	Default
service_container	ServiceContainer	Service container with all available services	required

Returns:

Type	Description
PluginContext	Plugin context with base services

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

def create_context(self, service_container: "ServiceContainer") -> PluginContext:
    """Create base context for plugin initialization.

    Args:
        service_container: Service container with all available services

    Returns:
        Plugin context with base services
    """
    context = PluginContext()

    # Set core services
    context.settings = service_container.settings
    context.http_pool_manager = service_container.get_pool_manager()
    context.logger = structlog.get_logger().bind(plugin=self.manifest.name)

    # Add explicit dependency injection services
    context.request_tracer = service_container.get_request_tracer()
    context.streaming_handler = cast(
        "StreamingMetrics", service_container.get_streaming_handler()
    )
    context.metrics = None  # Will be set by plugins if needed

    # Add CLI detection service
    context.cli_detection_service = service_container.get_cli_detection_service()

    # Add scheduler - not available in ServiceContainer, get from app state
    context.scheduler = None  # Will be set from app.state if needed

    # Add plugin registry - not directly in ServiceContainer, get from app state
    context.plugin_registry = None  # Will be set from app.state

    # Add OAuth registry for auth providers
    context.oauth_registry = service_container.get_oauth_registry()

    # Add hook registry and manager
    context.hook_registry = service_container.get_hook_registry()

    # Provide runtime helpers when available in the container
    try:
        from ccproxy.core.plugins.hooks.manager import HookManager

        context.hook_manager = service_container.get_service(HookManager)
    except (ValueError, ImportError):
        context.hook_manager = None

    try:
        context.app = service_container.get_service(FastAPI)
    except ValueError:
        context.app = None

    # Add service container directly
    context.service_container = service_container

    # Add plugin-specific config if available
    # ServiceContainer doesn't have get_plugin_config, so we'll get it from settings directly
    if self.manifest.config_class:
        plugin_config = service_container.settings.plugins.get(self.manifest.name)

        try:
            if plugin_config is None:
                # No explicit config provided; instantiate defaults
                context.config = self.manifest.config_class()
            else:
                # Validate (even if empty dict) to honor model defaults
                validated_config = self.manifest.config_class.model_validate(
                    plugin_config
                )
                context.config = validated_config
        except Exception as exc:  # pragma: no cover - defensive safety
            logger.warning(
                "plugin_config_initialization_failed",
                plugin=self.manifest.name,
                error=str(exc),
            )
            raise

    # Add format registry
    context.format_registry = service_container.get_format_registry()

    return context

ProviderPluginFactory

ProviderPluginFactory(manifest)

Bases: BasePluginFactory

Factory for provider plugins.

Provider plugins require additional components like adapters and detection services that must be created during initialization.

Parameters:

Name	Type	Description	Default
manifest	PluginManifest	Plugin manifest	required

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

def __init__(self, manifest: PluginManifest):
    """Initialize provider plugin factory.

    Args:
        manifest: Plugin manifest
    """
    # Local import to avoid circular dependency at module load time
    from .runtime import ProviderPluginRuntime

    super().__init__(manifest, ProviderPluginRuntime)

    # Validate this is a provider plugin
    if not manifest.is_provider:
        raise ValueError(
            f"Plugin {manifest.name} is not marked as provider but using ProviderPluginFactory"
        )

create_context

create_context(service_container)

Create context with provider-specific components.

Parameters:

Name	Type	Description	Default
core_services		Core services container	required

Returns:

Type	Description
PluginContext	Plugin context with provider components

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

def create_context(self, service_container: "ServiceContainer") -> PluginContext:
    """Create context with provider-specific components.

    Args:
        core_services: Core services container

    Returns:
        Plugin context with provider components
    """
    # Start with base context
    context = super().create_context(service_container)

    # Provider plugins need to create their own adapter and detection service
    # This is typically done in the specific plugin factory implementation
    # Here we just ensure the structure is correct

    return context

create_adapter abstractmethod async

create_adapter(context)

Create the adapter for this provider.

Parameters:

Name	Type	Description	Default
context	PluginContext	Plugin context	required

Returns:

Type	Description
Any	Provider adapter instance

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

@abstractmethod
async def create_adapter(self, context: PluginContext) -> Any:
    """Create the adapter for this provider.

    Args:
        context: Plugin context

    Returns:
        Provider adapter instance
    """
    ...

create_detection_service abstractmethod

create_detection_service(context)

Create the detection service for this provider.

Parameters:

Name	Type	Description	Default
context	PluginContext	Plugin context	required

Returns:

Type	Description
DetectionServiceProtocol | None	Detection service instance or None

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

@abstractmethod
def create_detection_service(
    self, context: PluginContext
) -> DetectionServiceProtocol | None:
    """Create the detection service for this provider.

    Args:
        context: Plugin context

    Returns:
        Detection service instance or None
    """
    ...

create_credentials_manager abstractmethod async

create_credentials_manager(context)

Create the credentials manager for this provider.

Parameters:

Name	Type	Description	Default
context	PluginContext	Plugin context	required

Returns:

Type	Description
BaseTokenManager[Any] | None	Credentials manager instance or None

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

@abstractmethod
async def create_credentials_manager(
    self, context: PluginContext
) -> BaseTokenManager[Any] | None:
    """Create the credentials manager for this provider.

    Args:
        context: Plugin context

    Returns:
        Credentials manager instance or None
    """
    ...

SystemPluginFactory

SystemPluginFactory(manifest)

Bases: BasePluginFactory

Factory for system plugins.

Parameters:

Name	Type	Description	Default
manifest	PluginManifest	Plugin manifest	required

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

def __init__(self, manifest: PluginManifest):
    """Initialize system plugin factory.

    Args:
        manifest: Plugin manifest
    """
    # Local import to avoid circular dependency at module load time
    from .runtime import SystemPluginRuntime

    super().__init__(manifest, SystemPluginRuntime)

    # Validate this is a system plugin
    if manifest.is_provider:
        raise ValueError(
            f"Plugin {manifest.name} is marked as provider but using SystemPluginFactory"
        )

AuthProviderPluginFactory

AuthProviderPluginFactory(manifest)

Bases: BasePluginFactory

Factory for authentication provider plugins.

Auth provider plugins provide OAuth authentication flows and token management without directly proxying requests to API providers.

Parameters:

Name	Type	Description	Default
manifest	PluginManifest	Plugin manifest	required

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

def __init__(self, manifest: PluginManifest):
    """Initialize auth provider plugin factory.

    Args:
        manifest: Plugin manifest
    """
    # Local import to avoid circular dependency at module load time
    from .runtime import AuthProviderPluginRuntime

    super().__init__(manifest, AuthProviderPluginRuntime)

    # Validate this is marked as a provider plugin (auth providers are a type of provider)
    if not manifest.is_provider:
        raise ValueError(
            f"Plugin {manifest.name} must be marked as provider for AuthProviderPluginFactory"
        )

create_context

create_context(service_container)

Create context with auth provider-specific components.

Parameters:

Name	Type	Description	Default
core_services		Core services container	required

Returns:

Type	Description
PluginContext	Plugin context with auth provider components

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

def create_context(self, service_container: "ServiceContainer") -> PluginContext:
    """Create context with auth provider-specific components.

    Args:
        core_services: Core services container

    Returns:
        Plugin context with auth provider components
    """
    # Start with base context
    context = super().create_context(service_container)

    # Auth provider plugins need to create their auth components
    # This is typically done in the specific plugin factory implementation

    return context

get_auth_manager_registry_name

get_auth_manager_registry_name()

Return registry key used for this auth manager.

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

def get_auth_manager_registry_name(self) -> str:
    """Return registry key used for this auth manager."""

    name = getattr(self, "auth_manager_name", None)
    return name or self.manifest.name

create_auth_provider abstractmethod

create_auth_provider(context=None)

Create the OAuth provider for this auth plugin.

Parameters:

Name	Type	Description	Default
context	PluginContext | None	Optional plugin context for initialization	None

Returns:

Type	Description
OAuthProviderProtocol	OAuth provider instance implementing OAuthProviderProtocol

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

@abstractmethod
def create_auth_provider(
    self, context: PluginContext | None = None
) -> OAuthProviderProtocol:
    """Create the OAuth provider for this auth plugin.

    Args:
        context: Optional plugin context for initialization

    Returns:
        OAuth provider instance implementing  OAuthProviderProtocol
    """
    ...

create_token_manager

create_token_manager()

Create the token manager for this auth plugin.

Returns:

Type	Description
BaseTokenManager[Any] | None	Token manager instance or None if not needed

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

def create_token_manager(self) -> BaseTokenManager[Any] | None:
    """Create the token manager for this auth plugin.

    Returns:
        Token manager instance or None if not needed
    """
    return None

create_storage

create_storage()

Create the storage implementation for this auth plugin.

Returns:

Type	Description
TokenStorage[Any] | None	Storage instance or None if using default

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

def create_storage(self) -> TokenStorage[Any] | None:
    """Create the storage implementation for this auth plugin.

    Returns:
        Storage instance or None if using default
    """
    return None

factory_type_name

factory_type_name(factory)

Return a stable type name for a plugin factory.

Returns one of: "auth_provider", "provider", "system", or "plugin" (fallback).

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

def factory_type_name(factory: PluginFactory) -> str:
    """Return a stable type name for a plugin factory.

    Returns one of: "auth_provider", "provider", "system", or "plugin" (fallback).
    """
    try:
        if isinstance(factory, AuthProviderPluginFactory):
            return "auth_provider"
        if isinstance(factory, ProviderPluginFactory):
            return "provider"
        if isinstance(factory, SystemPluginFactory):
            return "system"
    except Exception:
        pass
    return "plugin"