mindroot.lib package

Subpackages

Submodules

mindroot.lib.buchatlog module

class mindroot.lib.buchatlog.ChatLog(log_id=0, agent=None, context_length: int = 4096, user: str = None)[source]

Bases: object

add_message(message: Dict[str, str]) None[source]
delete_log() None[source]
get_history() List[Dict[str, str]][source]
get_recent(max_tokens: int = 4096) List[Dict[str, str]][source]
load_log(log_id=None) None[source]
save_log() None[source]

mindroot.lib.buchatlog2 module

class mindroot.lib.buchatlog2.ChatLog(log_id=0, agent=None, context_length: int = 4096, user: str = None)[source]

Bases: object

add_message(message: Dict[str, str]) None[source]
count_tokens() Dict[str, int][source]

Count tokens in the chat log, providing both sequence totals and cumulative request totals.

Returns:

  • input_tokens_sequence: Total tokens in all user messages

  • output_tokens_sequence: Total tokens in all assistant messages

  • input_tokens_total: Cumulative tokens sent to LLM across all requests

Return type:

Dict with the following keys

delete_log() None[source]
get_history() List[Dict[str, str]][source]
get_recent(max_tokens: int = 4096) List[Dict[str, str]][source]
load_log(log_id=None) None[source]
save_log() None[source]
mindroot.lib.buchatlog2.count_tokens_for_log_id(log_id: str) Dict[str, int][source]

Count tokens for a chat log identified by log_id, including any delegated tasks.

Parameters:

log_id – The log ID to count tokens for

Returns:

Dictionary with token counts or None if log not found

mindroot.lib.buchatlog2.extract_delegate_task_log_ids(messages: List[Dict]) List[str][source]

Extract log IDs from delegate_task commands in messages.

Parameters:

messages – List of chat messages

Returns:

List of log IDs found in delegate_task commands

mindroot.lib.buchatlog2.find_chatlog_file(log_id: str) str[source]

Find a chatlog file by its log_id.

Parameters:

log_id – The log ID to search for

Returns:

The full path to the chatlog file if found, None otherwise

mindroot.lib.buchatlog2.get_cache_dir() str[source]

Get the directory for token count cache files. Creates the directory if it doesn’t exist.

mindroot.lib.buchatlog2.get_cache_path(log_id: str) str[source]

Get the path to the cache file for a specific log_id.

mindroot.lib.buchatlog2.get_cached_token_counts(log_id: str, log_path: str) Dict[str, int][source]

Get cached token counts if available and valid.

Parameters:
  • log_id – The log ID

  • log_path – Path to the actual log file

Returns:

Cached token counts if valid, None otherwise

mindroot.lib.buchatlog2.save_token_counts_to_cache(log_id: str, token_counts: Dict[str, int]) None[source]

Save token counts to cache.

mindroot.lib.butemplates module

async mindroot.lib.butemplates.collect_content(template, blocks, template_type, data)[source]

Collect content from child templates.

Parameters:
  • template – Jinja2 template object

  • blocks (list) – List of block names

  • template_type (str) – Type of template (‘inject’ or ‘override’)

  • data (dict) – Template context data

Returns:

Collected content by block

Return type:

dict

async mindroot.lib.butemplates.find_parent_template(page_name, plugins)[source]

Find parent template in enabled plugins.

Parameters:
  • page_name (str) – Name of the template page

  • plugins (list) – List of enabled plugins

Returns:

Template path if found, None otherwise

Return type:

str

async mindroot.lib.butemplates.load_plugin_templates(page_name, plugins)[source]

Load templates from plugins.

Parameters:
  • page_name (str) – Name of the template page

  • plugins (list) – List of enabled plugins

Returns:

List of template info dictionaries

Return type:

list

async mindroot.lib.butemplates.render(page_name, context)[source]

Render a template with plugin injections and overrides.

Parameters:
  • page_name (str) – Name of the template page

  • context (dict) – Template context data

