feat: simple MCP server

docs/CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased] - ??
 
+### Added
+
+- Added `rsconnect mcp-server` command to expose rsconnect-python functionality to
+  LLM clients via the Model Context Protocol. The server provides a `get_command_info`
+  tool that returns parameter schemas for any rsconnect command, allowing LLMs
+  to more easily construct valid CLI commands.
+
 ### Fixed
 
 - Snowflake SPCS (Snowpark Container Services) authentication now properly handles API keys

docs/commands/mcp-server.md

@@ -0,0 +1,3 @@
+::: mkdocs-click
+    :module: rsconnect.main
+    :command: mcp_server

mkdocs.yml

@@ -50,6 +50,7 @@ nav:
       - system: commands/system.md
       - version: commands/version.md
       - write-manifest: commands/write-manifest.md
+      - mcp-server: commands/mcp-server.md
 
 
 theme:

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,123 @@ def version():
     click.echo(VERSION)
 
 
+@cli.command(
+    short_help="Start the Model Context Protocol (MCP) server.",
+    help=(
+        "Start a Model Context Protocol (MCP) server to expose rsconnect-python capabilities to AI applications "
+        "through a standardized protocol interface."
+        "\n\n"
+        "The MCP server exposes a single tool:\n\n"
+        "`get_command_info`:\n\n"
+        "  -  Provides detailed parameter schemas for any rsconnect command. "
+        "This provides context for an LLM to understand how to construct valid rsconnect "
+        "commands dynamically without hard-coded knowledge of the CLI."
+        "\n\n"
+        "System Requirements:\n\n"
+        "  - Python>=3.10\n"
+        "  - fastmcp"
+        "\n\n"
+        "The server runs in stdio mode, communicating via standard input/output streams."
+        "\n\n"
+        "Usage with popular LLM clients:\n\n"
+        "  -  [codex](https://developers.openai.com/codex/mcp/#configuration---cli)\n"
+        "  -  [claude code](https://docs.claude.com/en/docs/claude-code/mcp#option-3%3A-add-a-local-stdio-server)\n"
+        "  -  [VS Code](https://code.visualstudio.com/docs/copilot/customization/mcp-servers#_add-an-mcp-server)\n\n"
+        "The command `uvx --from rsconnect-python rsconnect mcp-server` is a simple option for use in each of "
+        "the above options."
+    ),
+)
+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)
+
+    def get_command_info(
+        command_path: str,
+    ) -> Dict[str, Any]:
+        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)}")
+
+    # dynamically build docstring with top level commands
+    # note: excluding mcp-server here
+    available_commands = sorted(cmd for cmd in all_commands_info["commands"].keys() if cmd != "mcp-server")
+    commands_list = "\n    ".join(f"- {cmd}" for cmd in available_commands)
+
+    get_command_info.__doc__ = f"""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.
+
+    Available top-level commands:
+    {commands_list}
+
+    :param command_path: space-separated command path (e.g., 'version', 'deploy notebook', 'content build add')
+    :return: dictionary with command parameter schema and execution metadata
+    """
+
+    mcp.tool(get_command_info)
+
+    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 +560,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))