ccproxy.core.plugins.runtime

ccproxy.core.plugins.runtime

Plugin runtime system for managing plugin instances.

This module defines runtime classes that manage plugin instances and lifecycle. Factory/loader utilities remain in their respective modules to avoid import cycles during consolidation. Import runtime classes from here, and import factories/loaders from their modules for now.

PluginContext

PluginContext()

Context provided to plugin runtime during initialization.

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

def __init__(self) -> None:
    """Initialize plugin context."""
    # Application settings
    self.settings: Settings | None = None
    self.http_client: httpx.AsyncClient | None = None
    self.logger: structlog.BoundLogger | None = None
    self.scheduler: Scheduler | None = None
    self.config: BaseModel | None = None
    self.cli_detection_service: CLIDetectionService | None = None
    self.plugin_registry: PluginRegistry | None = None

    # Core app and hook system
    self.app: FastAPI | None = None
    self.hook_registry: HookRegistry | None = None
    self.hook_manager: HookManager | None = None

    # Observability and streaming
    self.request_tracer: IRequestTracer | None = None
    self.streaming_handler: StreamingMetrics | None = None
    self.metrics: IMetricsCollector | None = None

    # Provider-specific
    self.adapter: BaseAdapter | None = None
    self.detection_service: Any = None
    self.credentials_manager: Any = None
    self.oauth_registry: OAuthRegistry | None = None
    self.http_pool_manager: Any = None
    self.service_container: Any = None
    self.auth_provider: Any = None
    self.token_manager: Any = None
    self.storage: Any = None

    self.format_registry: Any = None
    self.model_mapper: Any = None

    # Testing/utilities
    self.proxy_service: Any = None

    # Internal service mapping for type-safe access
    self._service_map: dict[type[Any], str] = {}
    self._initialize_service_map()

get_service

get_service(service_type)

Get a service instance by type with proper type safety.

Parameters:

Name	Type	Description	Default
service_type	type[T]	The type of service to retrieve	required

Returns:

Type	Description
T	The service instance

Raises:

Type	Description
ValueError	If the service is not available

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

def get_service(self, service_type: type[T]) -> T:
    """Get a service instance by type with proper type safety.

    Args:
        service_type: The type of service to retrieve

    Returns:
        The service instance

    Raises:
        ValueError: If the service is not available
    """
    # Create service mappings dynamically to access current values
    service_mappings: dict[type[Any], Any] = {}

    # Common concrete types
    if self.settings is not None:
        service_mappings[type(self.settings)] = self.settings
    if self.http_client is not None:
        service_mappings[httpx.AsyncClient] = self.http_client
    if self.logger is not None:
        service_mappings[structlog.BoundLogger] = self.logger
    if self.config is not None:
        service_mappings[type(self.config)] = self.config
        service_mappings[BaseModel] = self.config

    # Check if service type directly matches a known service
    if service_type in service_mappings:
        return service_mappings[service_type]  # type: ignore[no-any-return]

    # Check all attributes for an instance of the requested type
    for attr_name in dir(self):
        if not attr_name.startswith("_"):  # Skip private attributes
            attr_value = getattr(self, attr_name)
            if attr_value is not None and isinstance(attr_value, service_type):
                return attr_value  # type: ignore[no-any-return]

    # Service not found
    type_name = getattr(service_type, "__name__", str(service_type))
    raise ValueError(f"Service {type_name} not available in plugin context")

get

get(key_or_type, default=None)

Get service by type (new) or by string key (backward compatibility).

Parameters:

Name	Type	Description	Default
key_or_type	type[T] | str	Service type for type-safe access or string key for compatibility	required
default	Any	Default value for string-based access (ignored for type-safe access)	None

Returns:

Type	Description
T | Any	Service instance for type-safe access, or attribute value for string access

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

def get(self, key_or_type: type[T] | str, default: Any = None) -> T | Any:
    """Get service by type (new) or by string key (backward compatibility).

    Args:
        key_or_type: Service type for type-safe access or string key for compatibility
        default: Default value for string-based access (ignored for type-safe access)

    Returns:
        Service instance for type-safe access, or attribute value for string access
    """
    if isinstance(key_or_type, str):
        # Backward compatibility: string-based access
        return getattr(self, key_or_type, default)
    else:
        # Type-safe access
        return self.get_service(key_or_type)

get_attr