Returns:

Rendered HTML

Return type:

str

async mindroot.lib.butemplates.render_combined_template(page_name, plugins, context)[source]

Render combined template with injections and overrides.

Parameters:
  • page_name (str) – Name of the template page

  • plugins (list) – List of enabled plugins

  • context (dict) – Template context data

Returns:

Rendered HTML

Return type:

str

mindroot.lib.butemplates.setup_template_environment(plugins=None)[source]

Set up Jinja2 environment with multiple template paths.

Returns:

Configured Jinja2 environment

Return type:

Environment

mindroot.lib.chatcontext module

class mindroot.lib.chatcontext.BaseCommandSet(*args, **kwargs)[source]

Bases: Protocol

Base protocol for all command sets

class mindroot.lib.chatcontext.BaseService(*args, **kwargs)[source]

Bases: Protocol

Base protocol for all services

class mindroot.lib.chatcontext.ChatContext(command_manager_=None, service_manager_=None, user=None, log_id=None, parent_log_id=None)[source]

Bases: object

cmds(command_set: Type[CommandSetT]) CommandSetT[source]
async delete(cascade: bool = True)[source]
async classmethod delete_session_by_id(log_id: str, user: str, agent: str, cascade: bool = True, visited_log_ids: Set[str] = None)[source]
async load_context(log_id)[source]
proto(protocol_type: Type[ServiceT]) ServiceT[source]
async save_context()[source]
async save_context_data()[source]
async mindroot.lib.chatcontext.get_context(log_id, user)[source]

mindroot.lib.chatlog module

class mindroot.lib.chatlog.ChatLog(log_id=0, agent=None, parent_log_id=None, context_length: int = 4096, user: str = None)[source]

Bases: object

add_message(message: Dict[str, str]) None[source]

Synchronous version for backward compatibility

async add_message_async(message: Dict[str, str]) None[source]

Async version for new code that needs non-blocking operations

count_tokens() Dict[str, int][source]

Count tokens in the chat log, providing both sequence totals and cumulative request totals.

Returns:

  • input_tokens_sequence: Total tokens in all user messages

  • output_tokens_sequence: Total tokens in all assistant messages

  • input_tokens_total: Cumulative tokens sent to LLM across all requests

Return type:

Dict with the following keys

get_history() List[Dict[str, str]][source]
get_recent(max_tokens: int = 4096) List[Dict[str, str]][source]
async load_log(log_id=None) None[source]
async save_log() None[source]
async mindroot.lib.chatlog.build_token_hierarchy(log_id: str, user: str = None, visited: set = None) Dict[source]

Build a hierarchical token count structure for a log and its children.

Parameters:
  • log_id – The log ID to build hierarchy for

  • user – User for the log

  • visited – Set of already visited log IDs to prevent infinite recursion

Returns:

  • log_id: The log ID

  • individual_counts: Token counts for this log only

  • cumulative_counts: Token counts including all children

  • children: List of child hierarchies

Return type:

Dictionary with hierarchical structure containing

async mindroot.lib.chatlog.count_tokens_for_log_id(log_id: str, user: str = None, hierarchical: bool = False) Dict[str, int][source]

Count tokens for a chat log identified by log_id, including any delegated tasks.

Parameters:

log_id – The log ID to count tokens for

Returns:

Dictionary with token counts or None if log not found. If hierarchical=True, includes ‘hierarchy’ key with tree structure. If hierarchical=False (default), returns flat structure for backwards compatibility.

mindroot.lib.chatlog.extract_delegate_task_log_ids(messages: List[Dict]) List[str][source]

Extract log IDs from delegate_task and delegate_subtask commands in messages.

Parameters:

messages – List of chat messages

Returns:

List of log IDs found in delegate_task commands

async mindroot.lib.chatlog.find_chatlog_file(log_id: str) str[source]

Find a chatlog file by its log_id.

