climb.common package

Submodules

climb.common.data_structures module

class climb.common.data_structures.EngineConfig(*, episode_db: List[Dict[str, Any]], plan: List[str])[source]

Bases: BaseModel

episode_db: List[Dict[str, Any]]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plan: List[str]
class climb.common.data_structures.EngineParameter(*, name: str, description: str, kind: Literal['float', 'bool', 'enum', 'records'], default: str | float | bool | List[Dict] | None, enum_values: List[str] | None = None, min_value: float | None = None, max_value: float | None = None, value_set_by_static_method: str | None = None, disabled_set_by_static_method: str | None = None, enum_values_set_by_engine_config: str | None = None, disabled: bool | None = False, records_disabled_keys: List[str] | None = None)[source]

Bases: BaseModel

default: str | float | bool | List[Dict] | None
description: str
disabled: bool | None
disabled_set_by_static_method: str | None
enum_values: List[str] | None
enum_values_set_by_engine_config: str | None
kind: Literal['float', 'bool', 'enum', 'records']
max_value: float | None
min_value: float | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
records_disabled_keys: List[str] | None
value_set_by_static_method: str | None
class climb.common.data_structures.EngineState(*, streaming: bool, agent: Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], agent_switched: bool, agent_state: Dict[Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], Dict[str, Any]] = {}, executing_tool: str | None = None, user_message_requested: bool = False, response_kind: ResponseKind = ResponseKind.NOT_SET, tool_request: ToolCallRecord | None = None, ui_controlled: UIControlledState = UIControlledState(interaction_stage='reason', input_request=None, input_placeholder=None))[source]

Bases: BaseModel

agent: Literal['coordinator', 'worker', 'supervisor', 'simulated_user']
agent_state: Dict[Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], Dict[str, Any]]
agent_switched: bool
executing_tool: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property response_kind: ResponseKind
response_kind_value: ResponseKind
streaming: bool
property tool_request: ToolCallRecord | None
tool_request_value: ToolCallRecord | None
property ui_controlled: UIControlledState
ui_controlled_value: UIControlledState
user_message_requested: bool
class climb.common.data_structures.FileInfo(*, name: str, size: float, size_units: str, modified: datetime)[source]

Bases: BaseModel

property category: Literal['image', 'data', 'model', 'other']
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

modified: datetime
name: str
property previewable: bool
size: float
size_units: str
class climb.common.data_structures.KeyGeneration[source]

Bases: object

static generate_message_key(use_time: datetime | None = None) → str[source]
static generate_session_key(use_time: datetime | None = None) → str[source]
class climb.common.data_structures.Message(**data: Any)[source]

Bases: BaseModel

agent: Literal['coordinator', 'worker', 'supervisor', 'simulated_user']
property engine_state: EngineState | None
engine_state_value: EngineState | None
files_in: List[str] | None
files_out: List[str] | None
generated_code: str | None
generated_code_dependencies: List[str] | None
generated_code_stderr: str | None
generated_code_stdout: str | None
generated_code_success: bool | None
incoming_tool_calls: List[ToolCallRecord] | None
key: str
model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

new_reasoning_cycle: bool
outgoing_tool_call: ToolCallRecord | None
role: Literal['system', 'assistant', 'user', 'tool', 'code_execution']
summary_message: bool
text: str | None
token_counts: Dict[Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], int | None]
tool_call_logs: str | None
tool_call_return: str | None
tool_call_success: bool | None
tool_call_user_report: List[str | Figure | Figure] | None
visibility: Literal['all', 'ui_only', 'llm_only', 'llm_only_ephemeral', 'system_only']
class climb.common.data_structures.ResponseKind(value)[source]

Bases: Enum

An enumeration.

