Skip to content

UI Components API

hatchling.ui

Command System

hatchling.ui.abstract_commands

Abstract commands module for the chat interface.

This module provides the AbstractCommands base class that defines the common structure and shared functionality for all command handlers in the chat interface.

Classes

AbstractCommands

Bases: ABC

Abstract base class for chat command handlers.

This class defines the common structure and shared functionality that all command handlers should implement. Subclasses must implement the abstract methods to define their specific commands and behavior.

Source code in hatchling/ui/abstract_commands.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class AbstractCommands(ABC):
    """Abstract base class for chat command handlers.

    This class defines the common structure and shared functionality that all
    command handlers should implement. Subclasses must implement the abstract
    methods to define their specific commands and behavior.
    """
    def __init__(self, chat_session,
                 settings_registry: SettingsRegistry, style: Optional[Style] = None):
        """Initialize the command handler.

        Args:
            chat_session: The chat session this handler is associated with.
            settings (AppSettings): The chat settings to use.
            env_manager (HatchEnvironmentManager): The Hatch environment manager.
            style (Optional[Style]): Style for formatting command output.
        """
        self.chat_session = chat_session

        self.settings_registry = settings_registry
        self.settings = settings_registry.settings

        self.logger = logging_manager.get_session(self.__class__.__name__)

        # Set up styling - use provided style or create default
        self.style = style or Style.from_dict({
            'command.name': 'bold',
            'command.description': '',
            'header': 'bold underline',
        })

        # Initialize the commands dictionary
        self.commands = {}

        # Initialize the command registry
        self._register_commands()

        # Keep old format for backward compatibility
        self.sync_commands = {}
        self.async_commands = {}

        # Populate legacy command dictionaries
        self._build_legacy_commands()

    @abstractmethod
    def _register_commands(self) -> None:
        """Register all available commands with their handlers.

        Subclasses must implement this method to define their specific commands
        in the standardized format.
        """
        pass

    def _build_legacy_commands(self) -> None:
        """Build legacy command dictionaries for backward compatibility."""
        for cmd_name, cmd_info in self.commands.items():
            if cmd_info['is_async']:
                self.async_commands[cmd_name] = (cmd_info['handler'], cmd_info['description'])
            else:
                self.sync_commands[cmd_name] = (cmd_info['handler'], cmd_info['description'])

    def print_commands_help(self) -> None:
        """Print help for all available commands.

        Subclasses should implement this method to provide appropriate help text
        for their command set.
        """
        # Group commands by functionality and print them
        for cmd_name, cmd_info in sorted(self.commands.items()):
            formatted_cmd = self.format_command(cmd_name, cmd_info)
            print_formatted_text(FormattedText(formatted_cmd), style=self.style)

            # Show arguments if available
            if cmd_info.get('args'):
                for arg_name, arg_def in cmd_info['args'].items():
                    arg_text = [
                        ('', '\t'),
                        ('class:command.args', arg_name),
                        ('', ': '),
                        ('class:command.description', arg_def.get('description', 'No description'))
                    ]
                    if arg_def.get('required'):
                        arg_text.insert(2, ('class:command.args', ' (required)'))
                    print_formatted_text(FormattedText(arg_text), style=self.style)

    def format_command(self, cmd_name: str, cmd_info: Dict[str, Any], group: str = 'default') -> list:
        """Format a command as FormattedText.

        Can be overridden by subclasses to customize formatting.

        Args:
            cmd_name (str): Command name
            cmd_info (dict): Command information dictionary
            group (str): Command group name for styling

        Returns:
            list: FormattedText fragments
        """
        return [
            ('class:command.name', f"{cmd_name}"),
            ('', ' - '),
            ('class:command.description', f"{cmd_info['description']}")
        ]

    def reload_commands(self) -> Dict[str, Any]:
        """Reload all commands to apply the current language settings.

        Returns:
            Dict[str, Any]: The reloaded commands dictionary.
        """
        self._register_commands()  # Re-register commands to apply new language
        return self.commands

    def _print_command_help(self, command: str) -> None:
        """Print help for a specific command.

        Args:
            command (str): The command to print help for.
        """
        if command in self.commands:
            cmd_info = self.commands[command]
            formatted_cmd = self.format_command(command, cmd_info)
            print_formatted_text(FormattedText([('class:header', 'Command usage:')]), style=self.style)
            print_formatted_text(FormattedText(formatted_cmd), style=self.style)

            # Show arguments if available
            if cmd_info.get('args'):

                for arg_name, arg_def in cmd_info['args'].items():
                    arg_text = [
                        ('', '\t'),
                        ('class:command.args', arg_name),
                        ('', ': '),
                        ('class:command.description', arg_def.get('description', 'No description'))
                    ]
                    if arg_def.get('required'):
                        arg_text.insert(2, ('class:command.args', ' (required)'))
                    print_formatted_text(FormattedText(arg_text), style=self.style)
        else:
            print_formatted_text(FormattedText([
                ('class:command.description', f"No help available for command: {command}")
            ]), style=self.style)

    def _parse_args(self, args_str: str, arg_defs: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
        """Parse command arguments from a string.

        Args:
            args_str (str): The argument string to parse.
            arg_defs (Dict): Definitions of arguments to parse, including default values.

        Returns:
            Dict[str, Any]: Parsed arguments.
        """
        result = {}

        # Initialize with defaults
        for arg_name, arg_def in arg_defs.items():
            if 'default' in arg_def:
                result[arg_name] = arg_def['default']

        # Split by spaces, but respect quoted strings
        parts = []
        current_part = ""
        in_quotes = False
        quote_char = None

        for char in args_str:
            if char in ['"', "'"]:
                if not in_quotes:
                    in_quotes = True
                    quote_char = char
                elif char == quote_char:
                    in_quotes = False
                    quote_char = None
                else:
                    current_part += char
            elif char.isspace() and not in_quotes:
                if current_part:
                    parts.append(current_part)
                    current_part = ""
            else:
                current_part += char

        if current_part:
            parts.append(current_part)

        # Process positional and named arguments
        positionals = [arg_name for arg_name, arg_def in arg_defs.items() if arg_def.get('positional', False)]
        positional_idx = 0

        i = 0
        while i < len(parts):
            part = parts[i]

            # Handle named arguments (--arg or -a style)
            if part.startswith('--') or (part.startswith('-') and len(part) == 2):
                arg_name = part.lstrip('-')

                # Find the actual argument name if it's an alias
                for name, arg_def in arg_defs.items():
                    if arg_name == name or arg_name in arg_def.get('aliases', []):
                        arg_name = name
                        break

                # Check if this argument expects a value
                if i + 1 < len(parts) and not parts[i+1].startswith('-'):
                    result[arg_name] = parts[i+1]
                    i += 2
                else:
                    # Flag argument (boolean)
                    result[arg_name] = True
                    i += 1
            else:
                # Handle positional arguments
                if positional_idx < len(positionals):
                    result[positionals[positional_idx]] = part
                    positional_idx += 1
                i += 1

        return result

    def get_command_metadata(self) -> dict:
        """Get metadata for all registered commands for autocompletion.

        Returns:
            dict: Dictionary containing command metadata with the new standardized format
        """
        return self.commands
Functions
__init__(chat_session, settings_registry, style=None)

Initialize the command handler.

Parameters:

Name Type Description Default
chat_session

The chat session this handler is associated with.

 required
settings AppSettings

The chat settings to use.

 required
env_manager HatchEnvironmentManager

The Hatch environment manager.

 required
style Optional[Style]

Style for formatting command output.

 None
Source code in hatchling/ui/abstract_commands.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(self, chat_session,
             settings_registry: SettingsRegistry, style: Optional[Style] = None):
    """Initialize the command handler.

    Args:
        chat_session: The chat session this handler is associated with.
        settings (AppSettings): The chat settings to use.
        env_manager (HatchEnvironmentManager): The Hatch environment manager.
        style (Optional[Style]): Style for formatting command output.
    """
    self.chat_session = chat_session

    self.settings_registry = settings_registry
    self.settings = settings_registry.settings

    self.logger = logging_manager.get_session(self.__class__.__name__)

    # Set up styling - use provided style or create default
    self.style = style or Style.from_dict({
        'command.name': 'bold',
        'command.description': '',
        'header': 'bold underline',
    })

    # Initialize the commands dictionary
    self.commands = {}

    # Initialize the command registry
    self._register_commands()

    # Keep old format for backward compatibility
    self.sync_commands = {}
    self.async_commands = {}

    # Populate legacy command dictionaries
    self._build_legacy_commands()
format_command(cmd_name, cmd_info, group='default')

Format a command as FormattedText.

Can be overridden by subclasses to customize formatting.

Parameters:

Name Type Description Default
cmd_name str

Command name

 required
cmd_info dict

Command information dictionary

 required
group str

Command group name for styling

 'default'

Returns:

Name Type Description
list list

FormattedText fragments
Source code in hatchling/ui/abstract_commands.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def format_command(self, cmd_name: str, cmd_info: Dict[str, Any], group: str = 'default') -> list:
    """Format a command as FormattedText.

    Can be overridden by subclasses to customize formatting.

    Args:
        cmd_name (str): Command name
        cmd_info (dict): Command information dictionary
        group (str): Command group name for styling

    Returns:
        list: FormattedText fragments
    """
    return [
        ('class:command.name', f"{cmd_name}"),
        ('', ' - '),
        ('class:command.description', f"{cmd_info['description']}")
    ]
get_command_metadata()

Get metadata for all registered commands for autocompletion.

Returns:

Name Type Description
dict dict

Dictionary containing command metadata with the new standardized format
Source code in hatchling/ui/abstract_commands.py 
238
239
240
241
242
243
244
def get_command_metadata(self) -> dict:
    """Get metadata for all registered commands for autocompletion.

    Returns:
        dict: Dictionary containing command metadata with the new standardized format
    """
    return self.commands
print_commands_help()

Print help for all available commands.

Subclasses should implement this method to provide appropriate help text for their command set.

Source code in hatchling/ui/abstract_commands.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def print_commands_help(self) -> None:
    """Print help for all available commands.

    Subclasses should implement this method to provide appropriate help text
    for their command set.
    """
    # Group commands by functionality and print them
    for cmd_name, cmd_info in sorted(self.commands.items()):
        formatted_cmd = self.format_command(cmd_name, cmd_info)
        print_formatted_text(FormattedText(formatted_cmd), style=self.style)

        # Show arguments if available
        if cmd_info.get('args'):
            for arg_name, arg_def in cmd_info['args'].items():
                arg_text = [
                    ('', '\t'),
                    ('class:command.args', arg_name),
                    ('', ': '),
                    ('class:command.description', arg_def.get('description', 'No description'))
                ]
                if arg_def.get('required'):
                    arg_text.insert(2, ('class:command.args', ' (required)'))
                print_formatted_text(FormattedText(arg_text), style=self.style)
reload_commands()

Reload all commands to apply the current language settings.

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The reloaded commands dictionary.
Source code in hatchling/ui/abstract_commands.py 
121
122
123
124
125
126
127
128
def reload_commands(self) -> Dict[str, Any]:
    """Reload all commands to apply the current language settings.

    Returns:
        Dict[str, Any]: The reloaded commands dictionary.
    """
    self._register_commands()  # Re-register commands to apply new language
    return self.commands

hatchling.ui.base_commands

Base chat commands module for handling core chat interface commands.

Contains the BaseChatCommands class which provides basic command handling functionality for the chat interface, including help, exit, log control and tool management.

Classes

BaseChatCommands

Bases: AbstractCommands

Handles processing of command inputs in the chat interface.

Source code in hatchling/ui/base_commands.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class BaseChatCommands(AbstractCommands):
    """Handles processing of command inputs in the chat interface."""

    def _register_commands(self) -> None:
        """Register all available chat commands with their handlers."""
        # New standardized command registration format with i18n support
        self.commands = {
            'help': {
                'handler': self._cmd_help,
                'description': translate("commands.base.help_description"),
                'is_async': False,
                'args': {}
            },
            'exit': {
                'handler': self._cmd_exit,
                'description': translate("commands.base.exit_description"),
                'is_async': False,
                'args': {}
            },
            'quit': {
                'handler': self._cmd_exit,
                'description': translate("commands.base.quit_description"),
                'is_async': False,
                'args': {}
            },
            'clear': {
                'handler': self._cmd_clear,
                'description': translate("commands.base.clear_description"),
                'is_async': False,
                'args': {}
            },
            'show_logs': {
                'handler': self._cmd_show_logs,
                'description': translate("commands.base.show_logs_description"),
                'is_async': False,
                'args': {
                    'count': {
                        'positional': True,
                        'completer_type': 'suggestions',
                        'values': ['10', '20', '50', '100'],
                        'description': translate('commands.args.value_description'),
                        'required': False
                    }
                }
            },
            'set_log_level': {
                'handler': self._cmd_set_log_level,
                'description': translate("commands.base.set_log_level_description"),
                'is_async': False,
                'args': {
                    'level': {
                        'positional': True,
                        'completer_type': 'suggestions',
                        'values': ['debug', 'info', 'warning', 'error', 'critical'],
                        'description': translate('commands.args.value_description'),
                        'required': True
                    }
                }
            },
            'version': {
                'handler': self._cmd_version,
                'description': translate("commands.base.version_description"),
                'is_async': False,
                'args': {}
            }
        }

        # Keep old format for backward compatibility
        self.sync_commands = {}
        self.async_commands = {}

        for cmd_name, cmd_info in self.commands.items():
            if cmd_info['is_async']:
                self.async_commands[cmd_name] = (cmd_info['handler'], cmd_info['description'])
            else:
                self.sync_commands[cmd_name] = (cmd_info['handler'], cmd_info['description'])

    def print_commands_help(self) -> None:
        """Print help for all available chat commands."""
        print_formatted_text(FormattedText([
            ('class:header', "\n=== Base Chat Commands ===\n")
        ]), style=self.style)

        # Call parent class method to print formatted commands
        super().print_commands_help()

    def format_command(self, cmd_name: str, cmd_info: dict, group: str = 'base') -> list:
        """Format base commands with custom styling."""
        return [
            (f'class:command.name.{group}', f"{cmd_name}"),
            ('', ' - '),
            ('class:command.description', f"{cmd_info['description']}")
        ]

    def _cmd_help(self, _: str) -> bool:
        """
        This command is picked up by the ChatCommandHandler and not here.
        That's because it concerns all commands, not just base commands.
        """
        pass

    def _cmd_exit(self, _: str) -> bool:
        """Exit the chat session.

        Args:
            _ (str): Unused arguments.

        Returns:
            bool: False to end the chat session.
        """
        print("Ending chat session...")
        return False

    def _cmd_clear(self, _: str) -> bool:
        """Clear chat history.

        Args:
            _ (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        self.chat_session.history.clear()
        print("Chat history cleared!")
        return True

    def _cmd_show_logs(self, args: str) -> bool:
        """Display session logs.

        Args:
            args (str): Optional number of log entries to show.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            logs_to_show = int(args) if args.strip() else None
            print(self.chat_session.debug_log.get_logs(logs_to_show))
        except ValueError:
            print(f"Invalid number: {args}")
            print("Usage: show_logs [n]")
        return True

    def _cmd_set_log_level(self, args: str) -> bool:
        """Set the log level.

        Args:
            args (str): Log level name (debug, info, warning, error, critical).

        Returns:
            bool: True to continue the chat session.
        """
        level_name = args.strip().lower()
        level_map = {
            "debug": logging.DEBUG,
            "info": logging.INFO,
            "warning": logging.WARNING,
            "error": logging.ERROR,
            "critical": logging.CRITICAL
        }

        if level_name in level_map:
            logging_manager.set_log_level(level_map[level_name])
            self.logger.info(f"Log level set to {level_name}")
            if logging_manager.log_level > logging.INFO:
                # the only place where use a print given the change of log level might disable the logger
                print(f"Log level set to {level_name}")
        else:
            self.logger.error(f"Unknown log level: {level_name}. Available levels are: debug, info, warning, error, critical")
        return True

    def _cmd_version(self, _: str) -> bool:
        """Display the current version of Hatchling.

        Args:
            _ (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        self.logger.info(f"Hatchling version: {__version__}")
        return True
Functions
format_command(cmd_name, cmd_info, group='base')

Format base commands with custom styling.

Source code in hatchling/ui/base_commands.py 
104
105
106
107
108
109
110
def format_command(self, cmd_name: str, cmd_info: dict, group: str = 'base') -> list:
    """Format base commands with custom styling."""
    return [
        (f'class:command.name.{group}', f"{cmd_name}"),
        ('', ' - '),
        ('class:command.description', f"{cmd_info['description']}")
    ]
print_commands_help()

Print help for all available chat commands.

Source code in hatchling/ui/base_commands.py 
 95
 96
 97
 98
 99
100
101
102
def print_commands_help(self) -> None:
    """Print help for all available chat commands."""
    print_formatted_text(FormattedText([
        ('class:header', "\n=== Base Chat Commands ===\n")
    ]), style=self.style)

    # Call parent class method to print formatted commands
    super().print_commands_help()

Functions

hatchling.ui.chat_command_handler

Chat command handler module for processing user commands in the chat interface.

This module provides a central handler for all chat commands by combining base commands, Hatch-specific commands, and settings commands into a unified interface.

Classes

ChatCommandHandler

Handles processing of command inputs in the chat interface.

Source code in hatchling/ui/chat_command_handler.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class ChatCommandHandler:
    """Handles processing of command inputs in the chat interface."""    
    def __init__(self, chat_session, settings_registry: SettingsRegistry, style: Optional[Style] = None):
        """Initialize the command handler.

        Args:
            chat_session: The chat session this handler is associated with.
            settings_registry (SettingsRegistry): The settings registry containing configuration.
            style (Optional[Style]): Style for formatting command output.
        """


        self.settings_registry = settings_registry
        self.base_commands = BaseChatCommands(chat_session, settings_registry, style)
        self.hatch_commands = HatchCommands(chat_session, settings_registry, style)
        self.settings_commands = SettingsCommands(chat_session, settings_registry, style)
        self.mcp_commands = MCPCommands(chat_session, settings_registry, style)
        self.model_commands = ModelCommands(chat_session, settings_registry, style)

        self.logger = logging_manager.get_session("hatchling.core.chat.command_handler")

        self._register_commands()

    def _register_commands(self) -> None:
        """Register all available chat commands with their handlers."""
        # Combine all commands from all handlers
        self.commands = {}
        self.commands.update(self.base_commands.reload_commands())
        self.commands.update(self.hatch_commands.reload_commands())
        self.commands.update(self.settings_commands.reload_commands())
        self.commands.update(self.mcp_commands.reload_commands())
        self.commands.update(self.model_commands.reload_commands())

        self.command_completer = FuzzyCompleter(CommandCompleter(self.commands, mcp_manager.hatch_env_manager))
        self.command_lexer = ChatCommandLexer(self.commands)

        # Keep old format for backward compatibility
        self.sync_commands = {}
        self.async_commands = {}

        for cmd_name, cmd_info in self.commands.items():
            if cmd_info['is_async']:
                self.async_commands[cmd_name] = (cmd_info['handler'], cmd_info['description'])
            else:
                self.sync_commands[cmd_name] = (cmd_info['handler'], cmd_info['description'])

    def print_commands_help(self) -> None:
        """Print help for all available chat commands."""
        print("\n=== Chat Commands ===")
        print("Type 'help' for this help message")
        print()

        self.base_commands.print_commands_help()
        self.hatch_commands.print_commands_help()
        self.settings_commands.print_commands_help()
        self.mcp_commands.print_commands_help()
        self.model_commands.print_commands_help()

        print("======================\n")

    def set_commands_language(self, language_code: str) -> None:
        """Set the language for all commands.

        Args:
            language_code (str): The language code to set.
        """
        if not self.settings_registry:
            self.logger.error(translate("errors.settings_registry_not_available"))

        try:
            success = self.settings_registry.set_language(language_code)
            if success:
                self._register_commands()  # Re-register commands to apply new language
            else:
                self.logger.error(translate("errors.set_language_failed", language=language_code))
        except Exception as e:
            self.logger.error(str(e))


    async def process_command(self, user_input: str) -> Tuple[bool, bool]:
        """Process a potential command from user input.

        Args:
            user_input (str): The user's input text.

        Returns:
            Tuple[bool, bool]: (is_command, should_continue)
              - is_command: True if input was a command
              - should_continue: False if chat session should end
        """
        user_input = user_input.strip()

        # Handle empty input
        if not user_input:
            return True, True

        # Extract command and arguments
        parts = user_input.split(' ', 1)
        command = parts[0].lower()
        args = parts[1] if len(parts) > 1 else ""

        if command == "help":
            self.print_commands_help()
            return True, True

        if command == "settings:language:set":
            self.set_commands_language(args.strip())
            return True, True

        # Check if the input is a registered command
        if command in self.sync_commands:
            handler_func, _ = self.sync_commands[command]
            return True, handler_func(args)
        elif command in self.async_commands:
            async_handler_func, _ = self.async_commands[command]
            return True, await async_handler_func(args)

        # Not a command
        return False, True

    def get_all_command_metadata(self) -> dict:
        """Get all command metadata from both command handlers.

        Returns:
            dict: Combined command metadata from base and hatch commands.
        """
        return self.commands
Functions
__init__(chat_session, settings_registry, style=None)

Initialize the command handler.

Parameters:

Name Type Description Default
chat_session

The chat session this handler is associated with.

 required
settings_registry SettingsRegistry

The settings registry containing configuration.

 required
style Optional[Style]

Style for formatting command output.

 None
Source code in hatchling/ui/chat_command_handler.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(self, chat_session, settings_registry: SettingsRegistry, style: Optional[Style] = None):
    """Initialize the command handler.

    Args:
        chat_session: The chat session this handler is associated with.
        settings_registry (SettingsRegistry): The settings registry containing configuration.
        style (Optional[Style]): Style for formatting command output.
    """


    self.settings_registry = settings_registry
    self.base_commands = BaseChatCommands(chat_session, settings_registry, style)
    self.hatch_commands = HatchCommands(chat_session, settings_registry, style)
    self.settings_commands = SettingsCommands(chat_session, settings_registry, style)
    self.mcp_commands = MCPCommands(chat_session, settings_registry, style)
    self.model_commands = ModelCommands(chat_session, settings_registry, style)

    self.logger = logging_manager.get_session("hatchling.core.chat.command_handler")

    self._register_commands()
get_all_command_metadata()

Get all command metadata from both command handlers.

Returns:

Name Type Description
dict dict

Combined command metadata from base and hatch commands.
Source code in hatchling/ui/chat_command_handler.py 
145
146
147
148
149
150
151
def get_all_command_metadata(self) -> dict:
    """Get all command metadata from both command handlers.

    Returns:
        dict: Combined command metadata from base and hatch commands.
    """
    return self.commands
print_commands_help()

Print help for all available chat commands.

Source code in hatchling/ui/chat_command_handler.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def print_commands_help(self) -> None:
    """Print help for all available chat commands."""
    print("\n=== Chat Commands ===")
    print("Type 'help' for this help message")
    print()

    self.base_commands.print_commands_help()
    self.hatch_commands.print_commands_help()
    self.settings_commands.print_commands_help()
    self.mcp_commands.print_commands_help()
    self.model_commands.print_commands_help()

    print("======================\n")
process_command(user_input) async

Process a potential command from user input.

Parameters:

Name Type Description Default
user_input str

The user's input text.

 required

Returns:

Type Description
Tuple[bool, bool]

Tuple[bool, bool]: (is_command, should_continue) - is_command: True if input was a command - should_continue: False if chat session should end
Source code in hatchling/ui/chat_command_handler.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def process_command(self, user_input: str) -> Tuple[bool, bool]:
    """Process a potential command from user input.

    Args:
        user_input (str): The user's input text.

    Returns:
        Tuple[bool, bool]: (is_command, should_continue)
          - is_command: True if input was a command
          - should_continue: False if chat session should end
    """
    user_input = user_input.strip()

    # Handle empty input
    if not user_input:
        return True, True

    # Extract command and arguments
    parts = user_input.split(' ', 1)
    command = parts[0].lower()
    args = parts[1] if len(parts) > 1 else ""

    if command == "help":
        self.print_commands_help()
        return True, True

    if command == "settings:language:set":
        self.set_commands_language(args.strip())
        return True, True

    # Check if the input is a registered command
    if command in self.sync_commands:
        handler_func, _ = self.sync_commands[command]
        return True, handler_func(args)
    elif command in self.async_commands:
        async_handler_func, _ = self.async_commands[command]
        return True, await async_handler_func(args)

    # Not a command
    return False, True
set_commands_language(language_code)

Set the language for all commands.

Parameters:

Name Type Description Default
language_code str

The language code to set.

 required
Source code in hatchling/ui/chat_command_handler.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def set_commands_language(self, language_code: str) -> None:
    """Set the language for all commands.

    Args:
        language_code (str): The language code to set.
    """
    if not self.settings_registry:
        self.logger.error(translate("errors.settings_registry_not_available"))

    try:
        success = self.settings_registry.set_language(language_code)
        if success:
            self._register_commands()  # Re-register commands to apply new language
        else:
            self.logger.error(translate("errors.set_language_failed", language=language_code))
    except Exception as e:
        self.logger.error(str(e))

Functions

hatchling.ui.command_completion

Command completion module for the chat interface.

This module provides autocompletion functionality for chat commands using prompt_toolkit. It implements a three-phase approach: - Phase 2: Static command completion (command names, subcommands, argument flags) - Phase 3: Dynamic value completion (environment names, package names, file paths)

Classes

CommandCompleter

Bases: Completer

Main completer class that provides autocompletion for chat commands.

Source code in hatchling/ui/command_completion.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
class CommandCompleter(Completer):
    """Main completer class that provides autocompletion for chat commands."""

    def __init__(self, commands: Dict[str, Dict[str, Any]], env_manager: HatchEnvironmentManager):
        """Initialize the command completer.

        Args:
            commands: Dictionary containing command metadata from ChatCommandHandler
            env_manager: Hatch environment manager for dynamic completion
        """
        self.commands = commands
        self.env_manager = env_manager
        self.path_completer = PathCompleter()

        # Cache for dynamic completions to improve performance
        self._environment_cache = None
        self._package_cache = {}  # env_name -> package_list

    def get_completions(self, document: Document, complete_event) -> Iterable[Completion]:
        """Get completions for the current document position.

        Args:
            document: The current document
            complete_event: Completion event details

        Yields:
            Completion: Available completions
        """
        # Get the current line and cursor position
        text = document.text_before_cursor

        # If we're at the beginning or have no text, suggest command names
        if not text or text.isspace():
            yield from self._get_command_completions("")
            return

        # Split the input into parts
        parts = text.split()

        # If we're still typing the first word, complete command names
        if len(parts) == 1 and not text.endswith(' '):
            yield from self._get_command_completions(parts[0])
            return

        # If we have a command, complete its arguments
        if len(parts) >= 1:
            command = parts[0].lower()
            if command in self.commands:
                yield from self._get_argument_completions(command, parts[1:], text)
                return

        # Fallback - no completions available
        return []

    def set_commands(self, commands: Dict[str, Dict[str, Any]]):
        """Set the command metadata for this completer.

        Args:
            commands: Dictionary containing command metadata
        """
        self.commands = commands

    def _get_command_completions(self, prefix: str) -> Iterable[Completion]:
        """Get completions for command names.

        Args:
            prefix: The current command prefix being typed

        Yields:
            Completion: Command completions
        """
        for cmd_name, cmd_info in self.commands.items():
            if cmd_name.lower().startswith(prefix.lower()):
                # Calculate the start position for replacement
                start_position = -len(prefix) if prefix else 0

                yield Completion(
                    text=cmd_name,
                    start_position=start_position,
                    display=cmd_name,
                    display_meta=cmd_info.get('description', '')
                )

    def _get_argument_completions(self, command: str, args: List[str], full_text: str) -> Iterable[Completion]:
        """Get completions for command arguments.

        Args:
            command: The command name
            args: List of arguments already typed
            full_text: The full input text

        Yields:
            Completion: Argument completions
        """
        cmd_info = self.commands[command]
        arg_defs = cmd_info.get('args', {})

        if not arg_defs:
            return []

        # Check if we're completing a flag argument (starts with -)
        current_word = args[-1] if args and not full_text.endswith(' ') else ""

        if current_word.startswith('-'):
            yield from self._get_flag_completions(arg_defs, current_word)
            return

        # Check if the previous argument was a flag that expects a value
        if len(args) >= 1 and args[-1].startswith('-'):
            flag_name = args[-1].lstrip('-')
            yield from self._get_flag_value_completions(arg_defs, flag_name, current_word)
            return

        # Complete positional arguments
        yield from self._get_positional_completions(arg_defs, args, current_word)

        # Also suggest available flags
        if not current_word.startswith('-'):
            yield from self._get_available_flags(arg_defs, args)

    def _get_flag_completions(self, arg_defs: Dict[str, Dict], prefix: str) -> Iterable[Completion]:
        """Get completions for flag arguments.

        Args:
            arg_defs: Argument definitions
            prefix: Current flag prefix being typed

        Yields:
            Completion: Flag completions
        """
        for arg_name, arg_def in arg_defs.items():
            if arg_def.get('positional', False):
                continue

            # Complete long form (--argument)
            long_form = f"--{arg_name}"
            if long_form.startswith(prefix):
                start_position = -len(prefix)
                yield Completion(
                    text=long_form,
                    start_position=start_position,
                    display=long_form,
                    display_meta=arg_def.get('description', '')
                )

            # Complete short form aliases (-a)
            aliases = arg_def.get('aliases', [])
            for alias in aliases:
                short_form = f"-{alias}"
                if short_form.startswith(prefix):
                    start_position = -len(prefix)
                    yield Completion(
                        text=short_form,
                        start_position=start_position,
                        display=short_form,
                        display_meta=f"{arg_def.get('description', '')} (alias for --{arg_name})"
                    )

    def _get_flag_value_completions(self, arg_defs: Dict[str, Dict], flag_name: str, current_value: str) -> Iterable[Completion]:
        """Get completions for flag values.

        Args:
            arg_defs: Argument definitions
            flag_name: The flag name that expects a value
            current_value: Current value being typed

        Yields:
            Completion: Value completions
        """
        # Find the argument definition (could be by name or alias)
        arg_def = None
        for name, definition in arg_defs.items():
            if name == flag_name or flag_name in definition.get('aliases', []):
                arg_def = definition
                break

        if not arg_def:
            return []

        yield from self._get_value_completions(arg_def, current_value)

    def _get_positional_completions(self, arg_defs: Dict[str, Dict], args: List[str], current_word: str) -> Iterable[Completion]:
        """Get completions for positional arguments.

        Args:
            arg_defs: Argument definitions
            args: Arguments already provided
            current_word: Current word being typed

        Yields:
            Completion: Positional argument completions
        """
        # Find positional arguments in order
        positional_args = [
            (name, definition) for name, definition in arg_defs.items() 
            if definition.get('positional', False)
        ]

        # Determine which positional argument we're completing
        # Account for flags that might have consumed some arguments
        positional_index = len([arg for arg in args if not arg.startswith('-')])
        if current_word and not current_word.startswith('-'):
            positional_index -= 1

        if 0 <= positional_index < len(positional_args):
            _, arg_def = positional_args[positional_index]
            yield from self._get_value_completions(arg_def, current_word)

    def _get_available_flags(self, arg_defs: Dict[str, Dict], args: List[str]) -> Iterable[Completion]:
        """Get available flag suggestions.

        Args:
            arg_defs: Argument definitions
            args: Arguments already provided

        Yields:
            Completion: Available flag completions
        """
        # Get flags that have already been used
        used_flags = set()
        for arg in args:
            if arg.startswith('--'):
                used_flags.add(arg[2:])
            elif arg.startswith('-') and len(arg) == 2:
                # Find the flag name for this alias
                for name, definition in arg_defs.items():
                    if arg[1] in definition.get('aliases', []):
                        used_flags.add(name)
                        break

        # Suggest unused flags
        for arg_name, arg_def in arg_defs.items():
            if arg_def.get('positional', False) or arg_name in used_flags:
                continue

            long_form = f"--{arg_name}"
            yield Completion(
                text=long_form,
                start_position=0,
                display=long_form,
                display_meta=arg_def.get('description', '')
            )

    def _get_value_completions(self, arg_def: Dict[str, Any], current_value: str) -> Iterable[Completion]:
        """Get completions for argument values based on completer type.

        Args:
            arg_def: Argument definition
            current_value: Current value being typed

        Yields:
            Completion: Value completions
        """
        completer_type = arg_def.get('completer_type', 'none')
        start_position = -len(current_value) if current_value else 0

        if completer_type == 'suggestions':
            # Static suggestions
            suggestions = arg_def.get('values', [])
            for suggestion in suggestions:
                if suggestion.lower().startswith(current_value.lower()):
                    yield Completion(
                        text=suggestion,
                        start_position=start_position,
                        display=suggestion
                    )

        elif completer_type == 'environment':
            # Dynamic environment completions
            environments = self._get_environments()
            for env_name in environments:
                if env_name.lower().startswith(current_value.lower()):
                    yield Completion(
                        text=env_name,
                        start_position=start_position,
                        display=env_name,
                        display_meta="Hatch environment"
                    )

        elif completer_type == 'package':
            # Dynamic package completions
            packages = self._get_packages()
            for pkg_name in packages:
                if pkg_name.lower().startswith(current_value.lower()):
                    yield Completion(
                        text=pkg_name,
                        start_position=start_position,
                        display=pkg_name,
                        display_meta="Installed package"
                    )
        elif completer_type == 'path':
            # File path completions using prompt_toolkit's PathCompleter
            document = Document(current_value, len(current_value))
            for completion in self.path_completer.get_completions(document, None):
                yield completion

        elif completer_type == 'local_package':
            # Path completion with Hatch package detection/styling
            document = Document(current_value, len(current_value))

            # Get basic path completions first
            for completion in self.path_completer.get_completions(document, None):
                # Get full path by combining current path with completion
                full_path = self._get_full_path(current_value, completion.text)

                # Check if it's a Hatch package
                is_hatch_package = self._is_hatch_package(full_path)

                # Apply appropriate styling based on package status
                style = "fg:ansigreen bold" if is_hatch_package else "fg:ansired"
                display_meta = "Hatch Package" if is_hatch_package else "Directory"

                # Yield modified completion with styling
                yield Completion(
                    text=completion.text,
                    start_position=completion.start_position,
                    display=completion.display,
                    display_meta=display_meta,
                    style=style
                )

        elif completer_type == 'languages':
            languages = get_available_languages()
            for lang in languages:
                yield Completion(
                    text=lang['code'],
                    start_position=start_position,
                    display=lang['code'],
                    display_meta=lang.get('name', '')
                )

        # For 'none' type, no completions are provided

    def _get_environments(self) -> List[str]:
        """Get list of available Hatch environments.

        Returns:
            List[str]: Environment names
        """
        if self._environment_cache is None:
            try:
                environments = self.env_manager.list_environments()
                self._environment_cache = [env.get('name', '') for env in environments if env.get('name')]
            except Exception:
                self._environment_cache = []

        return self._environment_cache

    def _get_packages(self, env_name: Optional[str] = None) -> List[str]:
        """Get list of installed packages in an environment.

        Args:
            env_name: Environment name (uses current if None)

        Returns:
            List[str]: Package names
        """
        cache_key = env_name or 'current'

        if cache_key not in self._package_cache:
            try:
                packages = self.env_manager.list_packages(env_name)
                self._package_cache[cache_key] = [pkg.get('name', '') for pkg in packages if pkg.get('name')]
            except Exception:
                self._package_cache[cache_key] = []
        return self._package_cache[cache_key]

    def _is_hatch_package(self, path_str: str) -> bool:
        """Check if a directory contains hatch_metadata.json.

        Args:
            path_str: Path to the directory to check

        Returns:
            bool: True if directory contains hatch_metadata.json
        """
        try:
            path = Path(path_str)
            if path.is_dir():
                metadata_file = path / "hatch_metadata.json"
                return metadata_file.exists()
        except (OSError, PermissionError):
            pass
        return False

    def _get_full_path(self, current_input: str, completion_text: str) -> str:
        """Combine current input with completion to get full path.

        Args:
            current_input: The text the user has typed so far
            completion_text: The completion suggestion

        Returns:
            str: Full path combining input and completion
        """
        try:
            # Handle cases where current_input is empty or just whitespace
            if not current_input.strip():
                return completion_text

            # Handle relative paths correctly
            if current_input.endswith('/') or current_input.endswith('\\'):
                return current_input + completion_text

            # Get the directory part of the current input
            current_path = Path(current_input)
            if current_path.is_absolute():
                # For absolute paths, combine with parent directory
                return str(current_path.parent / completion_text)
            else:
                # For relative paths, get parent directory
                parent = current_path.parent
                if str(parent) == '.':
                    return completion_text
                return str(parent / completion_text)
        except (OSError, ValueError):
            # Fallback to simple concatenation if path operations fail
            return completion_text

    def invalidate_cache(self):
        """Invalidate cached dynamic completions."""
        self._environment_cache = None
        self._package_cache.clear()
Functions
__init__(commands, env_manager)

Initialize the command completer.

Parameters:

Name Type Description Default
commands Dict[str, Dict[str, Any]]

Dictionary containing command metadata from ChatCommandHandler

 required
env_manager HatchEnvironmentManager

Hatch environment manager for dynamic completion

 required
Source code in hatchling/ui/command_completion.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def __init__(self, commands: Dict[str, Dict[str, Any]], env_manager: HatchEnvironmentManager):
    """Initialize the command completer.

    Args:
        commands: Dictionary containing command metadata from ChatCommandHandler
        env_manager: Hatch environment manager for dynamic completion
    """
    self.commands = commands
    self.env_manager = env_manager
    self.path_completer = PathCompleter()

    # Cache for dynamic completions to improve performance
    self._environment_cache = None
    self._package_cache = {}  # env_name -> package_list
get_completions(document, complete_event)

Get completions for the current document position.

Parameters:

Name Type Description Default
document Document

The current document

 required
complete_event

Completion event details

 required

Yields:

Name Type Description
Completion Iterable[Completion]

Available completions
Source code in hatchling/ui/command_completion.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def get_completions(self, document: Document, complete_event) -> Iterable[Completion]:
    """Get completions for the current document position.

    Args:
        document: The current document
        complete_event: Completion event details

    Yields:
        Completion: Available completions
    """
    # Get the current line and cursor position
    text = document.text_before_cursor

    # If we're at the beginning or have no text, suggest command names
    if not text or text.isspace():
        yield from self._get_command_completions("")
        return

    # Split the input into parts
    parts = text.split()

    # If we're still typing the first word, complete command names
    if len(parts) == 1 and not text.endswith(' '):
        yield from self._get_command_completions(parts[0])
        return

    # If we have a command, complete its arguments
    if len(parts) >= 1:
        command = parts[0].lower()
        if command in self.commands:
            yield from self._get_argument_completions(command, parts[1:], text)
            return

    # Fallback - no completions available
    return []
invalidate_cache()

Invalidate cached dynamic completions.

Source code in hatchling/ui/command_completion.py 
437
438
439
440
def invalidate_cache(self):
    """Invalidate cached dynamic completions."""
    self._environment_cache = None
    self._package_cache.clear()
set_commands(commands)

Set the command metadata for this completer.

Parameters:

Name Type Description Default
commands Dict[str, Dict[str, Any]]

Dictionary containing command metadata

 required
Source code in hatchling/ui/command_completion.py 
72
73
74
75
76
77
78
def set_commands(self, commands: Dict[str, Dict[str, Any]]):
    """Set the command metadata for this completer.

    Args:
        commands: Dictionary containing command metadata
    """
    self.commands = commands

Functions

hatchling.ui.command_lexer

Command lexer for real-time syntax highlighting of chat commands.

This module provides a custom lexer that highlights commands, arguments, and values as the user types them in the chat interface.

Classes

ChatCommandLexer

Bases: Lexer

Custom lexer for highlighting chat commands in real-time.

Source code in hatchling/ui/command_lexer.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class ChatCommandLexer(Lexer):
    """Custom lexer for highlighting chat commands in real-time."""

    def __init__(self, commands: Dict[str, Dict[str, Any]]):
        """Initialize the lexer with command metadata.

        Args:
            commands: Dictionary containing command information including
                            command names, arguments, and their types.
        """
        self.commands = commands

        # Build command patterns
        self.command_names = set(commands.keys())

        # Build argument patterns for each command
        self.command_args = {}
        for cmd_name, cmd_info in commands.items():
            if 'args' in cmd_info:
                self.command_args[cmd_name] = cmd_info['args']

    def lex_document(self, document: Document) -> callable:
        """Lex the document and return a function that yields style/text tuples.

        Args:
            document: The document to lex.

        Returns:
            A function that takes a line number and yields (style, text) tuples.
        """
        def get_tokens(line_number: int):
            # Get the line content
            try:
                lines = document.text.split('\n')
                if line_number >= len(lines):
                    return []

                line_text = lines[line_number]
                if not line_text.strip():
                    return [('', line_text)]

                # Tokenize the input
                tokens = self._tokenize(line_text)

                result = []
                for token_type, token_text in tokens:
                    result.append((self._get_style_for_token(token_type), token_text))

                return result
            except Exception:
                # Fall back to plain text if tokenization fails
                return [('', line_text if 'line_text' in locals() else '')]

        return get_tokens

    def _tokenize(self, text: str) -> List[Tuple[str, str]]:
        """Tokenize the input text into command components.

        Args:
            text: Input text to tokenize.

        Returns:
            List of (token_type, token_text) tuples.
        """
        tokens = []

        # Simple tokenization - split by spaces but respect quotes
        parts = self._split_respecting_quotes(text)

        if not parts:
            return [('text', text)]

        # First part should be the command
        command = parts[0]

        # Check if it's a valid command
        if command in self.command_names:
            # Get command group for styling
            cmd_info = self.commands[command]
            group = 'hatch' if command.startswith('hatch:') else 'base'

            tokens.append((f'command.{group}', command))

            # Process arguments
            if len(parts) > 1:
                tokens.extend(self._tokenize_arguments(command, parts[1:], group))
        else:
            # Not a command, treat as regular text
            tokens.append(('text', text))

        return tokens

    def _split_respecting_quotes(self, text: str) -> List[str]:
        """Split text by spaces while respecting quoted strings.

        Args:
            text: Text to split.

        Returns:
            List of text parts.
        """
        parts = []
        current_part = ""
        in_quotes = False
        quote_char = None

        i = 0
        while i < len(text):
            char = text[i]

            if char in ['"', "'"]:
                if not in_quotes:
                    in_quotes = True
                    quote_char = char
                elif char == quote_char:
                    in_quotes = False
                    quote_char = None
                current_part += char
            elif char.isspace() and not in_quotes:
                if current_part:
                    parts.append(current_part)
                    current_part = ""
                # Add whitespace as separate token to preserve formatting
                while i < len(text) and text[i].isspace():
                    current_part += text[i]
                    i += 1
                if current_part:
                    parts.append(current_part)
                    current_part = ""
                i -= 1  # Adjust for the extra increment
            else:
                current_part += char

            i += 1

        if current_part:
            parts.append(current_part)

        return parts

    def _tokenize_arguments(self, command: str, arg_parts: List[str], group: str) -> List[Tuple[str, str]]:
        """Tokenize command arguments.

        Args:
            command: The command name.
            arg_parts: List of argument parts.
            group: Command group (base/hatch).

        Returns:
            List of (token_type, token_text) tuples.
        """
        tokens = []
        command_args = self.command_args.get(command, {})

        for i, part in enumerate(arg_parts):
            if part.isspace():
                tokens.append(('whitespace', part))
                continue

            # Check if it's a flag argument (starts with - or --)
            if part.startswith('-'):
                arg_name = part.lstrip('-')

                # Check if it's a valid argument for this command
                if arg_name in command_args:
                    tokens.append((f'argument.{group}', part))
                else:
                    # Check aliases
                    found_alias = False
                    for name, arg_def in command_args.items():
                        if arg_name in arg_def.get('aliases', []):
                            tokens.append((f'argument.{group}', part))
                            found_alias = True
                            break

                    if not found_alias:
                        tokens.append(('argument.invalid', part))
            else:
                # Could be a positional argument or a value
                # For simplicity, we'll style positional args as values
                # A more sophisticated approach would track argument expectations
                if self._looks_like_path(part):
                    tokens.append(('value.path', part))
                elif self._looks_like_number(part):
                    tokens.append(('value.number', part))
                elif part.startswith('"') or part.startswith("'"):
                    tokens.append(('value.string', part))
                else:
                    tokens.append(('value.generic', part))

        return tokens

    def _looks_like_path(self, text: str) -> bool:
        """Check if text looks like a file path."""
        return ('/' in text or '\\' in text or 
                text.startswith('./') or text.startswith('../') or
                text.endswith('.py') or text.endswith('.json') or
                text.endswith('.txt'))

    def _looks_like_number(self, text: str) -> bool:
        """Check if text looks like a number."""
        try:
            float(text)
            return True
        except ValueError:
            return False
    def _get_style_for_token(self, token_type: str) -> str:
        """Get the CSS style class for a token type.

        Args:
            token_type: The token type.

        Returns:
            CSS style class name.
        """
        # Map token types to CSS classes
        style_map = {
            'command.base': 'class:command.name.base',
            'command.hatch': 'class:command.name.hatch',
            'command': 'class:command.name',  # Fallback for generic commands
            'argument.base': 'class:command.args.base',
            'argument.hatch': 'class:command.args.hatch',
            'argument.invalid': 'class:command.args.invalid',
            'value.path': 'class:command.value.path',
            'value.number': 'class:command.value.number',
            'value.string': 'class:command.value.string',
            'value.generic': 'class:command.value.generic',
            'whitespace': '',
            'text': 'class:text.default',
        }

        return style_map.get(token_type, '')
Functions
__init__(commands)

Initialize the lexer with command metadata.

Parameters:

Name Type Description Default
commands Dict[str, Dict[str, Any]]

Dictionary containing command information including command names, arguments, and their types.

 required
Source code in hatchling/ui/command_lexer.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def __init__(self, commands: Dict[str, Dict[str, Any]]):
    """Initialize the lexer with command metadata.

    Args:
        commands: Dictionary containing command information including
                        command names, arguments, and their types.
    """
    self.commands = commands

    # Build command patterns
    self.command_names = set(commands.keys())

    # Build argument patterns for each command
    self.command_args = {}
    for cmd_name, cmd_info in commands.items():
        if 'args' in cmd_info:
            self.command_args[cmd_name] = cmd_info['args']
lex_document(document)

Lex the document and return a function that yields style/text tuples.

Parameters:

Name Type Description Default
document Document

The document to lex.

 required

Returns:

Type Description
callable

A function that takes a line number and yields (style, text) tuples.
Source code in hatchling/ui/command_lexer.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def lex_document(self, document: Document) -> callable:
    """Lex the document and return a function that yields style/text tuples.

    Args:
        document: The document to lex.

    Returns:
        A function that takes a line number and yields (style, text) tuples.
    """
    def get_tokens(line_number: int):
        # Get the line content
        try:
            lines = document.text.split('\n')
            if line_number >= len(lines):
                return []

            line_text = lines[line_number]
            if not line_text.strip():
                return [('', line_text)]

            # Tokenize the input
            tokens = self._tokenize(line_text)

            result = []
            for token_type, token_text in tokens:
                result.append((self._get_style_for_token(token_type), token_text))

            return result
        except Exception:
            # Fall back to plain text if tokenization fails
            return [('', line_text if 'line_text' in locals() else '')]

    return get_tokens

Chat Interface

hatchling.ui.cli_chat

Classes

CLIChat

Command-line interface for chat functionality.

Source code in hatchling/ui/cli_chat.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
class CLIChat:
    """Command-line interface for chat functionality."""

    def __init__(self, settings_registry: SettingsRegistry):
        """Initialize the CLI chat interface.

        Args:
            settings (SettingsRegistry): The settings management instance containing configuration.
        """
        # Store settings first
        self.settings_registry = settings_registry

        # Get a logger - styling is already configured at the application level
        self.logger = logging_manager.get_session("CLIChat")

        # Initialize prompt toolkit session with history
        history_dir = self.settings_registry.settings.paths.hatchling_cache_dir / 'histories'
        history_dir.mkdir(exist_ok=True, parents=True)

        # Setup persistent history with 500 entries limit
        try:
            self.prompt_session = PromptSession(
                history=FileHistory(str(history_dir / '.user_inputs')))
        except (IOError, OSError) as e:
            self.logger.warning(f"Could not create history file: {e}")
            self.logger.warning("Falling back to in-memory history")
            self.prompt_session = PromptSession(history=InMemoryHistory())

        # Define command styling for both help display and real-time input highlighting
        self.command_style = Style.from_dict({
            # Help display styles
            'command.name': 'bold #44ff00',          # Green bold for command names
            'command.description': "#ffffff",        # White for descriptions
            'command.args': 'italic #87afff',        # Light blue italic for arguments
            'header': 'bold #ff9d00 underline',      # Orange underline for headers

            # Group specific styles for help
            'command.name.hatch': 'bold #00b7c3',    # Teal for Hatch commands
            'command.name.base': 'bold #44ff00',     # Green for base commands
            'group.default': '',                     # Default group style

            # Real-time input highlighting styles
            'command.name': 'bold #44ff00',          # Command names - bright green
            'command.args.base': 'bold #87afff',     # Base command arguments - blue
            'command.args.hatch': 'bold #00b7c3',    # Hatch command arguments - teal
            'command.args.invalid': '#ff6b6b',       # Invalid arguments - red
            'command.value.path': '#ffb347',         # Path values - orange
            'command.value.number': '#98fb98',       # Number values - light green
            'command.value.string': '#dda0dd',       # String values - plum
            'command.value.generic': '#f0f0f0',      # Generic values - light gray
            'text.default': '#ffffff',               # Default text - white

            # Toolbar styles
            'toolbar.default': "#63818d",            # Sky blue for default toolbar
            'toolbar.tool': '#ffa500',               # Orange for tool execution
            'toolbar.error': '#ff6b6b',              # Red for errors
            'toolbar.info': "#49a949",               # Light green for info

            # Right prompt styles
            'right-prompt': '#d3d3d3',               # Light gray for right prompt
        })

        # Provider will be initialized during startup
        # Initialize the provider
        try:
            ProviderRegistry.get_provider(self.settings_registry.settings.llm.provider_enum)

        except Exception as e:
            msg = f"Failed to initialize {self.settings_registry.settings.llm.provider_enum} LLM provider: {e}"
            msg += "\nEnsure the LLM provider name is correct in your settings."
            msg += "\nYou can list providers compatible with Hatchling using `model:provider:list` command."
            msg += "\nEnsure you have switched to a supported provider before trying to use the chat interface."
            self.logger.warning(msg)

        finally:
            # Initialize chat session
            self.chat_session = ChatSession()

            # Initialize CLI event subscriber for UI state management
            self.cli_event_subscriber = CLIEventSubscriber()

            # Register CLI subscriber with chat session (decoupled)
            self.chat_session.register_subscriber(self.cli_event_subscriber)
            mcp_manager.publisher.subscribe(self.cli_event_subscriber)

            # Initialize command handler
            self.cmd_handler = ChatCommandHandler(self.chat_session, self.settings_registry, self.command_style)

            # Setup key bindings
            self.key_bindings = self._create_key_bindings()

    def _create_key_bindings(self) -> KeyBindings:
        """Create key bindings for UI control.

        Returns:
            KeyBindings: Configured key bindings for the CLI.
        """
        kb = KeyBindings()

        @kb.add('f2')
        def _(event):
            """Cycle toolbar view mode."""
            self.cli_event_subscriber.cycle_toolbar_view()
            # Force UI refresh by triggering a redraw
            event.app.invalidate()

        @kb.add('f3')
        def _(event):
            """Cycle right prompt view mode."""
            self.cli_event_subscriber.cycle_right_prompt_view()
            # Force UI refresh by triggering a redraw
            event.app.invalidate()

        @kb.add('f4')
        def _(event):
            """Clear current error/info messages."""
            self.cli_event_subscriber.current_error = None
            self.cli_event_subscriber.current_info = None
            event.app.invalidate()

        @kb.add('f10')
        def _(event):
            """Panic: forcibly reset UI state and allow user input."""
            self.logger.warning("F10 PANIC: Forcibly resetting UI state to allow user input.")
            self.cli_event_subscriber.set_processing_user_message(False)
            event.app.invalidate()

        @kb.add('f12')
        def _(event):
            """Show help for key bindings."""
            help_text = (
                "📋 Key Bindings:\n"
                "F2  - Cycle toolbar views\n"
                "F3  - Cycle right prompt views\n"
                "F4  - Clear messages\n"
                "F12 - Show this help\n"
            )
            print_pt(FormattedText([('class:toolbar.info', help_text)]))

        return kb

    def _get_bottom_toolbar(self) -> FormattedText:
        """Get bottom toolbar text with styling.

        Returns:
            FormattedText: Formatted toolbar text.
        """
        toolbar_text = self.cli_event_subscriber.get_toolbar_text()

        # Style based on content
        if toolbar_text.startswith('❌'):
            return FormattedText([('class:toolbar.error', toolbar_text)])
        elif toolbar_text.startswith('ℹ️'):
            return FormattedText([('class:toolbar.info', toolbar_text)])
        elif toolbar_text.startswith('🔧'):
            return FormattedText([('class:toolbar.tool', toolbar_text)])
        else:
            return FormattedText([('class:toolbar.default', toolbar_text)])

    def _get_right_prompt(self) -> FormattedText:
        """Get right prompt text with styling.

        Returns:
            FormattedText: Formatted right prompt text.
        """
        right_prompt_text = self.cli_event_subscriber.get_right_prompt_text()
        return FormattedText([('class:right-prompt', right_prompt_text)])

    async def start_interactive_session(self) -> None:
        """Run an interactive chat session with message history."""

        #async with aiohttp.ClientSession() as session:
        while True: 
            try:
                # Create formatted prompt
                prompt_message = [
                    # status_style,
                    ('green', 'You: ')
                ]
                # Use patch_stdout to prevent output interference
                with patch_stdout():
                    user_message = await self.prompt_session.prompt_async(
                        FormattedText(prompt_message),
                        completer=self.cmd_handler.command_completer,
                        lexer=self.cmd_handler.command_lexer,
                        style=self.command_style,
                        key_bindings=self.key_bindings,
                        bottom_toolbar=self._get_bottom_toolbar,
                        rprompt=self._get_right_prompt
                    )

                # Process as command if applicable
                is_command, should_continue = await self.cmd_handler.process_command(user_message)
                if is_command:
                    if not should_continue:
                        break
                    continue

                # Handle normal message
                if not user_message.strip():
                    # Skip empty input
                    continue

                # Mark that we're starting to process user message
                self.cli_event_subscriber.set_processing_user_message(True)

                # Clear previous content from event subscriber
                #self.cli_event_subscriber.clear_content_buffer()

                try:
                    # Send the message (this will trigger streaming events)
                    await self.chat_session.send_message(user_message)

                    # Wait for all processing to complete (tool chains, etc.)
                    await self._monitor_right_to_prompt()

                except Exception as send_error:
                    # Make sure to reset state on error
                    self.cli_event_subscriber.set_processing_user_message(False)
                    raise send_error

                print_pt('')  # Add an extra newline for readability

            except KeyboardInterrupt:
                print_pt(FormattedText([('red', '\nInterrupted. Ending chat session...')]))
                break
            except Exception as e:
                self.logger.error(f"Error: {e}")
                print_pt(FormattedText([('red', f'\nError: {e}')]))

    async def initialize_and_run(self) -> None:
        """Initialize the environment and run the interactive chat session."""
        try:
            # Start the interactive session
            await self.start_interactive_session()

        except Exception as e:
            error_msg = f"An error occurred: {e}"
            self.logger.error(error_msg)
            return

        finally:
            # Clean up any remaining MCP server processes only if tools were enabled
            if self.chat_session and len(mcp_manager.get_enabled_tools()) > 0:
                await mcp_manager.disconnect_all()

    async def _monitor_right_to_prompt(self) -> None:
        """
        Blocks until the all conditions are satisfied to finish a prompt loop
        and go back to async prompt input.
        """
        # Give a small delay to allow events to propagate
        await asyncio.sleep(0.1)

        while not self.cli_event_subscriber.is_ready_for_user_input():
            self.logger.debug("Waiting for user input readiness...")
            await asyncio.sleep(0.25)  # Check every 250ms
Functions
__init__(settings_registry)

Initialize the CLI chat interface.

Parameters:

Name Type Description Default
settings SettingsRegistry

The settings management instance containing configuration.

 required
Source code in hatchling/ui/cli_chat.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(self, settings_registry: SettingsRegistry):
    """Initialize the CLI chat interface.

    Args:
        settings (SettingsRegistry): The settings management instance containing configuration.
    """
    # Store settings first
    self.settings_registry = settings_registry

    # Get a logger - styling is already configured at the application level
    self.logger = logging_manager.get_session("CLIChat")

    # Initialize prompt toolkit session with history
    history_dir = self.settings_registry.settings.paths.hatchling_cache_dir / 'histories'
    history_dir.mkdir(exist_ok=True, parents=True)

    # Setup persistent history with 500 entries limit
    try:
        self.prompt_session = PromptSession(
            history=FileHistory(str(history_dir / '.user_inputs')))
    except (IOError, OSError) as e:
        self.logger.warning(f"Could not create history file: {e}")
        self.logger.warning("Falling back to in-memory history")
        self.prompt_session = PromptSession(history=InMemoryHistory())

    # Define command styling for both help display and real-time input highlighting
    self.command_style = Style.from_dict({
        # Help display styles
        'command.name': 'bold #44ff00',          # Green bold for command names
        'command.description': "#ffffff",        # White for descriptions
        'command.args': 'italic #87afff',        # Light blue italic for arguments
        'header': 'bold #ff9d00 underline',      # Orange underline for headers

        # Group specific styles for help
        'command.name.hatch': 'bold #00b7c3',    # Teal for Hatch commands
        'command.name.base': 'bold #44ff00',     # Green for base commands
        'group.default': '',                     # Default group style

        # Real-time input highlighting styles
        'command.name': 'bold #44ff00',          # Command names - bright green
        'command.args.base': 'bold #87afff',     # Base command arguments - blue
        'command.args.hatch': 'bold #00b7c3',    # Hatch command arguments - teal
        'command.args.invalid': '#ff6b6b',       # Invalid arguments - red
        'command.value.path': '#ffb347',         # Path values - orange
        'command.value.number': '#98fb98',       # Number values - light green
        'command.value.string': '#dda0dd',       # String values - plum
        'command.value.generic': '#f0f0f0',      # Generic values - light gray
        'text.default': '#ffffff',               # Default text - white

        # Toolbar styles
        'toolbar.default': "#63818d",            # Sky blue for default toolbar
        'toolbar.tool': '#ffa500',               # Orange for tool execution
        'toolbar.error': '#ff6b6b',              # Red for errors
        'toolbar.info': "#49a949",               # Light green for info

        # Right prompt styles
        'right-prompt': '#d3d3d3',               # Light gray for right prompt
    })

    # Provider will be initialized during startup
    # Initialize the provider
    try:
        ProviderRegistry.get_provider(self.settings_registry.settings.llm.provider_enum)

    except Exception as e:
        msg = f"Failed to initialize {self.settings_registry.settings.llm.provider_enum} LLM provider: {e}"
        msg += "\nEnsure the LLM provider name is correct in your settings."
        msg += "\nYou can list providers compatible with Hatchling using `model:provider:list` command."
        msg += "\nEnsure you have switched to a supported provider before trying to use the chat interface."
        self.logger.warning(msg)

    finally:
        # Initialize chat session
        self.chat_session = ChatSession()

        # Initialize CLI event subscriber for UI state management
        self.cli_event_subscriber = CLIEventSubscriber()

        # Register CLI subscriber with chat session (decoupled)
        self.chat_session.register_subscriber(self.cli_event_subscriber)
        mcp_manager.publisher.subscribe(self.cli_event_subscriber)

        # Initialize command handler
        self.cmd_handler = ChatCommandHandler(self.chat_session, self.settings_registry, self.command_style)

        # Setup key bindings
        self.key_bindings = self._create_key_bindings()
initialize_and_run() async

Initialize the environment and run the interactive chat session.

Source code in hatchling/ui/cli_chat.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
async def initialize_and_run(self) -> None:
    """Initialize the environment and run the interactive chat session."""
    try:
        # Start the interactive session
        await self.start_interactive_session()

    except Exception as e:
        error_msg = f"An error occurred: {e}"
        self.logger.error(error_msg)
        return

    finally:
        # Clean up any remaining MCP server processes only if tools were enabled
        if self.chat_session and len(mcp_manager.get_enabled_tools()) > 0:
            await mcp_manager.disconnect_all()
start_interactive_session() async

Run an interactive chat session with message history.

Source code in hatchling/ui/cli_chat.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
async def start_interactive_session(self) -> None:
    """Run an interactive chat session with message history."""

    #async with aiohttp.ClientSession() as session:
    while True: 
        try:
            # Create formatted prompt
            prompt_message = [
                # status_style,
                ('green', 'You: ')
            ]
            # Use patch_stdout to prevent output interference
            with patch_stdout():
                user_message = await self.prompt_session.prompt_async(
                    FormattedText(prompt_message),
                    completer=self.cmd_handler.command_completer,
                    lexer=self.cmd_handler.command_lexer,
                    style=self.command_style,
                    key_bindings=self.key_bindings,
                    bottom_toolbar=self._get_bottom_toolbar,
                    rprompt=self._get_right_prompt
                )

            # Process as command if applicable
            is_command, should_continue = await self.cmd_handler.process_command(user_message)
            if is_command:
                if not should_continue:
                    break
                continue

            # Handle normal message
            if not user_message.strip():
                # Skip empty input
                continue

            # Mark that we're starting to process user message
            self.cli_event_subscriber.set_processing_user_message(True)

            # Clear previous content from event subscriber
            #self.cli_event_subscriber.clear_content_buffer()

            try:
                # Send the message (this will trigger streaming events)
                await self.chat_session.send_message(user_message)

                # Wait for all processing to complete (tool chains, etc.)
                await self._monitor_right_to_prompt()

            except Exception as send_error:
                # Make sure to reset state on error
                self.cli_event_subscriber.set_processing_user_message(False)
                raise send_error

            print_pt('')  # Add an extra newline for readability

        except KeyboardInterrupt:
            print_pt(FormattedText([('red', '\nInterrupted. Ending chat session...')]))
            break
        except Exception as e:
            self.logger.error(f"Error: {e}")
            print_pt(FormattedText([('red', f'\nError: {e}')]))

hatchling.ui.cli_event_subscriber

CLI Event Subscriber for managing UI state and display.

Classes

CLIEventSubscriber

Bases: EventSubscriber

CLI Event Subscriber for managing UI state based on stream events.

This subscriber maintains state for all UI elements: - Bottom toolbar: tool execution status and server/tool counts - Right prompt: LLM info and token statistics - Error/info overlays: transient messages

Uses the Observer pattern to react to events and update UI state.

Source code in hatchling/ui/cli_event_subscriber.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
class CLIEventSubscriber(EventSubscriber):
    """CLI Event Subscriber for managing UI state based on stream events.

    This subscriber maintains state for all UI elements:
    - Bottom toolbar: tool execution status and server/tool counts
    - Right prompt: LLM info and token statistics
    - Error/info overlays: transient messages

    Uses the Observer pattern to react to events and update UI state.
    """

    def __init__(self, settings: Optional[AppSettings] = None):
        """Initialize the CLI event subscriber with default state."""
        self.logger = logging_manager.get_session("CLIEventSubscriber")

        # UI State
        self.server_status = ServerStatus()
        self.token_stats = TokenStats()

        self.settings = settings or AppSettings.get_instance()

        # UI Control State
        self.toolbar_view_mode = "default"  # default, tools, servers
        self.right_prompt_view_mode = "default"  # default, tokens, model

        # Error/Info Display
        self.current_error: Optional[str] = None
        self.current_info: Optional[str] = None
        self.message_timeout: float = 5.0  # seconds
        self.last_message_time: float = 0.0

        # UI state flags manager
        self.ui_state = UIStateManager()

    def on_event(self, event: Event) -> None:
        """Handle stream events and update UI state.

        Args:
            event (Event): The event to handle.
        """

        try:
            # Tool Chaining Events
            if event.type == EventType.TOOL_CHAIN_START:
                self.logger.debug(f"Handling TOOL_CHAIN_START event: {event.data}")
                self._handle_tool_chain_start(event)
            elif event.type == EventType.TOOL_CHAIN_ITERATION_START:
                self._handle_tool_chain_iteration_start(event)
            elif event.type == EventType.TOOL_CHAIN_ITERATION_END:
                self._handle_tool_chain_iteration_end(event)
            elif event.type == EventType.TOOL_CHAIN_END:
                self.logger.debug(f"Handling TOOL_CHAIN_END event: {event.data}")
                self._handle_tool_chain_end(event)
            elif event.type == EventType.TOOL_CHAIN_LIMIT_REACHED:
                self.logger.debug(f"Handling TOOL_CHAIN_LIMIT_REACHED event: {event.data}")
                self._handle_tool_chain_limit_reached(event)
            elif event.type == EventType.TOOL_CHAIN_ERROR:
                self.logger.debug(f"Handling TOOL_CHAIN_ERROR event: {event.data}")
                self._handle_tool_chain_error(event)

            # Tool Execution Events
            elif event.type == EventType.LLM_TOOL_CALL_REQUEST:
                self.logger.debug(f"Handling LLM_TOOL_CALL_REQUEST event: {event.data}")
                self._handle_llm_tool_call_request(event)
            elif event.type == EventType.MCP_TOOL_CALL_DISPATCHED:
                self.logger.debug(f"Handling MCP_TOOL_CALL_DISPATCHED event: {event.data}")
                self._handle_mcp_tool_call_dispatched(event)
            elif event.type == EventType.MCP_TOOL_CALL_RESULT:
                self.logger.debug(f"Handling MCP_TOOL_CALL_RESULT event: {event.data}")
                self._handle_mcp_tool_call_result(event)
            elif event.type == EventType.MCP_TOOL_CALL_ERROR:
                self.logger.debug(f"Handling MCP_TOOL_CALL_ERROR event: {event.data}")
                self._handle_mcp_tool_call_error(event)

            # MCP Server Events
            elif event.type == EventType.MCP_SERVER_UP:
                self.logger.debug(f"Handling MCP_SERVER_UP event: {event.data}")
                self._handle_mcp_server_up(event)
            elif event.type == EventType.MCP_SERVER_DOWN:
                self.logger.debug(f"Handling MCP_SERVER_DOWN event: {event.data}")
                self._handle_mcp_server_down(event)
            elif event.type == EventType.MCP_TOOL_ENABLED:
                self.logger.debug(f"Handling MCP_TOOL_ENABLED event: {event.data}")
                self._handle_mcp_tool_enabled(event)
            elif event.type == EventType.MCP_TOOL_DISABLED:
                self.logger.debug(f"Handling MCP_TOOL_DISABLED event: {event.data}")
                self._handle_mcp_tool_disabled(event)

            # LLM Events
            elif event.type == EventType.CONTENT:
                #self.logger.debug(f"Handling CONTENT event: {event.data}")
                self._handle_content(event)
            elif event.type == EventType.USAGE:
                self.logger.debug(f"Handling USAGE event: {event.data}")
                self._handle_usage(event)
            elif event.type == EventType.ERROR:
                self.logger.debug(f"Handling ERROR event: {event.data}")
                self._handle_error(event)
            elif event.type == EventType.FINISH:
                self.logger.debug(f"Handling FINISH event: {event.data}")
                self._handle_finish(event)

        except Exception as e:
            self.logger.error(f"Error handling event {event.type.value}: {e}")

    def get_subscribed_events(self) -> List[EventType]:
        """Return list of event types this subscriber is interested in.

        Returns:
            List[EventType]: Event types for UI updates.
        """
        return [
            # Tool Chaining Events
            EventType.TOOL_CHAIN_START,
            EventType.TOOL_CHAIN_END,
            EventType.TOOL_CHAIN_LIMIT_REACHED,
            EventType.TOOL_CHAIN_ERROR,

            # Tool Execution Events
            EventType.LLM_TOOL_CALL_REQUEST,
            EventType.MCP_TOOL_CALL_DISPATCHED,
            EventType.MCP_TOOL_CALL_RESULT,
            EventType.MCP_TOOL_CALL_ERROR,

            # MCP Server Events
            EventType.MCP_SERVER_UP,
            EventType.MCP_SERVER_DOWN,
            EventType.MCP_TOOL_ENABLED,
            EventType.MCP_TOOL_DISABLED,

            # LLM Events
            EventType.CONTENT,
            EventType.USAGE,
            EventType.ERROR,
            EventType.FINISH,
        ]

    # Tool Chaining Event Handlers
    def _handle_tool_chain_start(self, event: Event) -> None:
        """Handle tool chain start event."""
        data = event.data
        self.ui_state.set(UIStateFlags.TOOL_CHAIN_ACTIVE)
        self.ui_state.clear(UIStateFlags.USER_INPUT_READY)
        self.logger.debug(f"Tool chain started - TOOL_CHAIN_ACTIVE set, USER_INPUT_READY cleared")
        # Truncating initial query if it's too long
        initial_query = data.get("initial_query", "No query")
        if isinstance(initial_query, str) and len(initial_query) > 100:
            initial_query = initial_query[:100] + "..."
        self._set_info(
            f"[{data.get('tool_chain_id', 'ID unknown')}]\n" +
            f"Tool chaining started for initial query: {initial_query}\n" +
            f" {data.get('max_iterations', 0)} iterations allowed.")

    def _handle_tool_chain_iteration_start(self, event: Event) -> None:
        """Handle tool chain iteration event."""
        data = event.data
        self._set_info(
            f"[{data.get('tool_chain_id', 'ID unknown')}]\n" +
            f"Step {data.get('iteration', -1)}/{data.get('max_iterations', 0)}:\n" +
            f"Feeding back result of {data.get('tool_name', 'unknown')} to LLM.")
        self.ui_state.set(UIStateFlags.TOOL_CHAIN_ACTIVE)
        self.ui_state.clear(UIStateFlags.USER_INPUT_READY)

    def _handle_tool_chain_iteration_end(self, event: Event) -> None:
        """Handle tool chain iteration event."""
        data = event.data
        self._set_info(
            f"[{data.get('tool_chain_id', 'ID unknown')}]\n" +
            f"Step {data.get('iteration', -1)}/{data.get('max_iterations', 0)}:\n" +
            f"Result of {data.get('tool_name', 'unknown')} processed by the LLM.")

    def _handle_tool_chain_end(self, event: Event) -> None:
        """Handle tool chain end event."""
        data = event.data

        success = data.get("success", True)
        # Truncating initial query if it's too long
        initial_query = data.get("initial_query", "No query")
        if isinstance(initial_query, str) and len(initial_query) > 100:
            initial_query = initial_query[:100] + "..."

        if success:
            self._set_info(
                f"[{data.get('tool_chain_id', 'ID unknown')}]\n" +
                f"Tool chaining completed successfully for initial query: {initial_query}\n" +
                f"Iteration: {data.get('iteration', 0)}/{data.get('max_iterations', 0)}, Total time: {data.get('elapsed_time', 0):.2f} seconds"
            )
        else:
            self._set_error(
                f"[{data.get('tool_chain_id', 'ID unknown')}]\n" +
                f"Tool chaining failed for initial query: {initial_query}\n" +
                f"Iteration: {data.get('iteration', 0)}/{data.get('max_iterations', 0)}, Total time: {data.get('elapsed_time', 0):.2f} seconds"
            )

        self.current_chain = None
        self.ui_state.clear(UIStateFlags.TOOL_CHAIN_ACTIVE)
        self.logger.debug(f"Tool chain ended - TOOL_CHAIN_ACTIVE cleared")

        self._handle_finish(event)

    def _handle_tool_chain_limit_reached(self, event: Event) -> None:
        """Handle tool chain limit reached event."""
        data = event.data
        self._set_info(
            f"[{data.get('tool_chain_id', 'ID unknown')}]\n" +
            f"Tool chaining stopped: {data.get('limit_type', 'unknown')} ({data.get('iterations', 0)} steps, {data.get('elapsed_time', 0):.2f} seconds elapsed)"
        )
        self.logger.debug(f"Tool chain limit reached - TOOL_CHAIN_ACTIVE cleared")

    def _handle_tool_chain_error(self, event: Event) -> None:
        """Handle tool chain error event."""
        data = event.data
        self._set_error(
            f"[{data.get('tool_chain_id', 'ID unknown')}]\n" +
            f"Tool chaining failed at step {data.get('iteration', 0)}: {data.get('error', 'Unknown error')}"
        )
        self.ui_state.clear(UIStateFlags.TOOL_CHAIN_ACTIVE)
        self.logger.debug(f"Tool chain error - TOOL_CHAIN_ACTIVE cleared")

    # Tool Execution Event Handlers
    def _handle_llm_tool_call_request(self, event: Event) -> None:
        """Handle LLM tool call request event."""
        data = event.data
        # Set tool is running
        provider = ProviderRegistry.get_provider(event.provider)
        parsed_tool_call = provider.llm_to_hatchling_tool_call(event)
        self._set_info(
            f"[{parsed_tool_call.tool_call_id}]\n" +
            f"Tool call to {parsed_tool_call.function_name} requested with parameters:\n" +
            f"{', '.join([f'{k}={v}' for k, v in parsed_tool_call.arguments.items()])}"
        )


    def _handle_mcp_tool_call_dispatched(self, event: Event) -> None:
        """Handle MCP tool call dispatched event."""
        data = event.data
        self._set_info(
            f"[{data.get('tool_call_id', 'unknown')}]\n" +
            f"Tool {data.get('function_name', 'unknown')} dispatched with parameters:\n" +
            f"{', '.join([f'{k}={v}' for k, v in data.get('arguments', {}).items()])}"
        )

    def _handle_mcp_tool_call_result(self, event: Event) -> None:
        """Handle MCP tool call result event."""
        data = event.data
        # Truncate result if too long
        result = data.get("result", "No returned value")
        if isinstance(result, str) and len(result) > 100:
            result = result[:100] + "..."
        self._set_info(
            f"[{data.get('tool_call_id', 'unknown')}]\n" +
            f"Tool {data.get('function_name', 'unknown')} result: {result}"
        )

    def _handle_mcp_tool_call_error(self, event: Event) -> None:
        """Handle MCP tool call error event."""
        data = event.data
        tool_name = data.get("function_name", "unknown")
        error = data.get("error", "Unknown error")
        self._set_error(
            f"[{data.get('tool_call_id', 'ID unknown')}]\n" +
            f"Execution of tool {tool_name} failed: {error}"
        )
        self.current_tool = None
        # Don't clear any UI state flags here, as the tool call error
        # might be part of an ongoing tool chain. Hence we give the
        # LLM a chance to fix its mistake by retrying the tool call.
        # TODO: However, put up a warning
        #  TBD: Warning might be redundant with previous ones upstream
        #       in the code flow. 

    # MCP Server Event Handlers
    def _handle_mcp_server_up(self, event: Event) -> None:
        """Handle MCP server up event."""
        data = event.data
        self.server_status.servers_up += 1
        self.server_status.tools_total += data.get("tool_count", 0)
        server_path = data.get("server_path", "unknown")
        self._set_info(f"MCP server connected: {Path(server_path).name}")

    def _handle_mcp_server_down(self, event: Event) -> None:
        """Handle MCP server down event."""
        data = event.data
        self.server_status.servers_up -= 1
        self.server_status.tools_total -= data.get("tool_count", 0)
        server_path = data.get("server_path", "unknown")
        self._set_info(f"MCP server disconnected: {Path(server_path).name}")

    def _handle_mcp_tool_enabled(self, event: Event) -> None:
        """Handle MCP tool enabled event."""
        data = event.data
        self.server_status.tools_enabled += 1
        tool_info : MCPToolInfo = data.get("tool_info", {})
        self._set_info(f"Tool enabled: {tool_info.name} ({Path(tool_info.server_path).name})\n" +
                       f"\tDescription: {tool_info.description}\n" +
                       f"\tParameters: {', '.join([f'{k}={v}' for k, v in tool_info.schema.items()])}")

    def _handle_mcp_tool_disabled(self, event: Event) -> None:
        """Handle MCP tool disabled event."""
        data = event.data
        self.server_status.tools_enabled -= 1
        tool_info: MCPToolInfo = data.get("tool_info", {})
        if tool_info.reason == MCPToolStatusReason.FROM_SERVER_DOWN:
            self.server_status.tools_total -= 1 #If the server got disconnected, we decrement the tool count
        self._set_info(f"Tool disabled: {tool_info.name} ({Path(tool_info.server_path).name})")

    # LLM Event Handlers
    def _handle_usage(self, event: Event) -> None:
        """Handle usage statistics event.

        Args:
            event (Event): The event to handle.
        """
        usage_data = event.data.get("usage", {})
        self.token_stats.total_current = usage_data.get("total_tokens", 0)
        self.token_stats.total_tokens += self.token_stats.total_current
        self.token_stats.prompt_tokens = usage_data.get("prompt_tokens", 0)
        self.token_stats.completion_tokens = usage_data.get("completion_tokens", 0)
        # Optionally, print or log stats here if needed

    def _handle_content(self, event: Event) -> None:
        """Handle content event by accumulating content.

        Args:
            event (Event): The event to handle.
        """
        self.ui_state.clear(UIStateFlags.USER_INPUT_READY)
        self.ui_state.set(UIStateFlags.CONTENT_STREAMING)

        content = event.data.get("content", "")
        print_pt(content, end="", flush=True)

    def _handle_finish(self, event: Event) -> None:
        """Handle finish event.

        Args:
            event (Event): The event to handle.
        """
        if self.token_stats.start_time and not self.token_stats.end_time:
            self.token_stats.end_time = event.timestamp
        # Only if all other activities but content streaming are done
        # means that this finish event is the end of a content stream
        if self.ui_state.is_set(UIStateFlags.CONTENT_STREAMING):
            print() # new line
            if not self.ui_state.is_set(UIStateFlags.TOOL_CHAIN_ACTIVE):
                self.ui_state.clear(UIStateFlags.CONTENT_STREAMING)
                self.ui_state.set(UIStateFlags.USER_INPUT_READY)
                self.logger.debug("All content apparently finished, USER_INPUT_READY set")

    def _handle_error(self, event: Event) -> None:
        """Handle error event."""
        data = event.data
        error = data.get("error", "Unknown error")
        self._set_error(f"LLM Error: {error}")
        self.ui_state.clear(UIStateFlags.TOOL_CHAIN_ACTIVE)
        self.ui_state.set(UIStateFlags.USER_INPUT_READY)

    # UI State Management
    def _set_error(self, message: str) -> None:
        """Set error message with timestamp."""
        self.ui_state.set(UIStateFlags.ERROR_DISPLAYED)
        self.current_error = message
        self.current_info = None
        self.last_message_time = time.time()

    def _set_info(self, message: str) -> None:
        """Set info message with timestamp."""
        self.ui_state.set(UIStateFlags.INFO_DISPLAYED)
        self.current_info = message
        self.current_error = None
        self.last_message_time = time.time()

    def _clear_expired_messages(self) -> None:
        """Clear error/info messages after timeout."""
        if time.time() - self.last_message_time > self.message_timeout:
            self.current_error = None
            self.current_info = None

    # UI Display Methods
    def get_toolbar_text(self) -> str:
        """Get current toolbar text based on state and view mode."""
        self._clear_expired_messages()

        # Show error/info messages with priority
        if self.current_error:
            return f"❌ {self.current_error}"
        if self.current_info:
            return f"ℹ️ {self.current_info}"

        # Show server/tool status when idle
        return f"🌐 Servers: {self.server_status.servers_up} up | 🛠️ Tools: {self.server_status.tools_enabled} enabled / {self.server_status.tools_total} total"

    def get_right_prompt_text(self) -> str:
        """Get current right prompt text based on state and view mode."""
        duration = None
        tps = None
        if self.token_stats.start_time and self.token_stats.end_time and self.token_stats.completion_tokens > 0:
            duration = self.token_stats.end_time - self.token_stats.start_time
            tps = self.token_stats.completion_tokens / duration if duration > 0 else 0

        stats = ""
        if self.right_prompt_view_mode == "model":
            return f"🤖 {self.settings.llm.provider_name}\n{self.settings.llm.model}"

        if self.right_prompt_view_mode == "tokens":
            stats = f"📊 In: {self.token_stats.prompt_tokens}\nOut: {self.token_stats.completion_tokens}\nLast Query: {self.token_stats.total_current}"
            if tps is not None:
                stats += f"\n Rate: {tps:.1f}/s"
            stats += f"\nTotal: {self.token_stats.total_tokens}"
            return stats
        else:  # default
            stats = f"🤖 {self.settings.llm.model}\n📊 Last Query: {self.token_stats.total_current}"
            if tps is not None:
                stats += f"({tps:.1f}/s)"
            stats += f"\nTotal: {self.token_stats.total_tokens}"
            return stats

    def cycle_toolbar_view(self) -> None:
        """Cycle through toolbar view modes."""
        modes = ["default", "tools", "servers"]
        current_index = modes.index(self.toolbar_view_mode)
        self.toolbar_view_mode = modes[(current_index + 1) % len(modes)]

    def cycle_right_prompt_view(self) -> None:
        """Cycle through right prompt view modes."""
        modes = ["default", "tokens", "model"]
        current_index = modes.index(self.right_prompt_view_mode)
        self.right_prompt_view_mode = modes[(current_index + 1) % len(modes)]

    def is_ready_for_user_input(self) -> bool:
        """Check if the system is ready for user input using state flags."""
        # Not ready if tool chain or tool running
        not_ready_mask = UIStateFlags.TOOL_CHAIN_ACTIVE | UIStateFlags.CONTENT_STREAMING
        if self.ui_state.is_set(not_ready_mask):
            self.logger.debug(f"Not ready: {not_ready_mask} flag set:\n"
                              f" TOOL_CHAIN_ACTIVE={self.ui_state.is_set(UIStateFlags.TOOL_CHAIN_ACTIVE)}, "
                              f" CONTENT_STREAMING={self.ui_state.is_set(UIStateFlags.CONTENT_STREAMING)}")
            return False
        return self.ui_state.is_set(UIStateFlags.USER_INPUT_READY)

    def set_processing_user_message(self, processing: bool = True) -> None:
        """Set whether we're currently processing a user message using state flags.
        Args:
            processing (bool): True if processing, False if done.
        """
        if processing:
            self.ui_state.clear(UIStateFlags.USER_INPUT_READY)
            self.logger.debug("Set processing user message - USER_INPUT_READY cleared")
        else:
            self.ui_state.set(UIStateFlags.USER_INPUT_READY)
            self.logger.debug("Finished processing user message - USER_INPUT_READY set")
Functions
__init__(settings=None)

Initialize the CLI event subscriber with default state.

Source code in hatchling/ui/cli_event_subscriber.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def __init__(self, settings: Optional[AppSettings] = None):
    """Initialize the CLI event subscriber with default state."""
    self.logger = logging_manager.get_session("CLIEventSubscriber")

    # UI State
    self.server_status = ServerStatus()
    self.token_stats = TokenStats()

    self.settings = settings or AppSettings.get_instance()

    # UI Control State
    self.toolbar_view_mode = "default"  # default, tools, servers
    self.right_prompt_view_mode = "default"  # default, tokens, model

    # Error/Info Display
    self.current_error: Optional[str] = None
    self.current_info: Optional[str] = None
    self.message_timeout: float = 5.0  # seconds
    self.last_message_time: float = 0.0

    # UI state flags manager
    self.ui_state = UIStateManager()
cycle_right_prompt_view()

Cycle through right prompt view modes.

Source code in hatchling/ui/cli_event_subscriber.py 
493
494
495
496
497
def cycle_right_prompt_view(self) -> None:
    """Cycle through right prompt view modes."""
    modes = ["default", "tokens", "model"]
    current_index = modes.index(self.right_prompt_view_mode)
    self.right_prompt_view_mode = modes[(current_index + 1) % len(modes)]
cycle_toolbar_view()

Cycle through toolbar view modes.

Source code in hatchling/ui/cli_event_subscriber.py 
487
488
489
490
491
def cycle_toolbar_view(self) -> None:
    """Cycle through toolbar view modes."""
    modes = ["default", "tools", "servers"]
    current_index = modes.index(self.toolbar_view_mode)
    self.toolbar_view_mode = modes[(current_index + 1) % len(modes)]
get_right_prompt_text()

Get current right prompt text based on state and view mode.

Source code in hatchling/ui/cli_event_subscriber.py 
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
def get_right_prompt_text(self) -> str:
    """Get current right prompt text based on state and view mode."""
    duration = None
    tps = None
    if self.token_stats.start_time and self.token_stats.end_time and self.token_stats.completion_tokens > 0:
        duration = self.token_stats.end_time - self.token_stats.start_time
        tps = self.token_stats.completion_tokens / duration if duration > 0 else 0

    stats = ""
    if self.right_prompt_view_mode == "model":
        return f"🤖 {self.settings.llm.provider_name}\n{self.settings.llm.model}"

    if self.right_prompt_view_mode == "tokens":
        stats = f"📊 In: {self.token_stats.prompt_tokens}\nOut: {self.token_stats.completion_tokens}\nLast Query: {self.token_stats.total_current}"
        if tps is not None:
            stats += f"\n Rate: {tps:.1f}/s"
        stats += f"\nTotal: {self.token_stats.total_tokens}"
        return stats
    else:  # default
        stats = f"🤖 {self.settings.llm.model}\n📊 Last Query: {self.token_stats.total_current}"
        if tps is not None:
            stats += f"({tps:.1f}/s)"
        stats += f"\nTotal: {self.token_stats.total_tokens}"
        return stats
get_subscribed_events()

Return list of event types this subscriber is interested in.

Returns:

Type Description
List[EventType]

List[EventType]: Event types for UI updates.
Source code in hatchling/ui/cli_event_subscriber.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def get_subscribed_events(self) -> List[EventType]:
    """Return list of event types this subscriber is interested in.

    Returns:
        List[EventType]: Event types for UI updates.
    """
    return [
        # Tool Chaining Events
        EventType.TOOL_CHAIN_START,
        EventType.TOOL_CHAIN_END,
        EventType.TOOL_CHAIN_LIMIT_REACHED,
        EventType.TOOL_CHAIN_ERROR,

        # Tool Execution Events
        EventType.LLM_TOOL_CALL_REQUEST,
        EventType.MCP_TOOL_CALL_DISPATCHED,
        EventType.MCP_TOOL_CALL_RESULT,
        EventType.MCP_TOOL_CALL_ERROR,

        # MCP Server Events
        EventType.MCP_SERVER_UP,
        EventType.MCP_SERVER_DOWN,
        EventType.MCP_TOOL_ENABLED,
        EventType.MCP_TOOL_DISABLED,

        # LLM Events
        EventType.CONTENT,
        EventType.USAGE,
        EventType.ERROR,
        EventType.FINISH,
    ]
get_toolbar_text()

Get current toolbar text based on state and view mode.

Source code in hatchling/ui/cli_event_subscriber.py 
449
450
451
452
453
454
455
456
457
458
459
460
def get_toolbar_text(self) -> str:
    """Get current toolbar text based on state and view mode."""
    self._clear_expired_messages()

    # Show error/info messages with priority
    if self.current_error:
        return f"❌ {self.current_error}"
    if self.current_info:
        return f"ℹ️ {self.current_info}"

    # Show server/tool status when idle
    return f"🌐 Servers: {self.server_status.servers_up} up | 🛠️ Tools: {self.server_status.tools_enabled} enabled / {self.server_status.tools_total} total"
is_ready_for_user_input()

Check if the system is ready for user input using state flags.

Source code in hatchling/ui/cli_event_subscriber.py 
499
500
501
502
503
504
505
506
507
508
def is_ready_for_user_input(self) -> bool:
    """Check if the system is ready for user input using state flags."""
    # Not ready if tool chain or tool running
    not_ready_mask = UIStateFlags.TOOL_CHAIN_ACTIVE | UIStateFlags.CONTENT_STREAMING
    if self.ui_state.is_set(not_ready_mask):
        self.logger.debug(f"Not ready: {not_ready_mask} flag set:\n"
                          f" TOOL_CHAIN_ACTIVE={self.ui_state.is_set(UIStateFlags.TOOL_CHAIN_ACTIVE)}, "
                          f" CONTENT_STREAMING={self.ui_state.is_set(UIStateFlags.CONTENT_STREAMING)}")
        return False
    return self.ui_state.is_set(UIStateFlags.USER_INPUT_READY)
on_event(event)

Handle stream events and update UI state.

Parameters:

Name Type Description Default
event Event

The event to handle.

 required
Source code in hatchling/ui/cli_event_subscriber.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def on_event(self, event: Event) -> None:
    """Handle stream events and update UI state.

    Args:
        event (Event): The event to handle.
    """

    try:
        # Tool Chaining Events
        if event.type == EventType.TOOL_CHAIN_START:
            self.logger.debug(f"Handling TOOL_CHAIN_START event: {event.data}")
            self._handle_tool_chain_start(event)
        elif event.type == EventType.TOOL_CHAIN_ITERATION_START:
            self._handle_tool_chain_iteration_start(event)
        elif event.type == EventType.TOOL_CHAIN_ITERATION_END:
            self._handle_tool_chain_iteration_end(event)
        elif event.type == EventType.TOOL_CHAIN_END:
            self.logger.debug(f"Handling TOOL_CHAIN_END event: {event.data}")
            self._handle_tool_chain_end(event)
        elif event.type == EventType.TOOL_CHAIN_LIMIT_REACHED:
            self.logger.debug(f"Handling TOOL_CHAIN_LIMIT_REACHED event: {event.data}")
            self._handle_tool_chain_limit_reached(event)
        elif event.type == EventType.TOOL_CHAIN_ERROR:
            self.logger.debug(f"Handling TOOL_CHAIN_ERROR event: {event.data}")
            self._handle_tool_chain_error(event)

        # Tool Execution Events
        elif event.type == EventType.LLM_TOOL_CALL_REQUEST:
            self.logger.debug(f"Handling LLM_TOOL_CALL_REQUEST event: {event.data}")
            self._handle_llm_tool_call_request(event)
        elif event.type == EventType.MCP_TOOL_CALL_DISPATCHED:
            self.logger.debug(f"Handling MCP_TOOL_CALL_DISPATCHED event: {event.data}")
            self._handle_mcp_tool_call_dispatched(event)
        elif event.type == EventType.MCP_TOOL_CALL_RESULT:
            self.logger.debug(f"Handling MCP_TOOL_CALL_RESULT event: {event.data}")
            self._handle_mcp_tool_call_result(event)
        elif event.type == EventType.MCP_TOOL_CALL_ERROR:
            self.logger.debug(f"Handling MCP_TOOL_CALL_ERROR event: {event.data}")
            self._handle_mcp_tool_call_error(event)

        # MCP Server Events
        elif event.type == EventType.MCP_SERVER_UP:
            self.logger.debug(f"Handling MCP_SERVER_UP event: {event.data}")
            self._handle_mcp_server_up(event)
        elif event.type == EventType.MCP_SERVER_DOWN:
            self.logger.debug(f"Handling MCP_SERVER_DOWN event: {event.data}")
            self._handle_mcp_server_down(event)
        elif event.type == EventType.MCP_TOOL_ENABLED:
            self.logger.debug(f"Handling MCP_TOOL_ENABLED event: {event.data}")
            self._handle_mcp_tool_enabled(event)
        elif event.type == EventType.MCP_TOOL_DISABLED:
            self.logger.debug(f"Handling MCP_TOOL_DISABLED event: {event.data}")
            self._handle_mcp_tool_disabled(event)

        # LLM Events
        elif event.type == EventType.CONTENT:
            #self.logger.debug(f"Handling CONTENT event: {event.data}")
            self._handle_content(event)
        elif event.type == EventType.USAGE:
            self.logger.debug(f"Handling USAGE event: {event.data}")
            self._handle_usage(event)
        elif event.type == EventType.ERROR:
            self.logger.debug(f"Handling ERROR event: {event.data}")
            self._handle_error(event)
        elif event.type == EventType.FINISH:
            self.logger.debug(f"Handling FINISH event: {event.data}")
            self._handle_finish(event)

    except Exception as e:
        self.logger.error(f"Error handling event {event.type.value}: {e}")
set_processing_user_message(processing=True)

Set whether we're currently processing a user message using state flags. Args: processing (bool): True if processing, False if done.

Source code in hatchling/ui/cli_event_subscriber.py 
510
511
512
513
514
515
516
517
518
519
520
def set_processing_user_message(self, processing: bool = True) -> None:
    """Set whether we're currently processing a user message using state flags.
    Args:
        processing (bool): True if processing, False if done.
    """
    if processing:
        self.ui_state.clear(UIStateFlags.USER_INPUT_READY)
        self.logger.debug("Set processing user message - USER_INPUT_READY cleared")
    else:
        self.ui_state.set(UIStateFlags.USER_INPUT_READY)
        self.logger.debug("Finished processing user message - USER_INPUT_READY set")

ServerStatus dataclass

Status information for MCP servers and tools.

Source code in hatchling/ui/cli_event_subscriber.py 
50
51
52
53
54
55
56
57
@dataclass
class ServerStatus:
    """Status information for MCP servers and tools."""

    servers_up: int = 0
    servers_total: int = 0
    tools_enabled: int = 0
    tools_total: int = 0

TokenStats dataclass

Token usage statistics.

Source code in hatchling/ui/cli_event_subscriber.py 
60
61
62
63
64
65
66
67
68
@dataclass
class TokenStats:
    """Token usage statistics."""
    total_tokens: int = 0
    total_current: int = 0
    prompt_tokens: int = 0
    completion_tokens: int = 0
    start_time: float = None
    end_time: float = None

UIStateManager

Utility for managing UI state flags.

Source code in hatchling/ui/cli_event_subscriber.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class UIStateManager:
    """Utility for managing UI state flags."""
    def __init__(self):
        self.flags = UIStateFlags.USER_INPUT_READY

    def set(self, flag: UIStateFlags):
        self.flags |= flag

    def clear(self, flag: UIStateFlags):
        self.flags &= ~flag

    def is_set(self, flag: UIStateFlags) -> bool:
        return bool(self.flags & flag)

    def reset(self):
        self.flags = UIStateFlags.NONE

    def set_only(self, flag: UIStateFlags):
        self.flags = flag

Specific Commands

hatchling.ui.hatch_commands

Hatch package manager commands module for the chat interface.

This module provides commands for interacting with the Hatch package manager, including environment management, package operations, and template creation.

Classes

HatchCommands

Bases: AbstractCommands

Handles Hatch package manager commands in the chat interface.

Source code in hatchling/ui/hatch_commands.py 
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52
  53
  54
  55
  56
  57
  58
  59
  60
  61
  62
  63
  64
  65
  66
  67
  68
  69
  70
  71
  72
  73
  74
  75
  76
  77
  78
  79
  80
  81
  82
  83
  84
  85
  86
  87
  88
  89
  90
  91
  92
  93
  94
  95
  96
  97
  98
  99
 100
 101
 102
 103
 104
 105
 106
 107
 108
 109
 110
 111
 112
 113
 114
 115
 116
 117
 118
 119
 120
 121
 122
 123
 124
 125
 126
 127
 128
 129
 130
 131
 132
 133
 134
 135
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
class HatchCommands(AbstractCommands):
    """Handles Hatch package manager commands in the chat interface."""

    def _register_commands(self) -> None:
        """Register all available Hatch package manager commands."""

        self.commands = {
            # Environment commands
            'hatch:env:list': {
                'handler': self._cmd_env_list,
                'description': translate('commands.hatch.env_list_description'),
                'is_async': False,
                'args': {}
            },
            'hatch:env:create': {
                'handler': self._cmd_env_create,
                'description': translate('commands.hatch.env_create_description'),
                'is_async': False,
                'args': {
                    'name': {
                        'positional': True,
                        'completer_type': 'none',
                        'description': translate('commands.args.env_name_description'),
                        'required': True
                    },
                    'description': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.env_description_description'),
                        'aliases': ['D'],
                        'default': '',
                        'required': False
                    },
                    'python-version': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.python_version_description'),
                        'default': None,
                        'required': False
                    },
                    'no-python': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.no_python_description'),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'no-hatch-mcp-server': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.no_hatch_mcp_server_description'),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'hatch_mcp_server_tag': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.hatch_mcp_server_tag_description'),
                        'default': None,
                        'required': False
                    }
                }
            },
            'hatch:env:remove': {
                'handler': self._cmd_env_remove,
                'description': translate('commands.hatch.env_remove_description'),
                'is_async': False,
                'args': {
                    'name': {
                        'positional': True,
                        'completer_type': 'environment',
                        'description': translate('commands.args.env_remove_name_description'),
                        'required': True
                    }
                }
            },
            'hatch:env:current': {
                'handler': self._cmd_env_current,
                'description': translate('commands.hatch.env_current_description'),
                'is_async': False,
                'args': {}
            },
            'hatch:env:use': {
                'handler': self._cmd_env_use,
                'description': translate('commands.hatch.env_use_description'),
                'is_async': True,
                'args': {
                    'name': {
                        'positional': True,
                        'completer_type': 'environment',
                        'description': translate('commands.args.env_use_name_description'),
                        'required': True
                    }
                }
            },
            # Package commands
            'hatch:pkg:add': {
                'handler': self._cmd_pkg_add,
                'description': translate('commands.hatch.pkg_add_description'),
                'is_async': False,
                'args': {
                    'package_path_or_name': {
                        'positional': True,
                        'completer_type': 'local_package',
                        'description': translate('commands.args.package_path_or_name_description'),
                        'required': True
                    },
                    'env': {
                        'positional': False,
                        'completer_type': 'environment',
                        'description': translate('commands.args.env_target_description'),
                        'aliases': ['e'],
                        'default': None,
                        'required': False
                    },
                    'version': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.package_version_description'),
                        'aliases': ['v'],
                        'default': None,
                        'required': False
                    },
                    'force-download': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.force_download_description'),
                        'aliases': ['f'],
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'refresh-registry': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.refresh_registry_description'),
                        'aliases': ['r'],
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'auto-approve': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.auto_approve_description'),
                        'aliases': ['y'],
                        'default': False,
                        'is_flag': True,
                        'required': False
                    }
                }
            },
            'hatch:pkg:remove': {
                'handler': self._cmd_pkg_remove,
                'description': translate('commands.hatch.pkg_remove_description'),
                'is_async': False,
                'args': {
                    'package_name': {
                        'positional': True,
                        'completer_type': 'package',
                        'description': translate('commands.args.package_name_description'),
                        'required': True
                    },
                    'env': {
                        'positional': False,
                        'completer_type': 'environment',
                        'description': translate('commands.args.env_remove_package_description'),
                        'aliases': ['e'],
                        'default': None,
                        'required': False
                    }
                }
            },
            'hatch:pkg:list': {
                'handler': self._cmd_pkg_list,
                'description': translate('commands.hatch.pkg_list_description'),
                'is_async': False,
                'args': {
                    'env': {
                        'positional': False,
                        'completer_type': 'environment',
                        'description': translate('commands.args.env_list_packages_description'),
                        'aliases': ['e'],
                        'default': None,
                        'required': False
                    }
                }
            },
            # Package creation command
            'hatch:pkg:create': {
                'handler': self._cmd_create_package,
                'description': translate('commands.hatch.pkg_create_description'),
                'is_async': False,
                'args': {
                    'name': {
                        'positional': True,
                        'completer_type': 'none',
                        'description': translate('commands.args.package_create_name_description'),
                        'required': True
                    },
                    'dir': {
                        'positional': False,
                        'completer_type': 'path',
                        'description': translate('commands.args.dir_description'),
                        'aliases': ['d'],
                        'default': '.',
                        'required': False
                    },
                    'description': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.package_description_description'),
                        'aliases': ['D'],
                        'default': '',
                        'required': False
                    }
                }
            },
            # Package validation command
            'hatch:pkg:validate': {
                'handler': self._cmd_validate_package,
                'description': translate('commands.hatch.pkg_validate_description'),
                'is_async': False,
                'args': {
                    'package_dir': {
                        'positional': True,
                        'completer_type': 'path',
                        'description': translate('commands.args.package_dir_description'),
                        'required': True
                    }
                }
            },
            # Python environment management commands
            'hatch:env:python:init': {
                'handler': self._cmd_env_python_init,
                'description': translate('commands.hatch.env_python_init_description'),
                'is_async': False,
                'args': {
                    'hatch_env': {
                        'positional': False,
                        'completer_type': 'environment',
                        'description': translate('commands.args.hatch_env_description'),
                        'default': None,
                        'required': False
                    },
                    'python-version': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.python_version_description'),
                        'default': None,
                        'required': False
                    },
                    'force': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.force_recreation_description'),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'no-hatch-mcp-server': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.no_hatch_mcp_server_wrapper_description'),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'hatch_mcp_server_tag': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.hatch_mcp_server_tag_description'),
                        'default': None,
                        'required': False
                    }
                }
            },
            'hatch:env:python:info': {
                'handler': self._cmd_env_python_info,
                'description': translate('commands.hatch.env_python_info_description'),
                'is_async': False,
                'args': {
                    'hatch_env': {
                        'positional': False,
                        'completer_type': 'environment',
                        'description': translate('commands.args.hatch_env_description'),
                        'default': None,
                        'required': False
                    },
                    'detailed': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.detailed_description'),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    }
                }
            },
            'hatch:env:python:remove': {
                'handler': self._cmd_env_python_remove,
                'description': translate('commands.hatch.env_python_remove_description'),
                'is_async': False,
                'args': {
                    'hatch_env': {
                        'positional': False,
                        'completer_type': 'environment',
                        'description': translate('commands.args.hatch_env_description'),
                        'default': None,
                        'required': False
                    },
                    'force': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.force_removal_description'),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    }
                }
            },
            'hatch:env:python:shell': {
                'handler': self._cmd_env_python_shell,
                'description': translate('commands.hatch.env_python_shell_description'),
                'is_async': False,
                'args': {
                    'hatch_env': {
                        'positional': False,
                        'completer_type': 'environment',
                        'description': translate('commands.args.hatch_env_description'),
                        'default': None,
                        'required': False
                    },
                    'cmd': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.cmd_description'),
                        'default': None,
                        'required': False
                    }
                }
            },
            'hatch:env:python:add-hatch-mcp': {
                'handler': self._cmd_env_python_add_hatch_mcp,
                'description': translate('commands.hatch.env_python_add_hatch_mcp_description'),
                'is_async': False,
                'args': {
                    'hatch_env': {
                        'positional': False,
                        'completer_type': 'environment',
                        'description': translate('commands.args.hatch_env_special_description'),
                        'default': None,
                        'required': False
                    },
                    'tag': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.args.tag_description'),
                        'default': None,
                        'required': False
                    }
                }
            }
        }

    def print_commands_help(self) -> None:
        """Print help for all available chat commands."""
        print_formatted_text(FormattedText([
            ('class:header', "\n=== Hatch Chat Commands ===\n")
        ]), style=self.style)

        super().print_commands_help()

    def format_command(self, cmd_name: str, cmd_info: Dict[str, Any], group: str = 'hatch') -> list:
        """Format Hatch commands with custom styling."""
        return [
            (f'class:command.name.{group}', f"{cmd_name}"),
            ('', ' - '),
            ('class:command.description', f"{cmd_info['description']}")
        ]

    def _cmd_env_list(self, _: str) -> bool:
        """List all available Hatch environments.

        Args:
            _ (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            environments = mcp_manager.hatch_env_manager.list_environments()

            if not environments:
                print("No Hatch environments found.")
                return True

            print("Available Hatch environments:")
            for env in environments:
                current_marker = "* " if env.get("is_current") else "  "
                description = f" - {env.get('description')}" if env.get("description") else ""
                print(f"{current_marker}{env.get('name')}{description}")

        except Exception as e:
            self.logger.error(f"Error listing environments: {e}")

        return True

    def _cmd_env_create(self, args: str) -> bool:
        """Create a new Hatch environment.

        Creates a new Hatch environment with optional Python environment support and
        automatic hatch_mcp_server installation. The MCP server installation can be
        controlled via command flags.

        Args:
            args (str): Environment name and optional parameters including:
                      - description: Environment description
                      - python-version: Specific Python version
                      - no-python: Skip Python environment creation
                      - no-hatch-mcp-server: Skip MCP server installation
                      - hatch_mcp_server_tag: Git tag/branch for MCP server

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'name': {'positional': True},
            'description': {'aliases': ['D'], 'default': ''},
            'python-version': {'default': None},
            'no-python': {'default': False, 'action': 'store_true'},
            'no-hatch-mcp-server': {'default': False, 'action': 'store_true'},
            'hatch_mcp_server_tag': {'default': None}
        }

        parsed_args = self._parse_args(args, arg_defs)

        if 'name' not in parsed_args or not parsed_args['name']:
            self.logger.error("Environment name is required.")
            self._print_command_help('hatch:env:create')
            return True

        try:
            name = parsed_args['name']
            description = parsed_args.get('description', '')
            python_version = parsed_args.get('python-version')
            create_python_env = not parsed_args.get('no-python', False)
            no_hatch_mcp_server = parsed_args.get('no-hatch-mcp-server', False)
            hatch_mcp_server_tag = parsed_args.get('hatch_mcp_server_tag')

            if mcp_manager.hatch_env_manager.create_environment(
                name, 
                description, 
                python_version=python_version, 
                create_python_env=create_python_env,
                no_hatch_mcp_server=no_hatch_mcp_server,
                hatch_mcp_server_tag=hatch_mcp_server_tag
            ):                
                self.logger.info(f"Environment created: {name}")
                if create_python_env and python_version:
                    self.logger.info(f"Python environment initialized with version: {python_version}")
                elif create_python_env:
                    self.logger.info("Python environment initialized with default version")
                else:
                    self.logger.info("Python environment creation skipped (--no-python)")

                # Provide feedback about MCP server installation
                if create_python_env and not no_hatch_mcp_server:
                    if hatch_mcp_server_tag:
                        self.logger.info(f"hatch_mcp_server installed with tag/branch: {hatch_mcp_server_tag}")
                    else:
                        self.logger.info("hatch_mcp_server installed with default branch")
                elif no_hatch_mcp_server:
                    self.logger.info("hatch_mcp_server installation skipped (--no-hatch-mcp-server)")
                elif not create_python_env:
                    self.logger.info("hatch_mcp_server installation skipped (no Python environment)")
            else:
                self.logger.error(f"Failed to create environment: {name}")

        except Exception as e:
            self.logger.error(f"Error creating environment: {e}")

        return True

    def _cmd_env_remove(self, args: str) -> bool:
        """Remove a Hatch environment.

        Args:
            args (str): Environment name.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'name': {'positional': True}
        }

        parsed_args = self._parse_args(args, arg_defs)

        if 'name' not in parsed_args or not parsed_args['name']:
            self.logger.error("Environment name is required.")
            self._print_command_help('hatch:env:remove')
            return True

        try:
            name = parsed_args['name']

            if mcp_manager.hatch_env_manager.remove_environment(name):
                self.logger.info(f"Environment removed: {name}")
            else:
                self.logger.error(f"Failed to remove environment: {name}")

        except Exception as e:
            self.logger.error(f"Error removing environment: {e}")

        return True

    def _cmd_env_current(self, _: str) -> bool:
        """Show the current Hatch environment.

        Args:
            _ (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            current_env = mcp_manager.hatch_env_manager.get_current_environment()
            if current_env:
                self.logger.info(f"Current environment: {current_env}")
            else:
                self.logger.info("No current environment set.")

        except Exception as e:
            self.logger.error(f"Error getting current environment: {e}")

        return True

    async def _cmd_env_use(self, args: str) -> bool:
        """Set the current Hatch environment.

        Args:
            args (str): Environment name.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'name': {'positional': True}
        }

        parsed_args = self._parse_args(args, arg_defs)
        if 'name' not in parsed_args or not parsed_args['name']:
            self.logger.error(f"Environment name is required.")
            self._print_command_help('hatch:env:use')
            return True

        try:
            name = parsed_args['name']

            if mcp_manager.hatch_env_manager.set_current_environment(name):
                self.logger.info(f"Current environment set to: {name}")

                # When changing the current environment, we must handle
                # disconnecting from the previous environment's tools if any,
                # and connecting to the new environment's tools.
                if mcp_manager.is_connected:

                    # Disconnection
                    await mcp_manager.disconnect_all()
                    self.logger.info("Disconnected from previous environment's tools.")

                    # Get the new environment's entry points for the MCP servers
                    mcp_servers_url = mcp_manager.hatch_env_manager.get_servers_entry_points(name)

                    if mcp_servers_url:
                        # Reconnect to the new environment's tools
                        connected = await mcp_manager.connect_to_servers(mcp_servers_url)
                        if not connected:
                            self.logger.error("Failed to connect to new environment's MCP servers. Tools not enabled.")
                        else:
                            self.logger.info("Connected to new environment's MCP servers successfully!")

            else:
                self.logger.error(f"Failed to set environment: {name}")

        except Exception as e:
            self.logger.error(f"Error setting current environment: {e}")

        return True

    def _cmd_pkg_add(self, args: str) -> bool:
        """Add a package to an environment.

        Args:
            args (str): Package path or name and options.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'package_path_or_name': {'positional': True},
            'env': {'aliases': ['e'], 'default': None},
            'version': {'aliases': ['v'], 'default': None},
            'force-download': {'aliases': ['f'], 'default': False, 'action': 'store_true'},
            'refresh-registry': {'aliases': ['r'], 'default': False, 'action': 'store_true'},
            'auto-approve': {'aliases': ['y'], 'default': False, 'action': 'store_true'}
        }

        parsed_args = self._parse_args(args, arg_defs)
        if 'package_path_or_name' not in parsed_args or not parsed_args['package_path_or_name']:
            self.logger.error("Package path or name is required.")
            self._print_command_help('hatch:pkg:add')
            return True

        try:
            package = parsed_args['package_path_or_name']
            env = parsed_args.get('env')
            version = parsed_args.get('version')
            force_download = parsed_args.get('force-download', False)
            refresh_registry = parsed_args.get('refresh-registry', False)
            auto_approve = parsed_args.get('auto-approve', False)

            if mcp_manager.hatch_env_manager.add_package_to_environment(package, env, version, force_download, refresh_registry, auto_approve):
                self.logger.info(f"Successfully added package: {package}")
            else:
                self.logger.error(f"Failed to add package: {package}")

        except Exception as e:
            self.logger.error(f"Error adding package: {e}")

        return True

    def _cmd_pkg_remove(self, args: str) -> bool:
        """Remove a package from an environment.

        Args:
            args (str): Package name and options.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'package_name': {'positional': True},
            'env': {'aliases': ['e'], 'default': None}
        }

        parsed_args = self._parse_args(args, arg_defs)
        if 'package_name' not in parsed_args or not parsed_args['package_name']:
            self.logger.error("Package name is required.")
            self._print_command_help('hatch:pkg:remove')
            return True

        try:
            package_name = parsed_args['package_name']
            env = parsed_args.get('env')

            if mcp_manager.hatch_env_manager.remove_package(package_name, env):
                self.logger.info(f"Successfully removed package: {package_name}")
            else:
                self.logger.error(f"Failed to remove package: {package_name}")

        except Exception as e:
            self.logger.error(f"Error removing package: {e}")

        return True

    def _cmd_pkg_list(self, args: str) -> bool:
        """List packages in an environment.

        Args:
            args (str): Environment options.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'env': {'aliases': ['e'], 'default': None}
        }

        parsed_args = self._parse_args(args, arg_defs)
        env = parsed_args.get('env')

        try:
            packages = mcp_manager.hatch_env_manager.list_packages(env)
            if not packages:
                env_name = env if env else "current environment"
                self.logger.info(f"No packages found in {env_name}.")
                return True

            env_name = env if env else "current environment"
            self.logger.info(f"Listing {len(packages)} packages in {env_name}")
            print(f"Packages in {env_name}:")
            for pkg in packages:
                print(f"{pkg['name']} ({pkg['version']})  Hatch compliant: {pkg['hatch_compliant']} Source: {pkg['source']['uri']}  Location: {pkg['source']['path']}")

        except Exception as e:
            self.logger.error(f"Error listing packages: {e}")

        return True

    def _cmd_create_package(self, args: str) -> bool:
        """Create a new package template.

        Args:
            args (str): Package name and options.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'name': {'positional': True},
            'dir': {'aliases': ['d'], 'default': '.'},
            'description': {'aliases': ['D'], 'default': ''}
        }

        parsed_args = self._parse_args(args, arg_defs)
        if 'name' not in parsed_args or not parsed_args['name']:
            self.logger.error("Package name is required.")
            self._print_command_help('hatch:create')
            return True

        try:
            name = parsed_args['name']
            target_dir = Path(parsed_args.get('dir', '.')).resolve()
            description = parsed_args.get('description', '')

            package_dir = create_package_template(
                target_dir=target_dir,
                package_name=name,
                description=description
            )

            self.logger.info(f"Package template created at: {package_dir}")

        except Exception as e:
            self.logger.error(f"Error creating package template: {e}")

        return True

    def _cmd_validate_package(self, args: str) -> bool:
        """Validate a package.

        Args:
            args (str): Package directory.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'package_dir': {'positional': True}
        }

        parsed_args = self._parse_args(args, arg_defs)
        if 'package_dir' not in parsed_args or not parsed_args['package_dir']:
            self.logger.error("Package directory is required.")
            self._print_command_help('hatch:validate')
            return True

        try:
            package_path = Path(parsed_args['package_dir']).resolve()

            # Create validator with registry data from environment manager
            from hatch_validator import HatchPackageValidator
            validator = HatchPackageValidator(
                version="latest",
                allow_local_dependencies=True,
                registry_data=mcp_manager.hatch_env_manager.registry_data
            )

            # Validate the package
            is_valid, validation_results = validator.validate_package(package_path)

            if is_valid:
                self.logger.info(f"Package validation SUCCESSFUL: {package_path}")
            else:
                self.logger.warning(f"Package validation FAILED: {package_path}")

                # Print detailed validation results if available
                if validation_results and isinstance(validation_results, dict):
                    for category, result in validation_results.items():
                        if category != 'valid' and category != 'metadata' and isinstance(result, dict):
                            if not result.get('valid', True) and result.get('errors'):
                                self.logger.warning(f"\n{category.replace('_', ' ').title()} errors:")
                                for error in result['errors']:
                                    self.logger.warning(f"  - {error}")

        except Exception as e:
            self.logger.error(f"Error validating package: {e}")

        return True

    def _cmd_env_python_init(self, args: str) -> bool:
        """Initialize Python environment for a Hatch environment.

        Creates a Python environment using conda/mamba for the specified Hatch environment,
        with optional hatch_mcp_server wrapper installation. The MCP server installation
        can be controlled via command flags.

        Args:
            args (str): Environment options including:
                      - hatch_env: Hatch environment name (optional)
                      - python-version: Python version (optional)
                      - force: Force recreation if exists
                      - no-hatch-mcp-server: Skip MCP server installation
                      - hatch_mcp_server_tag: Git tag/branch for MCP server

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'hatch_env': {'positional': False, 'default': None},
            'python-version': {'default': None},
            'force': {'default': False, 'is_flag': True},
            'no-hatch-mcp-server': {'default': False, 'action': 'store_true'},
            'hatch_mcp_server_tag': {'default': None}
        }

        parsed_args = self._parse_args(args, arg_defs)
        hatch_env = parsed_args.get('hatch_env')
        python_version = parsed_args.get('python-version')
        force = parsed_args.get('force', False)
        no_hatch_mcp_server = parsed_args.get('no-hatch-mcp-server', False)
        hatch_mcp_server_tag = parsed_args.get('hatch_mcp_server_tag')

        try:
            if mcp_manager.hatch_env_manager.create_python_environment_only(
                hatch_env, 
                python_version, 
                force,
                no_hatch_mcp_server=no_hatch_mcp_server,
                hatch_mcp_server_tag=hatch_mcp_server_tag
            ):
                env_name = hatch_env if hatch_env else "current environment"
                self.logger.info(f"Python environment initialized for: {env_name}")

                if python_version:
                    self.logger.info(f"Python version: {python_version}")

                # Show Python environment info
                python_info = mcp_manager.hatch_env_manager.get_python_environment_info(hatch_env)
                if python_info:
                    self.logger.info(f"  Python executable: {python_info['python_executable']}")
                    self.logger.info(f"  Python version: {python_info.get('python_version', 'Unknown')}")
                    self.logger.info(f"  Conda environment: {python_info.get('conda_env_name', 'N/A')}")

                # Provide feedback about MCP server installation
                if not no_hatch_mcp_server:
                    if hatch_mcp_server_tag:
                        self.logger.info(f"hatch_mcp_server installed with tag/branch: {hatch_mcp_server_tag}")
                    else:
                        self.logger.info("hatch_mcp_server installed with default branch")
                else:
                    self.logger.info("hatch_mcp_server installation skipped (--no-hatch-mcp-server)")
            else:
                env_name = hatch_env if hatch_env else "current environment"
                self.logger.error(f"Failed to initialize Python environment for: {env_name}")

        except Exception as e:
            self.logger.error(f"Error initializing Python environment: {e}")

        return True

    def _cmd_env_python_info(self, args: str) -> bool:
        """Show Python environment information.

        Args:
            args (str): Environment options.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'hatch_env': {'positional': False, 'default': None},
            'detailed': {'default': False, 'is_flag': True}
        }

        parsed_args = self._parse_args(args, arg_defs)
        hatch_env = parsed_args.get('hatch_env')
        detailed = parsed_args.get('detailed', False)

        try:
            env_name = hatch_env if hatch_env else "current environment"

            if detailed:
                # Get detailed diagnostics
                python_info = mcp_manager.hatch_env_manager.get_python_environment_diagnostics(hatch_env)
                if python_info:
                    self.logger.info(f"Detailed Python environment diagnostics for {env_name}:")
                    for key, value in python_info.items():
                        if isinstance(value, dict):
                            self.logger.info(f"  {key}:")
                            for sub_key, sub_value in value.items():
                                self.logger.info(f"    {sub_key}: {sub_value}")
                        else:
                            self.logger.info(f"  {key}: {value}")
                else:
                    self.logger.info(f"No detailed Python environment diagnostics found for {env_name}.")
            else:
                # Get basic info
                python_info = mcp_manager.hatch_env_manager.get_python_environment_info(hatch_env)
                if python_info:
                    self.logger.info(f"Python environment information for {env_name}:")
                    for key, value in python_info.items():
                        if key == "packages":
                            continue
                        self.logger.info(f"  {key}: {value}")
                    # List packages separately
                    self.logger.info("  Packages:")
                    for _pkg in python_info.get("packages", []):
                        self.logger.info(f"    - {_pkg['name']} ({_pkg['version']})") 
                else:
                    self.logger.info(f"No Python environment information found for {env_name}.")

        except Exception as e:
            self.logger.error(f"Error getting Python environment information: {e}")

        return True

    def _cmd_env_python_remove(self, args: str) -> bool:
        """Remove Python environment.

        Args:
            args (str): Environment options.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'hatch_env': {'positional': False, 'default': None},
            'force': {'default': False, 'is_flag': True}
        }

        parsed_args = self._parse_args(args, arg_defs)
        hatch_env = parsed_args.get('hatch_env')
        force = parsed_args.get('force', False)

        try:
            env_name = hatch_env if hatch_env else "current environment"

            if not force:
                # Ask for confirmation if not forced
                self.logger.warning(f"This will remove the Python environment for {env_name}. Use --force to skip confirmation.")
                return True

            if mcp_manager.hatch_env_manager.remove_python_environment_only(hatch_env):
                self.logger.info(f"Python environment removed for Hatch environment: {env_name}")
            else:
                self.logger.error(f"Failed to remove Python environment for Hatch environment: {env_name}")

        except Exception as e:
            self.logger.error(f"Error removing Python environment: {e}")

        return True

    def _cmd_env_python_shell(self, args: str) -> bool:
        """Launch Python shell in environment.

        Args:
            args (str): Environment options.

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'hatch_env': {'positional': False, 'default': None},
            'cmd': {'default': None}
        }

        parsed_args = self._parse_args(args, arg_defs)
        hatch_env = parsed_args.get('hatch_env')
        cmd = parsed_args.get('cmd')

        try:
            env_name = hatch_env if hatch_env else "current environment"

            if mcp_manager.hatch_env_manager.launch_python_shell(hatch_env, cmd):
                if cmd:
                    self.logger.info(f"Executing command '{cmd}' in Python environment for {env_name}")
                else:
                    self.logger.info(f"Python shell launched in Hatch environment: {env_name}")
            else:
                self.logger.error(f"Failed to launch Python shell in Hatch environment: {env_name}")

        except Exception as e:
            self.logger.error(f"Error launching Python shell: {e}")

        return True

    def _cmd_env_python_add_hatch_mcp(self, args: str) -> bool:
        """Add hatch_mcp_server wrapper to the environment.

        Installs the hatch_mcp_server wrapper in an existing Python environment
        within a Hatch environment. The environment must have a valid Python
        environment already initialized.

        Args:
            args (str): Installation options including:
                      - hatch_env: Hatch environment name (optional)
                      - tag: Git tag/branch reference for installation

        Returns:
            bool: True to continue the chat session.
        """
        arg_defs = {
            'hatch_env': {'positional': False, 'default': None},
            'tag': {'default': None}
        }

        parsed_args = self._parse_args(args, arg_defs)
        hatch_env = parsed_args.get('hatch_env')
        tag = parsed_args.get('tag')

        try:
            env_name = hatch_env if hatch_env else mcp_manager.hatch_env_manager.get_current_environment()

            if mcp_manager.hatch_env_manager.install_mcp_server(hatch_env, tag):
                self.logger.info(f"hatch_mcp_server wrapper installed successfully in environment: {env_name}")
                if tag:
                    self.logger.info(f"Using tag/branch: {tag}")
                else:
                    self.logger.info("Using default branch")
            else:
                self.logger.error(f"Failed to install hatch_mcp_server wrapper in environment: {env_name}")

        except Exception as e:
            self.logger.error(f"Error installing hatch_mcp_server wrapper: {e}")

        return True
Functions
format_command(cmd_name, cmd_info, group='hatch')

Format Hatch commands with custom styling.

Source code in hatchling/ui/hatch_commands.py 
396
397
398
399
400
401
402
def format_command(self, cmd_name: str, cmd_info: Dict[str, Any], group: str = 'hatch') -> list:
    """Format Hatch commands with custom styling."""
    return [
        (f'class:command.name.{group}', f"{cmd_name}"),
        ('', ' - '),
        ('class:command.description', f"{cmd_info['description']}")
    ]
print_commands_help()

Print help for all available chat commands.

Source code in hatchling/ui/hatch_commands.py 
388
389
390
391
392
393
394
def print_commands_help(self) -> None:
    """Print help for all available chat commands."""
    print_formatted_text(FormattedText([
        ('class:header', "\n=== Hatch Chat Commands ===\n")
    ]), style=self.style)

    super().print_commands_help()

Functions

hatchling.ui.mcp_commands

MCP Server Management Commands.

This module provides CLI commands for managing MCP servers, tools, and debugging. Commands follow the format 'mcp:action:target' for clarity and consistency.

Classes

MCPCommands

Bases: AbstractCommands

CLI commands for MCP server and tool management.

Source code in hatchling/ui/mcp_commands.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
class MCPCommands(AbstractCommands):
    """CLI commands for MCP server and tool management."""

    def _register_commands(self) -> None:
        """Register all MCP-related commands."""

        self.commands = {
            'mcp:server:list': {
                'handler': self._cmd_server_list,
                'description': translate('commands.mcp.server_list_description'),
                'is_async': False,
                'args': {}
            },
            'mcp:server:status': {
                'handler': self._cmd_server_status,
                'description': translate('commands.mcp.server_status_description'),
                'is_async': False,
                'args': {
                    'server_path': {
                        'positional': True,
                        'completer_type': 'none',
                        'description': translate('commands.mcp.server_path_arg_description'),
                        'required': True
                    }
                }
            },
            'mcp:server:connect': {
                'handler': self._cmd_server_connect,
                'description': translate('commands.mcp.server_connect_description'),
                'is_async': True,
                'args': {
                    'server_paths': {
                        'positional': True,
                        'completer_type': 'none',
                        'description': translate('commands.mcp.server_paths_arg_description'),
                        'required': False
                    }
                }
            },
            'mcp:server:disconnect': {
                'handler': self._cmd_server_disconnect,
                'description': translate('commands.mcp.server_disconnect_description'),
                'is_async': True,
                'args': {}
            },
            'mcp:tool:list': {
                'handler': self._cmd_tool_list,
                'description': translate('commands.mcp.tool_list_description'),
                'is_async': False,
                'args': {
                    'server_path': {
                        'positional': False,
                        'completer_type': 'none',
                        'description': translate('commands.mcp.server_path_arg_description_optional'),
                        'required': False
                    }
                }
            },
            'mcp:tool:info': {
                'handler': self._cmd_tool_info,
                'description': translate('commands.mcp.tool_info_description'),
                'is_async': False,
                'args': {
                    'tool_name': {
                        'positional': True,
                        'completer_type': 'none',
                        'description': translate('commands.mcp.tool_name_arg_description'),
                        'required': True
                    }
                }
            },
            'mcp:tool:enable': {
                'handler': self._cmd_tool_enable,
                'description': translate('commands.mcp.tool_enable_description'),
                'is_async': False,
                'args': {
                    'tool_name': {
                        'positional': True,
                        'completer_type': 'none',
                        'description': translate('commands.mcp.tool_name_arg_description_enable'),
                        'required': True
                    }
                }
            },
            'mcp:tool:disable': {
                'handler': self._cmd_tool_disable,
                'description': translate('commands.mcp.tool_disable_description'),
                'is_async': False,
                'args': {
                    'tool_name': {
                        'positional': True,
                        'completer_type': 'none',
                        'description': translate('commands.mcp.tool_name_arg_description_disable'),
                        'required': True
                    }
                }
            },
            # 'mcp:tool:execute': {
            #     'handler': self._cmd_tool_execute,
            #     'description': translate('commands.mcp.tool_execute_description'),
            #     'is_async': True,
            #     'args': {
            #         'tool_name': {
            #             'positional': True,
            #             'completer_type': 'none',
            #             'description': translate('commands.mcp.tool_name_arg_description_execute'),
            #             'required': True
            #         }
            #     }
            # },
            # 'mcp:tool:schema': {
            #     'handler': self._cmd_tool_schema,
            #     'description': translate('commands.mcp.tool_schema_description'),
            #     'is_async': False,
            #     'args': {
            #         'tool_name': {
            #             'positional': True,
            #             'completer_type': 'none',
            #             'description': translate('commands.mcp.tool_name_arg_description'),
            #             'required': True
            #         }
            #     }
            # },
            'mcp:health': {
                'handler': self._cmd_health,
                'description': translate('commands.mcp.health_description'),
                'is_async': False,
                'args': {}
            },
            # 'mcp:citations': {
            #     'handler': self._cmd_citations,
            #     'description': translate('commands.mcp.citations_description'),
            #     'is_async': True,
            #     'args': {}
            # },
            # 'mcp:reset': {
            #     'handler': self._cmd_reset,
            #     'description': translate('commands.mcp.reset_description'),
            #     'is_async': False,
            #     'args': {}
            # }
        }
    def _cmd_server_list(self, args: str) -> bool:
        """List all configured MCP servers and their status.

        Args:
            args (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            servers = MCPServerAPI.get_server_list()

            if not servers:
                print("No MCP servers configured.")
                return True

            print("MCP Servers:")
            for server in servers:
                status_display = server.status.value.upper()
                tool_info = f"({server.enabled_tool_count}/{server.tool_count} tools enabled)"
                print(f"  {server.path} - {status_display} {tool_info}")

        except Exception as e:
            self.logger.error(f"Error in server list command: {e}")

        return True

    def _cmd_server_status(self, args: str) -> bool:
        """Show detailed status for a specific MCP server.

        Args:
            args (str): Server path argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            server_path = args.strip()
            if not server_path:
                print("Error: Server path is required")
                return True

            server_info = MCPServerAPI.get_server_status(server_path)

            print(f"Server: {server_info.path}")
            print(f"  Status: {server_info.status.value.upper()}")
            print(f"  Tools: {server_info.enabled_tool_count}/{server_info.tool_count} enabled")

            if server_info.error_message:
                print(f"  Error: {server_info.error_message}")

            if server_info.last_connected:
                import time
                connected_time = time.strftime('%Y-%m-%d %H:%M:%S', 
                                             time.localtime(server_info.last_connected))
                print(f"  Last Connected: {connected_time}")

        except Exception as e:
            self.logger.error(f"Error in server status command: {e}")

        return True

    async def _cmd_server_connect(self, args: str) -> bool:
        """Connect to MCP servers.

        Args:
            args (str): Comma-separated server paths. If

        Returns:
            bool: True to continue the chat session.
        """
        try:
            server_paths_str = args.strip()
            if not server_paths_str:
                server_paths = None
            else:
                server_paths = [path.strip() for path in server_paths_str.split(',')]

            success = await MCPServerAPI.connect_servers(server_paths)

            if success:
                print("Successfully connected to MCP servers")
            else:
                print("Failed to connect to MCP servers")

        except Exception as e:
            self.logger.error(f"Error connecting to servers: {e}")

        return True

    async def _cmd_server_disconnect(self, args: str) -> bool:
        """Disconnect from all MCP servers.

        Args:
            args (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            print("Disconnecting from all MCP servers...")
            await MCPServerAPI.disconnect_all_servers()
            print("Disconnected from all MCP servers")

        except Exception as e:
            self.logger.error(f"Error disconnecting from servers: {e}")

        return True

    # =============================================================================
    # Tool Management Commands
    # =============================================================================

    def _cmd_tool_list(self, args: str) -> bool:
        """List all available MCP tools.

        Args:
            args (str): Optional server path filter.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            server_path_filter = args.strip() if args.strip() else None

            if server_path_filter:
                tools = MCPServerAPI.get_tools_by_server(server_path_filter)
                print(f"MCP Tools from {server_path_filter}:")
            else:
                tools = MCPServerAPI.get_all_tools()
                print("All MCP Tools:")

            if not tools:
                print("  No tools found")
                return True

            # Group by server for better organization
            by_server = {}
            for tool in tools:
                if tool.server_path not in by_server:
                    by_server[tool.server_path] = []
                by_server[tool.server_path].append(tool)

            for server_path, server_tools in by_server.items():
                if not server_path_filter:  # Only show server headers when not filtering
                    print(f"  Server: {server_path}")

                for tool in server_tools:
                    status = "ENABLED" if tool.status == MCPToolStatus.ENABLED else "DISABLED"
                    description = f" - {tool.description}" if tool.description else ""
                    indent = "    " if not server_path_filter else "  "
                    print(f"{indent}{tool.name} ({status}){description}")

        except Exception as e:
            print(f"Error listing tools: {e}")
            self.logger.error(f"Error in tool list command: {e}")

        return True

    def _cmd_tool_info(self, args: str) -> bool:
        """Show detailed information about a specific tool.

        Args:
            args (str): Tool name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            tool_name = args.strip()
            if not tool_name:
                print("Error: Tool name is required")
                return True

            tool_info = MCPServerAPI.get_tool_info(tool_name)

            if not tool_info:
                print(f"Tool '{tool_name}' not found")
                return True

            print(f"Tool: {tool_info.name}")
            print(f"  Server: {tool_info.server_path}")
            print(f"  Status: {tool_info.status.value.upper()}")

            if tool_info.description:
                print(f"  Description: {tool_info.description}")

            if tool_info.last_updated:
                import time
                updated_time = time.strftime('%Y-%m-%d %H:%M:%S', 
                                           time.localtime(tool_info.last_updated))
                print(f"  Last Updated: {updated_time}")

        except Exception as e:
            self.logger.error(f"Error getting tool info: {e}")

        return True

    def _cmd_tool_enable(self, args: str) -> bool:
        """Enable a specific MCP tool.

        Args:
            args (str): Tool name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            tool_name = args.strip()
            if not tool_name:
                print("Error: Tool name is required")
                return True

            MCPServerAPI.enable_tool(tool_name)

        except Exception as e:
            print(f"Error enabling tool: {e}")
            self.logger.error(f"Error in tool enable command: {e}")

        return True

    def _cmd_tool_disable(self, args: str) -> bool:
        """Disable a specific MCP tool.

        Args:
            args (str): Tool name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            tool_name = args.strip()
            if not tool_name:
                print("Error: Tool name is required")
                return True

            success = MCPServerAPI.disable_tool(tool_name)

            if success:
                print(f"Tool '{tool_name}' disabled successfully")
            else:
                print(f"Failed to disable tool '{tool_name}' (already disabled or not found)")

        except Exception as e:
            self.logger.error(f"Error disabling tool: {e}")

        return True

    async def _cmd_tool_execute(self, args: str) -> bool:
        """Execute an MCP tool manually for debugging.

        Args:
            args (str): Tool name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            tool_name = args.strip()
            if not tool_name:
                print("Error: Tool name is required")
                return True

            # Get tool schema to guide argument input
            schema = MCPServerAPI.get_tool_schema(tool_name)
            if not schema:
                print(f"Tool '{tool_name}' not found or schema unavailable")
                return True

            print(f"Executing tool: {tool_name}")
            print("Tool arguments schema:")
            print(json.dumps(schema, indent=2))

            # Parse arguments dynamically based on schema
            arguments = await self._parse_tool_arguments(schema)
            if arguments is None:
                print("Execution cancelled")
                return True

            print(f"Executing with arguments: {json.dumps(arguments, indent=2)}")

            # Execute the tool
            success, result, error = await MCPServerAPI.execute_tool_manually(tool_name, arguments)

            if success:
                print("Execution successful:")
                if isinstance(result, (dict, list)):
                    print(json.dumps(result, indent=2))
                else:
                    print(str(result))
            else:
                print(f"Execution failed: {error}")

        except Exception as e:
            self.logger.error(f"Error executing tool: {e}")

        return True

    def _cmd_tool_schema(self, args: str) -> bool:
        """Show the JSON schema for a tool's arguments.

        Args:
            args (str): Tool name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            tool_name = args.strip()
            if not tool_name:
                print("Error: Tool name is required")
                return True

            schema = MCPServerAPI.get_tool_schema(tool_name)

            if not schema:
                print(f"Schema not available for tool '{tool_name}'")
                return True

            print(f"Schema for tool '{tool_name}':")
            print(json.dumps(schema, indent=2))

        except Exception as e:
            self.logger.error(f"Error getting tool schema: {e}")

        return True

    # =============================================================================
    # System Commands
    # =============================================================================

    def _cmd_health(self, args: str) -> bool:
        """Show overall MCP system health summary.

        Args:
            args (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            health = MCPServerAPI.get_health_summary()

            print("MCP System Health:")
            print(f"  Connected Servers: {health['connected_servers']}")
            print(f"  Total Tools: {health['total_tools']}")
            print(f"  Enabled Tools: {health['enabled_tools']}")
            print(f"  Disabled Tools: {health['disabled_tools']}")

            if health['server_details']:
                print("\nServer Details:")
                for server in health['server_details']:
                    print(f"  {server['path']} ({server['status']}) - {server['enabled_tools']}/{server['tools']} tools")

        except Exception as e:
            self.logger.error(f"Error getting health summary: {e}")

        return True

    async def _cmd_citations(self, args: str) -> bool:
        """Show MCP server citations for current session.

        Args:
            args (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            citations = await MCPServerAPI.get_session_citations()

            if not citations:
                print("No MCP servers used in current session")
                return True

            print("MCP Server Citations:")
            for server_path, citation_info in citations.items():
                print(f"  {server_path}:")
                for key, value in citation_info.items():
                    print(f"    {key}: {value}")

        except Exception as e:
            self.logger.error(f"Error getting citations: {e}")

        return True

    def _cmd_reset(self, args: str) -> bool:
        """Reset MCP session tracking.

        Args:
            args (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            MCPServerAPI.reset_session_tracking()
            print("MCP session tracking reset")

        except Exception as e:
            self.logger.error(f"Error resetting session tracking: {e}")

        return True

    # =============================================================================
    # Helper Methods
    # =============================================================================

    async def _parse_tool_arguments(self, schema: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """Parse tool arguments interactively based on JSON schema.

        Args:
            schema (Dict[str, Any]): The JSON schema for the tool arguments.

        Returns:
            Optional[Dict[str, Any]]: Parsed arguments or None if cancelled.
        """
        try:
            arguments = {}
            properties = schema.get('properties', {})
            required = schema.get('required', [])

            print("\nEnter arguments (press Enter with empty value to skip optional args):")

            for prop_name, prop_schema in properties.items():
                is_required = prop_name in required
                prop_type = prop_schema.get('type', 'string')
                description = prop_schema.get('description', '')

                # Show prompt
                required_marker = "*" if is_required else ""
                type_hint = f" ({prop_type})" if prop_type != 'string' else ""
                desc_hint = f" - {description}" if description else ""

                while True:
                    user_input = input(f"  {prop_name}{required_marker}{type_hint}{desc_hint}: ").strip()

                    # Handle empty input
                    if not user_input:
                        if is_required:
                            print(f"    Error: {prop_name} is required")
                            continue
                        else:
                            break  # Skip optional parameter

                    # Handle cancellation
                    if user_input.lower() in ['cancel', 'quit', 'exit']:
                        return None

                    # Convert value based on type
                    try:
                        converted_value = self._convert_argument_value(user_input, prop_type, prop_schema)
                        arguments[prop_name] = converted_value
                        break

                    except ValueError as e:
                        self.logger.error(f"Invalid value for {prop_name}: {e}")
                        continue

            return arguments

        except KeyboardInterrupt:
            self.logger.info("Tool argument input cancelled by user")
            return None

    def _convert_argument_value(self, value: str, prop_type: str, prop_schema: Dict[str, Any]) -> Any:
        """Convert string input to the appropriate type based on schema.

        Args:
            value (str): User input value.
            prop_type (str): Expected property type.
            prop_schema (Dict[str, Any]): Full property schema.

        Returns:
            Any: Converted value.

        Raises:
            ValueError: If conversion fails.
        """
        if prop_type == 'string':
            return value
        elif prop_type == 'number':
            try:
                return float(value)
            except ValueError:
                self.logger.error(f"Error parsing {value} as number: must be a valid float")
        elif prop_type == 'integer':
            try:
                return int(value)
            except ValueError:
                self.logger.error(f"Error parsing {value} as integer: must be a valid integer")
        elif prop_type == 'boolean':
            if value.lower() in ['true', '1', 'yes', 'y']:
                return True
            elif value.lower() in ['false', '0', 'no', 'n']:
                return False
            else:
                self.logger.error(f"Error parsing {value} as boolean: must be true/false or 1/0")
        elif prop_type == 'array':
            # Simple array parsing - split by comma
            try:
                if value.startswith('[') and value.endswith(']'):
                    # JSON array format
                    return json.loads(value)
                else:
                    # Comma-separated values
                    return [item.strip() for item in value.split(',') if item.strip()]
            except json.JSONDecodeError as e:
                self.logger.error(f"Error parsing {value} as array: {e}")
        elif prop_type == 'object':
            # Parse as JSON
            try:
                return json.loads(value)
            except json.JSONDecodeError as e:
                self.logger.error(f"Error parsing {value} as object: {e}")
        else:
            # Default to string for unknown types
            return value

Functions

hatchling.ui.model_commands

LLM Model Management Commands.

This module provides CLI commands for managing LLM models and providers. Commands follow the format 'llm:target:action' for clarity and consistency.

Classes

ModelCommands

Bases: AbstractCommands

CLI commands for LLM model and provider management.

Source code in hatchling/ui/model_commands.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
class ModelCommands(AbstractCommands):
    """CLI commands for LLM model and provider management."""

    def _register_commands(self) -> None:
        """Register all model-related commands."""

        from hatchling.config.i18n import translate
        self.commands = {
            # Provider management commands
            'llm:provider:supported': {
                'handler': self._cmd_provider_supported,
                'description': translate('commands.llm.provider_supported_description'),
                'is_async': False,
                'args': {}
            },
            'llm:provider:status': {
                'handler': self._cmd_provider_status,
                'description': translate('commands.llm.provider_status_description'),
                'is_async': True,
                'args': {
                    'provider-name': {
                        'positional': False,
                        'completer_type': 'suggestions',
                        'values': self.settings.llm.provider_names,
                        'description': translate('commands.llm.provider_name_arg_description'),
                        'required': False
                    }
                }
            },
            # Model management commands
            'llm:model:list': {
                'handler': self._cmd_model_list,
                'description': translate('commands.llm.model_list_description'),
                'is_async': True,
                'args': {}
            },
            'llm:model:add': {
                'handler': self._cmd_model_add,
                'description': translate('commands.llm.model_add_description'),
                'is_async': True,
                'args': {
                    'provider-name': {
                        'positional': False,
                        'completer_type': 'suggestions',
                        'values': self.settings.llm.provider_names,
                        'description': translate('commands.llm.provider_name_arg_description'),
                        'required': False
                    },
                    'model-name': {
                        'positional': True,
                        'description': translate('commands.llm.model_name_arg_description'),
                        'required': True
                    }
                }
            },
            'llm:model:use': {
                'handler': self._cmd_model_use,
                'description': translate('commands.llm.model_use_description'),
                'is_async': False,
                'args': {
                    'model-name': {
                        'positional': True,
                        'completer_type': 'suggestions',
                        'values': [model.name for model in self.settings.llm.models],
                        'description': translate('commands.llm.model_name_arg_description'),
                        'required': True
                    },
                    'force-confirmed':{
                        'positional': False,
                        'completer_type': 'boolean',
                        'description': translate('commands.llm.force_confirmed_arg_description'),
                        'required': False,
                        'default': False,
                        'is_flag': True
                    }
                }
            },
            'llm:model:remove': {
                'handler': self._cmd_model_remove,
                'description': translate('commands.llm.model_remove_description'),
                'is_async': False,
                'args': {
                    'model-name': {
                        'positional': True,
                        'completer_type': 'suggestions',
                        'values': [model.name for model in self.settings.llm.models],
                        'description': translate('commands.llm.model_name_arg_description'),
                        'required': True
                    }
                }
            }
        }

    def print_commands_help(self) -> None:
        """Print help for all available chat commands."""
        print_formatted_text(FormattedText([
            ('class:header', "\n=== Model Commands ===\n")
        ]), style=self.style)

        # Call parent class method to print formatted commands
        super().print_commands_help()

    # =============================================================================
    # Provider Management Commands
    # =============================================================================

    def _cmd_provider_supported(self, args: str) -> bool:
        """List all supported LLM providers.
        This command retrieves and displays all LLM providers supported by the system.

        Args:
            args (str): Unused arguments.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            providers = ModelManagerAPI.list_providers()

            if not providers:
                print("No LLM providers found")
                return True

            print("Available LLM Providers:")
            for provider in providers:
                print(f"  {provider}")

        except Exception as e:
            print(f"Error listing providers: {e}")
            self.logger.error(f"Error in provider list command: {e}")

        return True

    async def _cmd_provider_status(self, args: str) -> bool:
        """Check status of a specific provider.

        Args:
            args (str): Provider name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            args_def = self.commands['llm:provider:status']['args']
            parsed_args = self._parse_args(args, args_def)
            provider_name = parsed_args.get('provider-name', '')
            providers = self.settings.llm.provider_enums

            if provider_name:
                # Get provider health
                providers = [self.settings.llm.to_provider_enum(provider_name)] 

            for provider in providers:
                is_healthy = await ModelManagerAPI.check_provider_health(provider)
                if is_healthy:
                    print(f"Provider: {provider} - Status: AVAILABLE")
                    models = await ModelManagerAPI.list_available_models(provider)
                    print(f"  - Models: {[model.name for model in models]}")
                else:
                    print(f"Provider: {provider.value} - Status: UNAVAILABLE")

        except Exception as e:
            self.logger.error(f"Error in provider status command: {e}")

        finally:
            return True

    # =============================================================================
    # Model Management Commands
    # =============================================================================

    async def _cmd_model_list(self, args: str) -> bool:
        """List all available models, optionally filtered by provider or search query.

        Args:
            args (str): Optional provider name or search query to filter models.

        Returns:
            bool: True to continue the chat session.
        """

        #TODO: Implement filtering by provider name or search query

        print("Available LLM Models:")
        for model_info in self.settings.llm.models:
            print(f"  - {model_info.provider.value} {model_info.name}")

        return True

    async def _cmd_model_add(self, args: str) -> bool:
        """Pull/download a model (Ollama only).

        Args:
            args (str): Model name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            args_def = self.commands['llm:model:add']['args']
            parsed_args = self._parse_args(args, args_def)

            model_name = parsed_args.get('model-name', '')
            provider_name = parsed_args.get('provider-name', self.settings.llm.provider_enum.value)

            if not model_name:
                self.logger.error("Positional argument 'model-name' is required for pulling a model.")
                return True

            success = await ModelManagerAPI.pull_model(model_name, LLMSettings.to_provider_enum(provider_name))

            if success:
                # We update the commands args value suggestion for the autocompletion
                self.commands['llm:model:use']['args']['model-name']['values'] = [model.name for model in self.settings.llm.models]
                self.commands['llm:model:remove']['args']['model-name']['values'] = [model.name for model in self.settings.llm.models]

        except Exception as e:
            self.logger.error(f"Error in model pull command: {e}")

        return True

    def _cmd_model_use(self, args: str) -> bool:
        """Set the default model to use for the current session.

        Args:
            args (str): Model name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            args_def = self.commands['llm:model:use']['args']
            parsed_args = self._parse_args(args, args_def)
            model_name = parsed_args.get('model-name', '')
            force_confirmed = parsed_args.get('force-confirmed', False)

            if not model_name:
                self.logger.error("Positional argument 'model-name' is required to set the default model.")
                return True

            # Check if the model exists in the available models
            model_info = None
            for model in self.settings.llm.models:
                if model.name == model_name:
                    model_info = model
                    break

            if not model_info:
                self.logger.warning(f"Model '{model_name}' not found in available models. No action taken.")
                return True

            # Set the default model in the settings
            self.settings_registry.set_setting(
                "llm", "model", model_info.name, force=force_confirmed
            )
            self.settings_registry.set_setting(
                "llm", "provider_enum", model_info.provider, force=force_confirmed
            )

        except Exception as e:
            self.logger.error(f"Error in model use command: {e}")

        return True

    def _cmd_model_remove(self, args: str) -> bool:
        """Remove a model from the list of available models.

        Args:
            args (str): Model name argument.

        Returns:
            bool: True to continue the chat session.
        """
        try:
            args_def = self.commands['llm:model:remove']['args']
            parsed_args = self._parse_args(args, args_def)

            model_name = parsed_args.get('model-name', '')
            if not model_name:
                self.logger.error("Positional argument 'model-name' is required to remove a model.")
                return True

            # Find and remove the model
            for model_info in self.settings.llm.models:
                if model_info.name == model_name:
                    self.settings.llm.models.remove(model_info)
                    self.logger.info(f"Model '{model_name}' removed successfully.")

                    # Update the command args values for autocompletion
                    self.commands['llm:model:use']['args']['model-name']['values'] = [model.name for model in self.settings.llm.models]
                    self.commands['llm:model:remove']['args']['model-name']['values'] = [model.name for model in self.settings.llm.models]
                    return True

            self.logger.warning(f"Model '{model_name}' not found in available models. No action taken.")
            return True

        except Exception as e:
            self.logger.error(f"Error in model remove command: {e}")

        return True
Functions
print_commands_help()

Print help for all available chat commands.

Source code in hatchling/ui/model_commands.py 
107
108
109
110
111
112
113
114
def print_commands_help(self) -> None:
    """Print help for all available chat commands."""
    print_formatted_text(FormattedText([
        ('class:header', "\n=== Model Commands ===\n")
    ]), style=self.style)

    # Call parent class method to print formatted commands
    super().print_commands_help()

hatchling.ui.settings_commands

Settings commands module for handling settings-related CLI operations.

This module provides SettingsCommands class which handles all settings-related command operations including list, get, set, reset, import, export, and language management through the chat interface.

Classes

SettingsCommands

Bases: AbstractCommands

Handles settings-related command operations in the chat interface.

Source code in hatchling/ui/settings_commands.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
class SettingsCommands(AbstractCommands):
    """Handles settings-related command operations in the chat interface."""

    def _register_commands(self) -> None:
        """Register all settings-related commands."""
        self.commands = {
            'settings:list': {
                'handler': self._cmd_settings_list,
                'description': translate("commands.settings.list_description"),
                'is_async': False,
                'args': {
                    'filter': {
                        'positional': True,
                        'description': translate("commands.args.filter_description"),
                        'default': None,
                        'required': False
                    },
                    'format': {
                        'positional': False,
                        'description': translate("commands.args.format_description"),
                        'default': "table",
                        'completer_type': 'suggestions',
                        'values': ["table", "json", "yaml"],
                        'required': False
                    }
                }
            },
            'settings:get': {
                'handler': self._cmd_settings_get,
                'description': translate("commands.settings.get_description"),
                'is_async': False,
                'args': {
                    'setting': {
                        'positional': True,
                        'completer_type': 'suggestions',
                        'values': self._get_available_settings(),
                        'description': translate("commands.args.setting_description"),
                        'required': True
                    }
                }
            },
            'settings:set': {
                'handler': self._cmd_settings_set,
                'description': translate("commands.settings.set_description"),
                'is_async': True,
                'args': {
                    'setting': {
                        'positional': True,
                        'completer_type': 'suggestions',
                        'values': self._get_available_settings(),
                        'description': translate("commands.args.setting_description"),
                        'required': True
                    },
                    'value': {
                        'positional': True,
                        'description': translate("commands.args.value_description"),
                        'required': True
                    },
                    'force-confirm': {
                        'positional': False,
                        'description': translate("commands.args.force_description"),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'force-protected': {
                        'positional': False,
                        'description': translate("commands.args.force_protected_description"),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    }
                }
            },
            'settings:reset': {
                'handler': self._cmd_settings_reset,
                'description': translate("commands.settings.reset_description"),
                'is_async': True,
                'args': {
                    'setting': {
                        'positional': True,
                        'completer_type': 'suggestions',
                        'values': self._get_available_settings(),
                        'description': translate("commands.args.setting_description"),
                        'required': True
                    },
                    'force-confirmed': {
                        'positional': False,
                        'description': translate("commands.args.force_description"),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'force-protected': {
                        'positional': False,
                        'description': translate("commands.args.force_protected_description"),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    }
                }
            },
            'settings:export': {
                'handler': self._cmd_settings_export,
                'description': translate("commands.settings.export_description"),
                'is_async': False,
                'args': {
                    'file': {
                        'positional': True,
                        'completer_type': 'path',
                        'description': translate("commands.args.file_description"),
                        'required': True
                    },
                    'format': {
                        'positional': False,
                        'completer_type': 'suggestions',
                        'values': ["json", "yaml", "toml"],
                        'description': translate("commands.args.format_description"),
                        'default': "toml",
                        'required': False
                    },
                    'all': {
                        'positional': False,
                        'description': translate("commands.args.all_settings_description"),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    }
                }
            },
            'settings:save': {
                'handler': self._cmd_settings_save,
                'description': translate("commands.settings.save_description"),
                'is_async': False,
                'args': {
                    'format': {
                            'positional': False,
                            'completer_type': 'suggestions',
                            'values': ["json", "yaml", "toml"],
                            'description': translate("commands.args.format_description"),
                            'default': "toml",
                            'required': False
                        }
                }
            },
            'settings:import': {
                'handler': self._cmd_settings_import,
                'description': translate("commands.settings.import_description"),
                'is_async': True,
                'args': {
                    'file': {
                        'positional': True,
                        'completer_type': 'path',
                        'description': translate("commands.args.file_description"),
                        'required': True
                    },
                    'force-confirm': {
                        'positional': False,
                        'description': translate("commands.args.force_description"),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    },
                    'force-protected': {
                        'positional': False,
                        'description': translate("commands.args.force_protected_description"),
                        'default': False,
                        'is_flag': True,
                        'required': False
                    }
                }
            },
            'settings:language:list': {
                'handler': self._cmd_language_list,
                'description': translate("commands.settings.language_list_description"),
                'is_async': False,
                'args': {}
            },
            'settings:language:set': {
                'handler': self._cmd_language_set,
                'description': translate("commands.settings.language_set_description"),
                'is_async': False,
                'args': {
                    'language': {
                        'positional': True,
                        'completer_type': 'languages',
                        'description': translate("commands.args.language_description"),
                        'required': True
                    }
                }
            }
        }

    def print_commands_help(self) -> None:
        """Print help for all available chat commands."""
        print_formatted_text(FormattedText([
            ('class:header', "\n=== Settings Commands ===\n")
        ]), style=self.style)

        # Call parent class method to print formatted commands
        super().print_commands_help()

    def _cmd_settings_list(self, args: str) -> bool:
        """List all available settings with optional filtering.

        Args:
            args (str): Command arguments as a string.

        Returns:
            bool: True to continue the chat session, False to exit.
        """
        arg_defs = self.commands['settings:list']['args']

        parsed_args = self._parse_args(args, arg_defs)
        filter_pattern = parsed_args.get('filter')
        output_format = parsed_args.get('format', "table")

        if not self.settings_registry:
            self._print_error(translate("errors.settings_registry_not_available"))
            return True

        try:
            settings = self.settings_registry.list_settings(filter_pattern)
            self._output_settings_list(settings, output_format)
        except Exception as e:
            self._print_error(translate("errors.list_settings_failed", error=str(e)))
        return True

    def _cmd_settings_get(self, args: str) -> bool:
        """Get the value and metadata for a specific setting.

        Args:
            args (str): Command arguments as a string.

        Returns:
            bool: True to continue the chat session, False to exit.
        """
        arg_defs = self.commands['settings:get']['args']

        parsed_args = self._parse_args(args, arg_defs)
        setting_path = parsed_args.get('setting')

        if not setting_path:
            self._print_error(translate("errors.setting_name_required"))
            return True

        if not self.settings_registry:
            self._print_error(translate("errors.settings_registry_not_available"))
            return True

        try:
            category, name = self._parse_setting_path(setting_path)
            setting_info = self.settings_registry.get_setting(category, name)
            self._output_setting_info(setting_info)
        except ValueError as e:
            self._print_error(str(e))
        except Exception as e:
            self._print_error(translate("errors.get_setting_failed", error=str(e)))
        return True

    async def _cmd_settings_set(self, args: str) -> bool:
        """Set the value of a specific setting.

        Args:
            args (str): Command arguments as a string.

        Returns:
            bool: True to continue the chat session, False to exit.
        """
        arg_defs = self.commands['settings:set']['args']

        parsed_args = self._parse_args(args, arg_defs)
        setting_path = parsed_args.get('setting')
        value = parsed_args.get('value')
        force_confirm = parsed_args.get('force-confirm', False)
        force_protected = parsed_args.get('force-protected', False)

        if not setting_path or value is None:
            self._print_error(translate("errors.setting_and_value_required"))
            return True

        if not self.settings_registry:
            self._print_error(translate("errors.settings_registry_not_available"))
            return True

        try:
            category, name = self._parse_setting_path(setting_path)

            if not force_confirm:
                if not await self._request_user_consent(translate("prompts.confirm_set", setting=f"{category}:{name}", value=value)):
                    self._print_info(translate("info.operation_cancelled"))
                    return True

            current_setting = self.settings_registry.get_setting(category, name)
            typed_value = self._convert_value(value, current_setting["current_value"])
            success = self.settings_registry.set_setting(category, name, typed_value, force=force_protected)
            if success:
                self._print_success(translate("info.setting_updated",
                                              setting=f"{category}:{name}",
                                              value=str(typed_value)))
            else:
                self._print_error(translate("errors.set_setting_failed", setting=f"{category}:{name}"))
        except ValueError as e:
            self._print_error(str(e))
        except Exception as e:
            self._print_error(translate("errors.set_setting_failed", error=str(e)))
        return True

    async def _cmd_settings_reset(self, args: str) -> bool:
        """Reset a setting to its default value.

        Args:
            args (str): Command arguments as a string.

        Returns:
            bool: True to continue the chat session, False to exit.
        """
        arg_defs = self.commands['settings:reset']['args']

        parsed_args = self._parse_args(args, arg_defs)
        setting_path = parsed_args.get('setting')
        force_confirm = parsed_args.get('force-confirm', False)
        force_protected = parsed_args.get('force-protected', False)

        if not setting_path:
            self._print_error(translate("errors.setting_name_required"))
            return True

        if not self.settings_registry:
            self._print_error(translate("errors.settings_registry_not_available"))
            return True

        try:
            category, name = self._parse_setting_path(setting_path)
            if not force_confirm:
                if not await self._request_user_consent(translate("prompts.confirm_reset", setting=f"{category}:{name}")):
                    self._print_info(translate("info.operation_cancelled"))
                    return True

            success = self.settings_registry.reset_setting(category, name, force=force_protected)
            if success:
                self._print_success(translate("info.setting_reset", setting=f"{category}:{name}"))
            else:
                self._print_error(translate("errors.reset_setting_failed", setting=f"{category}:{name}"))
        except ValueError as e:
            self._print_error(str(e))
        except Exception as e:
            self._print_error(translate("errors.reset_setting_failed", error=str(e)))
        return True

    def _cmd_settings_export(self, args: str) -> bool:
        """Export settings to a file.

        Args:
            args (str): Command arguments as a string.

        Returns:
            bool: True to continue the chat session, False to exit.
        """
        arg_defs = self.commands['settings:export']['args']

        parsed_args = self._parse_args(args, arg_defs)
        file_path = parsed_args.get('file')
        file_format = parsed_args.get('format')
        all_settings = parsed_args.get('all', False)

        if not file_path:
            self._print_error(translate("errors.file_path_required"))
            return True

        if not self.settings_registry:
            self._print_error(translate("errors.settings_registry_not_available"))
            return True

        file_path_obj = Path(file_path)
        if not file_format:
            file_format = self._detect_format(file_path_obj)

        try:
            success = self.settings_registry.export_settings_to_file(str(file_path_obj), file_format, include_read_only=all_settings)
            if success:
                self._print_success(translate("info.settings_exported", file=str(file_path_obj)))
            else:
                self._print_error(translate("errors.export_settings_failed", file=str(file_path_obj)))
        except Exception as e:
            self._print_error(translate("errors.export_settings_failed", error=str(e)))
        return True

    def _cmd_settings_save(self, args: str) -> bool:
        """Save current settings to the configured file.

        Args:
            args (str): Command arguments as a string.

        Returns:
            bool: True to continue the chat session, False to exit.
        """
        arg_defs = self.commands['settings:save']['args']

        parsed_args = self._parse_args(args, arg_defs)
        file_format = parsed_args.get('format', "toml")

        if not self.settings_registry:
            self._print_error(translate("errors.settings_registry_not_available"))
            return True

        try:
            success = self.settings_registry.save_persistent_settings(file_format)
            if success:
                self._print_success(translate("info.settings_saved", file=self.settings_registry.get_persistent_settings_file_path()))
        except Exception as e:
            self._print_error(translate("errors.save_settings_failed", error=str(e)))
        return True

    async def _cmd_settings_import(self, args: str) -> bool:
        """Import settings from a file.

        Args:
            args (str): Command arguments as a string.

        Returns:
            bool: True to continue the chat session, False to exit.
        """
        arg_defs = self.commands['settings:import']['args']

        parsed_args = self._parse_args(args, arg_defs)
        file_path = parsed_args.get('file')
        force_confirm = parsed_args.get('force-confirm', False)
        force_protected = parsed_args.get('force-protected', False)

        if not file_path:
            self._print_error(translate("errors.file_path_required"))
            return True

        if not self.settings_registry:
            self._print_error(translate("errors.settings_registry_not_available"))
            return True

        file_path_obj = Path(file_path)
        if not file_path_obj.exists():
            self._print_error(translate("errors.file_not_found", file=str(file_path_obj)))
            return True

        try:
            if not force_confirm:
                if not await self._request_user_consent(translate("prompts.confirm_import", file=str(file_path_obj))):
                    self._print_info(translate("info.operation_cancelled"))
                    return True

            success = self.settings_registry.import_settings_from_file(str(file_path_obj), force=force_protected)
            if success:
                self._print_success(translate("info.settings_imported", file=str(file_path_obj)))
            else:
                self._print_error(translate("errors.import_settings_failed", file=str(file_path_obj)))
        except Exception as e:
            self._print_error(translate("errors.import_settings_failed", error=str(e)))
        return True

    def _cmd_language_list(self, args: str) -> bool:
        """List available languages.

        Args:
            args (str): Command arguments as a string.

        Returns:
            bool: True to continue the chat session, False to exit.
        """
        if not self.settings_registry:
            self._print_error(translate("errors.settings_registry_not_available"))
            return True

        try:
            languages = self.settings_registry.get_available_languages()
            current_language = self.settings_registry.get_current_language()

            self._print_header(translate("headers.available_languages"))
            for lang in languages:
                marker = "* " if lang["code"] == current_language else "  "
                self._print_info(f"{marker}{lang['code']}: {lang['name']}")
        except Exception as e:
            self._print_error(translate("errors.list_languages_failed", error=str(e)))
        return True

    def _cmd_language_set(self, args: str) -> bool:
        """
        List all available languages for settings commands.
        This command is picked up by the ChatCommandHandler and not here.
        That's because it concerns all commands, not just settings commands.
        """
        pass

    # Helper methods

    async def _request_user_consent(self, message: str) -> bool:
        """Request user consent for the installation plan.

        Args:
            message (str): Message to display for confirmation.

        Returns:
            bool: True if user approves, False otherwise.
        """        
        # Request confirmation
        session = PromptSession()
        while True:
            response = (await session.prompt_async(f"\n{message} [y/N]: ")).strip().lower()
            if response in ['y', 'yes']:
                return True
            elif response in ['n', 'no', '']:
                return False
            else:
                print("Please enter 'y' for yes or 'n' for no.")

    def _parse_setting_path(self, setting_path: str) -> tuple[str, str]:
        """Parse a setting path in format 'category:name'.

        Args:
            setting_path (str): Setting path to parse.

        Returns:
            tuple[str, str]: Tuple of (category, name).

        Raises:
            ValueError: If path format is invalid.
        """
        if ":" not in setting_path:
            raise ValueError(translate("errors.invalid_setting_path", path=setting_path))

        parts = setting_path.split(":", 1)
        if len(parts) != 2 or not parts[0] or not parts[1]:
            raise ValueError(translate("errors.invalid_setting_path", path=setting_path))

        return parts[0], parts[1]

    def _convert_value(self, value: str, current_value: Any) -> Any:
        """Convert string value to appropriate type based on current value.
        However, if the value is "None" (as a string), it will return None
        (not a string).

        Args:
            value (str): String value to convert.
            current_value (Any): Current value to infer type from.

        Returns:
            Any: Converted value.
        """
        if value == "None":
            return None

        if isinstance(current_value, bool):
            return value.lower() in ("true", "1", "yes", "on", "enabled")
        elif isinstance(current_value, int):
            return int(value)
        elif isinstance(current_value, float):
            return float(value)
        else:
            return value

    def _detect_format(self, file_path: Path) -> str:
        """Detect file format from extension.

        Args:
            file_path (Path): File path to check.

        Returns:
            str: Detected format ("toml", "json", or "yaml").
        """
        suffix = file_path.suffix.lower()
        if suffix == ".toml":
            return "toml"
        elif suffix == ".json":
            return "json"
        elif suffix in (".yaml", ".yml"):
            return "yaml"
        else:
            return "toml"  # Default to TOML

    def _output_settings_list(self, settings: List[Dict[str, Any]], format_type: str) -> None:
        """Output settings list in specified format.

        Args:
            settings (List[Dict[str, Any]]): List of settings to output.
            format_type (str): Output format ("table", "json", or "yaml").
        """
        if format_type == "json":
            print(json.dumps(self.settings_registry.make_serializable(settings), indent=2))
        elif format_type == "yaml" and yaml:
            print(yaml.dump(self.settings_registry.make_serializable(settings), default_flow_style=False))
        else:
            # Table format (default)
            self._print_header(translate("headers.settings_list"))

            for setting in settings:
                # Group by category
                category_display_name = setting.get("category_display_name", setting["category_name"])
                self._print_subheader(f"[{category_display_name}] ({setting['category_name']})")

                # Format setting info
                display_name = setting.get("display_name", setting["name"])
                access_level = setting["access_level"].value if hasattr(setting["access_level"], "value") else setting["access_level"]

                self._print_info(f"  {display_name} ({setting['name']})")
                self._print_detail(f"    {translate('common.description')}: {setting['description']}")
                self._print_detail(f"    {translate('common.current')}: {setting['current_value']}")
                self._print_detail(f"    {translate('common.default')}: {setting['default_value']}")
                self._print_detail(f"    Access: {access_level}")

                if setting.get("hint"):
                    self._print_detail(f"    Hint: {setting['hint']}")
                print()

    def _output_setting_info(self, setting: Dict[str, Any]) -> None:
        """Output detailed information for a single setting.

        Args:
            setting (Dict[str, Any]): Setting information to output.
        """
        display_name = setting.get("display_name", setting["name"])
        category_name = setting.get("category_display_name", setting["category_name"])
        access_level = setting["access_level"].value if hasattr(setting["access_level"], "value") else setting["access_level"]

        self._print_header(f"{category_name}: {display_name}")
        self._print_info(f"{translate('common.name')}: {setting['name']}")
        self._print_info(f"{translate('common.category')}: {setting['category_name']}")
        self._print_info(f"{translate('common.description')}: {setting['description']}")
        self._print_info(f"{translate('common.current')}: {setting['current_value']}")
        self._print_info(f"{translate('common.default')}: {setting['default_value']}")
        self._print_info(f"Access Level: {access_level}")
        self._print_info(f"Type: {setting['type']}")

        if setting.get("hint"):
            self._print_info(f"Hint: {setting['hint']}")

    def _get_available_settings(self) -> List[str]:
        """Get a list of all available settings in the registry.

        Returns:
            List[str]: List of setting names.
        """
        if not self.settings_registry:
            return []

        return [f"{setting['category_name']}:{setting['name']}" for setting in self.settings_registry.list_settings()]

    def _print_header(self, text: str) -> None:
        """Print a formatted header."""
        print_formatted_text(FormattedText([("class:header", text)]), style=self.style)

    def _print_subheader(self, text: str) -> None:
        """Print a formatted subheader."""
        print_formatted_text(FormattedText([("class:subheader", text)]), style=self.style)

    def _print_info(self, text: str) -> None:
        """Print informational text."""
        print_formatted_text(FormattedText([("class:info", text)]), style=self.style)

    def _print_detail(self, text: str) -> None:
        """Print detail text."""
        print_formatted_text(FormattedText([("class:detail", text)]), style=self.style)

    def _print_success(self, text: str) -> None:
        """Print success message."""
        print_formatted_text(FormattedText([("class:success", f"✓ {text}")]), style=self.style)

    def _print_error(self, text: str) -> None:
        """Print error message."""
        self.logger.error(text)

    def _print_warning(self, text: str) -> None:
        """Print warning message."""
        self.logger.warning(text)
Functions
print_commands_help()

Print help for all available chat commands.

Source code in hatchling/ui/settings_commands.py 
214
215
216
217
218
219
220
221
def print_commands_help(self) -> None:
    """Print help for all available chat commands."""
    print_formatted_text(FormattedText([
        ('class:header', "\n=== Settings Commands ===\n")
    ]), style=self.style)

    # Call parent class method to print formatted commands
    super().print_commands_help()

Functions