initial command

Author: joshyam-k

pyproject.toml

@@ -5,15 +5,17 @@ description = "The Posit Connect command-line interface."
 authors = [{ name = "Posit, PBC", email = "[email protected]" }]
 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",
     "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",
+    "posit-sdk"
 ]
 
 dynamic = ["version"]

rsconnect/main.py

@@ -392,6 +392,62 @@ 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
+
+    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.json()
+        except Exception as e:
+            raise ToolError(f"Failed to get content: {e}")
+
+
+    mcp = FastMCP("Connect MCP")
+
+    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