ccproxy.plugins.max_tokens.adapter

ccproxy.plugins.max_tokens.adapter

Max tokens adapter implementation using hook system.

MaxTokensHook

MaxTokensHook(config, service)

Bases: Hook

Hook that enforces max_tokens limits before provider dispatch.

Source code in ccproxy/plugins/max_tokens/adapter.py

def __init__(self, config: MaxTokensConfig, service: TokenLimitsService):
    """Initialize max tokens hook."""
    self.config = config
    self.service = service

MaxTokensAdapter

MaxTokensAdapter(config)

Max tokens adapter using hook-based request interception.

Source code in ccproxy/plugins/max_tokens/adapter.py

def __init__(self, config: MaxTokensConfig):
    """Initialize max tokens adapter."""
    self.config = config
    self.service = TokenLimitsService(config)
    self.hook = MaxTokensHook(config, self.service)
    self._initialized = False

initialize async

initialize()

Initialize the adapter and register hooks.

Source code in ccproxy/plugins/max_tokens/adapter.py

async def initialize(self) -> None:
    """Initialize the adapter and register hooks."""
    if self._initialized:
        return

    if not self.config.enabled:
        logger.debug("max_tokens_adapter_disabled")
        return

    # Initialize the service
    self.service.initialize()

    # Register hook with hook manager
    try:
        from ccproxy.services.container import ServiceContainer

        container = ServiceContainer.get_current(strict=False)
        if container:
            try:
                hook_registry = container.get_hook_registry()
                if hook_registry:
                    hook_registry.register(self.hook)
                    logger.debug(
                        "max_tokens_hook_registered",
                        hook_name=self.hook.name,
                        events=[event.value for event in self.hook.events],
                        priority=self.hook.priority,
                    )
                else:
                    logger.warning("no_hook_registry_available")
            except Exception as service_error:
                logger.warning(
                    "hook_registry_service_not_available",
                    error=str(service_error),
                )
        else:
            logger.warning("no_service_container_available")

    except Exception as e:
        logger.error(
            "failed_to_register_max_tokens_hook",
            error=str(e),
            exc_info=e,
        )
        # Continue without hook registration - plugin will be effectively disabled

    self._initialized = True
    logger.info("max_tokens_adapter_initialized")

cleanup async

cleanup()

Cleanup resources.

Source code in ccproxy/plugins/max_tokens/adapter.py

async def cleanup(self) -> None:
    """Cleanup resources."""
    if not self._initialized:
        return

    try:
        # Unregister hook
        from ccproxy.services.container import ServiceContainer

        container = ServiceContainer.get_current(strict=False)
        if container:
            try:
                hook_registry = container.get_hook_registry()
                if hook_registry:
                    hook_registry.unregister(self.hook)
                    logger.debug("max_tokens_hook_unregistered")
            except Exception as service_error:
                logger.debug(
                    "hook_registry_service_not_available_during_cleanup",
                    error=str(service_error),
                )

    except Exception as e:
        logger.error(
            "failed_to_unregister_max_tokens_hook",
            error=str(e),
            exc_info=e,
        )

    self._initialized = False
    logger.debug("max_tokens_adapter_cleanup_completed")

get_modification_stats

get_modification_stats()

Get statistics about max_tokens modifications.

Source code in ccproxy/plugins/max_tokens/adapter.py

def get_modification_stats(self) -> dict[str, Any]:
    """Get statistics about max_tokens modifications."""
    # This could be enhanced to track modification statistics
    return {
        "adapter_initialized": self._initialized,
        "config_enabled": self.config.enabled,
        "target_providers": self.config.target_providers,
        "fallback_max_tokens": self.config.fallback_max_tokens,
    }