Skip to content

mcp_registry_client.commands

Command pattern implementation for CLI commands.

BaseCommand

Bases: ABC

Abstract base class for CLI commands.

Provides a standard interface for command validation, execution, and output formatting.

Source code in mcp_registry_client/commands/base.py 
12
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
class BaseCommand(ABC):
    """Abstract base class for CLI commands.

    Provides a standard interface for command validation, execution,
    and output formatting.
    """

    def __init__(self, args: argparse.Namespace) -> None:
        """Initialize command with parsed arguments.

        Args:
            args: Parsed command line arguments

        """
        self.args = args

    @abstractmethod
    def validate_args(self) -> None:
        """Validate command arguments.

        Raises:
            ValueError: If arguments are invalid

        """

    @abstractmethod
    async def execute(self) -> Any:  # noqa: ANN401
        """Execute the command logic.

        Returns:
            Command execution result

        Raises:
            Exception: Various exceptions based on command implementation

        """

    @abstractmethod
    def format_output(self, result: Any) -> None:  # noqa: ANN401
        """Format and display command output.

        Args:
            result: Result from execute() method

        """

    async def run(self) -> int:
        """Run the complete command workflow.

        Returns:
            Exit code (0 for success, non-zero for error)

        """
        try:
            self.validate_args()
            result: Any = await self.execute()
            self.format_output(result)
        except KeyboardInterrupt:
            raise
        except (ValueError, RegistryClientError, OSError, RuntimeError) as e:
            return self._handle_error(e)
        else:
            return 0

    def _handle_error(self, exc: Exception) -> int:
        """Handle command errors with consistent messaging.

        Args:
            exc: The exception that occurred

        Returns:
            Appropriate exit code for the error type

        """
        # Handle ValueError specially (user input validation errors)
        if isinstance(exc, ValueError):
            print_error(f'Error: {exc}')
            return 1

        context = f'{self.__class__.__name__.lower().replace("command", "")} operation'
        return handle_command_error(exc, context)

__init__(args)

Initialize command with parsed arguments.

Parameters:

Name Type Description Default
args Namespace

Parsed command line arguments

 required
Source code in mcp_registry_client/commands/base.py 
19
20
21
22
23
24
25
26
def __init__(self, args: argparse.Namespace) -> None:
    """Initialize command with parsed arguments.

    Args:
        args: Parsed command line arguments

    """
    self.args = args

execute() abstractmethod async

Execute the command logic.

Returns:

Type Description
Any

Command execution result

Raises:

Type Description
Exception

Various exceptions based on command implementation
Source code in mcp_registry_client/commands/base.py 
37
38
39
40
41
42
43
44
45
46
47
@abstractmethod
async def execute(self) -> Any:  # noqa: ANN401
    """Execute the command logic.

    Returns:
        Command execution result

    Raises:
        Exception: Various exceptions based on command implementation

    """

format_output(result) abstractmethod

Format and display command output.

Parameters:

Name Type Description Default
result Any

Result from execute() method

 required
Source code in mcp_registry_client/commands/base.py 
49
50
51
52
53
54
55
56
@abstractmethod
def format_output(self, result: Any) -> None:  # noqa: ANN401
    """Format and display command output.

    Args:
        result: Result from execute() method

    """

run() async

Run the complete command workflow.

Returns:

Type Description
int

Exit code (0 for success, non-zero for error)
Source code in mcp_registry_client/commands/base.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def run(self) -> int:
    """Run the complete command workflow.

    Returns:
        Exit code (0 for success, non-zero for error)

    """
    try:
        self.validate_args()
        result: Any = await self.execute()
        self.format_output(result)
    except KeyboardInterrupt:
        raise
    except (ValueError, RegistryClientError, OSError, RuntimeError) as e:
        return self._handle_error(e)
    else:
        return 0

validate_args() abstractmethod

Validate command arguments.

Raises:

Type Description
ValueError

If arguments are invalid
Source code in mcp_registry_client/commands/base.py 
28
29
30
31
32
33
34
35
@abstractmethod
def validate_args(self) -> None:
    """Validate command arguments.

    Raises:
        ValueError: If arguments are invalid

    """

InfoCommand

Bases: BaseCommand

Command to get detailed information about a specific server.

Source code in mcp_registry_client/commands/info.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
class InfoCommand(BaseCommand):
    """Command to get detailed information about a specific server."""

    def __init__(self, args: argparse.Namespace) -> None:
        """Initialize info command.

        Args:
            args: Parsed command line arguments containing 'server_name' and 'json'

        """
        super().__init__(args)

    def validate_args(self) -> None:
        """Validate info command arguments.

        Raises:
            ValueError: If server name is invalid

        """
        validate_server_name(self.args.server_name)

    async def execute(self) -> Server:
        """Execute the info command.

        Returns:
            Server information from the registry client

        Raises:
            RegistryAPIError: If API request fails
            RegistryClientError: If client processing fails
            httpx.RequestError: If HTTP request fails
            ValidationError: If response validation fails
            ValueError: If server is not found

        """
        async with RegistryClient() as client:
            server = await client.get_server_by_name(self.args.server_name)

            if server is None:
                msg = f'Server "{self.args.server_name}" not found'
                raise ValueError(msg)

            return server

    def format_output(self, result: Server) -> None:
        """Format and display server information.

        Args:
            result: Server object with detailed information

        """
        cli_config = get_cli_config()

        if self.args.json:
            data = format_server_detailed(result)
            print_json(data, indent=cli_config.json_indent)
        else:
            print_server_info_human_readable(result)

