Skip to content

Commands

commands

Command registry for unified local and remote execution.

This module provides a centralized registry of commands used by tools, enabling consistent execution across local and remote systems.

CommandGroup

Bases: BaseModel

Group of related commands for multi-command tool operations.

Attributes:

Name Type Description
commands Mapping[str, CommandSpec]

Named commands within the group.
Source code in src/linux_mcp_server/commands.py 
73
74
75
76
77
78
79
80
81
82
class CommandGroup(BaseModel):
    """Group of related commands for multi-command tool operations.

    Attributes:
        commands: Named commands within the group.
    """

    model_config = ConfigDict(frozen=True)

    commands: Mapping[str, CommandSpec]

CommandSpec

Bases: BaseModel

Specification for a single command with optional fallback.

Attributes:

Name Type Description
args tuple[str, ...]

Command arguments as a tuple of strings.
fallback tuple[str, ...] | None

Alternative command arguments if primary fails.
optional_flags Mapping[str, tuple[str, ...]] | None

Maps parameter names to flag arguments that are added when the parameter is truthy. For example: {"unit": ["--unit", "{unit}"]} adds "--unit " when unit is provided.
Source code in src/linux_mcp_server/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
class CommandSpec(BaseModel):
    """Specification for a single command with optional fallback.

    Attributes:
        args: Command arguments as a tuple of strings.
        fallback: Alternative command arguments if primary fails.
        optional_flags: Maps parameter names to flag arguments that are added
            when the parameter is truthy. For example:
            {"unit": ["--unit", "{unit}"]} adds "--unit <value>" when unit is provided.
    """

    model_config = ConfigDict(frozen=True)

    args: tuple[str, ...]
    fallback: tuple[str, ...] | None = None
    optional_flags: Mapping[str, tuple[str, ...]] | None = None

    async def run(self, host: str | None = None, **kwargs: object) -> tuple[int, str, str]:
        """Run the command with optional fallback.

        Args:
            host: Optional remote host address.
            **kwargs: Additional arguments passed to substitute_command_args.
        """
        args = list(substitute_command_args(self.args, **kwargs))
        if self.optional_flags:
            for param_name, flag_args in self.optional_flags.items():
                if kwargs.get(param_name):
                    args.extend(substitute_command_args(flag_args, **kwargs))

        returncode, stdout, stderr = await execute_with_fallback(tuple(args), fallback=self.fallback, host=host)
        stdout = stdout if isinstance(stdout, str) else stdout.decode("utf-8", errors="replace")
        stderr = stderr if isinstance(stderr, str) else stderr.decode("utf-8", errors="replace")
        return returncode, stdout, stderr

    async def run_bytes(self, host: str | None = None, **kwargs: object) -> tuple[int, bytes, bytes]:
        """Run the command with optional fallback and return raw bytes.

        Args:
            host: Optional remote host address.
            **kwargs: Additional arguments passed to substitute_command_args.
        """
        args = list(substitute_command_args(self.args, **kwargs))
        if self.optional_flags:
            for param_name, flag_args in self.optional_flags.items():
                if kwargs.get(param_name):
                    args.extend(substitute_command_args(flag_args, **kwargs))

        returncode, stdout, stderr = await execute_with_fallback(
            tuple(args), fallback=self.fallback, host=host, encoding=None
        )
        stdout = stdout if isinstance(stdout, bytes) else stdout.encode("utf-8")
        stderr = stderr if isinstance(stderr, bytes) else stderr.encode("utf-8")
        return returncode, stdout, stderr

run async 

run(
    host: str | None = None, **kwargs: object
) -> tuple[int, str, str]

Run the command with optional fallback.

Parameters:

Name Type Description Default
host str | None

Optional remote host address.

 None
**kwargs object

Additional arguments passed to substitute_command_args.

 {}
Source code in src/linux_mcp_server/commands.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
async def run(self, host: str | None = None, **kwargs: object) -> tuple[int, str, str]:
    """Run the command with optional fallback.

    Args:
        host: Optional remote host address.
        **kwargs: Additional arguments passed to substitute_command_args.
    """
    args = list(substitute_command_args(self.args, **kwargs))
    if self.optional_flags:
        for param_name, flag_args in self.optional_flags.items():
            if kwargs.get(param_name):
                args.extend(substitute_command_args(flag_args, **kwargs))

    returncode, stdout, stderr = await execute_with_fallback(tuple(args), fallback=self.fallback, host=host)
    stdout = stdout if isinstance(stdout, str) else stdout.decode("utf-8", errors="replace")
    stderr = stderr if isinstance(stderr, str) else stderr.decode("utf-8", errors="replace")
    return returncode, stdout, stderr

run_bytes async 

run_bytes(
    host: str | None = None, **kwargs: object
) -> tuple[int, bytes, bytes]

Run the command with optional fallback and return raw bytes.