Parameters:

log_id – The log ID to search for

Returns:

The full path to the chatlog file if found, None otherwise

async mindroot.lib.chatlog.find_child_logs_by_parent_id(parent_log_id: str) List[str][source]

Find all chat logs that have the given parent_log_id.

Parameters:

parent_log_id – The parent log ID to search for

Returns:

List of log IDs that have this parent_log_id

async mindroot.lib.chatlog.get_cache_dir() str[source]

Get the directory for token count cache files. Creates the directory if it doesn’t exist.

async mindroot.lib.chatlog.get_cache_path(log_id: str) str[source]

Get the path to the cache file for a specific log_id.

async mindroot.lib.chatlog.get_cached_token_counts(log_id: str, log_path: str) Dict[str, int][source]

Get cached token counts if available and valid.

Parameters:
  • log_id – The log ID

  • log_path – Path to the actual log file

Returns:

Cached token counts if valid, None otherwise

async mindroot.lib.chatlog.save_token_counts_to_cache(log_id: str, token_counts: Dict[str, int]) None[source]

Save token counts to cache.

mindroot.lib.chatlog_optimized module

class mindroot.lib.chatlog_optimized.ChatLog(log_id=0, agent=None, parent_log_id=None, context_length: int = 4096, user: str = None)[source]

Bases: object

add_message(message: Dict[str, str]) None[source]
count_tokens() Dict[str, int][source]
get_history() List[Dict[str, str]][source]
get_recent(max_tokens: int = 4096) List[Dict[str, str]][source]
load_log(log_id=None) None[source]
save_log() None[source]
mindroot.lib.chatlog_optimized.count_tokens_for_log_id(log_id: str) Dict[str, int][source]

Count tokens for a chat log identified by log_id, including any delegated tasks. This is the optimized version that uses caching and batch operations.

mindroot.lib.chatlog_optimized.count_tokens_for_log_id_optimized(log_id: str) Dict[str, int][source]

Optimized version of count_tokens_for_log_id that uses caching and batch operations.

Parameters:

log_id – The log ID to count tokens for

Returns:

Dictionary with token counts or None if log not found

mindroot.lib.chatlog_optimized.extract_delegate_task_log_ids(messages: List[Dict]) List[str][source]

Extract log IDs from delegate_task commands in messages.

Parameters:

messages – List of chat messages

Returns:

List of log IDs found in delegate_task commands

mindroot.lib.chatlog_optimized.find_chatlog_file(log_id: str) str | None[source]

Find a chatlog file by its log_id using the cached index.

Parameters:

log_id – The log ID to search for

Returns:

The full path to the chatlog file if found, None otherwise

mindroot.lib.chatlog_optimized.find_child_logs_by_parent_id(parent_log_id: str) List[str][source]

Find all chat logs that have the given parent_log_id using the cached index.

Parameters:

parent_log_id – The parent log ID to search for

Returns:

List of log IDs that have this parent_log_id

mindroot.lib.chatlog_optimized.get_cache_dir() str[source]

Get the directory for token count cache files. Creates the directory if it doesn’t exist.

mindroot.lib.chatlog_optimized.get_cache_path(log_id: str) str[source]

Get the path to the cache file for a specific log_id.

mindroot.lib.chatlog_optimized.get_cached_token_counts(log_id: str, log_path: str) Dict[str, int][source]

Get cached token counts if available and valid.

Parameters:
  • log_id – The log ID

  • log_path – Path to the actual log file

Returns:

Cached token counts if valid, None otherwise

mindroot.lib.chatlog_optimized.save_token_counts_to_cache(log_id: str, token_counts: Dict[str, int]) None[source]

Save token counts to cache.

mindroot.lib.json_escape module

mindroot.lib.json_escape.escape_for_json(s: str) str[source]

Escape a string to make it safe for use as a JSON property value or JSON string.

Args:

s (str): The input string to escape

Returns:

