API Reference

Complete reference for Kollab CLI's core APIs, classes, and interfaces for plugin development and customization.

Core Components

Kollab CLI is built on a modular architecture with these core components:

Event Bus

Central messaging system for hook registration and event dispatch.

core.events.EventBus

State Manager

Persistent state storage and retrieval.

core.storage.StateManager

Config Manager

Configuration loading and access with dot notation.

core.config.ConfigManager

Terminal Renderer

Terminal output, visual effects, and UI rendering.

core.io.Renderer

EventBus API

The EventBus is the central component for hook registration and event dispatch.

register_hook(hook: Hook) -> bool

await event_bus.register_hook(my_hook)

emit(event: Event) -> dict

Emit an event and execute all registered hooks. Returns aggregated results.

event = Event(
    type=EventType.USER_INPUT_PRE,
    data={"message": user_input},
    source="input_handler"
)
result = await event_bus.emit(event)

unregister_hook(name: str) -> bool

Remove a hook by name. Returns True if found and removed.

await event_bus.unregister_hook("my_hook_name")

Hook Class

The Hook dataclass defines a callback to execute when an event occurs.

from core.events import Hook, EventType, HookPriority

hook = Hook(
    name="unique_hook_name",        # Required: Unique identifier
    plugin_name="my_plugin",        # Required: Owning plugin name
    event_type=EventType.USER_INPUT_PRE,  # Required: Event to listen for
    priority=HookPriority.NORMAL.value,   # Required: Execution order
    callback=async_handler_fn,      # Required: Async callback function
    enabled=True,                   # Optional: Enable/disable hook
    timeout=30,                     # Optional: Max seconds to execute
    retry_attempts=3,               # Optional: Retry on failure
    error_action="continue"         # Optional: "continue" or "stop"
)

Event Class

The Event dataclass contains event data and metadata.

from core.events import Event, EventType

event = Event(
    type=EventType.LLM_REQUEST_PRE,  # Required: Event type
    data={                            # Required: Event-specific data
        "messages": messages,
        "model": "gpt-4o"
    },
    source="llm_service"              # Required: Source component
)

# Event properties (read-only)
event.timestamp     # datetime: When created
event.processed     # bool: Has been processed
event.cancelled     # bool: Was cancelled
event.result        # dict: Hook results

CommandDefinition

Define custom slash commands with the CommandDefinition class.

from core.events.models import CommandDefinition, CommandMode, CommandCategory

command = CommandDefinition(
    name="mycommand",
    description="My custom command",
    handler=self.handle_command,
    plugin_name=self.name,
    aliases=["mc", "mycmd"],
    mode=CommandMode.INSTANT,
    category=CommandCategory.CUSTOM,
    icon="*"
)

Command Modes

Mode	Description
`INSTANT`	Executes immediately (/clear)
`INLINE_INPUT`	Takes inline parameters (/save filename)
`MENU_POPUP`	Shows command menu overlay
`STATUS_TAKEOVER`	Takes over status area
`MODAL`	Modal overlay mode
`LIVE_MODAL`	Live-updating modal

KollaborPluginSDK

The SDK provides utilities for cross-plugin communication and service registration.

from core.llm.plugin_sdk import KollaborPluginSDK

sdk = KollaborPluginSDK()

# Register a service for other plugins
sdk.register_service("my_service", self.service_handler)

# Discover available services
services = await sdk.discover_services()

# Call a service from another plugin
result = await sdk.call_service("other_plugin_service", data)

PluginFactory

The factory handles plugin discovery and loading.

from core.plugins.factory import PluginFactory

factory = PluginFactory()

# Get all loaded plugins
plugins = await factory.get_loaded_plugins()

# Get a specific plugin
plugin = await factory.get_plugin("plugin_name")

# Check if plugin is loaded
is_loaded = factory.is_plugin_loaded("plugin_name")

Configuration API

Access configuration values using dot notation:

# Get configuration value
api_key = config.get("core.llm.api_key")
temperature = config.get("core.llm.temperature", 0.7)  # With default

# Check if key exists
has_key = config.has("plugins.my_plugin.enabled")

# Get nested configuration
llm_config = config.get_section("core.llm")

# Environment variable override
# KOLLABOR_API_KEY overrides core.llm.api_key

State Manager API

Persist and retrieve state across sessions:

# Save state
await state_manager.set("my_plugin.last_run", datetime.now().isoformat())
await state_manager.set("my_plugin.settings", {"key": "value"})

# Load state
last_run = await state_manager.get("my_plugin.last_run")
settings = await state_manager.get("my_plugin.settings", default={})

# Delete state
await state_manager.delete("my_plugin.temp_data")

# Check existence
exists = await state_manager.exists("my_plugin.cache")

Environment Variables

Variable	Description
`KOLLABOR_API_ENDPOINT`	LLM API endpoint URL
`KOLLABOR_API_TOKEN`	API authentication token
`KOLLABOR_API_MODEL`	Model name (e.g., gpt-4o)
`KOLLABOR_API_MAX_TOKENS`	Maximum response tokens
`KOLLABOR_API_TEMPERATURE`	Response randomness (0-2)
`KOLLABOR_SYSTEM_PROMPT`	System prompt string
`KOLLABOR_SYSTEM_PROMPT_FILE`	Path to system prompt file