__init__(args)

Initialize info command.

Parameters:

Name Type Description Default
args Namespace

Parsed command line arguments containing 'server_name' and 'json'

 required
Source code in mcp_registry_client/commands/info.py 
21
22
23
24
25
26
27
28
def __init__(self, args: argparse.Namespace) -> None:
    """Initialize info command.

    Args:
        args: Parsed command line arguments containing 'server_name' and 'json'

    """
    super().__init__(args)

execute() async

Execute the info command.

Returns:

Type Description
Server

Server information from the registry client

Raises:

Type Description
RegistryAPIError

If API request fails
RegistryClientError

If client processing fails
RequestError

If HTTP request fails
ValidationError

If response validation fails
ValueError

If server is not found
Source code in mcp_registry_client/commands/info.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def execute(self) -> Server:
    """Execute the info command.

    Returns:
        Server information from the registry client

    Raises:
        RegistryAPIError: If API request fails
        RegistryClientError: If client processing fails
        httpx.RequestError: If HTTP request fails
        ValidationError: If response validation fails
        ValueError: If server is not found

    """
    async with RegistryClient() as client:
        server = await client.get_server_by_name(self.args.server_name)

        if server is None:
            msg = f'Server "{self.args.server_name}" not found'
            raise ValueError(msg)

        return server

format_output(result)

Format and display server information.

Parameters:

Name Type Description Default
result Server

Server object with detailed information

 required
Source code in mcp_registry_client/commands/info.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def format_output(self, result: Server) -> None:
    """Format and display server information.

    Args:
        result: Server object with detailed information

    """
    cli_config = get_cli_config()

    if self.args.json:
        data = format_server_detailed(result)
        print_json(data, indent=cli_config.json_indent)
    else:
        print_server_info_human_readable(result)

validate_args()

Validate info command arguments.

Raises:

Type Description
ValueError

If server name is invalid
Source code in mcp_registry_client/commands/info.py 
30
31
32
33
34
35
36
37
def validate_args(self) -> None:
    """Validate info command arguments.

    Raises:
        ValueError: If server name is invalid

    """
    validate_server_name(self.args.server_name)

SearchCommand

Bases: BaseCommand

Command to search for MCP servers in the registry.

Source code in mcp_registry_client/commands/search.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
class SearchCommand(BaseCommand):
    """Command to search for MCP servers in the registry."""

    def __init__(self, args: argparse.Namespace) -> None:
        """Initialize search command.

        Args:
            args: Parsed command line arguments containing 'name' and 'json'

        """
        super().__init__(args)

    def validate_args(self) -> None:
        """Validate search command arguments.

        Raises:
            ValueError: If search term is invalid

        """
        validate_search_term(self.args.name)

    async def execute(self) -> SearchResponse:
        """Execute the search command.

        Returns:
            Search result from the registry client

        Raises:
            RegistryAPIError: If API request fails
            RegistryClientError: If client processing fails
            httpx.RequestError: If HTTP request fails
            ValidationError: If response validation fails

        """
        async with RegistryClient() as client:
            return await client.search_servers(name=self.args.name)

    def format_output(self, result: SearchResponse) -> None:
        """Format and display search results.

        Args:
            result: Search result containing servers list

        """
        cli_config = get_cli_config()

        if self.args.json:
            data = [format_server_summary(server) for server in result.servers]
            print_json(data, indent=cli_config.json_indent)
        else:
            print_table(result.servers, cli_config.table_max_description_width)

__init__(args)

Initialize search command.

Parameters:

Name Type Description Default
args Namespace

Parsed command line arguments containing 'name' and 'json'

 required
Source code in mcp_registry_client/commands/search.py 
17
18
19
20
21
22
23
24
def __init__(self, args: argparse.Namespace) -> None:
    """Initialize search command.

    Args:
        args: Parsed command line arguments containing 'name' and 'json'

    """
    super().__init__(args)

execute() async

Execute the search command.

Returns:

Type Description
SearchResponse

Search result from the registry client

Raises:

Type Description
RegistryAPIError

If API request fails
RegistryClientError

If client processing fails
RequestError

If HTTP request fails
ValidationError

If response validation fails
Source code in mcp_registry_client/commands/search.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
async def execute(self) -> SearchResponse:
    """Execute the search command.

    Returns:
        Search result from the registry client

    Raises:
        RegistryAPIError: If API request fails
        RegistryClientError: If client processing fails
        httpx.RequestError: If HTTP request fails
        ValidationError: If response validation fails

    """
    async with RegistryClient() as client:
        return await client.search_servers(name=self.args.name)

format_output(result)

Format and display search results.

Parameters:

Name Type Description Default
result SearchResponse

Search result containing servers list

 required
Source code in mcp_registry_client/commands/search.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def format_output(self, result: SearchResponse) -> None:
    """Format and display search results.

    Args:
        result: Search result containing servers list

    """
    cli_config = get_cli_config()

    if self.args.json:
        data = [format_server_summary(server) for server in result.servers]
        print_json(data, indent=cli_config.json_indent)
    else:
        print_table(result.servers, cli_config.table_max_description_width)

validate_args()

Validate search command arguments.

Raises:

Type Description
ValueError

If search term is invalid
Source code in mcp_registry_client/commands/search.py 
26
27
28
29
30
31
32
33
def validate_args(self) -> None:
    """Validate search command arguments.

    Raises:
        ValueError: If search term is invalid

    """
    validate_search_term(self.args.name)