ccproxy.core.plugins.declaration

ccproxy.core.plugins.declaration

Plugin declaration system for static plugin specification.

This module provides the declaration layer of the plugin system, allowing plugins to specify their requirements and capabilities at declaration time (app creation) rather than runtime (lifespan).

FormatAdapterSpec dataclass

FormatAdapterSpec(
    from_format,
    to_format,
    adapter_factory,
    priority=100,
    description="",
)

Specification for format adapter registration.

format_pair property

format_pair

Get the format pair tuple.

MiddlewareLayer

Bases: IntEnum

Middleware layers for ordering.

MiddlewareSpec dataclass

MiddlewareSpec(
    middleware_class, priority=APPLICATION, kwargs=dict()
)

Specification for plugin middleware.

RouterSpec dataclass

RouterSpec(
    router, prefix, tags=list(), dependencies=list()
)

Specification for individual routers in a plugin.

RouteSpec dataclass

RouteSpec(router, prefix, tags=list(), dependencies=list())

Specification for plugin routes.

TaskSpec dataclass

TaskSpec(
    task_name,
    task_type,
    task_class,
    interval_seconds,
    enabled=True,
    kwargs=dict(),
)

Specification for scheduled tasks.

HookSpec dataclass

HookSpec(hook_class, kwargs=dict())

Specification for plugin hooks.

AuthCommandSpec dataclass

AuthCommandSpec(
    command_name, description, handler, options=dict()
)

Specification for auth commands.

CliCommandSpec dataclass

CliCommandSpec(
    command_name,
    command_function,
    help_text="",
    parent_command=None,
)

Specification for plugin CLI commands.

CliArgumentSpec dataclass

CliArgumentSpec(
    target_command,
    argument_name,
    argument_type=str,
    help_text="",
    default=None,
    required=False,
    typer_kwargs=dict(),
)

Specification for adding arguments to existing commands.

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
    ]

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
    ]

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."""
    ...