str: The escaped string safe for JSON

Examples:
>>> escape_for_json('Hello "world"')
'Hello \"world\"'
>>> escape_for_json('Line 1
Line 2’)

‘Line 1nLine 2’

mindroot.lib.model_selector module

mindroot.lib.parent_templates module

Parent template environment with single template per filename.

class mindroot.lib.parent_templates.FirstMatchLoader(loaders)[source]

Bases: ChoiceLoader

Custom loader that only returns the first matching template.

get_source(environment, template)[source]

Get template source, returning only first match for each template name.

mindroot.lib.parent_templates.get_parent_templates_env(plugins=None)[source]

Set up Jinja2 environment that only loads first matching template.

Parameters:

plugins (list, optional) – List of plugin names. Defaults to enabled plugins.

Returns:

Configured Jinja2 environment with FirstMatchLoader

Return type:

Environment

mindroot.lib.plugins module

Plugin system for managing, installing, and loading plugins.

This module provides the main interface for the plugin system, including: - Plugin installation and updates - Plugin loading and initialization - Manifest management - Path resolution

Typical usage:

from lib.plugins import load, plugin_install, toggle_plugin_state

# Install a plugin plugin_install(‘my-plugin’, source=’github’, source_path=’user/repo’)

# Load plugins in FastAPI app await load(app)

mindroot.lib.plugins.check_plugin_dependencies(plugin_path)[source]

Check if all plugin dependencies are met.

Parameters:

plugin_path (str) – Path to plugin directory

Returns:

True if all dependencies are met

Return type:

bool

mindroot.lib.plugins.get_plugin_import_path(plugin_name)[source]

Get the Python import path for a plugin.

Parameters:

plugin_name (str) – Name of the plugin

Returns:

Import path for the plugin or None if not found

Return type:

str

mindroot.lib.plugins.get_plugin_path(plugin_name)[source]

Get the filesystem path for a plugin.

Parameters:

plugin_name (str) – Name of the plugin

Returns:

Absolute path to the plugin directory or None if not found

Return type:

str

mindroot.lib.plugins.install_plugin_dependencies(plugin_path)[source]

Install plugin dependencies from requirements.txt.

Parameters:

plugin_path (str) – Path to plugin directory

Returns:

True if installation successful

Return type:

bool

mindroot.lib.plugins.list_enabled(include_category=True)[source]

List all enabled plugins.

Parameters:

include_category (bool) – Whether to include the category in results

Returns:

List of enabled plugins, optionally with categories

Return type:

list

async mindroot.lib.plugins.load(app=None)[source]

Load all enabled plugins with l8n translation support.

Parameters:

app (FastAPI, optional) – The FastAPI application instance

Raises:

Exception – If no FastAPI instance is provided or found

mindroot.lib.plugins.load_plugin_manifest()[source]

Load the plugin manifest file.

Returns:

The manifest data structure

Return type:

dict

async mindroot.lib.plugins.plugin_install(plugin_name, source='pypi', source_path=None, remote_source=None)[source]

Install a plugin from various sources.

Parameters:
  • plugin_name (str) – Name of the plugin

  • source (str) – Installation source (‘pypi’, ‘local’, ‘github’)

  • source_path (str, optional) – Path or GitHub repo reference

  • remote_source (str, optional) – Remote source for the plugin (e.g., GitHub repo path)

Returns:

True if installation successful

Return type:

bool

mindroot.lib.plugins.plugin_update(plugin_name)[source]

Update an installed plugin.

Parameters:

plugin_name (str) – Name of the plugin to update

Returns:

True if update successful

Return type:

bool

async mindroot.lib.plugins.pre_load(app=None)[source]
mindroot.lib.plugins.save_plugin_manifest(manifest)[source]

Save the plugin manifest file.

Parameters:

manifest (dict) – The manifest data structure to save

mindroot.lib.plugins.toggle_plugin_state(plugin_name, enabled)[source]

