From e579430a326b6b3fd87d6485d93bf3aa6d7f9f66 Mon Sep 17 00:00:00 2001 From: joshyam-k Date: Thu, 25 Sep 2025 11:08:37 -0700 Subject: [PATCH 1/7] initial command --- pyproject.toml | 6 ++- rsconnect/main.py | 108 +++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 111 insertions(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 7768298e..6a72025a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ 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" dependencies = [ "typing-extensions>=4.8.0", @@ -13,7 +13,9 @@ 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'", + "fastmcp", + "posit-sdk" ] dynamic = ["version"] diff --git a/rsconnect/main.py b/rsconnect/main.py index 357d2b9a..c3c28a99 100644 --- a/rsconnect/main.py +++ b/rsconnect/main.py @@ -3,12 +3,24 @@ import functools import json import os +import subprocess import sys import textwrap 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, + List, + Literal, + Optional, + Sequence, + TypeVar, + cast, +) import click @@ -392,6 +404,100 @@ def version(): click.echo(VERSION) +@cli.command(help="Start the MCP server") +@click.option( + "--server", + "-s", + envvar="CONNECT_SERVER", + help="Posit Connect server URL" +) +@click.option( + "--api-key", + "-k", + envvar="CONNECT_API_KEY", + help="The API key to use to authenticate with Posit Connect." +) +def mcp_server(server: str, api_key: str): + from fastmcp import FastMCP + from fastmcp.exceptions import ToolError + from posit.connect import Client + + mcp = FastMCP("Connect MCP") + + def get_content_logs(app_guid: str): + try: + client = Client(server, api_key) + response = client.get(f"v1/content/{app_guid}/jobs") + jobs = response.json() + # first job key is the most recent one + key = jobs[0]["key"] + logs = client.get(f"v1/content/{app_guid}/jobs/{key}/log") + return logs.json() + except Exception as e: + raise ToolError(f"Failed to get logs: {e}") + + def list_content(): + try: + client = Client(server, api_key) + response = client.get("v1/content") + return response.json() + except Exception as e: + raise ToolError(f"Failed to list content: {e}") + + def get_content_item(app_guid: str): + try: + client = Client(server, api_key) + response = client.content.get(app_guid) + return response + except Exception as e: + raise ToolError(f"Failed to get content: {e}") + + @mcp.tool() + async def deploy_shiny( + directory: str, + name: Optional[str] = None, + title: Optional[str] = None + ) -> Dict[str, Any]: + """Deploy a Shiny application to Posit Connect""" + + # Build the CLI command + args = ["deploy", "shiny", directory] + + if name: + args.extend(["--name", name]) + + if title: + args.extend(["--title", title]) + + args.extend(["--server", server]) + args.extend(["--api-key", api_key]) + + try: + result = subprocess.run( + args, + check=True, + capture_output=True, + text=True + ) + return { + "success": True, + "message": "Deployment completed successfully", + "stdout": result.stdout, + "stderr": result.stderr + } + except subprocess.CalledProcessError as e: + raise ToolError(f"Deployment failed with exit code {e.returncode}: {e.stderr}") + except Exception as e: + raise ToolError(f"Command failed with error: {e}") + + + mcp.tool(description="Get content logs from Posit Connect")(get_content_logs) + mcp.tool(description="List content from Posit Connect")(list_content) + mcp.tool(description="Get content item from Posit Connect")(get_content_item) + + 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 From 358171135a3821a7e947f8506fb81380886b36d1 Mon Sep 17 00:00:00 2001 From: joshyam-k Date: Mon, 6 Oct 2025 10:09:15 -0700 Subject: [PATCH 2/7] testing new approach --- pyproject.toml | 3 +- rsconnect/main.py | 178 ++++++++++++++++++-------------- rsconnect/mcp_deploy_context.py | 116 +++++++++++++++++++++ 3 files changed, 218 insertions(+), 79 deletions(-) create mode 100644 rsconnect/mcp_deploy_context.py diff --git a/pyproject.toml b/pyproject.toml index 6a72025a..e3a98388 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,8 +14,7 @@ dependencies = [ "pyjwt>=2.4.0", "click>=8.0.0", "toml>=0.10; python_version < '3.11'", - "fastmcp", - "posit-sdk" + "fastmcp" ] dynamic = ["version"] diff --git a/rsconnect/main.py b/rsconnect/main.py index c3c28a99..f312ec2c 100644 --- a/rsconnect/main.py +++ b/rsconnect/main.py @@ -3,7 +3,7 @@ import functools import json import os -import subprocess +import shlex import sys import textwrap import traceback @@ -14,7 +14,6 @@ Callable, Dict, ItemsView, - List, Literal, Optional, Sequence, @@ -405,95 +404,120 @@ def version(): @cli.command(help="Start the MCP server") -@click.option( - "--server", - "-s", - envvar="CONNECT_SERVER", - help="Posit Connect server URL" -) -@click.option( - "--api-key", - "-k", - envvar="CONNECT_API_KEY", - help="The API key to use to authenticate with Posit Connect." -) -def mcp_server(server: str, api_key: str): +def mcp_server(): from fastmcp import FastMCP from fastmcp.exceptions import ToolError - from posit.connect import Client mcp = FastMCP("Connect MCP") - def get_content_logs(app_guid: str): - try: - client = Client(server, api_key) - response = client.get(f"v1/content/{app_guid}/jobs") - jobs = response.json() - # first job key is the most recent one - key = jobs[0]["key"] - logs = client.get(f"v1/content/{app_guid}/jobs/{key}/log") - return logs.json() - except Exception as e: - raise ToolError(f"Failed to get logs: {e}") - - def list_content(): - try: - client = Client(server, api_key) - response = client.get("v1/content") - return response.json() - except Exception as e: - raise ToolError(f"Failed to list content: {e}") + # Discover all deploy commands at startup + from .mcp_deploy_context import discover_deploy_commands + deploy_commands_info = discover_deploy_commands(cli) - def get_content_item(app_guid: str): - try: - client = Client(server, api_key) - response = client.content.get(app_guid) - return response - except Exception as e: - raise ToolError(f"Failed to get content: {e}") + @mcp.tool() + def list_servers() -> Dict[str, Any]: + """ + Show the stored information about each known server nickname. + + :return: a dictionary containing the command to run and instructions. + """ + return { + "type": "command", + "command": "rsconnect list" + } @mcp.tool() - async def deploy_shiny( - directory: str, - name: Optional[str] = None, - title: Optional[str] = None + def add_server( + server_url: str, + nickname: str, + api_key: str, ) -> Dict[str, Any]: - """Deploy a Shiny application to Posit Connect""" - - # Build the CLI command - args = ["deploy", "shiny", directory] + """ + Prompt user to add a Posit Connect server using rsconnect add. - if name: - args.extend(["--name", name]) + 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. - if title: - args.extend(["--title", title]) + :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}" - args.extend(["--server", server]) - args.extend(["--api-key", api_key]) + return { + "type": "command", + "command": command + } - try: - result = subprocess.run( - args, - check=True, - capture_output=True, - text=True + @mcp.tool() + def build_deploy_command( + application_type: str, + parameters: Dict[str, Any] + ) -> Dict[str, Any]: + """ + Build a deploy command for any content type using discovered parameters. + + This tool automatically builds the correct rsconnect deploy command + with proper parameter names and types based on the command type. + + :param application_type: the type of content to deploy (e.g., 'shiny', 'notebook', 'quarto', etc.) + :param parameters: dictionary of parameter names and their values. + :return: dictionary with the generated command and instructions. + """ + # use the deploy commands info discovered at startup + deploy_info = deploy_commands_info + + if application_type not in deploy_info["app_type"]: + available = ", ".join(deploy_info["app_type"].keys()) + raise ToolError( + f"Unknown application type '{application_type}'. " + f"Available types: {available}" ) - return { - "success": True, - "message": "Deployment completed successfully", - "stdout": result.stdout, - "stderr": result.stderr - } - except subprocess.CalledProcessError as e: - raise ToolError(f"Deployment failed with exit code {e.returncode}: {e.stderr}") - except Exception as e: - raise ToolError(f"Command failed with error: {e}") - - - mcp.tool(description="Get content logs from Posit Connect")(get_content_logs) - mcp.tool(description="List content from Posit Connect")(list_content) - mcp.tool(description="Get content item from Posit Connect")(get_content_item) + + cmd_parts = ["rsconnect", "deploy", application_type] + cmd_info = deploy_info["app_type"][application_type] + + arguments = [p for p in cmd_info["parameters"] if p.get("param_type") == "argument"] + options = [p for p in cmd_info["parameters"] if p.get("param_type") == "option"] + + # add arguments + for arg in arguments: + if arg["name"] in parameters: + value = parameters[arg["name"]] + if value is not None: + cmd_parts.append(shlex.quote(str(value))) + + # add options + for opt in options: + param_name = opt["name"] + if param_name not in parameters: + continue + + value = parameters[param_name] + if value is None: + continue + + cli_flag = max(opt.get("cli_flags", []), key=len) if opt.get("cli_flags") else f"--{param_name.replace('_', '-')}" + + if opt["type"] == "boolean": + if value: + cmd_parts.append(cli_flag) + elif opt["type"] == "array" and isinstance(value, list): + for item in value: + cmd_parts.extend([cli_flag, shlex.quote(str(item))]) + else: + cmd_parts.extend([cli_flag, shlex.quote(str(value))]) + + command = " ".join(cmd_parts) + + return { + "type": "command", + "command": command, + "application_type": application_type, + "description": cmd_info["description"] + } mcp.run() diff --git a/rsconnect/mcp_deploy_context.py b/rsconnect/mcp_deploy_context.py new file mode 100644 index 00000000..7794063d --- /dev/null +++ b/rsconnect/mcp_deploy_context.py @@ -0,0 +1,116 @@ +""" +Programmatically discover all parameters for rsconnect deploy commands. +This helps MCP tools understand exactly how to use `rsconnect deploy ...` +""" + +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: + info["default"] = param.default + + # required params + info["required"] = param.required + + return info + + +def discover_deploy_commands(cli_group: click.Group) -> Dict[str, Any]: + """Discover all deploy commands and their parameters.""" + + if "deploy" not in cli_group.commands: + return {"error": "deploy command group not found"} + + deploy_group = cli_group.commands["deploy"] + + if not isinstance(deploy_group, click.Group): + return {"error": "deploy is not a command group"} + + result = { + "group_name": "deploy", + "description": deploy_group.help or "Deploy content to Posit Connect, Posit Cloud, or shinyapps.io.", + "app_type": {} + } + + for cmd_name, cmd in deploy_group.commands.items(): + cmd_info = { + "name": cmd_name, + "description": cmd.help or cmd.short_help or f"Deploy {cmd_name}", + "short_help": cmd.short_help, + "parameters": [] + } + for param in cmd.params: + if isinstance(param, click.Context): + continue + + if param.name in ["verbose", "v"]: + continue + + param_info = extract_parameter_info(param) + cmd_info["parameters"].append(param_info) + + result["app_type"][cmd_name] = cmd_info + + return result + + +if __name__ == "__main__": + from rsconnect.main import cli + + deploy_commands_info = discover_deploy_commands(cli)["app_type"]["shiny"] + print(json.dumps(deploy_commands_info, indent=2)) From 92a7431556f658a82ad767f1d1f8ff110a017fd0 Mon Sep 17 00:00:00 2001 From: joshyam-k Date: Tue, 7 Oct 2025 08:51:05 -0700 Subject: [PATCH 3/7] fix ordering of args and opts --- rsconnect/main.py | 89 ++++++++------------------------- rsconnect/mcp_deploy_context.py | 25 ++++----- 2 files changed, 29 insertions(+), 85 deletions(-) diff --git a/rsconnect/main.py b/rsconnect/main.py index f312ec2c..67e88e79 100644 --- a/rsconnect/main.py +++ b/rsconnect/main.py @@ -405,6 +405,8 @@ def version(): @cli.command(help="Start the MCP server") def mcp_server(): + import subprocess + from fastmcp import FastMCP from fastmcp.exceptions import ToolError @@ -415,16 +417,13 @@ def mcp_server(): deploy_commands_info = discover_deploy_commands(cli) @mcp.tool() - def list_servers() -> Dict[str, Any]: - """ - Show the stored information about each known server nickname. - - :return: a dictionary containing the command to run and instructions. - """ - return { - "type": "command", - "command": "rsconnect list" - } + 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( @@ -452,71 +451,23 @@ def add_server( } @mcp.tool() - def build_deploy_command( - application_type: str, - parameters: Dict[str, Any] + def deploy_command_context( + content_type: str ) -> Dict[str, Any]: """ - Build a deploy command for any content type using discovered parameters. + Get the parameter schema for rsconnect deploy commands that should be executed via bash. - This tool automatically builds the correct rsconnect deploy command - with proper parameter names and types based on the command type. + Returns information about the parameters needed to construct an rsconnect deploy + command that can be executed in a bash shell. - :param application_type: the type of content to deploy (e.g., 'shiny', 'notebook', 'quarto', etc.) - :param parameters: dictionary of parameter names and their values. - :return: dictionary with the generated command and instructions. + :param content_type: the type of content (e.g., 'shiny', 'notebook', 'quarto') + :return: dictionary with command parameter schema and execution metadata """ - # use the deploy commands info discovered at startup - deploy_info = deploy_commands_info - - if application_type not in deploy_info["app_type"]: - available = ", ".join(deploy_info["app_type"].keys()) - raise ToolError( - f"Unknown application type '{application_type}'. " - f"Available types: {available}" - ) - - cmd_parts = ["rsconnect", "deploy", application_type] - cmd_info = deploy_info["app_type"][application_type] - - arguments = [p for p in cmd_info["parameters"] if p.get("param_type") == "argument"] - options = [p for p in cmd_info["parameters"] if p.get("param_type") == "option"] - - # add arguments - for arg in arguments: - if arg["name"] in parameters: - value = parameters[arg["name"]] - if value is not None: - cmd_parts.append(shlex.quote(str(value))) - - # add options - for opt in options: - param_name = opt["name"] - if param_name not in parameters: - continue - - value = parameters[param_name] - if value is None: - continue - - cli_flag = max(opt.get("cli_flags", []), key=len) if opt.get("cli_flags") else f"--{param_name.replace('_', '-')}" - - if opt["type"] == "boolean": - if value: - cmd_parts.append(cli_flag) - elif opt["type"] == "array" and isinstance(value, list): - for item in value: - cmd_parts.extend([cli_flag, shlex.quote(str(item))]) - else: - cmd_parts.extend([cli_flag, shlex.quote(str(value))]) - - command = " ".join(cmd_parts) - + ctx = deploy_commands_info["content_type"][content_type] return { - "type": "command", - "command": command, - "application_type": application_type, - "description": cmd_info["description"] + "context": ctx, + "command_usage": "rsconnect deploy COMMAND [OPTIONS] DIRECTORY [ARGS]...", + "shell": "bash" } mcp.run() diff --git a/rsconnect/mcp_deploy_context.py b/rsconnect/mcp_deploy_context.py index 7794063d..25863f9d 100644 --- a/rsconnect/mcp_deploy_context.py +++ b/rsconnect/mcp_deploy_context.py @@ -62,7 +62,10 @@ def extract_parameter_info(param: click.Parameter) -> Dict[str, Any]: # defaults (important to avoid noise in returned command) if param.default is not None and not param.is_flag: - info["default"] = param.default + 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 @@ -73,38 +76,28 @@ def extract_parameter_info(param: click.Parameter) -> Dict[str, Any]: def discover_deploy_commands(cli_group: click.Group) -> Dict[str, Any]: """Discover all deploy commands and their parameters.""" - if "deploy" not in cli_group.commands: - return {"error": "deploy command group not found"} - deploy_group = cli_group.commands["deploy"] - if not isinstance(deploy_group, click.Group): - return {"error": "deploy is not a command group"} - result = { "group_name": "deploy", - "description": deploy_group.help or "Deploy content to Posit Connect, Posit Cloud, or shinyapps.io.", - "app_type": {} + "description": deploy_group.help, + "content_type": {} } for cmd_name, cmd in deploy_group.commands.items(): cmd_info = { "name": cmd_name, - "description": cmd.help or cmd.short_help or f"Deploy {cmd_name}", - "short_help": cmd.short_help, + "description": cmd.help, "parameters": [] } for param in cmd.params: - if isinstance(param, click.Context): - continue - if param.name in ["verbose", "v"]: continue param_info = extract_parameter_info(param) cmd_info["parameters"].append(param_info) - result["app_type"][cmd_name] = cmd_info + result["content_type"][cmd_name] = cmd_info return result @@ -112,5 +105,5 @@ def discover_deploy_commands(cli_group: click.Group) -> Dict[str, Any]: if __name__ == "__main__": from rsconnect.main import cli - deploy_commands_info = discover_deploy_commands(cli)["app_type"]["shiny"] + deploy_commands_info = discover_deploy_commands(cli)["content_type"]["shiny"] print(json.dumps(deploy_commands_info, indent=2)) From cd9687feaaca099ae83a336d9d5b233101d6aa94 Mon Sep 17 00:00:00 2001 From: joshyam-k Date: Tue, 14 Oct 2025 11:32:23 -0700 Subject: [PATCH 4/7] generalize. tests written by claude --- pyproject.toml | 2 +- rsconnect/main.py | 117 ++++++++++--------- rsconnect/mcp_deploy_context.py | 60 ++++++---- tests/test_mcp_deploy_context.py | 192 +++++++++++++++++++++++++++++++ 4 files changed, 294 insertions(+), 77 deletions(-) create mode 100644 tests/test_mcp_deploy_context.py diff --git a/pyproject.toml b/pyproject.toml index e3a98388..c8667da6 100644 --- a/pyproject.toml +++ b/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"] diff --git a/rsconnect/main.py b/rsconnect/main.py index 67e88e79..ccba739b 100644 --- a/rsconnect/main.py +++ b/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( diff --git a/rsconnect/mcp_deploy_context.py b/rsconnect/mcp_deploy_context.py index 25863f9d..674b2273 100644 --- a/rsconnect/mcp_deploy_context.py +++ b/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)) diff --git a/tests/test_mcp_deploy_context.py b/tests/test_mcp_deploy_context.py new file mode 100644 index 00000000..8ffe8ec1 --- /dev/null +++ b/tests/test_mcp_deploy_context.py @@ -0,0 +1,192 @@ +"""Tests for MCP deploy context.""" + +from unittest import TestCase + +from rsconnect.main import cli +from rsconnect.mcp_deploy_context import discover_all_commands + + +class TestDiscoverAllCommands(TestCase): + def test_discover_rsconnect_cli(self): + result = discover_all_commands(cli) + + self.assertIn("commands", result) + self.assertIsNotNone(result["description"]) + + def test_top_level_commands(self): + result = discover_all_commands(cli) + + expected = ["version", "mcp-server", "add", "list", "remove", "details", "info", "deploy", "write-manifest", "content", "system", "bootstrap"] + for cmd in expected: + self.assertIn(cmd, result["commands"]) + + def test_deploy_is_command_group(self): + result = discover_all_commands(cli) + self.assertIn("commands", result["commands"]["deploy"]) + + def test_deploy_subcommands(self): + result = discover_all_commands(cli) + + deploy = result["commands"]["deploy"] + expected = ["notebook", "voila", "manifest", "quarto", "tensorflow", "html", "api", "flask", "fastapi", "dash", "streamlit", "bokeh", "shiny", "gradio"] + for subcmd in expected: + self.assertIn(subcmd, deploy["commands"]) + + def test_content_is_command_group(self): + result = discover_all_commands(cli) + self.assertIn("commands", result["commands"]["content"]) + + def test_content_subcommands(self): + result = discover_all_commands(cli) + + content = result["commands"]["content"] + expected = ["search", "describe", "download-bundle", "build"] + for subcmd in expected: + self.assertIn(subcmd, content["commands"]) + + def test_content_build_nested_group(self): + result = discover_all_commands(cli) + + build = result["commands"]["content"]["commands"]["build"] + self.assertIn("commands", build) + + expected = ["add", "rm", "ls", "history", "logs", "run"] + for subcmd in expected: + self.assertIn(subcmd, build["commands"]) + + def test_system_caches_nested_group(self): + result = discover_all_commands(cli) + + caches = result["commands"]["system"]["commands"]["caches"] + self.assertIn("commands", caches) + + expected = ["list", "delete"] + for subcmd in expected: + self.assertIn(subcmd, caches["commands"]) + + def test_write_manifest_is_command_group(self): + result = discover_all_commands(cli) + self.assertIn("commands", result["commands"]["write-manifest"]) + + def test_version_is_simple_command(self): + result = discover_all_commands(cli) + + version = result["commands"]["version"] + self.assertNotIn("commands", version) + self.assertIn("parameters", version) + + def test_mcp_server_command_exists(self): + result = discover_all_commands(cli) + self.assertIn("mcp-server", result["commands"]) + self.assertIn("parameters", result["commands"]["mcp-server"]) + + def test_deploy_notebook_has_parameters(self): + result = discover_all_commands(cli) + + notebook = result["commands"]["deploy"]["commands"]["notebook"] + param_names = [p["name"] for p in notebook["parameters"]] + + self.assertIn("file", param_names) + self.assertIn("name", param_names) + self.assertIn("server", param_names) + self.assertIn("api_key", param_names) + + def test_add_command_has_parameters(self): + result = discover_all_commands(cli) + + add = result["commands"]["add"] + param_names = [p["name"] for p in add["parameters"]] + + self.assertIn("name", param_names) + self.assertIn("server", param_names) + self.assertIn("api_key", param_names) + self.assertIn("insecure", param_names) + + def test_parameter_has_required_fields(self): + result = discover_all_commands(cli) + + for param in result["commands"]["add"]["parameters"]: + self.assertIn("name", param) + self.assertIn("param_type", param) + self.assertIn("required", param) + + if param["param_type"] == "option": + self.assertIn("cli_flags", param) + self.assertGreater(len(param["cli_flags"]), 0) + + def test_boolean_flags_identified(self): + result = discover_all_commands(cli) + + add = result["commands"]["add"] + insecure = next((p for p in add["parameters"] if p["name"] == "insecure"), None) + + self.assertIsNotNone(insecure) + self.assertEqual(insecure["type"], "boolean") + + def test_parameters_have_descriptions(self): + result = discover_all_commands(cli) + + add = result["commands"]["add"] + server = next((p for p in add["parameters"] if p["name"] == "server"), None) + + self.assertIsNotNone(server) + self.assertIn("description", server) + self.assertGreater(len(server["description"]), 0) + + def test_verbose_parameters_excluded(self): + result = discover_all_commands(cli) + + param_names = [p["name"] for p in result["commands"]["add"]["parameters"]] + self.assertNotIn("verbose", param_names) + self.assertNotIn("v", param_names) + + def test_all_commands_have_valid_structure(self): + def validate_command(cmd_info, path=""): + self.assertIn("name", cmd_info) + + if "commands" in cmd_info: + self.assertIsInstance(cmd_info["commands"], dict) + for subcmd_name, subcmd_info in cmd_info["commands"].items(): + validate_command(subcmd_info, f"{path}/{subcmd_name}") + else: + self.assertIn("parameters", cmd_info) + self.assertIsInstance(cmd_info["parameters"], list) + + for param in cmd_info["parameters"]: + self.assertIn("name", param) + self.assertIn("param_type", param) + self.assertIn("required", param) + + result = discover_all_commands(cli) + validate_command(result, "cli") + + def test_multiple_value_parameters(self): + result = discover_all_commands(cli) + + quarto = result["commands"]["deploy"]["commands"]["quarto"] + exclude = next((p for p in quarto["parameters"] if p["name"] == "exclude"), None) + + self.assertIsNotNone(exclude) + self.assertEqual(exclude["type"], "array") + + def test_required_parameters_marked(self): + result = discover_all_commands(cli) + + describe = result["commands"]["content"]["commands"]["describe"] + guid = next((p for p in describe["parameters"] if p["name"] == "guid"), None) + + self.assertIsNotNone(guid) + self.assertTrue(guid["required"]) + + def test_cli_flags_format(self): + result = discover_all_commands(cli) + + add = result["commands"]["add"] + name = next((p for p in add["parameters"] if p["name"] == "name"), None) + + self.assertIsNotNone(name) + self.assertIn("cli_flags", name) + self.assertGreater(len(name["cli_flags"]), 0) + + for flag in name["cli_flags"]: + self.assertTrue(flag.startswith("-")) From 63197bf74f8769ac0efc07ddbfdbd6c42e10ea2c Mon Sep 17 00:00:00 2001 From: joshyam-k Date: Tue, 14 Oct 2025 13:31:05 -0700 Subject: [PATCH 5/7] change fastmcp dependency approach --- pyproject.toml | 5 ++-- rsconnect/main.py | 27 ++++++++++---------- rsconnect/mcp_deploy_context.py | 18 ++++--------- tests/test_mcp_deploy_context.py | 43 ++++++++++++++++++++++++++++---- 4 files changed, 60 insertions(+), 33 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index c8667da6..b163ffcd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ 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.10" +requires-python = ">=3.8" dependencies = [ "typing-extensions>=4.8.0", @@ -14,7 +14,6 @@ dependencies = [ "pyjwt>=2.4.0", "click>=8.0.0", "toml>=0.10; python_version < '3.11'", - "fastmcp==2.12.4" ] dynamic = ["version"] @@ -38,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", diff --git a/rsconnect/main.py b/rsconnect/main.py index ccba739b..535ce182 100644 --- a/rsconnect/main.py +++ b/rsconnect/main.py @@ -404,13 +404,20 @@ def version(): @cli.command(help="Start the MCP server") def mcp_server(): - from fastmcp import FastMCP - from fastmcp.exceptions import ToolError + 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() @@ -431,10 +438,7 @@ def get_command_info( 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 - } + return {"error": "Command path cannot be empty", "available_commands": available_commands} current_info = all_commands_info current_path = [] @@ -451,11 +455,8 @@ def get_command_info( # 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 - } + 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) @@ -467,7 +468,7 @@ def get_command_info( "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." + "message": f"The '{' '.join(parts)}' command requires a subcommand.", } else: return { @@ -476,7 +477,7 @@ def get_command_info( "name": current_info.get("name", parts[-1]), "description": current_info.get("description"), "parameters": current_info.get("parameters", []), - "shell": "bash" + "shell": "bash", } except Exception as e: raise ToolError(f"Failed to retrieve command info: {str(e)}") diff --git a/rsconnect/mcp_deploy_context.py b/rsconnect/mcp_deploy_context.py index 674b2273..b47639bd 100644 --- a/rsconnect/mcp_deploy_context.py +++ b/rsconnect/mcp_deploy_context.py @@ -15,7 +15,7 @@ def extract_parameter_info(param: click.Parameter) -> 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('-', '_') + mcp_arg_name = max(param.opts, key=len).lstrip("-").replace("-", "_") info["name"] = mcp_arg_name info["cli_flags"] = param.opts info["param_type"] = "option" @@ -25,7 +25,7 @@ def extract_parameter_info(param: click.Parameter) -> Dict[str, Any]: info["param_type"] = "argument" # extract help text for added context - help_text = getattr(param, 'help', None) + help_text = getattr(param, "help", None) if help_text: info["description"] = help_text @@ -36,7 +36,7 @@ def extract_parameter_info(param: click.Parameter) -> Dict[str, Any]: info["default"] = param.default or False # choices - elif param.type and hasattr(param.type, 'choices'): + elif param.type and hasattr(param.type, "choices"): info["type"] = "string" info["choices"] = list(param.type.choices) @@ -75,11 +75,7 @@ def extract_parameter_info(param: click.Parameter) -> Dict[str, Any]: 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": [] - } + cmd_info = {"name": cmd.name, "description": cmd.help, "parameters": []} for param in cmd.params: if param.name in ["verbose", "v"]: @@ -93,11 +89,7 @@ def discover_single_command(cmd: click.Command) -> Dict[str, Any]: 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": {} - } + result = {"name": group.name, "description": group.help, "commands": {}} for cmd_name, cmd in group.commands.items(): if isinstance(cmd, click.Group): diff --git a/tests/test_mcp_deploy_context.py b/tests/test_mcp_deploy_context.py index 8ffe8ec1..39a98ce3 100644 --- a/tests/test_mcp_deploy_context.py +++ b/tests/test_mcp_deploy_context.py @@ -1,9 +1,14 @@ """Tests for MCP deploy context.""" -from unittest import TestCase +import pytest -from rsconnect.main import cli -from rsconnect.mcp_deploy_context import discover_all_commands +# Skip entire module if fastmcp is not available (requires Python 3.10+) +pytest.importorskip("fastmcp", reason="fastmcp library not installed (requires Python 3.10+)") + +from unittest import TestCase # noqa + +from rsconnect.main import cli # noqa +from rsconnect.mcp_deploy_context import discover_all_commands # noqa class TestDiscoverAllCommands(TestCase): @@ -16,7 +21,20 @@ def test_discover_rsconnect_cli(self): def test_top_level_commands(self): result = discover_all_commands(cli) - expected = ["version", "mcp-server", "add", "list", "remove", "details", "info", "deploy", "write-manifest", "content", "system", "bootstrap"] + expected = [ + "version", + "mcp-server", + "add", + "list", + "remove", + "details", + "info", + "deploy", + "write-manifest", + "content", + "system", + "bootstrap", + ] for cmd in expected: self.assertIn(cmd, result["commands"]) @@ -28,7 +46,22 @@ def test_deploy_subcommands(self): result = discover_all_commands(cli) deploy = result["commands"]["deploy"] - expected = ["notebook", "voila", "manifest", "quarto", "tensorflow", "html", "api", "flask", "fastapi", "dash", "streamlit", "bokeh", "shiny", "gradio"] + expected = [ + "notebook", + "voila", + "manifest", + "quarto", + "tensorflow", + "html", + "api", + "flask", + "fastapi", + "dash", + "streamlit", + "bokeh", + "shiny", + "gradio", + ] for subcmd in expected: self.assertIn(subcmd, deploy["commands"]) From 2b5d6c52dfe6a90c3d46e4ecd8c8f09c35a55ac9 Mon Sep 17 00:00:00 2001 From: joshyam-k Date: Wed, 15 Oct 2025 14:33:57 -0700 Subject: [PATCH 6/7] improve docstrings --- rsconnect/main.py | 29 +++++++++++++++++++---------- 1 file changed, 19 insertions(+), 10 deletions(-) diff --git a/rsconnect/main.py b/rsconnect/main.py index 535ce182..cc5b6504 100644 --- a/rsconnect/main.py +++ b/rsconnect/main.py @@ -420,19 +420,9 @@ def mcp_server(): 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() @@ -482,6 +472,25 @@ def get_command_info( 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() From fc384453a29893e0c1c887e5bff21010b1d5d475 Mon Sep 17 00:00:00 2001 From: joshyam-k Date: Fri, 17 Oct 2025 08:43:41 -0700 Subject: [PATCH 7/7] add better docs --- docs/commands/mcp-server.md | 3 +++ mkdocs.yml | 1 + rsconnect/main.py | 27 ++++++++++++++++++++++++++- 3 files changed, 30 insertions(+), 1 deletion(-) create mode 100644 docs/commands/mcp-server.md diff --git a/docs/commands/mcp-server.md b/docs/commands/mcp-server.md new file mode 100644 index 00000000..972c027d --- /dev/null +++ b/docs/commands/mcp-server.md @@ -0,0 +1,3 @@ +::: mkdocs-click + :module: rsconnect.main + :command: mcp_server diff --git a/mkdocs.yml b/mkdocs.yml index 96dd18fd..541da548 100644 --- a/mkdocs.yml +++ b/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: diff --git a/rsconnect/main.py b/rsconnect/main.py index cc5b6504..9f96b798 100644 --- a/rsconnect/main.py +++ b/rsconnect/main.py @@ -402,7 +402,32 @@ def version(): click.echo(VERSION) -@cli.command(help="Start the MCP server") +@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