generalize. tests written by claude

Author: joshyam-k

pyproject.toml

@@ -14,7 +14,7 @@ dependencies = [
     "pyjwt>=2.4.0",
     "click>=8.0.0",
     "toml>=0.10; python_version < '3.11'",
-    "fastmcp"
+    "fastmcp==2.12.4"
 ]
 
 dynamic = ["version"]

rsconnect/main.py

@@ -3,7 +3,6 @@
 import functools
 import json
 import os
-import shlex
 import sys
 import textwrap
 import traceback
@@ -405,70 +404,82 @@ def version():
 
 @cli.command(help="Start the MCP server")
 def mcp_server():
-    import subprocess
-
     from fastmcp import FastMCP
     from fastmcp.exceptions import ToolError
 
     mcp = FastMCP("Connect MCP")
 
-    # Discover all deploy commands at startup
-    from .mcp_deploy_context import discover_deploy_commands
-    deploy_commands_info = discover_deploy_commands(cli)
-
-    @mcp.tool()
-    def list_servers():
-        """Show the stored information about each known server nickname."""
-        try:
-            result = subprocess.run(["rsconnect", "list"], capture_output=True, text=True, check=True)
-            return result.stdout
-        except subprocess.CalledProcessError as e:
-            raise ToolError(f"Command failed with error: {e}")
-
-    @mcp.tool()
-    def add_server(
-        server_url: str,
-        nickname: str,
-        api_key: str,
-    ) -> Dict[str, Any]:
-        """
-        Prompt user to add a Posit Connect server using rsconnect add.
-
-        This tool guides users through registering a Connect server so they can use
-        rsconnect deployment commands. The server nickname allows you to reference
-        the server in future deployment commands.
-
-        :param server_url: the URL of your Posit Connect server (e.g., https://my.connect.server/)
-        :param nickname: a nickname you choose for the server (e.g., myServer)
-        :param api_key: your personal API key for authentication
-        :return: a dictionary containing the command to run and instructions.
-        """
-        command = f"rsconnect add --server {server_url} --name {nickname} --api-key {api_key}"
-
-        return {
-            "type": "command",
-            "command": command
-        }
+    # Discover all commands at startup
+    from .mcp_deploy_context import discover_all_commands
+    all_commands_info = discover_all_commands(cli)
 
     @mcp.tool()
