Skip to content

ccproxy.core.plugins.loader

ccproxy.core.plugins.loader

Centralized plugin loader.

Provides a single entry to discover factories, build a PluginRegistry, and prepare MiddlewareManager based on settings. This isolates loader usage to one place and reinforces import boundaries (core should not import concrete plugin modules directly).

load_plugin_system 

load_plugin_system(settings)

Discover plugins and build a registry + middleware manager.

This function is the single entry point to set up the plugin layer for the application factory. It avoids scattering discovery/registry logic.

Parameters:

Name Type Description Default
settings Settings

Application settings (with plugin config)

 required

Returns:

Type Description
tuple[PluginRegistry, MiddlewareManager]

Tuple of (PluginRegistry, MiddlewareManager)
Source code in ccproxy/core/plugins/loader.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def load_plugin_system(settings: Settings) -> tuple[PluginRegistry, MiddlewareManager]:
    """Discover plugins and build a registry + middleware manager.

    This function is the single entry point to set up the plugin layer for
    the application factory. It avoids scattering discovery/registry logic.

    Args:
        settings: Application settings (with plugin config)

    Returns:
        Tuple of (PluginRegistry, MiddlewareManager)
    """
    # Discover factories (filesystem + entry points) with existing helper
    factories: dict[str, PluginFactory] = discover_and_load_plugins(settings)

    # Create registry and register all factories
    registry = PluginRegistry()
    for _name, factory in factories.items():
        registry.register_factory(factory)

    # Prepare middleware manager; plugins will populate via manifests during
    # app creation (manifest population stage) and at runtime as needed
    middleware_manager = MiddlewareManager()

    logger.debug(
        "plugin_system_loaded",
        factory_count=len(factories),
        plugins=list(factories.keys()),
        category="plugin",
    )

    return registry, middleware_manager

load_cli_plugins 

load_cli_plugins(
    settings, auth_provider=None, allow_plugins=None
)

Load filtered plugins for CLI operations.

This function creates a lightweight plugin registry for CLI commands that: - Includes only CLI-safe plugins (marked with cli_safe = True) - Optionally includes a specific auth provider plugin if requested - Excludes heavy provider plugins that cause DuckDB locks, task manager errors, etc.

Parameters:

Name Type Description Default
settings Any

Application settings

 required
auth_provider str | None

Name of auth provider to include (e.g., "codex", "claude-api")

 None
allow_plugins list[str] | None

Additional plugins to explicitly allow (beyond cli_safe ones)

 None

Returns:

Type Description
PluginRegistry

Filtered PluginRegistry containing only CLI-appropriate plugins
Source code in ccproxy/core/plugins/loader.py 
 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
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def load_cli_plugins(
    settings: Any,
    auth_provider: str | None = None,
    allow_plugins: list[str] | None = None,
) -> PluginRegistry:
    """Load filtered plugins for CLI operations.

    This function creates a lightweight plugin registry for CLI commands that:
    - Includes only CLI-safe plugins (marked with cli_safe = True)
    - Optionally includes a specific auth provider plugin if requested
    - Excludes heavy provider plugins that cause DuckDB locks, task manager errors, etc.

    Args:
        settings: Application settings
        auth_provider: Name of auth provider to include (e.g., "codex", "claude-api")
        allow_plugins: Additional plugins to explicitly allow (beyond cli_safe ones)

    Returns:
        Filtered PluginRegistry containing only CLI-appropriate plugins
    """
    # Discover all available factories
    all_factories: dict[str, PluginFactory] = discover_and_load_plugins(settings)

    # Start with CLI-safe plugins
    cli_factories: dict[str, PluginFactory] = {}

    for name, factory in all_factories.items():
        # Include plugins explicitly marked as CLI-safe
        if getattr(factory, "cli_safe", False):
            cli_factories[name] = factory

    # Add specific auth provider if requested
    if auth_provider:
        auth_plugin_name = _resolve_auth_provider_plugin_name(auth_provider)
        if auth_plugin_name and auth_plugin_name in all_factories:
            cli_factories[auth_plugin_name] = all_factories[auth_plugin_name]
        else:
            logger.warning(
                "auth_provider_not_found",
                provider=auth_provider,
                resolved_name=auth_plugin_name,
                available_auth_providers=[
                    name
                    for name, factory in all_factories.items()
                    if isinstance(factory, AuthProviderPluginFactory)
                ],
            )

    # Add explicitly allowed plugins
    if allow_plugins:
        for plugin_name in allow_plugins:
            if plugin_name in all_factories and plugin_name not in cli_factories:
                cli_factories[plugin_name] = all_factories[plugin_name]

    # Create filtered registry
    registry = PluginRegistry()
    for _name, factory in cli_factories.items():
        registry.register_factory(factory)

    logger.debug(
        "cli_plugin_system_loaded",
        total_available=len(all_factories),
        cli_safe_count=len(
            [f for f in all_factories.values() if getattr(f, "cli_safe", False)]
        ),
        loaded_count=len(cli_factories),
        loaded_plugins=list(cli_factories.keys()),
        auth_provider=auth_provider,
        allow_plugins=allow_plugins or [],
        category="plugin",
    )

    return registry