CODE_GENERATION = 4
NOT_SET = 1
TEXT_MESSAGE = 2
TOOL_REQUEST = 3
class climb.common.data_structures.Session(*, session_key: str, working_directory: str, started_at: ~datetime.datetime = datetime.datetime(2025, 12, 19, 16, 17, 18, 295716), friendly_name: str = '', engine_name: str, engine_params: ~typing.Dict[str, str | float | bool | ~typing.List[~typing.Dict]] = {}, messages: ~typing.List[~climb.common.data_structures.Message] = [], engine_state: ~climb.common.data_structures.EngineState = EngineState(     streaming=False, agent='worker', agent_switched=False, agent_state={}, executing_tool=None, user_message_requested=False, response_kind=<ResponseKind.NOT_SET: 1>, tool_request=None, ui_controlled=UIControlledState(         interaction_stage='reason', input_request=None, input_placeholder=None     ) ), engine_config: ~climb.common.data_structures.EngineConfig = EngineConfig(episode_db=[], plan=[]), session_settings: ~climb.common.data_structures.SessionSettings = SessionSettings(show_tool_call_logs=True, show_tool_call_return=True, show_code=True, show_code_out=True, show_planning_details=False, show_full_message_history=False, show_message_history_length=15))[source]

Bases: BaseModel

engine_config: EngineConfig
engine_name: str
engine_params: Dict[str, str | float | bool | List[Dict]]
engine_state: EngineState
friendly_name: str
messages: List[Message]
model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

session_key: str
session_settings: SessionSettings
started_at: datetime
working_directory: str
class climb.common.data_structures.SessionSettings(*, show_tool_call_logs: bool = True, show_tool_call_return: bool = True, show_code: bool = True, show_code_out: bool = True, show_planning_details: bool = False, show_full_message_history: bool = False, show_message_history_length: int = 15)[source]

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

show_code: bool
show_code_out: bool
show_full_message_history: bool
show_message_history_length: int
show_planning_details: bool
show_tool_call_logs: bool
show_tool_call_return: bool
class climb.common.data_structures.ToolCallRecord(*, name: str, arguments: str, engine_id: str | None = None)[source]

Bases: BaseModel

arguments: str
engine_id: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
class climb.common.data_structures.UIControlledState(*, interaction_stage: Literal['reason', 'output', 'await_user_input'] = 'reason', input_request: UserInputRequest | None = None, input_placeholder: str | None = None)[source]

Bases: BaseModel

input_placeholder: str | None
input_request: UserInputRequest | None
interaction_stage: Literal['reason', 'output', 'await_user_input']
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class climb.common.data_structures.UploadedFileAbstraction(*, name: str, content: bytes)[source]

Bases: BaseModel

content: bytes
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
class climb.common.data_structures.UserInputRequest(*, key: str, kind: Literal['text', 'file', 'multiple_files'], description: str | None = None, extra: Dict[str, Any] = {})[source]

Bases: BaseModel

check_received_input() → None[source]
description: str | None
extra: Dict[str, Any]
key: str
kind: Literal['text', 'file', 'multiple_files']
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

property received_input: Any
class climb.common.data_structures.UserSettings(*, user_name: str = 'docs', disclaimer_shown: bool = False, default_session_settings: SessionSettings = SessionSettings(show_tool_call_logs=True, show_tool_call_return=True, show_code=True, show_code_out=True, show_planning_details=False, show_full_message_history=False, show_message_history_length=15), active_session: str | None = None)[source]

Bases: BaseModel

active_session: str | None
default_session_settings: SessionSettings
disclaimer_shown: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

user_name: str
climb.common.data_structures.create_new_session(session_name: str | None, engine_name: str, engine_params: Dict[str, str | float | bool | List[Dict]], db: DB) → Session[source]
climb.common.data_structures.get_category_from_name(name: str) → Literal['image', 'data', 'model', 'other'][source]
climb.common.data_structures.get_previewable(filetype: Literal['image', 'data', 'model', 'other']) → bool[source]

climb.common.disclaimer module

climb.common.exc module

CliMB exceptions.

exception climb.common.exc.ClimbConfigurationError[source]

Bases: Exception

CliMB configuration-related error.

climb.common.plan_files module

climb.common.plan_files.load_plan_and_template_files(return_as: Literal['absolute_paths', 'relative_paths', 'filenames'] = 'filenames') → dict[str, list[str]][source]

Load all .json files from TEMPLATES_DIR and PLAN_FILES_DIR.