-    def deploy_command_context(
-        content_type: str
+    def get_command_info(
+        command_path: str,
     ) -> Dict[str, Any]:
         """
-        Get the parameter schema for rsconnect deploy commands that should be executed via bash.
+        Get the parameter schema for any rsconnect command.
 
-        Returns information about the parameters needed to construct an rsconnect deploy
-        command that can be executed in a bash shell.
+        Returns information about the parameters needed to construct an rsconnect command
+        that can be executed in a bash shell. Supports nested command groups of arbitrary depth.
 
-        :param content_type: the type of content (e.g., 'shiny', 'notebook', 'quarto')
+        :param command_path: space-separated command path (e.g., 'version', 'deploy notebook', 'content build add')
         :return: dictionary with command parameter schema and execution metadata
         """
-        ctx = deploy_commands_info["content_type"][content_type]
-        return {
-            "context": ctx,
-            "command_usage": "rsconnect deploy COMMAND [OPTIONS] DIRECTORY [ARGS]...",
-            "shell": "bash"
-        }
+        try:
+            # split the command path into parts
+            parts = command_path.strip().split()
+            if not parts:
+                available_commands = list(all_commands_info["commands"].keys())
+                return {
+                    "error": "Command path cannot be empty",
+                    "available_commands": available_commands
+                }
+
+            current_info = all_commands_info
+            current_path = []
+
+            for _, part in enumerate(parts):
+                # error if we find unexpected additional subcommands
+                if "commands" not in current_info:
+                    return {
+                        "error": f"'{' '.join(current_path)}' is not a command group. Unexpected part: '{part}'",
+                        "type": "command",
+                        "command_path": f"rsconnect {' '.join(current_path)}",
+                    }
+
+                # try to return useful messaging for invalid subcommands
+                if part not in current_info["commands"]:
+                    available = list(current_info["commands"].keys())
+                    path_str = ' '.join(current_path) if current_path else "top level"
+                    return {
+                        "error": f"Command '{part}' not found in {path_str}",
+                        "available_commands": available
+                    }
+
+                current_info = current_info["commands"][part]
+                current_path.append(part)
+
+            # still return something useful if additional subcommands are needed
+            if "commands" in current_info:
+                return {
+                    "type": "command_group",
+                    "name": current_info.get("name", parts[-1]),
+                    "description": current_info.get("description"),
+                    "available_subcommands": list(current_info["commands"].keys()),
+                    "message": f"The '{' '.join(parts)}' command requires a subcommand."
+                }
+            else:
+                return {
+                    "type": "command",
+                    "command_path": f"rsconnect {' '.join(parts)}",
+                    "name": current_info.get("name", parts[-1]),
+                    "description": current_info.get("description"),
+                    "parameters": current_info.get("parameters", []),
+                    "shell": "bash"
+                }
+        except Exception as e:
+            raise ToolError(f"Failed to retrieve command info: {str(e)}")
 
     mcp.run()
 
@@ -514,7 +525,7 @@ def _test_spcs_creds(server: SPCSConnectServer):
 
 @cli.command(
     short_help="Create an initial admin user to bootstrap a Connect instance.",
-    help="Creates an initial admin user to bootstrap a Connect instance. Returns the provisionend API key.",
+    help="Creates an initial admin user to bootstrap a Connect instance. Returns the provisioned API key.",
     no_args_is_help=True,
 )
 @click.option(

rsconnect/mcp_deploy_context.py

@@ -1,6 +1,6 @@
 """
-Programmatically discover all parameters for rsconnect deploy commands.
-This helps MCP tools understand exactly how to use `rsconnect deploy ...`
+Programmatically discover all parameters for rsconnect commands.
+This helps MCP tools understand how to use the cli.
 """
 
 import json
@@ -73,37 +73,51 @@ def extract_parameter_info(param: click.Parameter) -> Dict[str, Any]:
     return info
 
 
-def discover_deploy_commands(cli_group: click.Group) -> Dict[str, Any]:
-    """Discover all deploy commands and their parameters."""
+def discover_single_command(cmd: click.Command) -> Dict[str, Any]:
+    """Discover a single command and its parameters."""
+    cmd_info = {
+        "name": cmd.name,
+        "description": cmd.help,
+        "parameters": []
+    }
+
+    for param in cmd.params:
+        if param.name in ["verbose", "v"]:
+            continue
+
+        param_info = extract_parameter_info(param)
+        cmd_info["parameters"].append(param_info)
+
+    return cmd_info
 
-    deploy_group = cli_group.commands["deploy"]
 
+def discover_command_group(group: click.Group) -> Dict[str, Any]:
+    """Discover all commands in a command group and their parameters."""
     result = {
-        "group_name": "deploy",
-        "description": deploy_group.help,
-        "content_type": {}
+        "name": group.name,
+        "description": group.help,
+        "commands": {}
     }
 
-    for cmd_name, cmd in deploy_group.commands.items():
-        cmd_info = {
-            "name": cmd_name,
-            "description": cmd.help,
-            "parameters": []
-        }
-        for param in cmd.params:
-            if param.name in ["verbose", "v"]:
-                continue
+    for cmd_name, cmd in group.commands.items():
+        if isinstance(cmd, click.Group):
+            # recursively discover nested command groups
+            result["commands"][cmd_name] = discover_command_group(cmd)
+        else:
+            result["commands"][cmd_name] = discover_single_command(cmd)
 
-            param_info = extract_parameter_info(param)
-            cmd_info["parameters"].append(param_info)
+    return result
 
-        result["content_type"][cmd_name] = cmd_info
 
-    return result
+def discover_all_commands(cli: click.Group) -> Dict[str, Any]:
+    """Discover all commands in the CLI and their parameters."""
+    return discover_command_group(cli)
 
 
 if __name__ == "__main__":
     from rsconnect.main import cli
 
-    deploy_commands_info = discover_deploy_commands(cli)["content_type"]["shiny"]
-    print(json.dumps(deploy_commands_info, indent=2))
+    # Discover all commands in the CLI
+    # use this for testing/debugging
+    all_commands = discover_all_commands(cli)
+    print(json.dumps(all_commands, indent=2))