get_attr(key, default=None)

Get attribute by string name - for backward compatibility.

Parameters:

Name	Type	Description	Default
key	str	String attribute name	required
default	Any	Default value if attribute not found	None

Returns:

Type	Description
Any	Attribute value or default

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

def get_attr(self, key: str, default: Any = None) -> Any:
    """Get attribute by string name - for backward compatibility.

    Args:
        key: String attribute name
        default: Default value if attribute not found

    Returns:
        Attribute value or default
    """
    return getattr(self, key, default)

keys

keys()

Backward compatibility: Return list of available service keys.

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

def keys(self) -> list[str]:
    """Backward compatibility: Return list of available service keys."""
    return [
        attr
        for attr in dir(self)
        if not attr.startswith("_")
        and not callable(getattr(self, attr))
        and getattr(self, attr) is not None
    ]

PluginManifest dataclass

PluginManifest(
    name,
    version,
    description="",
    dependencies=list(),
    is_provider=False,
    provides=list(),
    requires=list(),
    optional_requires=list(),
    middleware=list(),
    routes=list(),
    tasks=list(),
    hooks=list(),
    auth_commands=list(),
    config_class=None,
    tool_accumulator_class=None,
    oauth_client_factory=None,
    oauth_provider_factory=None,
    token_manager_factory=None,
    oauth_config_class=None,
    oauth_routes=list(),
    format_adapters=list(),
    requires_format_adapters=list(),
    cli_commands=list(),
    cli_arguments=list(),
)

Complete static declaration of a plugin's capabilities.

This manifest is created at module import time and contains all static information needed to integrate the plugin into the application.

validate_dependencies

validate_dependencies(available_plugins)

Validate that all dependencies are available.

Parameters:

Name	Type	Description	Default
available_plugins	set[str]	Set of available plugin names	required

Returns:

Type	Description
list[str]	List of missing dependencies

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

def validate_dependencies(self, available_plugins: set[str]) -> list[str]:
    """Validate that all dependencies are available.

    Args:
        available_plugins: Set of available plugin names

    Returns:
        List of missing dependencies
    """
    return [dep for dep in self.dependencies if dep not in available_plugins]

validate_service_dependencies

validate_service_dependencies(available_services)

Validate that required services are available.

Parameters:

Name	Type	Description	Default
available_services	set[str]	Set of available service names	required

Returns:

Type	Description
list[str]	List of missing required services

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

def validate_service_dependencies(self, available_services: set[str]) -> list[str]:
    """Validate that required services are available.

    Args:
        available_services: Set of available service names

    Returns:
        List of missing required services
    """
    missing = []
    for required in self.requires:
        if required not in available_services:
            missing.append(required)
    return missing

get_sorted_middleware

get_sorted_middleware()

Get middleware sorted by priority.

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

def get_sorted_middleware(self) -> list[MiddlewareSpec]:
    """Get middleware sorted by priority."""
    return sorted(self.middleware)

validate_format_adapter_requirements

validate_format_adapter_requirements(available_adapters)

Validate that required format adapters are available.

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

def validate_format_adapter_requirements(
    self, available_adapters: set[FormatPair]
) -> list[FormatPair]:
    """Validate that required format adapters are available."""
    return [
        req
        for req in self.requires_format_adapters
        if req not in available_adapters
    ]

PluginRuntimeProtocol

Bases: Protocol

Protocol for plugin runtime instances.

initialize async

initialize(context)

Initialize the plugin with runtime context.

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

async def initialize(self, context: PluginContext) -> None:
    """Initialize the plugin with runtime context."""
    ...

shutdown async

shutdown()

Cleanup on shutdown.

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

async def shutdown(self) -> None:
    """Cleanup on shutdown."""
    ...

validate async

validate()

Validate plugin is ready.

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

async def validate(self) -> bool:
    """Validate plugin is ready."""
    ...

health_check async

health_check()

Perform health check.

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

async def health_check(self) -> dict[str, Any]:
    """Perform health check."""
    ...

BasePluginRuntime

BasePluginRuntime(manifest)

Bases: PluginRuntimeProtocol

Base implementation of plugin runtime.

This class provides common functionality for all plugin runtimes. Specific plugin types (system, provider) can extend this base class.

Parameters:

Name	Type	Description	Default
manifest	PluginManifest	Plugin manifest with static declarations	required

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

