feat: simple MCP server

pyproject.toml

@@ -13,7 +13,7 @@ dependencies = [
     "semver>=2.0.0,<4.0.0",
     "pyjwt>=2.4.0",
     "click>=8.0.0",
-    "toml>=0.10; python_version < '3.11'"
+    "toml>=0.10; python_version < '3.11'",
 ]
 
 dynamic = ["version"]
@@ -37,8 +37,10 @@ test = [
     "setuptools_scm[toml]>=3.4",
     "twine",
     "types-Flask",
+    "fastmcp==2.12.4; python_version >= '3.10'",
 ]
 snowflake = ["snowflake-cli"]
+mcp = ["fastmcp==2.12.4; python_version >= '3.10'"]
 docs = [
     "mkdocs-material",
     "mkdocs-click",

rsconnect/main.py

@@ -8,7 +8,17 @@
 import traceback
 from functools import wraps
 from os.path import abspath, dirname, exists, isdir, join
-from typing import Callable, ItemsView, Literal, Optional, Sequence, TypeVar, cast
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    ItemsView,
+    Literal,
+    Optional,
+    Sequence,
+    TypeVar,
+    cast,
+)
 
 import click
 
@@ -392,6 +402,89 @@ def version():
     click.echo(VERSION)
 
 
+@cli.command(help="Start the MCP server")
+def mcp_server():
+    try:
+        from fastmcp import FastMCP
+        from fastmcp.exceptions import ToolError
+    except ImportError:
+        raise RSConnectException(
+            "The fastmcp package is required for MCP server functionality. "
+            "Install it with: pip install rsconnect-python[mcp]"
+        )
+
+    mcp = FastMCP("Connect MCP")
+
+    # Discover all commands at startup
+    from .mcp_deploy_context import discover_all_commands
+
+    all_commands_info = discover_all_commands(cli)
+
+    @mcp.tool()
+    def get_command_info(
+        command_path: str,
+    ) -> Dict[str, Any]:
+        """
+        Get the parameter schema for any rsconnect command.
+
+        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 command_path: space-separated command path (e.g., 'version', 'deploy notebook', 'content build add')
+        :return: dictionary with command parameter schema and execution metadata
+        """
+        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()
+
+
 def _test_server_and_api(server: str, api_key: str, insecure: bool, ca_cert: str | None):
     """
     Test the specified server information to make sure it works.  If so, a
@@ -433,7 +526,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

@@ -0,0 +1,115 @@
+"""
+Programmatically discover all parameters for rsconnect commands.
+This helps MCP tools understand how to use the cli.
+"""
+
+import json
+from typing import Any, Dict
+
+import click
+
+
+def extract_parameter_info(param: click.Parameter) -> Dict[str, Any]:
+    """Extract detailed information from a Click parameter."""
+    info: Dict[str, Any] = {}
+
+    if isinstance(param, click.Option) and param.opts:
+        # Use the longest option name (usually the full form without dashes)
+        mcp_arg_name = max(param.opts, key=len).lstrip("-").replace("-", "_")
+        info["name"] = mcp_arg_name
+        info["cli_flags"] = param.opts
+        info["param_type"] = "option"
+    else:
+        info["name"] = param.name
+        if isinstance(param, click.Argument):
+            info["param_type"] = "argument"
+
+    # extract help text for added context
+    help_text = getattr(param, "help", None)
+    if help_text:
+        info["description"] = help_text
+
+    if isinstance(param, click.Option):
+        # Boolean flags
+        if param.is_flag:
+            info["type"] = "boolean"
+            info["default"] = param.default or False
+
+        # choices
+        elif param.type and hasattr(param.type, "choices"):
+            info["type"] = "string"
+            info["choices"] = list(param.type.choices)
+
+        # multiple
+        elif param.multiple:
+            info["type"] = "array"
+            info["items"] = {"type": "string"}
+
+        # files
+        elif isinstance(param.type, click.Path):
+            info["type"] = "string"
+            info["format"] = "path"
+            if param.type.exists:
+                info["path_must_exist"] = True
+            if param.type.file_okay and not param.type.dir_okay:
+                info["path_type"] = "file"
+            elif param.type.dir_okay and not param.type.file_okay:
+                info["path_type"] = "directory"
+
+        # default
+        else:
+            info["type"] = "string"
+
+        # defaults (important to avoid noise in returned command)
+        if param.default is not None and not param.is_flag:
+            if isinstance(param.default, tuple):
+                info["default"] = list(param.default)
+            elif isinstance(param.default, (str, int, float, bool, list, dict)):
+                info["default"] = param.default
+
+    # required params
+    info["required"] = param.required
+
+    return info
+
+
+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
+
+
+def discover_command_group(group: click.Group) -> Dict[str, Any]:
+    """Discover all commands in a command group and their parameters."""
+    result = {"name": group.name, "description": group.help, "commands": {}}
+
+    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)
+
+    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
+
+    # Discover all commands in the CLI
+    # use this for testing/debugging
+    all_commands = discover_all_commands(cli)
+    print(json.dumps(all_commands, indent=2))