API Reference

Reference for Kollab's core packages, classes, and interfaces for plugin development and customization.

Core Components

Kollab is a monorepo built on these extracted packages:

Event System

Event bus, hook registry and execution, event and command models.

kollabor_events

Plugin System

Plugin factory with dependency injection and the plugin SDK.

kollabor_plugins

Configuration

Config loading and access with dot notation and defaults.

kollabor_config

Terminal UI

Terminal rendering, input handling, widgets, and the design system.

kollabor_tui

EventBus API

The EventBus coordinates hook registration and event processing. Import it from kollabor_events.bus.

await register_hook(hook: Hook) -> bool

Register a hook with the event bus. Config defaults for timeout and retries are applied at execution time, not registration. Returns True on success.

await event_bus.register_hook(my_hook)

await emit_with_hooks(event_type, data, source) -> dict

Emit an event and run its hooks. For EventType enum values, the mapped pre, main, and post phases run. For string event types (plugin custom events), only the main phase runs. Returns the aggregated hook results.

result = await event_bus.emit_with_hooks(
    EventType.USER_INPUT_PRE,
    {"message": user_input},
    "input_handler",
)

await unregister_hook(plugin_name: str, hook_name: str) -> bool

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

await event_bus.unregister_hook("my_plugin", "my_hook_name")

The bus also provides enable_hook(plugin_name, hook_name) and disable_hook(plugin_name, hook_name) for toggling hooks at runtime, get_hooks_for_event(event_type) for hook counts, and register_service(name, service) / get_service(name) for sharing service instances across plugins.

Hook Class

The Hook dataclass defines a callback to execute when an event occurs. The callback receives two arguments: the event data dict and the Event object.

from kollabor_events import EventType, Hook, 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.PREPROCESSING.value,  # Required: higher runs first
    callback=self.handle_input,                 # Required: async fn(event.data, event)
    enabled=True,             # Optional: enable or disable the hook
    timeout=30,               # Optional: max seconds (None = config default, then 30)
    retry_attempts=3,         # Optional: retries on failure (None = config default, then 3)
    error_action="continue",  # Optional: "continue" or "stop" (None = config default)
)

await event_bus.register_hook(hook)

Hook Priorities

Hooks run in priority order, higher numbers first.

Priority	Value	Description
`SYSTEM`	1000	System-level hooks
`SECURITY`	900	Permission checks and security policy
`PREPROCESSING`	500	Input and data preprocessing
`LLM`	100	LLM request and response handling
`POSTPROCESSING`	50	Post-processing of results
`DISPLAY`	10	Rendering and display

Event Class

The Event dataclass carries event data through the pipeline. Events are built by the event processor when you call emit_with_hooks; hook callbacks receive the data dict and the Event.

from kollabor_events import Event, EventType

event = Event(
    type=EventType.LLM_REQUEST_PRE,  # EventType enum, or str for custom events
    data={                           # Event-specific data
        "messages": messages,
        "model": "gpt-4o"
    },
    source="llm_coordinator"         # Component that generated the event
)

# Event state
event.timestamp  # datetime: when the event was created
event.processed  # bool: whether processing finished
event.cancelled  # bool: whether processing was cancelled
event.result     # dict: results from hook processing

CommandDefinition

Define custom slash commands with the CommandDefinition class and register them through the command registry.

from kollabor_events import (
    CommandCategory,
    CommandDefinition,
    CommandMode,
    SubcommandInfo,
)

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,
    subcommands=[
        SubcommandInfo("action1", "", "Execute action 1"),
        SubcommandInfo("action2", "<arg>", "Execute action 2 with an argument"),
    ],
    icon="*"
)

Command Modes

Mode	Description
`NORMAL`	Regular input mode
`INSTANT`	Executes immediately (/clear)
`INLINE_INPUT`	Takes inline parameters (/save filename)
`MENU_POPUP`	Shows command menu overlay
`STATUS_TAKEOVER`	Takes over the status area (/agents)
`STATUS_MODAL`	Modal within the status area only
`MODAL`	Modal overlay mode
`ALTVIEW`	AltView fullscreen mode

KollaborPluginSDK

The SDK scaffolds plugin files and manages custom LLM tools. Plugins follow a lifecycle contract: initialize(), register_hooks(), get_status_line(), and shutdown(), plus a static get_default_config() and an optional process_command(command, params).

from kollabor_plugins.plugin_sdk import KollaborPluginSDK

sdk = KollaborPluginSDK()

# Scaffold a plugin file in plugins/
path = sdk.create_plugin_template("my_plugin", "Does one thing well")

# Register a custom tool for LLM use
# (name, description, and handler are required; handler takes a params dict)
sdk.register_custom_tool({
    "name": "my_tool",
    "description": "What the tool does",
    "handler": my_handler,  # async or sync
})

# Execute a registered tool
result = await sdk.execute_custom_tool("my_tool", {"key": "value"})

Also available: list_custom_tools(), enable_tool(name) and disable_tool(name), and validate_plugin(path) which checks structure and flags dangerous calls.

PluginFactory

The factory instantiates plugin classes with dependency injection. Plugins are constructed with (name, event_bus, renderer, config). Class names ending in "Plugin" are stored under both the full class name and a cleaned lowercase name.

from kollabor_plugins.factory import PluginFactory

factory = PluginFactory()

# Instantiate a single plugin
instance = factory.instantiate_plugin(
    MyPlugin, "MyPlugin", event_bus, renderer, config
)

# Instantiate every discovered plugin class
instances = factory.instantiate_all(plugin_classes, event_bus, renderer, config)

# Look up instances
plugin = factory.get_instance("my")   # cleaned name for MyPlugin
all_instances = factory.get_all_instances()

Configuration API

Read and write configuration values using dot notation:

# Get a configuration value, with a default if unset
max_history = config.get("kollabor.llm.max_history", 90)

# Set a configuration value at runtime
config.set("plugins.my_plugin.enabled", False)

Plugins declare their defaults in a static get_default_config() method; user overrides live in ~/.kollab/config.json (global) or .kollab/config.json (project-local). Only the diff from defaults is persisted.

Environment Variables

LLM profile fields resolve from environment variables before falling back to config: KOLLAB_{PROFILE}_{FIELD} beats KOLLAB_{FIELD}, which beats the config value, which beats the default. Per-profile example: KOLLAB_CLAUDE_API_KEY.

Variable	Description
`KOLLAB_API_KEY`	API key for the provider
`KOLLAB_MODEL`	Model name
`KOLLAB_MAX_TOKENS`	Maximum response tokens (integer)
`KOLLAB_TEMPERATURE`	Sampling temperature (0.0 to 2.0)
`KOLLAB_TIMEOUT`	Request timeout in seconds (integer)
`KOLLAB_SYSTEM_PROMPT`	System prompt string
`KOLLAB_SYSTEM_PROMPT_FILE`	Path to a system prompt file

Profile-scoped variants support the same fields, plus PROVIDER, BASE_URL, TOP_P, EFFORT, and STREAMING.