def __init__(self, manifest: PluginManifest):
    """Initialize runtime with manifest.

    Args:
        manifest: Plugin manifest with static declarations
    """
    self.manifest = manifest
    self.context: PluginContext | None = None
    self.initialized = False

name property

name

Plugin name from manifest.

version property

version

Plugin version from manifest.

initialize async

initialize(context)

Initialize the plugin with runtime context.

Parameters:

Name	Type	Description	Default
context	PluginContext	Runtime context with services and configuration	required

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

async def initialize(self, context: PluginContext) -> None:
    """Initialize the plugin with runtime context.

    Args:
        context: Runtime context with services and configuration
    """
    if self.initialized:
        logger.warning(
            "plugin_already_initialized", plugin=self.name, category="plugin"
        )
        return

    self.context = context

    # Allow subclasses to perform custom initialization
    await self._on_initialize()

    self.initialized = True
    logger.debug(
        "plugin_initialized",
        plugin=self.name,
        version=self.version,
        category="plugin",
    )

shutdown async

shutdown()

Cleanup on shutdown.

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

async def shutdown(self) -> None:
    """Cleanup on shutdown."""
    if not self.initialized:
        return

    # Allow subclasses to perform custom cleanup
    await self._on_shutdown()

    self.initialized = False
    logger.info("plugin_shutdown", plugin=self.name, category="plugin")

validate async

validate()

Validate plugin is ready.

Returns:

Type	Description
bool	True if plugin is ready, False otherwise

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

async def validate(self) -> bool:
    """Validate plugin is ready.

    Returns:
        True if plugin is ready, False otherwise
    """
    # Basic validation - plugin is initialized
    if not self.initialized:
        return False

    # Allow subclasses to add custom validation
    return await self._on_validate()

health_check async

health_check()

Perform health check.

Returns:

Type	Description
dict[str, Any]	Health check result following IETF format

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

async def health_check(self) -> dict[str, Any]:
    """Perform health check.

    Returns:
        Health check result following IETF format
    """
    try:
        # Start with basic health check
        is_healthy = await self.validate()

        # Allow subclasses to provide detailed health info
        details = await self._get_health_details()

        return {
            "status": "pass" if is_healthy else "fail",
            "componentId": self.name,
            "componentType": "provider_plugin"
            if self.manifest.is_provider
            else "system_plugin",
            "version": self.version,
            "details": details,
        }
    except Exception as e:
        logger.error(
            "plugin_health_check_failed",
            plugin=self.name,
            error=str(e),
            exc_info=e,
            category="plugin",
        )
        return {
            "status": "fail",
            "componentId": self.name,
            "componentType": "provider_plugin"
            if self.manifest.is_provider
            else "system_plugin",
            "version": self.version,
            "output": str(e),
        }

SystemPluginRuntime

SystemPluginRuntime(manifest)

Bases: BasePluginRuntime

Runtime for system plugins (non-provider plugins).

System plugins provide functionality like logging, monitoring, permissions, etc., but don't proxy to external providers.

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

def __init__(self, manifest: PluginManifest):
    """Initialize runtime with manifest.

    Args:
        manifest: Plugin manifest with static declarations
    """
    self.manifest = manifest
    self.context: PluginContext | None = None
    self.initialized = False

AuthProviderPluginRuntime

AuthProviderPluginRuntime(manifest)

Bases: BasePluginRuntime

Runtime for authentication provider plugins.

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

Parameters:

Name	Type	Description	Default
manifest	PluginManifest	Plugin manifest with static declarations	required

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

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

    Args:
        manifest: Plugin manifest with static declarations
    """
    super().__init__(manifest)
    self.auth_provider: Any | None = None  # OAuthProviderProtocol
    self.token_manager: Any | None = None
    self.storage: Any | None = None

ProviderPluginRuntime

ProviderPluginRuntime(manifest)

Bases: BasePluginRuntime

Runtime for provider plugins.

Provider plugins proxy requests to external API providers and require additional components like adapters and detection services.

Parameters:

Name	Type	Description	Default
manifest	PluginManifest	Plugin manifest with static declarations	required

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

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

    Args:
        manifest: Plugin manifest with static declarations
    """
    super().__init__(manifest)
    self.adapter: Any | None = None  # BaseAdapter
    self.detection_service: Any | None = None
    self.credentials_manager: Any | None = None