Parameters:

return_as – Whether to return the paths of the files instead of just the filenames. - “absolute_paths”: Return the absolute paths of the files. - “relative_paths”: Return the relative paths of the files. - “filenames”: Return the filenames only.

Returns:

dict with keys “plan_files” and “template_plan_files”, each containing a list of .json filenames (without path, just the filename).

climb.common.plan_files.load_plan_file(plan_file: str | Path, relative_path: bool = False) → List[Dict[str, Any]][source]

climb.common.serialization module

climb.common.serialization.decode_enum(s: str) → Enum[source]

Recover the module and the enum name and value from the string, instantiate the enum and return it.

Parameters:

s (str) – The encoded string.

Returns:

The decoded enum object.

Return type:

enum.Enum

climb.common.serialization.encode_enum(obj: Enum) → str[source]

Store the module and the enum name and value in a string separated by a slash.

Parameters:

obj (enum.Enum) – The enum object to encode.

Returns:

The encoded string.

Return type:

str

climb.common.serialization.message_from_serializable_dict(message_dict: Dict[str, Any]) → Message[source]
climb.common.serialization.message_to_serializable_dict(message: Message, session_path: str) → Dict[str, Any][source]
climb.common.serialization.session_from_serializable_dict(session_dict: Dict[str, Any]) → Session[source]
climb.common.serialization.session_to_serializable_dict(session: Session) → Dict[str, Any][source]

climb.common.utils module

climb.common.utils.analyze_df_modifications(df_before: DataFrame, df_after: DataFrame, row_similarity_threshold: float = 0.8) → Dict[str, Any][source]
climb.common.utils.attempt_imputation_match(df1: DataFrame, df2: DataFrame, nan_sentinel: Any) → List[int][source]
climb.common.utils.check_extra_available() → bool[source]
climb.common.utils.convert_size(size_bytes: float) → Tuple[float, str][source]

Convert file size to a more readable format.

climb.common.utils.d2m(d: Dict[str, Any], model: T) → T[source]

