Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,16 @@ description = "The Posit Connect command-line interface."
authors = [{ name = "Posit, PBC", email = "rsconnect@posit.co" }]
license = { file = "LICENSE.md" }
readme = { file = "README.md", content-type = "text/markdown" }
requires-python = ">=3.8"
requires-python = ">=3.10"
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fastmcp requires >=3.10 which would be a breaking change... what's the protocol around doing something like this? Is this reason enough to move this work off of rsconnect-python?


dependencies = [
"typing-extensions>=4.8.0",
"pip>=10.0.0",
"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'",
"fastmcp==2.12.4"
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fastmcp recommends strict versioning because of the rapidly evolving MCP ecosystem. This is just the version I have installed.

]

dynamic = ["version"]
Expand Down
96 changes: 94 additions & 2 deletions rsconnect/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -392,6 +402,88 @@ def version():
click.echo(VERSION)


@cli.command(help="Start the MCP server")
def mcp_server():
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

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
Expand Down Expand Up @@ -433,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(
Expand Down
123 changes: 123 additions & 0 deletions rsconnect/mcp_deploy_context.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,123 @@
"""
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))
Loading
Loading