Max tokens plugin for automatic token limit enforcement.
This plugin intercepts requests and automatically sets max_tokens based on
model limits from the pricing data when no max_tokens is provided.
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
|
async
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")
|
async
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 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,
}
|
Bases: BaseModel
Configuration for the max_tokens plugin.
classmethod
Validate configuration values.
Source code in ccproxy/plugins/max_tokens/config.py
| @model_validator(mode="before")
@classmethod
def validate_config(cls, data: dict[str, Any]) -> dict[str, Any]:
"""Validate configuration values."""
# Ensure target_providers is a list
if isinstance(data.get("target_providers"), str):
data["target_providers"] = [data["target_providers"]]
return data
|
should_process_provider(provider)
Check if plugin should process requests for given provider.
Source code in ccproxy/plugins/max_tokens/config.py
| def should_process_provider(self, provider: str) -> bool:
"""Check if plugin should process requests for given provider."""
if self.apply_to_all_providers:
return True
return provider in self.target_providers
|
get_modification_reason(reason_type)
Get modification reason text for given reason type.
Source code in ccproxy/plugins/max_tokens/config.py
| def get_modification_reason(self, reason_type: str) -> str:
"""Get modification reason text for given reason type."""
return self.modification_reasons.get(
reason_type, f"Unknown reason: {reason_type}"
)
|