Toggle a plugin’s enabled state.

Parameters:
  • plugin_name (str) – Name of the plugin

  • enabled (bool) – New enabled state

Returns:

True if successful, False if plugin not found

Return type:

bool

mindroot.lib.plugins.update_plugin_manifest(plugin_name, source, source_path, remote_source=None, version='0.0.1', metadata=None)[source]

Update or add a plugin entry in the manifest.

Parameters:
  • plugin_name (str) – Name of the plugin

  • source (str) – Source type (‘core’, ‘local’, ‘github’)

  • source_path (str) – Path to the plugin

  • remote_source (str, optional) – GitHub repository reference

  • version (str, optional) – Plugin version

  • metadata (dict, optional) – Plugin metadata including commands and services

mindroot.lib.plugins_install module

mindroot.lib.plugins_install.install_plugins(plugin_list)[source]
mindroot.lib.plugins_install.main()[source]
mindroot.lib.plugins_install.update_plugins_json(installed_plugins, json_path)[source]

mindroot.lib.route_decorators module

mindroot.lib.route_decorators.add_public_static(path_start)[source]

Add a path prefix to the public static routes

mindroot.lib.route_decorators.public_route()[source]

Decorator to mark a route as public (no authentication required)

mindroot.lib.route_decorators.requires_all_roles(*roles: str)[source]
mindroot.lib.route_decorators.requires_any_role(*roles: str)[source]
mindroot.lib.route_decorators.requires_role(role: str)[source]

mindroot.lib.session_files module

async mindroot.lib.session_files.load_session_data(session_id: str, property: str)[source]
async mindroot.lib.session_files.save_session_data(session_id: str, property: str, value: str)[source]

mindroot.lib.streamcmd module

async mindroot.lib.streamcmd.read_stream(stream: StreamReader, callback: Callable[[str], Any])[source]

Read from stream line by line and call the callback for each line.

async mindroot.lib.streamcmd.run_command_with_streaming(cmd: List[str], stdout_callback: Callable[[str], Any], stderr_callback: Callable[[str], Any], cwd: str | None = None, env: dict | None = None) int[source]

Run a command asynchronously and stream its output.

Parameters:
  • cmd – Command to run as a list of strings

  • stdout_callback – Callback for stdout lines

  • stderr_callback – Callback for stderr lines

  • cwd – Working directory for the command

  • env – Environment variables for the command

Returns:

Exit code of the command

async mindroot.lib.streamcmd.stream_command_as_events(cmd: List[str], cwd: str | None = None, env: dict | None = None) AsyncGenerator[Dict[str, str], None][source]

Run a command and yield its output as events.

Parameters:
  • cmd – Command to run as a list of strings

  • cwd – Working directory for the command

  • env – Environment variables for the command

Yields:

Events with type and data

mindroot.lib.templates module

mindroot.lib.templates.apply_translations_to_content(content, template_path=None)[source]

Apply l8n translations to template content.

Parameters:
  • content (str) – Template content with __TRANSLATE_key__ placeholders

  • template_path (str) – Path to the template file for plugin context

Returns:

Content with translations applied, or None if translations are incomplete

Return type:

str or None

mindroot.lib.templates.check_for_localized_template(template_path)[source]

Check if a localized version of a template exists.

Parameters:

template_path (str) – Original template path

Returns:

Path to localized template if it exists, None otherwise

Return type:

str

async mindroot.lib.templates.collect_content(template, blocks, template_type, data, assume_blank=False)[source]

Collect content from child templates.

Parameters:
  • template – Jinja2 template object

  • blocks (list) – List of block names

  • template_type (str) – Type of template (‘inject’ or ‘override’)

  • data (dict) – Template context data

  • assume_blank (bool) – If True and type is ‘override’, unspecified blocks are blanked

Returns:

Collected content by block

Return type:

dict

async mindroot.lib.templates.find_parent_template(page_name, plugins)[source]