Parameters:

Name Type Description Default
host str | None

Optional remote host address.

 None
**kwargs object

Additional arguments passed to substitute_command_args.

 {}
Source code in src/linux_mcp_server/commands.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def run_bytes(self, host: str | None = None, **kwargs: object) -> tuple[int, bytes, bytes]:
    """Run the command with optional fallback and return raw bytes.

    Args:
        host: Optional remote host address.
        **kwargs: Additional arguments passed to substitute_command_args.
    """
    args = list(substitute_command_args(self.args, **kwargs))
    if self.optional_flags:
        for param_name, flag_args in self.optional_flags.items():
            if kwargs.get(param_name):
                args.extend(substitute_command_args(flag_args, **kwargs))

    returncode, stdout, stderr = await execute_with_fallback(
        tuple(args), fallback=self.fallback, host=host, encoding=None
    )
    stdout = stdout if isinstance(stdout, bytes) else stdout.encode("utf-8")
    stderr = stderr if isinstance(stderr, bytes) else stderr.encode("utf-8")
    return returncode, stdout, stderr

get_command 

get_command(
    name: str, subcommand: str = "default"
) -> CommandSpec

Get a command spec from the registry.

Parameters:

Name Type Description Default
name str

The command name in the registry.

 required
subcommand str

The subcommand key within the group (default: "default").

 'default'

Returns:

Type Description
CommandSpec

The CommandSpec for the given name and subcommand.

Raises:

Type Description
KeyError

If the command name or subcommand is not found.
Source code in src/linux_mcp_server/commands.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def get_command(name: str, subcommand: str = "default") -> CommandSpec:
    """Get a command spec from the registry.

    Args:
        name: The command name in the registry.
        subcommand: The subcommand key within the group (default: "default").

    Returns:
        The CommandSpec for the given name and subcommand.

    Raises:
        KeyError: If the command name or subcommand is not found.
    """
    group = get_command_group(name)
    try:
        return group.commands[subcommand]
    except KeyError as e:
        available = ", ".join(sorted(group.commands.keys()))
        raise KeyError(f"Subcommand '{subcommand}' not found for '{name}'. Available: {available}") from e

get_command_group 

get_command_group(name: str) -> CommandGroup

Get a command group from the registry.

Use this when you need to iterate over all subcommands in a group.

Parameters:

Name Type Description Default
name str

The command group name in the registry.

 required

Returns:

Type Description
CommandGroup

The CommandGroup for the given name.

Raises:

Type Description
KeyError

If the command name is not found in the registry.
Source code in src/linux_mcp_server/commands.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def get_command_group(name: str) -> CommandGroup:
    """Get a command group from the registry.

    Use this when you need to iterate over all subcommands in a group.

    Args:
        name: The command group name in the registry.

    Returns:
        The CommandGroup for the given name.

    Raises:
        KeyError: If the command name is not found in the registry.
    """
    try:
        return COMMANDS[name]
    except KeyError as e:
        available = ", ".join(sorted(COMMANDS.keys()))
        raise KeyError(f"Command '{name}' not found in registry. Available: {available}") from e

substitute_command_args 

substitute_command_args(
    args: Sequence[str], **kwargs: object
) -> tuple[str, ...]

Substitute placeholder values in command arguments.

Parameters:

Name Type Description Default
args Sequence[str]

Sequence of command arguments, possibly with {placeholder} values.

 required
**kwargs object

Key-value pairs to substitute into placeholders.

 {}

Returns:

Type Description
tuple[str, ...]

Tuple of command arguments with placeholders replaced.

Raises:

Type Description
ValueError

If any placeholders are missing from kwargs or remain unsubstituted after replacement.
Example

substitute_command_args(("ps", "-p", "{pid}"), pid=1234) ("ps", "-p", "1234")

Source code in src/linux_mcp_server/commands.py 
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
def substitute_command_args(args: Sequence[str], **kwargs: object) -> tuple[str, ...]:
    """Substitute placeholder values in command arguments.

    Args:
        args: Sequence of command arguments, possibly with {placeholder} values.
        **kwargs: Key-value pairs to substitute into placeholders.

    Returns:
        Tuple of command arguments with placeholders replaced.

    Raises:
        ValueError: If any placeholders are missing from kwargs or remain
            unsubstituted after replacement.

    Example:
        >>> substitute_command_args(("ps", "-p", "{pid}"), pid=1234)
        ("ps", "-p", "1234")
    """
    try:
        result = tuple(arg.format(**kwargs) for arg in args)
    except KeyError as e:
        raise ValueError(f"Missing required placeholder: {e}") from e

    # Validate all placeholders were replaced (catches edge cases like nested braces)
    for arg in result:
        if "{" in arg and "}" in arg:
            raise ValueError(f"Unsubstituted placeholder in command argument: {arg}")

    return result