Add Claude Desktop configuration via skyvern run mcp (#2045)

pyproject.toml

@@ -52,6 +52,7 @@ pyotp = "^2.9.0"
 asyncpg = "^0.30.0"
 json-repair = "^0.34.0"
 pypdf = "^5.1.0"
+fastmcp = "^0.4.1"
 
 [tool.poetry.group.dev.dependencies]
 isort = "^5.13.2"

skyvern/cli/commands.py

@@ -1,11 +1,17 @@
+import json
+import os
 import shutil
 import subprocess
 import time
 from typing import Optional
 
 import typer
+from click import Choice
+from dotenv import load_dotenv
 
-from skyvern.utils import migrate_db
+from skyvern.utils import detect_os, get_windows_appdata_roaming, migrate_db
+
+load_dotenv()
 
 app = typer.Typer()
 run_app = typer.Typer()
@@ -129,6 +135,300 @@ def migrate() -> None:
     migrate_db()
 
 
+def get_claude_config_path(host_system: str) -> str:
+    """Get the Claude Desktop config file path for the current OS."""
+    if host_system == "wsl":
+        roaming_path = get_windows_appdata_roaming()
+        if roaming_path is None:
+            raise RuntimeError("Could not locate Windows AppData\\Roaming path from WSL")
+        return os.path.join(str(roaming_path), "Claude", "claude_desktop_config.json")
+
+    base_paths = {
+        "darwin": ["~/Library/Application Support/Claude"],
+        "linux": ["~/.config/Claude", "~/.local/share/Claude", "~/Claude"],
+    }
+
+    if host_system == "darwin":
+        base_path = os.path.expanduser(base_paths["darwin"][0])
+        return os.path.join(base_path, "claude_desktop_config.json")
+
+    if host_system == "linux":
+        for path in base_paths["linux"]:
+            full_path = os.path.expanduser(path)
+            if os.path.exists(full_path):
+                return os.path.join(full_path, "claude_desktop_config.json")
+
+    raise Exception(f"Unsupported host system: {host_system}")
+
+
+def get_claude_command_config(
+    host_system: str, path_to_env: str, path_to_server: str, env_vars: str
+) -> tuple[str, list]:
+    """Get the command and arguments for Claude Desktop configuration."""
+    base_env_vars = f"{env_vars} ENABLE_OPENAI=true LOG_LEVEL=CRITICAL"
+    artifacts_path = os.path.join(os.path.abspath("./"), "artifacts")
+
+    if host_system == "wsl":
+        env_vars = f"{base_env_vars} ARTIFACT_STORAGE_PATH={artifacts_path} BROWSER_TYPE=chromium-headless"
+        return "wsl.exe", ["bash", "-c", f"{env_vars} {path_to_env} {path_to_server}"]
+
+    if host_system in ["linux", "darwin"]:
+        env_vars = f"{base_env_vars} ARTIFACT_STORAGE_PATH={artifacts_path}"
+        return path_to_env, [path_to_server]
+
+    raise Exception(f"Unsupported host system: {host_system}")
+
+
+def is_claude_desktop_installed(host_system: str) -> bool:
+    """Check if Claude Desktop is installed by looking for its config directory."""
+    try:
+        config_path = os.path.dirname(get_claude_config_path(host_system))
+        return os.path.exists(config_path)
+    except Exception:
+        return False
+
+
+def get_cursor_config_path(host_system: str) -> str:
+    """Get the Cursor config file path for the current OS."""
+    if host_system == "wsl":
+        roaming_path = get_windows_appdata_roaming()
+        if roaming_path is None:
+            raise RuntimeError("Could not locate Windows AppData\\Roaming path from WSL")
+        return os.path.join(str(roaming_path), ".cursor", "mcp.json")
+
+    # For both darwin and linux, use ~/.cursor/mcp.json
+    return os.path.expanduser("~/.cursor/mcp.json")
+
+
+def is_cursor_installed(host_system: str) -> bool:
+    """Check if Cursor is installed by looking for its config directory."""
+    try:
+        config_dir = os.path.expanduser("~/.cursor")
+        return os.path.exists(config_dir)
+    except Exception:
+        return False
+
+
+def setup_cursor_mcp(host_system: str, path_to_env: str, path_to_server: str, env_vars: str) -> None:
+    """Set up Cursor MCP configuration."""
+    if not is_cursor_installed(host_system):
+        print("Cursor is not installed. Skipping Cursor MCP setup.")
+        return
+
+    try:
+        path_cursor_config = get_cursor_config_path(host_system)
+    except Exception as e:
+        print(f"Error setting up Cursor: {e}")
+        return
+
+    # Get command configuration
+    try:
+        command, args = get_claude_command_config(host_system, path_to_env, path_to_server, env_vars)
+    except Exception as e:
+        print(f"Error configuring Cursor command: {e}")
+        return
+
+    # Create or update Cursor config file
+    os.makedirs(os.path.dirname(path_cursor_config), exist_ok=True)
+    config = {"Skyvern": {"command": command, "args": args}}
+
+    if os.path.exists(path_cursor_config):
+        try:
+            with open(path_cursor_config, "r") as f:
+                existing_config = json.load(f)
+                existing_config.update(config)
+                config = existing_config
+        except json.JSONDecodeError:
+            pass  # Use default config if file is corrupted
+
+    with open(path_cursor_config, "w") as f:
+        json.dump(config, f, indent=2)
+
+    print("Cursor MCP configuration updated successfully.")
+
+
+def setup_claude_desktop(host_system: str, path_to_env: str, path_to_server: str) -> None:
+    """Set up Claude Desktop configuration for Skyvern MCP."""
+    if not is_claude_desktop_installed(host_system):
+        print("Claude Desktop is not installed. Skipping MCP setup.")
+        return
+
+    # Get config file path
+    try:
+        path_claude_config = get_claude_config_path(host_system)
+    except Exception as e:
+        print(f"Error setting up Claude Desktop: {e}")
+        return
+
+    # Setup environment variables
+    env_vars = ""
+    for key in ["OPENAI_API_KEY", "ANTHROPIC_API_KEY"]:
+        value = os.getenv(key)
+        if value is None:
+            value = typer.prompt(f"Enter your {key}")
+        env_vars += f"{key}={value} "
+
+    # Get command configuration
+    try:
+        claude_command, claude_args = get_claude_command_config(host_system, path_to_env, path_to_server, env_vars)
+    except Exception as e:
+        print(f"Error configuring Claude Desktop command: {e}")
+        return
+
+    # Create or update Claude config file
+    os.makedirs(os.path.dirname(path_claude_config), exist_ok=True)
+    if not os.path.exists(path_claude_config):
+        with open(path_claude_config, "w") as f:
+            json.dump({"mcpServers": {}}, f, indent=2)
+
+    with open(path_claude_config, "r") as f:
+        claude_config = json.load(f)
+        claude_config["mcpServers"].pop("Skyvern", None)
+        claude_config["mcpServers"]["Skyvern"] = {"command": claude_command, "args": claude_args}
+
+    with open(path_claude_config, "w") as f:
+        json.dump(claude_config, f, indent=2)
+
+    print("Claude Desktop configuration updated successfully.")
+
+
+def get_mcp_server_url(deployment_type: str, host: str = "") -> str:
+    """Get the MCP server URL based on deployment type."""
+    if deployment_type in ["local", "cloud"]:
+        return os.path.join(os.path.abspath("./skyvern/mcp"), "server.py")
+    else:
+        raise ValueError(f"Invalid deployment type: {deployment_type}")
+
+
+def setup_mcp_config(host_system: str, deployment_type: str, host: str = "") -> tuple[str, str]:
+    """Set up MCP configuration based on deployment type."""
+    if deployment_type in ["local", "cloud"]:
+        # For local deployment, we need Python environment
+        python_path = shutil.which("python")
+        if python_path:
+            path_to_env = python_path
+        else:
+            path_to_env = typer.prompt("Enter the full path to your configured python environment")
+        return path_to_env, get_mcp_server_url(deployment_type)
+    else:
+        raise NotImplementedError()
+
+
+def get_command_config(host_system: str, command: str, target: str, env_vars: str) -> tuple[str, list]:
+    """Get the command and arguments for MCP configuration."""
+    base_env_vars = f"{env_vars} ENABLE_OPENAI=true LOG_LEVEL=CRITICAL"
+    artifacts_path = os.path.join(os.path.abspath("./"), "artifacts")
+
+    if host_system == "wsl":
+        env_vars = f"{base_env_vars} ARTIFACT_STORAGE_PATH={artifacts_path} BROWSER_TYPE=chromium-headless"
+        return "wsl.exe", ["bash", "-c", f"{env_vars} {command} {target}"]
+
+    if host_system in ["linux", "darwin"]:
+        env_vars = f"{base_env_vars} ARTIFACT_STORAGE_PATH={artifacts_path}"
+        if target.startswith("http"):
+            return command, ["-X", "POST", target]
+        return command, [target]
+
+    raise Exception(f"Unsupported host system: {host_system}")
+
+
 @run_app.command(name="mcp")
 def run_mcp() -> None:
-    pass
+    """Configure MCP for different Skyvern deployments."""
+    host_system = detect_os()
+
+    # Prompt for deployment type
+    deployment_types = ["local", "cloud"]
+    deployment_type = typer.prompt("Select Skyvern deployment type", type=Choice(deployment_types), default="local")
+
+    try:
+        command, target = setup_mcp_config(host_system, deployment_type)
+    except Exception as e:
+        print(f"Error setting up MCP configuration: {e}")
+        return
+
+    # Cloud deployment variables
+    env_vars = ""
+    if deployment_type == "cloud":
+        for key in ["SKYVERN_MCP_CLOUD_URL", "SKYVERN_MCP_API_KEY"]:
+            value = os.getenv(key)
+            if value is None:
+                value = typer.prompt(f"Enter your {key}")
+            env_vars += f"{key}={value} "
+
+    # Setup environment variables
+    for key in ["OPENAI_API_KEY", "ANTHROPIC_API_KEY"]:
+        value = os.getenv(key)
+        if value is None:
+            value = typer.prompt(f"Enter your {key}")
+        env_vars += f"{key}={value} "
+
+    # Configure both Claude Desktop and Cursor
+    success = False
+    success |= setup_claude_desktop_config(host_system, command, target, env_vars)
+    success |= setup_cursor_config(host_system, command, target, env_vars)
+
+    if not success:
+        print("Neither Claude Desktop nor Cursor is installed. Please install at least one of them.")
+
+
+def setup_claude_desktop_config(host_system: str, command: str, target: str, env_vars: str) -> bool:
+    """Set up Claude Desktop configuration with given command and args."""
+    if not is_claude_desktop_installed(host_system):
+        return False
+
+    try:
+        claude_command, claude_args = get_command_config(host_system, command, target, env_vars)
+        path_claude_config = get_claude_config_path(host_system)
+
+        os.makedirs(os.path.dirname(path_claude_config), exist_ok=True)
+        if not os.path.exists(path_claude_config):
+            with open(path_claude_config, "w") as f:
+                json.dump({"mcpServers": {}}, f, indent=2)
+
+        with open(path_claude_config, "r") as f:
+            claude_config = json.load(f)
+            claude_config["mcpServers"].pop("Skyvern", None)
+            claude_config["mcpServers"]["Skyvern"] = {"command": claude_command, "args": claude_args}
+
+        with open(path_claude_config, "w") as f:
+            json.dump(claude_config, f, indent=2)
+
+        print("Claude Desktop configuration updated successfully.")
+        return True
+
+    except Exception as e:
+        print(f"Error configuring Claude Desktop: {e}")
+        return False
+
+
+def setup_cursor_config(host_system: str, command: str, target: str, env_vars: str) -> bool:
+    """Set up Cursor configuration with given command and args."""
+    if not is_cursor_installed(host_system):
+        return False
+
+    try:
+        cursor_command, cursor_args = get_command_config(host_system, command, target, env_vars)
+        path_cursor_config = get_cursor_config_path(host_system)
+
+        os.makedirs(os.path.dirname(path_cursor_config), exist_ok=True)
+        config = {"Skyvern": {"command": cursor_command, "args": cursor_args}}
+
+        if os.path.exists(path_cursor_config):
+            try:
+                with open(path_cursor_config, "r") as f:
+                    existing_config = json.load(f)
+                    existing_config.update(config)
+                    config = existing_config
+            except json.JSONDecodeError:
+                pass
+
+        with open(path_cursor_config, "w") as f:
+            json.dump(config, f, indent=2)
+
+        print(f"Cursor configuration updated successfully at {path_cursor_config}")
+        return True
+
+    except Exception as e:
+        print(f"Error configuring Cursor: {e}")
+        return False

skyvern/mcp/server.py

@@ -0,0 +1,30 @@
+import os
+
+from mcp.server.fastmcp import FastMCP
+
+from skyvern.agent import SkyvernAgent
+
+mcp = FastMCP("Skyvern")
+
+if "SKYVERN_MCP_CLOUD_URL" in os.environ and "SKYVERN_MCP_API_KEY" in os.environ:
+    skyvern_agent = SkyvernAgent(
+        base_url=os.environ.get("SKYVERN_MCP_CLOUD_URL"), api_key=os.environ.get("SKYVERN_MCP_API_KEY")
+    )
+else:
+    skyvern_agent = SkyvernAgent()
+
+
+@mcp.tool()
+async def skyvern_run_task(prompt: str, url: str) -> str:
+    """Browse the internet using a browser to achieve a user goal.
+
+    Args:
+        prompt: brief description of what the user wants to accomplish
+        url: the target website for the user goal
+    """
+    res = await skyvern_agent.run_task(prompt=prompt, url=url)
+    return res.model_dump()["output"]
+
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")

skyvern/utils/__init__.py

@@ -1,3 +1,8 @@
+import platform
+import subprocess
+from pathlib import Path
+from typing import Optional
+
 from alembic import command
 from alembic.config import Config
 from skyvern.constants import REPO_ROOT_DIR
@@ -8,3 +13,50 @@ def migrate_db() -> None:
     path = f"{REPO_ROOT_DIR}/alembic"
     alembic_cfg.set_main_option("script_location", path)
     command.upgrade(alembic_cfg, "head")
+
+
+def detect_os() -> str:
+    """
+    Detects the operating system.
+
+    Returns:
+        str: The name of the OS in lowercase.
+             Returns 'wsl' for Windows Subsystem for Linux,
+             'linux' for native Linux,
+             or the lowercase name of other platforms (e.g., 'windows', 'darwin').
+    """
+    system = platform.system()
+    if system == "Linux":
+        try:
+            with open("/proc/version", "r") as f:
+                version_info = f.read().lower()
+                if "microsoft" in version_info:
+                    return "wsl"
+        except Exception:
+            pass
+        return "linux"
+    else:
+        return system.lower()
+
+
+def get_windows_appdata_roaming() -> Optional[Path]:
+    """
+    Retrieves the Windows 'AppData\\Roaming' directory path from WSL.
+
+    Returns:
+        Optional[Path]: A Path object representing the translated Linux-style path
+                        to the Windows AppData\\Roaming folder, or None if retrieval fails.
+    """
+    try:
+        output = (
+            subprocess.check_output(
+                ["powershell.exe", "-NoProfile", "-Command", "[Environment]::GetFolderPath('ApplicationData')"],
+                stderr=subprocess.DEVNULL,
+            )
+            .decode("utf-8")
+            .strip()
+        )
+        linux_path = "/mnt/" + output[0].lower() + output[2:].replace("\\", "/")
+        return Path(linux_path)
+    except Exception:
+        return None