d(ictionary)_to_m(odel). Convert a dictionary to a Pydantic model instance`.

Parameters:

  • d (Dict[str, Any]) – The dictionary to convert.

  • model (pydantic.BaseModel) – The Pydantic model class to convert to.

Returns:

The Pydantic model instance.

Return type:

pydantic.BaseModel

climb.common.utils.dedent(text: str) → str[source]

Remove the same amount of leading spaces from each line of the text.

Parameters:

text (str) – Multiline string to process.

Returns:

Processed text with leading spaces removed.

Return type:

str

climb.common.utils.engine_log(*args: Any, **kwargs) → None[source]
climb.common.utils.filter_out_lines(in_str: str) → str[source]
climb.common.utils.fix_windows_path_backslashes(path: str) → str[source]

Modify the path string for Windows with correct escape sequences. Converts single backslashes to double backslashes.

Parameters:

path (str) – The original file path string.

Returns:

Modified path with double backslashes where needed.

Return type:

str

climb.common.utils.log_messages_to_file(messages: List[Dict], tools: List[Dict] | None, metadata: Dict | None, path: str) → None[source]
climb.common.utils.m2d(model: Any) → Dict[str, Any][source]

m(odel)_to_d(ictionary). Convert a Pydantic model instance to a dictionary.

Parameters:

model (Any) – The Pydantic model instance to convert.

Returns:

The dictionary.

Return type:

Dict[str, Any]

climb.common.utils.make_filename_path_safe(s: str, remove_slashes: bool = False) → str[source]

Make a string safe to use as a path. A simple implementation, only applicable to file name, not the full path.

climb.common.utils.raise_if_extra_not_available() → NoReturn | None[source]
climb.common.utils.replace_str_from_dict(s: str, d: Dict[str, Any]) → str[source]

Replace all instances of keys in a string with their corresponding values. The dictionary values that are not strings will be converted to strings str(value) before replacement.

Parameters:

  • s (str) – The string to modify.

  • d (Dict[str, Any]) – The dictionary that contains the replacement information. Keys are the strings to be replaced, and values are the strings to replace them with.

Returns:

The modified string.

Return type:

str

climb.common.utils.similar(a: Any, b: Any) → float[source]
climb.common.utils.truncate_dict_values(d: Dict, max_len: int = 50) → Dict[source]

Recursively truncates dictionary values whose string representation exceeds max_len characters. To be used for shorter representations of possibly large nested dictionaries.

Parameters:

  • d (Dict) – Input dictionary, possibly nested.

  • max_len (int) – Maximum length of the string representation of the values.

Returns:

New dictionary with possibly truncated values.

Return type:

Dict

climb.common.utils.ui_log(*args: Any, **kwargs) → None[source]
climb.common.utils.update_templates(body_text: str, templates: Dict[str, str]) → str[source]

Module contents

Common data structures and functions used by climb.

class climb.common.EngineConfig(*, episode_db: List[Dict[str, Any]], plan: List[str])[source]

Bases: BaseModel

episode_db: List[Dict[str, Any]]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plan: List[str]
class climb.common.EngineParameter(*, name: str, description: str, kind: Literal['float', 'bool', 'enum', 'records'], default: str | float | bool | List[Dict] | None, enum_values: List[str] | None = None, min_value: float | None = None, max_value: float | None = None, value_set_by_static_method: str | None = None, disabled_set_by_static_method: str | None = None, enum_values_set_by_engine_config: str | None = None, disabled: bool | None = False, records_disabled_keys: List[str] | None = None)[source]

Bases: BaseModel

default: str | float | bool | List[Dict] | None
description: str
disabled: bool | None
disabled_set_by_static_method: str | None
enum_values: List[str] | None
enum_values_set_by_engine_config: str | None
kind: Literal['float', 'bool', 'enum', 'records']
max_value: float | None
min_value: float | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
records_disabled_keys: List[str] | None
value_set_by_static_method: str | None
class climb.common.EngineState(*, streaming: bool, agent: Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], agent_switched: bool, agent_state: Dict[Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], Dict[str, Any]] = {}, executing_tool: str | None = None, user_message_requested: bool = False, response_kind: ResponseKind = ResponseKind.NOT_SET, tool_request: ToolCallRecord | None = None, ui_controlled: UIControlledState = UIControlledState(interaction_stage='reason', input_request=None, input_placeholder=None))[source]

Bases: BaseModel

agent: Literal['coordinator', 'worker', 'supervisor', 'simulated_user']
agent_state: Dict[Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], Dict[str, Any]]
agent_switched: bool
executing_tool: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property response_kind: ResponseKind
response_kind_value: ResponseKind
streaming: bool
property tool_request: ToolCallRecord | None
tool_request_value: ToolCallRecord | None
property ui_controlled: UIControlledState
ui_controlled_value: UIControlledState
user_message_requested: bool
class climb.common.FileInfo(*, name: str, size: float, size_units: str, modified: datetime)[source]

Bases: BaseModel

property category: Literal['image', 'data', 'model', 'other']
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

modified: datetime
name: str
property previewable: bool
size: float
size_units: str
class climb.common.KeyGeneration[source]

Bases: object

static generate_message_key(use_time: datetime | None = None) → str[source]
static generate_session_key(use_time: datetime | None = None) → str[source]
class climb.common.Message(*, key: str, role: Literal['system', 'assistant', 'user', 'tool', 'code_execution'], visibility: Literal['all', 'ui_only', 'llm_only', 'llm_only_ephemeral', 'system_only'] = 'all', agent: Literal['coordinator', 'worker', 'supervisor', 'simulated_user'] = 'worker', new_reasoning_cycle: bool = False, summary_message: bool = False, text: str | None, token_counts: Dict[Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], int | None] = {}, incoming_tool_calls: List[ToolCallRecord] | None = None, outgoing_tool_call: ToolCallRecord | None = None, tool_call_success: bool | None = None, tool_call_logs: str | None = None, tool_call_return: str | None = None, tool_call_user_report: List[str | Figure | Figure] | None = None, generated_code_dependencies: List[str] | None = None, generated_code: str | None = None, generated_code_success: bool | None = None, generated_code_stdout: str | None = None, generated_code_stderr: str | None = None, files_in: List[str] | None = None, files_out: List[str] | None = None, engine_state: EngineState | None = None)[source]

Bases: BaseModel

agent: Literal['coordinator', 'worker', 'supervisor', 'simulated_user']
property engine_state: EngineState | None
engine_state_value: EngineState | None
files_in: List[str] | None
files_out: List[str] | None
generated_code: str | None
generated_code_dependencies: List[str] | None
generated_code_stderr: str | None
generated_code_stdout: str | None
generated_code_success: bool | None
incoming_tool_calls: List[ToolCallRecord] | None
key: str
model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

new_reasoning_cycle: bool
outgoing_tool_call: ToolCallRecord | None
role: Literal['system', 'assistant', 'user', 'tool', 'code_execution']
summary_message: bool
text: str | None
token_counts: Dict[Literal['coordinator', 'worker', 'supervisor', 'simulated_user'], int | None]
tool_call_logs: str | None
tool_call_return: str | None
tool_call_success: bool | None
tool_call_user_report: List[str | Figure | Figure] | None
visibility: Literal['all', 'ui_only', 'llm_only', 'llm_only_ephemeral', 'system_only']
class climb.common.ResponseKind(value)[source]

Bases: Enum

An enumeration.

CODE_GENERATION = 4
NOT_SET = 1
TEXT_MESSAGE = 2
TOOL_REQUEST = 3
class climb.common.Session(*, session_key: str, working_directory: str, started_at: ~datetime.datetime = datetime.datetime(2025, 12, 19, 16, 17, 18, 295716), friendly_name: str = '', engine_name: str, engine_params: ~typing.Dict[str, str | float | bool | ~typing.List[~typing.Dict]] = {}, messages: ~typing.List[~climb.common.data_structures.Message] = [], engine_state: ~climb.common.data_structures.EngineState = EngineState(     streaming=False, agent='worker', agent_switched=False, agent_state={}, executing_tool=None, user_message_requested=False, response_kind=<ResponseKind.NOT_SET: 1>, tool_request=None, ui_controlled=UIControlledState(         interaction_stage='reason', input_request=None, input_placeholder=None     ) ), engine_config: ~climb.common.data_structures.EngineConfig = EngineConfig(episode_db=[], plan=[]), session_settings: ~climb.common.data_structures.SessionSettings = SessionSettings(show_tool_call_logs=True, show_tool_call_return=True, show_code=True, show_code_out=True, show_planning_details=False, show_full_message_history=False, show_message_history_length=15))[source]

Bases: BaseModel

engine_config: EngineConfig
engine_name: str
engine_params: Dict[str, str | float | bool | List[Dict]]
engine_state: EngineState
friendly_name: str
messages: List[Message]
model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

session_key: str
session_settings: SessionSettings
started_at: datetime
working_directory: str
class climb.common.ToolCallRecord(*, name: str, arguments: str, engine_id: str | None = None)[source]

Bases: BaseModel

arguments: str
engine_id: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
class climb.common.UIControlledState(*, interaction_stage: Literal['reason', 'output', 'await_user_input'] = 'reason', input_request: UserInputRequest | None = None, input_placeholder: str | None = None)[source]

Bases: BaseModel

input_placeholder: str | None
input_request: UserInputRequest | None
interaction_stage: Literal['reason', 'output', 'await_user_input']
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class climb.common.UserSettings(*, user_name: str = 'docs', disclaimer_shown: bool = False, default_session_settings: SessionSettings = SessionSettings(show_tool_call_logs=True, show_tool_call_return=True, show_code=True, show_code_out=True, show_planning_details=False, show_full_message_history=False, show_message_history_length=15), active_session: str | None = None)[source]

Bases: BaseModel

active_session: str | None
default_session_settings: SessionSettings
disclaimer_shown: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

user_name: str
climb.common.create_new_session(session_name: str | None, engine_name: str, engine_params: Dict[str, str | float | bool | List[Dict]], db: DB) → Session[source]