Find parent template in enabled plugins.

Parameters:
  • page_name (str) – Name of the template page

  • plugins (list) – List of enabled plugins

Returns:

Template path if found, None otherwise

Return type:

str

async mindroot.lib.templates.find_plugin_template(page_name, plugins)[source]

Find a template in a plugin’s templates directory.

Parameters:
  • page_name (str) – Name of the template page

  • plugins (list) – List of enabled plugins

Returns:

(template_path, template_obj) if found, (None, None) otherwise

Return type:

tuple

mindroot.lib.templates.get_current_language()[source]

Get the current language for translation.

async mindroot.lib.templates.load_plugin_templates(page_name, plugins)[source]

Load templates from plugins with translation support.

Parameters:
  • page_name (str) – Name of the template page

  • plugins (list) – List of enabled plugins

Returns:

List of template info dictionaries

Return type:

list

mindroot.lib.templates.load_template_with_translation(template_path)[source]

Load a template and apply translations if available and complete.

If translations are missing for the current language, this function will fall back to serving the original template to avoid showing __TRANSLATE_key__ placeholders to users.

Parameters:

template_path (str) – Path to the template file

Returns:

Template content with translations applied, or original content

if translations are incomplete for the current language

Return type:

str

async mindroot.lib.templates.render(page_name, context)[source]

Render a template with plugin injections, overrides, and translation support. If no parent template exists, tries to render a template directly from a plugin.

Parameters:
  • page_name (str) – Name of the template page

  • context (dict) – Template context data (can be None)

Returns:

Rendered HTML

Return type:

str

async mindroot.lib.templates.render_combined_template(page_name, plugins, context)[source]

Render combined template with injections and overrides, including translation support.

Parameters:
  • page_name (str) – Name of the template page

  • plugins (list) – List of enabled plugins

  • context (dict) – Template context data (can be None)

Returns:

Rendered HTML

Return type:

str

async mindroot.lib.templates.render_direct_template(template_path, context)[source]

Render a template directly without combining with a parent template, with translation support.

Parameters:
  • template_path (str) – Path to the template

  • context (dict) – Template context data

Returns:

Rendered HTML

Return type:

str

mindroot.lib.templates.setup_template_environment(plugins=None)[source]

Set up Jinja2 environment with multiple template paths.

Returns:

Configured Jinja2 environment

Return type:

Environment

mindroot.lib.token_counter module

async mindroot.lib.token_counter.count_tokens_for_log_id(log_id: str) Dict[str, int][source]

Count tokens for a chat log identified by log_id, including any delegated tasks.

Parameters:

log_id – The log ID to count tokens for

Returns:

Dictionary with token counts or None if log not found

mindroot.lib.token_counter.extract_delegate_task_log_ids(messages: List[Dict]) List[str][source]

Extract log IDs from delegate_task commands in messages.

Parameters:

messages – List of chat messages

Returns:

List of log IDs found in delegate_task commands

async mindroot.lib.token_counter.find_chatlog_file(log_id: str) str[source]

Find a chatlog file by its log_id.

Parameters:

log_id – The log ID to search for

Returns:

The full path to the chatlog file if found, None otherwise

async mindroot.lib.token_counter.get_cache_dir() str[source]

Get the directory for token count cache files. Creates the directory if it doesn’t exist.

async mindroot.lib.token_counter.get_cache_path(log_id: str) str[source]

Get the path to the cache file for a specific log_id.

async mindroot.lib.token_counter.get_cached_token_counts(log_id: str, log_path: str) Dict[str, int][source]

Get cached token counts if available and valid.

Parameters:
  • log_id – The log ID

  • log_path – Path to the actual log file

Returns:

Cached token counts if valid, None otherwise

async mindroot.lib.token_counter.save_token_counts_to_cache(log_id: str, token_counts: Dict[str, int]) None[source]

Save token counts to cache.

Module contents