mirror of
https://github.com/Skyvern-AI/skyvern.git
synced 2026-04-28 03:30:10 +00:00
Add MCP API key switcher for local clients [SKY-8218] (#5060)
This commit is contained in:
parent
fa892f152f
commit
af91183d75
9 changed files with 1567 additions and 38 deletions
|
|
@ -151,10 +151,16 @@ skyvern setup claude-code --global # Force global Claude Code config
|
|||
skyvern setup claude # Register with Claude Desktop
|
||||
skyvern setup cursor # Register with Cursor
|
||||
skyvern setup windsurf # Register with Windsurf
|
||||
skyvern mcp switch # Interactively switch existing Skyvern MCP configs
|
||||
skyvern mcp switch --dry-run # Preview changes without writing files
|
||||
skyvern mcp profile list # List saved switch sources
|
||||
skyvern mcp profile save work-prod --api-key YOUR_KEY --base-url https://api.skyvern.com
|
||||
```
|
||||
|
||||
For the local self-hosted path, `skyvern quickstart` or `skyvern init` can also configure Claude Code during the interactive MCP step. In a project directory that flow writes `.mcp.json`, installs `.claude/skills/qa`, and keeps the MCP connection fully local for localhost testing.
|
||||
|
||||
`skyvern mcp switch` updates existing Skyvern entries in Claude Code, Claude Desktop, Cursor, Windsurf, and Codex configs. It lets you choose a source from env, saved profiles, existing configs already on disk, or manual entry; creates backups before writing; and preserves the config's current transport shape. If a tool has no Skyvern entry yet, run `skyvern setup` first.
|
||||
|
||||
### Other
|
||||
|
||||
```bash
|
||||
|
|
|
|||
|
|
@ -133,6 +133,31 @@ x-api-key = "SKYVERN_API_KEY"
|
|||
|
||||
That's it — no Python, no `pip install`, no local server. Your AI assistant connects directly to Skyvern Cloud over HTTPS. Claude Desktop uses [mcp-remote](https://www.npmjs.com/package/mcp-remote) to bridge the connection (requires Node.js >= 20).
|
||||
|
||||
## Switch an existing MCP config
|
||||
|
||||
If you already have a Skyvern MCP entry and just need to switch to another API key or base URL, use the CLI instead of hand-editing JSON or TOML:
|
||||
|
||||
```bash
|
||||
pip install skyvern
|
||||
skyvern mcp switch --dry-run
|
||||
skyvern mcp switch
|
||||
```
|
||||
|
||||
The switcher scans supported local config files created by `skyvern setup` or manual setup, shows which ones already contain a Skyvern entry, lets you choose which apps to update, and writes a backup before changing anything.
|
||||
|
||||
Today it can update:
|
||||
|
||||
- Claude Code project `.mcp.json`
|
||||
- Claude Code global `~/.claude.json`
|
||||
- Claude Desktop JSON config
|
||||
- Cursor `~/.cursor/mcp.json`
|
||||
- Windsurf `~/.codeium/windsurf/mcp_config.json`
|
||||
- Codex `~/.codex/config.toml`
|
||||
|
||||
`skyvern mcp switch` preserves the existing transport shape. Remote configs keep using `https://.../mcp/`; local stdio configs keep `SKYVERN_BASE_URL` and `SKYVERN_API_KEY` in the launched process environment. After switching, restart your AI client.
|
||||
|
||||
If a config file exists but does not already contain a Skyvern entry, run `skyvern setup` or use the manual setup snippets on this page first.
|
||||
|
||||
## Alternative: Local mode (self-hosted)
|
||||
|
||||
Use this if you are self-hosting Skyvern and want the MCP server to talk to your own instance instead of Skyvern Cloud. This runs a lightweight Python process on your machine that connects to your local Skyvern server. Requires Python 3.11, 3.12, or 3.13.
|
||||
|
|
@ -278,7 +303,7 @@ Replace `/usr/bin/python3` with the output of `which python3`. For Skyvern Cloud
|
|||
<Accordion title="Invalid API key or 401 errors">
|
||||
Double-check that your API key is correct. You can find or regenerate it at [Settings](https://app.skyvern.com). Make sure there are no extra spaces or newlines when pasting the key.
|
||||
|
||||
If you recently regenerated your API key, update it in your MCP config and restart your AI client.
|
||||
If you recently regenerated your API key, run `skyvern mcp switch` if you installed the CLI, or update the MCP config manually, then restart your AI client.
|
||||
</Accordion>
|
||||
|
||||
<Accordion title="Tools not responding or timing out">
|
||||
|
|
|
|||
|
|
@ -161,6 +161,10 @@ skyvern setup claude-code --skip-skills # MCP only, no skills
|
|||
skyvern setup claude # Configure Claude Desktop only
|
||||
skyvern setup cursor # Configure Cursor only
|
||||
skyvern setup windsurf # Configure Windsurf only
|
||||
skyvern mcp switch # Interactively switch existing Skyvern MCP configs
|
||||
skyvern mcp switch --dry-run # Preview changes without writing files
|
||||
skyvern mcp profile list # List saved switch sources
|
||||
skyvern mcp profile save work-prod --api-key YOUR_KEY --base-url https://api.skyvern.com
|
||||
```
|
||||
|
||||
Use bare `skyvern setup` to configure everything at once. The per-tool subcommands are for when you only want to target one tool.
|
||||
|
|
@ -169,6 +173,8 @@ Use bare `skyvern setup` to configure everything at once. The per-tool subcomman
|
|||
|
||||
For Claude Desktop on macOS or Windows, the recommended no-Node path is the downloadable `.mcpb` bundle in [MCP Server](/integrations/mcp). `skyvern setup claude` still writes the manual `mcp-remote` JSON bridge, so remote mode requires Node.js unless you use the bundle.
|
||||
|
||||
`skyvern mcp switch` updates existing Skyvern entries in Claude Code, Claude Desktop, Cursor, Windsurf, and Codex configs. It lets you choose a source from env, saved profiles, existing configs already on disk, or manual entry; creates backups before writing; and preserves the config's current transport shape. If a tool has no Skyvern entry yet, run `skyvern setup` first.
|
||||
|
||||
### Self-hosted setup
|
||||
|
||||
These commands are for running Skyvern on your own infrastructure. If you're using Skyvern Cloud, you don't need them — `skyvern login` + `skyvern setup` is all you need.
|
||||
|
|
|
|||
|
|
@ -150,6 +150,31 @@ That's it — no Python, no `pip install`, no local server. Your AI assistant co
|
|||
Prefer a CLI? For Skyvern Cloud, `pip install skyvern` then run `skyvern setup`. For the local self-hosted path, run `skyvern quickstart` or `skyvern init` and choose Claude Code during the MCP step.
|
||||
</Tip>
|
||||
|
||||
## Switch an existing MCP config
|
||||
|
||||
If you already have a Skyvern MCP entry and just need to switch to another API key or base URL, use the CLI instead of hand-editing JSON or TOML:
|
||||
|
||||
```bash
|
||||
pip install skyvern
|
||||
skyvern mcp switch --dry-run
|
||||
skyvern mcp switch
|
||||
```
|
||||
|
||||
The switcher scans supported local config files created by `skyvern setup` or manual setup, shows which ones already contain a Skyvern entry, lets you choose which apps to update, and writes a backup before changing anything.
|
||||
|
||||
Today it can update:
|
||||
|
||||
- Claude Code project `.mcp.json`
|
||||
- Claude Code global `~/.claude.json`
|
||||
- Claude Desktop JSON config
|
||||
- Cursor `~/.cursor/mcp.json`
|
||||
- Windsurf `~/.codeium/windsurf/mcp_config.json`
|
||||
- Codex `~/.codex/config.toml`
|
||||
|
||||
`skyvern mcp switch` preserves the existing transport shape. Remote configs keep using `https://.../mcp/`; local stdio configs keep `SKYVERN_BASE_URL` and `SKYVERN_API_KEY` in the launched process environment. After switching, restart your AI client.
|
||||
|
||||
If a config file exists but does not already contain a Skyvern entry, run `skyvern setup` or use the manual setup snippets on this page first.
|
||||
|
||||
## Alternative: Local mode (self-hosted)
|
||||
|
||||
Use this if you are self-hosting Skyvern and want the MCP server to talk to your own instance instead of Skyvern Cloud. This runs a lightweight Python process on your machine that connects to your local Skyvern server.
|
||||
|
|
@ -351,7 +376,7 @@ The `/qa` skill uses `skyvern_evaluate` for fast DOM assertions (~10ms each) and
|
|||
<Accordion title="Invalid API key or 401 errors">
|
||||
Double-check that your API key is correct. You can find or regenerate it at [Settings](https://app.skyvern.com). Make sure there are no extra spaces or newlines when pasting the key.
|
||||
|
||||
If you recently regenerated your API key, update it in your MCP config and restart your AI client.
|
||||
If you recently regenerated your API key, run `skyvern mcp switch` if you installed the CLI, or update the MCP config manually, then restart your AI client.
|
||||
</Accordion>
|
||||
|
||||
<Accordion title="Tools not responding or timing out">
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ from ..credential import credential_app
|
|||
from ..credentials import credentials_app
|
||||
from ..docs import docs_app
|
||||
from ..init_command import init_browser, init_env
|
||||
from ..mcp_commands import mcp_app
|
||||
from ..quickstart import quickstart_app
|
||||
from ..run_commands import run_app
|
||||
from ..setup_commands import setup_app
|
||||
|
|
@ -89,6 +90,7 @@ cli_app.command(name="signup", hidden=True)(signup_command) # backwards compat
|
|||
|
||||
# Browser automation commands
|
||||
cli_app.add_typer(browser_app, name="browser", help="Browser automation commands.")
|
||||
cli_app.add_typer(mcp_app, name="mcp", help="Switch local MCP client configs and manage optional saved profiles.")
|
||||
cli_app.add_typer(skill_app, name="skill", help="Manage bundled skill reference files.")
|
||||
cli_app.add_typer(setup_app, name="setup", help="Register Skyvern MCP with AI coding tools.")
|
||||
|
||||
|
|
|
|||
836
skyvern/cli/mcp_commands.py
Normal file
836
skyvern/cli/mcp_commands.py
Normal file
|
|
@ -0,0 +1,836 @@
|
|||
"""CLI commands for switching local MCP client configs and managing optional saved profiles."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
from urllib.parse import urlparse, urlunparse
|
||||
|
||||
import toml
|
||||
import typer
|
||||
from rich.panel import Panel
|
||||
from rich.prompt import Confirm, Prompt
|
||||
from rich.syntax import Syntax
|
||||
from rich.table import Table
|
||||
|
||||
from .console import console
|
||||
from .setup_commands import (
|
||||
_backup_config_path,
|
||||
_claude_code_global_config_path,
|
||||
_claude_code_project_config_path,
|
||||
_claude_desktop_config_path,
|
||||
_codex_config_path,
|
||||
_cursor_config_path,
|
||||
_find_server_key,
|
||||
_get_env_credentials,
|
||||
_load_mcp_config,
|
||||
_mask_key,
|
||||
_mask_secrets,
|
||||
_windsurf_config_path,
|
||||
_write_mcp_config,
|
||||
)
|
||||
|
||||
mcp_app = typer.Typer(help="Manage local MCP configs and optional saved Skyvern profiles.", no_args_is_help=True)
|
||||
profile_app = typer.Typer(help="Manage saved Skyvern MCP profiles.", no_args_is_help=True)
|
||||
mcp_app.add_typer(profile_app, name="profile")
|
||||
|
||||
_MANUAL_SOURCE_NAME = "Manual entry"
|
||||
_DEFAULT_BASE_URL = "https://api.skyvern.com"
|
||||
_PROFILE_FILENAME_RE = re.compile(r"[^A-Za-z0-9._-]+")
|
||||
_ANSI_ESCAPE_RE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
|
||||
_CONFIG_FORMAT_JSON = "json"
|
||||
_CONFIG_FORMAT_CODEX = "codex_toml"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MCPProfile:
|
||||
name: str
|
||||
api_key: str
|
||||
base_url: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SwitchTarget:
|
||||
name: str
|
||||
config_path: Path
|
||||
entry_key: str | None
|
||||
entry: dict | None
|
||||
config_format: str = _CONFIG_FORMAT_JSON
|
||||
error: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProfileChoice:
|
||||
label: str
|
||||
profile: MCPProfile
|
||||
sources: list[str]
|
||||
saved_name: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SwitchTargetSpec:
|
||||
name: str
|
||||
config_path_fn: Callable[[], Path]
|
||||
config_format: str = _CONFIG_FORMAT_JSON
|
||||
|
||||
|
||||
def _sanitize_prompt_response(value: str) -> str:
|
||||
cleaned = _ANSI_ESCAPE_RE.sub("", value)
|
||||
return "".join(ch for ch in cleaned if ch.isprintable()).strip()
|
||||
|
||||
|
||||
def _prompt_text(prompt: str, *, default: str | None = None, password: bool = False) -> str:
|
||||
return _sanitize_prompt_response(Prompt.ask(prompt, default=default, password=password))
|
||||
|
||||
|
||||
def _ask_choice(prompt: str, *, choices: list[str], default: str | None = None) -> str:
|
||||
allowed = set(choices)
|
||||
while True:
|
||||
response = _prompt_text(prompt, default=default)
|
||||
if response in allowed:
|
||||
return response
|
||||
console.print(f"[red]Invalid choice. Enter one of: {', '.join(choices)}[/red]")
|
||||
|
||||
|
||||
def _profile_store_dir() -> Path:
|
||||
if platform.system() == "Windows":
|
||||
appdata = os.environ.get("APPDATA")
|
||||
if appdata:
|
||||
return Path(appdata) / "skyvern" / "mcp-profiles"
|
||||
|
||||
xdg_config_home = os.environ.get("XDG_CONFIG_HOME")
|
||||
if xdg_config_home:
|
||||
return Path(xdg_config_home) / "skyvern" / "mcp-profiles"
|
||||
|
||||
return Path.home() / ".config" / "skyvern" / "mcp-profiles"
|
||||
|
||||
|
||||
def _profile_slug(name: str) -> str:
|
||||
slug = _PROFILE_FILENAME_RE.sub("-", name.strip()).strip("-.").lower()
|
||||
if not slug:
|
||||
raise typer.BadParameter("Profile name must include at least one letter or number.")
|
||||
return slug
|
||||
|
||||
|
||||
def _profile_path(name: str) -> Path:
|
||||
return _profile_store_dir() / f"{_profile_slug(name)}.json"
|
||||
|
||||
|
||||
def _normalize_base_url(base_url: str) -> str:
|
||||
normalized = base_url.strip()
|
||||
if not normalized:
|
||||
raise typer.BadParameter("Base URL cannot be empty.")
|
||||
|
||||
parsed = urlparse(normalized)
|
||||
if parsed.scheme not in ("http", "https") or not parsed.netloc:
|
||||
raise typer.BadParameter("Base URL must start with http:// or https://")
|
||||
|
||||
path = parsed.path.rstrip("/")
|
||||
if path.endswith("/mcp"):
|
||||
path = path[: -len("/mcp")]
|
||||
|
||||
return urlunparse((parsed.scheme, parsed.netloc, path, "", "", "")).rstrip("/")
|
||||
|
||||
|
||||
def _profile_to_mcp_url(base_url: str) -> str:
|
||||
return f"{_normalize_base_url(base_url)}/mcp/"
|
||||
|
||||
|
||||
def _build_profile(name: str, api_key: str, base_url: str) -> MCPProfile:
|
||||
clean_name = name.strip()
|
||||
clean_key = api_key.strip()
|
||||
if not clean_name:
|
||||
raise typer.BadParameter("Profile name cannot be empty.")
|
||||
if not clean_key:
|
||||
raise typer.BadParameter("API key cannot be empty.")
|
||||
|
||||
return MCPProfile(
|
||||
name=clean_name,
|
||||
api_key=clean_key,
|
||||
base_url=_normalize_base_url(base_url),
|
||||
)
|
||||
|
||||
|
||||
def _load_profile_from_path(path: Path) -> MCPProfile:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError(f"{path} must contain a JSON object.")
|
||||
|
||||
name = data.get("name", "")
|
||||
api_key = data.get("api_key", "")
|
||||
base_url = data.get("base_url", "")
|
||||
for field_name, value in (("name", name), ("api_key", api_key), ("base_url", base_url)):
|
||||
if not isinstance(value, str):
|
||||
raise ValueError(f"{path} field '{field_name}' must be a string.")
|
||||
|
||||
return _build_profile(
|
||||
name,
|
||||
api_key,
|
||||
base_url,
|
||||
)
|
||||
|
||||
|
||||
def _list_profiles() -> list[MCPProfile]:
|
||||
store_dir = _profile_store_dir()
|
||||
if not store_dir.exists():
|
||||
return []
|
||||
|
||||
profiles: list[MCPProfile] = []
|
||||
for path in sorted(store_dir.glob("*.json")):
|
||||
try:
|
||||
profiles.append(_load_profile_from_path(path))
|
||||
except Exception as exc:
|
||||
console.print(f"[yellow]Skipping invalid profile {path.name}: {exc}[/yellow]")
|
||||
|
||||
return sorted(profiles, key=lambda profile: profile.name.lower())
|
||||
|
||||
|
||||
def _load_profile(name: str) -> MCPProfile:
|
||||
path = _profile_path(name)
|
||||
if not path.exists():
|
||||
console.print(f"[red]No saved MCP profile named '{name}'.[/red]")
|
||||
raise typer.Exit(code=1)
|
||||
try:
|
||||
return _load_profile_from_path(path)
|
||||
except Exception as exc:
|
||||
console.print(f"[red]Failed to load profile '{name}': {exc}[/red]")
|
||||
raise typer.Exit(code=1) from exc
|
||||
|
||||
|
||||
def _save_profile(profile: MCPProfile, overwrite: bool = False) -> Path:
|
||||
path = _profile_path(profile.name)
|
||||
if path.exists() and not overwrite:
|
||||
console.print(
|
||||
f"[red]Profile '{profile.name}' already exists at {path}. Re-run with --overwrite to replace it.[/red]"
|
||||
)
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if os.name != "nt":
|
||||
try:
|
||||
path.parent.chmod(0o700)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
payload = {
|
||||
"name": profile.name,
|
||||
"api_key": profile.api_key,
|
||||
"base_url": profile.base_url,
|
||||
}
|
||||
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
|
||||
if os.name != "nt":
|
||||
try:
|
||||
path.chmod(0o600)
|
||||
except OSError:
|
||||
pass
|
||||
return path
|
||||
|
||||
|
||||
def _prompt_for_manual_source() -> MCPProfile:
|
||||
env_key, env_base_url = _get_env_credentials()
|
||||
default_base_url = env_base_url or _DEFAULT_BASE_URL
|
||||
|
||||
api_key = _prompt_text("Skyvern API key", password=True)
|
||||
base_url = _prompt_text("Skyvern base URL (remote configs derive /mcp automatically)", default=default_base_url)
|
||||
return _build_profile(_MANUAL_SOURCE_NAME, api_key, base_url)
|
||||
|
||||
|
||||
def _extract_entry_base_url(entry: dict) -> str:
|
||||
location = _entry_location(entry).strip()
|
||||
if not location:
|
||||
return ""
|
||||
|
||||
try:
|
||||
return _normalize_base_url(location)
|
||||
except typer.BadParameter:
|
||||
return ""
|
||||
|
||||
|
||||
def _server_block_key(config_format: str) -> str:
|
||||
return "mcp_servers" if config_format == _CONFIG_FORMAT_CODEX else "mcpServers"
|
||||
|
||||
|
||||
def _load_codex_config(config_path: Path) -> tuple[dict | None, str | None]:
|
||||
if not config_path.exists():
|
||||
return {}, None
|
||||
|
||||
try:
|
||||
existing = toml.loads(config_path.read_text(encoding="utf-8"))
|
||||
except toml.TomlDecodeError:
|
||||
return None, f"Cannot parse {config_path}. Fix the TOML and re-run."
|
||||
|
||||
if not isinstance(existing, dict):
|
||||
return None, f"{config_path} must contain a top-level TOML table."
|
||||
|
||||
servers = existing.get("mcp_servers")
|
||||
if servers is not None and not isinstance(servers, dict):
|
||||
return None, f"{config_path} has invalid `mcp_servers`; expected a TOML table."
|
||||
|
||||
return existing, None
|
||||
|
||||
|
||||
def _load_switch_config(config_path: Path, config_format: str) -> tuple[dict | None, str | None]:
|
||||
if config_format == _CONFIG_FORMAT_CODEX:
|
||||
return _load_codex_config(config_path)
|
||||
return _load_mcp_config(config_path)
|
||||
|
||||
|
||||
def _write_codex_config(config_path: Path, config: dict, create_backup: bool = True) -> Path | None:
|
||||
backup_path: Path | None = None
|
||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if create_backup and config_path.exists():
|
||||
backup_path = _backup_config_path(config_path)
|
||||
shutil.copy2(config_path, backup_path)
|
||||
|
||||
content = toml.dumps(config)
|
||||
if not content.endswith("\n"):
|
||||
content += "\n"
|
||||
config_path.write_text(content, encoding="utf-8")
|
||||
return backup_path
|
||||
|
||||
|
||||
def _write_switch_config(config_path: Path, config: dict, config_format: str) -> Path | None:
|
||||
if config_format == _CONFIG_FORMAT_CODEX:
|
||||
return _write_codex_config(config_path, config, create_backup=True)
|
||||
return _write_mcp_config(config_path, config, create_backup=True)
|
||||
|
||||
|
||||
def _switch_target_specs() -> list[SwitchTargetSpec]:
|
||||
return [
|
||||
SwitchTargetSpec("Claude Code (global)", _claude_code_global_config_path),
|
||||
SwitchTargetSpec("Claude Code (project)", _claude_code_project_config_path),
|
||||
SwitchTargetSpec("Claude Desktop", _claude_desktop_config_path),
|
||||
SwitchTargetSpec("Cursor", _cursor_config_path),
|
||||
SwitchTargetSpec("Windsurf", _windsurf_config_path),
|
||||
SwitchTargetSpec("Codex", _codex_config_path, config_format=_CONFIG_FORMAT_CODEX),
|
||||
]
|
||||
|
||||
|
||||
def _entry_kind(entry: dict | None) -> str:
|
||||
if not entry:
|
||||
return "missing"
|
||||
|
||||
command_name = Path(str(entry.get("command", ""))).name.lower()
|
||||
args = entry.get("args", [])
|
||||
if command_name == "npx" and isinstance(args, list) and args and args[0] == "mcp-remote":
|
||||
return "mcp-remote bridge"
|
||||
|
||||
if isinstance(entry.get("env"), dict):
|
||||
return "local stdio"
|
||||
|
||||
if entry.get("type") == "http" or "url" in entry or isinstance(entry.get("http_headers"), dict):
|
||||
return "remote http"
|
||||
|
||||
return "unsupported"
|
||||
|
||||
|
||||
def _extract_entry_api_key(entry: dict) -> str:
|
||||
headers = entry.get("headers", {})
|
||||
if isinstance(headers, dict):
|
||||
api_key = headers.get("x-api-key")
|
||||
if isinstance(api_key, str):
|
||||
return api_key
|
||||
|
||||
http_headers = entry.get("http_headers", {})
|
||||
if isinstance(http_headers, dict):
|
||||
api_key = http_headers.get("x-api-key")
|
||||
if isinstance(api_key, str):
|
||||
return api_key
|
||||
|
||||
env = entry.get("env", {})
|
||||
if isinstance(env, dict):
|
||||
api_key = env.get("SKYVERN_API_KEY")
|
||||
if isinstance(api_key, str):
|
||||
return api_key
|
||||
|
||||
args = entry.get("args", [])
|
||||
if isinstance(args, list):
|
||||
for arg in args:
|
||||
if isinstance(arg, str) and arg.startswith("x-api-key:"):
|
||||
return arg[len("x-api-key:") :]
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
def _append_profile_choice(
|
||||
choices_by_key: dict[tuple[str, str], ProfileChoice],
|
||||
*,
|
||||
label: str,
|
||||
profile: MCPProfile,
|
||||
source: str,
|
||||
saved_name: str | None = None,
|
||||
) -> None:
|
||||
key = (profile.api_key, profile.base_url)
|
||||
existing = choices_by_key.get(key)
|
||||
if existing is None:
|
||||
choices_by_key[key] = ProfileChoice(
|
||||
label=label,
|
||||
profile=profile,
|
||||
sources=[source],
|
||||
saved_name=saved_name,
|
||||
)
|
||||
return
|
||||
|
||||
if source not in existing.sources:
|
||||
existing.sources.append(source)
|
||||
|
||||
if saved_name and existing.saved_name is None:
|
||||
existing.label = label
|
||||
existing.saved_name = saved_name
|
||||
|
||||
|
||||
def _profile_from_target(target: SwitchTarget) -> MCPProfile | None:
|
||||
if target.entry is None:
|
||||
return None
|
||||
|
||||
api_key = _extract_entry_api_key(target.entry).strip()
|
||||
base_url = _extract_entry_base_url(target.entry)
|
||||
if not api_key or not base_url:
|
||||
return None
|
||||
|
||||
return MCPProfile(name=f"{target.name} current", api_key=api_key, base_url=base_url)
|
||||
|
||||
|
||||
def _collect_profile_choices(discovered: list[SwitchTarget]) -> list[ProfileChoice]:
|
||||
choices_by_key: dict[tuple[str, str], ProfileChoice] = {}
|
||||
|
||||
env_key, env_base_url = _get_env_credentials()
|
||||
if env_key:
|
||||
env_profile = _build_profile("Current environment", env_key, env_base_url or _DEFAULT_BASE_URL)
|
||||
_append_profile_choice(
|
||||
choices_by_key,
|
||||
label="Current environment",
|
||||
profile=env_profile,
|
||||
source="env/.env",
|
||||
)
|
||||
|
||||
for saved_profile in _list_profiles():
|
||||
_append_profile_choice(
|
||||
choices_by_key,
|
||||
label=saved_profile.name,
|
||||
profile=saved_profile,
|
||||
source="saved profile",
|
||||
saved_name=saved_profile.name,
|
||||
)
|
||||
|
||||
for target in discovered:
|
||||
discovered_profile = _profile_from_target(target)
|
||||
if discovered_profile is None:
|
||||
continue
|
||||
_append_profile_choice(
|
||||
choices_by_key,
|
||||
label=f"{target.name} current config",
|
||||
profile=discovered_profile,
|
||||
source=target.name,
|
||||
)
|
||||
|
||||
return sorted(choices_by_key.values(), key=lambda choice: (choice.label.lower(), choice.profile.base_url))
|
||||
|
||||
|
||||
def _select_profile(profile_name: str | None, discovered: list[SwitchTarget]) -> MCPProfile:
|
||||
if profile_name:
|
||||
return _load_profile(profile_name)
|
||||
|
||||
choices = _collect_profile_choices(discovered)
|
||||
|
||||
table = Table(show_header=True, header_style="bold")
|
||||
table.add_column("#", style="cyan", justify="right")
|
||||
table.add_column("Source")
|
||||
table.add_column("From")
|
||||
table.add_column("Base URL")
|
||||
table.add_column("API Key")
|
||||
|
||||
for index, choice in enumerate(choices, start=1):
|
||||
table.add_row(
|
||||
str(index),
|
||||
choice.label,
|
||||
", ".join(choice.sources),
|
||||
choice.profile.base_url,
|
||||
_mask_key(choice.profile.api_key),
|
||||
)
|
||||
|
||||
manual_choice_index = len(choices) + 1
|
||||
table.add_row(
|
||||
str(manual_choice_index),
|
||||
"Enter manually",
|
||||
"prompt",
|
||||
"-",
|
||||
"-",
|
||||
)
|
||||
|
||||
console.print("\n[bold]Available switch sources[/bold]")
|
||||
console.print(table)
|
||||
|
||||
selected = _ask_choice(
|
||||
"Choose a source number for the Skyvern API key/base URL",
|
||||
choices=[str(index) for index in range(1, manual_choice_index + 1)],
|
||||
default="1" if choices else str(manual_choice_index),
|
||||
)
|
||||
|
||||
if int(selected) == manual_choice_index:
|
||||
return _prompt_for_manual_source()
|
||||
|
||||
return choices[int(selected) - 1].profile
|
||||
|
||||
|
||||
def _entry_location(entry: dict) -> str:
|
||||
kind = _entry_kind(entry)
|
||||
if kind == "local stdio":
|
||||
env = entry.get("env", {})
|
||||
if isinstance(env, dict):
|
||||
return str(env.get("SKYVERN_BASE_URL", ""))
|
||||
if kind == "remote http":
|
||||
return str(entry.get("url", ""))
|
||||
if kind == "mcp-remote bridge":
|
||||
args = entry.get("args", [])
|
||||
if isinstance(args, list) and len(args) > 1 and isinstance(args[1], str):
|
||||
return args[1]
|
||||
return ""
|
||||
|
||||
|
||||
def _discover_switch_targets() -> tuple[list[SwitchTarget], list[tuple[str, Path]]]:
|
||||
discovered: list[SwitchTarget] = []
|
||||
missing: list[tuple[str, Path]] = []
|
||||
|
||||
for spec in _switch_target_specs():
|
||||
path = spec.config_path_fn()
|
||||
if not path.exists():
|
||||
missing.append((spec.name, path))
|
||||
continue
|
||||
|
||||
config, error = _load_switch_config(path, spec.config_format)
|
||||
if error:
|
||||
discovered.append(
|
||||
SwitchTarget(
|
||||
name=spec.name,
|
||||
config_path=path,
|
||||
entry_key=None,
|
||||
entry=None,
|
||||
config_format=spec.config_format,
|
||||
error=error,
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
servers = (config or {}).get(_server_block_key(spec.config_format), {})
|
||||
server_key = _find_server_key(servers, preferred="skyvern") if isinstance(servers, dict) else None
|
||||
entry = servers.get(server_key) if server_key else None
|
||||
discovered.append(
|
||||
SwitchTarget(
|
||||
name=spec.name,
|
||||
config_path=path,
|
||||
entry_key=server_key,
|
||||
entry=entry,
|
||||
config_format=spec.config_format,
|
||||
error=None,
|
||||
)
|
||||
)
|
||||
|
||||
return discovered, missing
|
||||
|
||||
|
||||
def _print_discovery_results(discovered: list[SwitchTarget], missing: list[tuple[str, Path]]) -> None:
|
||||
if discovered:
|
||||
table = Table(show_header=True, header_style="bold")
|
||||
table.add_column("App")
|
||||
table.add_column("Config Path")
|
||||
table.add_column("Status")
|
||||
table.add_column("Current")
|
||||
|
||||
for target in discovered:
|
||||
if target.error:
|
||||
status = "[red]Invalid config[/red]"
|
||||
current = "-"
|
||||
elif target.entry is None:
|
||||
status = "[yellow]Config found, no Skyvern entry[/yellow]"
|
||||
current = "-"
|
||||
else:
|
||||
status = f"[green]{_entry_kind(target.entry)}[/green]"
|
||||
location = _entry_location(target.entry)
|
||||
current = f"{_mask_key(_extract_entry_api_key(target.entry))} {location}".strip()
|
||||
table.add_row(target.name, str(target.config_path), status, current)
|
||||
|
||||
console.print(table)
|
||||
|
||||
if missing:
|
||||
missing_names = ", ".join(f"{name} ({path})" for name, path in missing)
|
||||
console.print(f"[dim]Config not found: {missing_names}[/dim]")
|
||||
|
||||
|
||||
def _select_targets(discovered: list[SwitchTarget]) -> list[SwitchTarget]:
|
||||
selectable = [
|
||||
target for target in discovered if target.entry is not None and _entry_kind(target.entry) != "unsupported"
|
||||
]
|
||||
if not selectable:
|
||||
console.print(
|
||||
"[red]No switchable Skyvern MCP entries were found. Add a Skyvern MCP entry for the client first, "
|
||||
"then re-run `skyvern mcp switch`.[/red]"
|
||||
)
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
if len(selectable) == 1:
|
||||
return selectable
|
||||
|
||||
table = Table(show_header=True, header_style="bold")
|
||||
table.add_column("#", style="cyan", justify="right")
|
||||
table.add_column("App")
|
||||
table.add_column("Config Path")
|
||||
table.add_column("Transport")
|
||||
|
||||
for index, target in enumerate(selectable, start=1):
|
||||
table.add_row(str(index), target.name, str(target.config_path), _entry_kind(target.entry))
|
||||
|
||||
console.print("\n[bold]Switchable Skyvern configs[/bold]")
|
||||
console.print(table)
|
||||
|
||||
raw_choice = _prompt_text(
|
||||
"Which apps should use the selected profile? Enter numbers separated by commas or 'all'",
|
||||
default="all",
|
||||
)
|
||||
if raw_choice.strip().lower() == "all":
|
||||
return selectable
|
||||
|
||||
chosen_indexes: list[int] = []
|
||||
for item in raw_choice.split(","):
|
||||
stripped = item.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
try:
|
||||
index = int(stripped)
|
||||
except ValueError as exc:
|
||||
raise typer.BadParameter("Selections must be numbers separated by commas or 'all'.") from exc
|
||||
if index < 1 or index > len(selectable):
|
||||
raise typer.BadParameter(f"Selection {index} is out of range.")
|
||||
if index not in chosen_indexes:
|
||||
chosen_indexes.append(index)
|
||||
|
||||
if not chosen_indexes:
|
||||
raise typer.BadParameter("Choose at least one app to update.")
|
||||
|
||||
return [selectable[index - 1] for index in chosen_indexes]
|
||||
|
||||
|
||||
def _patch_entry_with_profile(
|
||||
entry: dict,
|
||||
profile: MCPProfile,
|
||||
*,
|
||||
config_format: str = _CONFIG_FORMAT_JSON,
|
||||
) -> dict:
|
||||
patched = copy.deepcopy(entry)
|
||||
kind = _entry_kind(entry)
|
||||
|
||||
if kind == "local stdio":
|
||||
env = dict(patched.get("env") or {})
|
||||
env["SKYVERN_API_KEY"] = profile.api_key
|
||||
env["SKYVERN_BASE_URL"] = profile.base_url
|
||||
patched["env"] = env
|
||||
return patched
|
||||
|
||||
if kind == "remote http":
|
||||
target_url = _profile_to_mcp_url(profile.base_url)
|
||||
if config_format == _CONFIG_FORMAT_CODEX or "http_headers" in patched:
|
||||
headers = dict(patched.get("http_headers") or {})
|
||||
headers["x-api-key"] = profile.api_key
|
||||
patched["http_headers"] = headers
|
||||
patched["url"] = target_url
|
||||
return patched
|
||||
|
||||
headers = dict(patched.get("headers") or {})
|
||||
headers["x-api-key"] = profile.api_key
|
||||
patched["headers"] = headers
|
||||
patched["type"] = "http"
|
||||
patched["url"] = target_url
|
||||
return patched
|
||||
|
||||
if kind == "mcp-remote bridge":
|
||||
args = list(patched.get("args") or [])
|
||||
target_url = _profile_to_mcp_url(profile.base_url)
|
||||
if not args or args[0] != "mcp-remote":
|
||||
raise ValueError("Unsupported mcp-remote entry format.")
|
||||
|
||||
if len(args) == 1:
|
||||
args.append(target_url)
|
||||
else:
|
||||
args[1] = target_url
|
||||
|
||||
cleaned_args: list[object] = []
|
||||
skip_next = False
|
||||
for index, arg in enumerate(args):
|
||||
if skip_next:
|
||||
skip_next = False
|
||||
continue
|
||||
if arg == "--header" and index + 1 < len(args):
|
||||
next_arg = args[index + 1]
|
||||
if isinstance(next_arg, str) and next_arg.startswith("x-api-key:"):
|
||||
skip_next = True
|
||||
continue
|
||||
if isinstance(arg, str) and arg.startswith("x-api-key:"):
|
||||
continue
|
||||
cleaned_args.append(arg)
|
||||
|
||||
cleaned_args.extend(["--header", f"x-api-key:{profile.api_key}"])
|
||||
patched["args"] = cleaned_args
|
||||
return patched
|
||||
|
||||
raise ValueError(f"Unsupported Skyvern MCP entry format: {kind}")
|
||||
|
||||
|
||||
def _render_patched_entry(target: SwitchTarget, patched: dict) -> Syntax:
|
||||
masked = _mask_secrets(patched)
|
||||
if target.config_format == _CONFIG_FORMAT_CODEX and target.entry_key:
|
||||
snippet = toml.dumps({_server_block_key(target.config_format): {target.entry_key: masked}})
|
||||
return Syntax(snippet, "toml")
|
||||
return Syntax(json.dumps(masked, indent=2), "json")
|
||||
|
||||
|
||||
def _apply_profile_to_target(
|
||||
target: SwitchTarget,
|
||||
profile: MCPProfile,
|
||||
*,
|
||||
dry_run: bool = False,
|
||||
) -> tuple[bool, Path | None]:
|
||||
if target.entry is None or target.entry_key is None:
|
||||
raise ValueError(f"{target.name} does not have a switchable Skyvern MCP entry.")
|
||||
|
||||
config, error = _load_switch_config(target.config_path, target.config_format)
|
||||
if error:
|
||||
raise ValueError(error)
|
||||
|
||||
existing = config or {}
|
||||
server_block = _server_block_key(target.config_format)
|
||||
servers = existing.setdefault(server_block, {})
|
||||
current = servers.get(target.entry_key)
|
||||
if not isinstance(current, dict):
|
||||
raise ValueError(f"{target.name} does not have a valid Skyvern MCP entry.")
|
||||
|
||||
patched = _patch_entry_with_profile(current, profile, config_format=target.config_format)
|
||||
if patched == current:
|
||||
return False, None
|
||||
|
||||
if dry_run:
|
||||
console.print(f"\n[bold]{target.name}[/bold] -> {target.config_path}")
|
||||
console.print(_render_patched_entry(target, patched))
|
||||
return True, None
|
||||
|
||||
servers[target.entry_key] = patched
|
||||
backup_path = _write_switch_config(target.config_path, existing, target.config_format)
|
||||
return True, backup_path
|
||||
|
||||
|
||||
@profile_app.command("save")
|
||||
def save_profile_command(
|
||||
name: str = typer.Argument(..., help="Profile name, for example 'work-prod'."),
|
||||
api_key: str | None = typer.Option(None, "--api-key", "-k", help="Skyvern API key (reads env/.env if omitted)"),
|
||||
base_url: str | None = typer.Option(
|
||||
None,
|
||||
"--base-url",
|
||||
help="Skyvern API base URL (reads env/.env if omitted, default: https://api.skyvern.com)",
|
||||
),
|
||||
overwrite: bool = typer.Option(False, "--overwrite", help="Replace an existing profile with the same name."),
|
||||
) -> None:
|
||||
env_key, env_base_url = _get_env_credentials()
|
||||
resolved_api_key = api_key or env_key
|
||||
if not resolved_api_key:
|
||||
resolved_api_key = _prompt_text("Skyvern API key", password=True)
|
||||
|
||||
resolved_base_url = base_url or env_base_url or _DEFAULT_BASE_URL
|
||||
profile = _build_profile(name, resolved_api_key, resolved_base_url)
|
||||
path = _save_profile(profile, overwrite=overwrite)
|
||||
console.print(f"[green]Saved MCP profile '{profile.name}' at {path}[/green]")
|
||||
console.print(f"[dim]Saved MCP profiles store API keys in plaintext JSON under {path.parent}[/dim]")
|
||||
|
||||
|
||||
@profile_app.command("list")
|
||||
def list_profiles_command() -> None:
|
||||
profiles = _list_profiles()
|
||||
if not profiles:
|
||||
console.print(f"[yellow]No saved MCP profiles found in {_profile_store_dir()}[/yellow]")
|
||||
return
|
||||
|
||||
table = Table(show_header=True, header_style="bold")
|
||||
table.add_column("Profile")
|
||||
table.add_column("Base URL")
|
||||
table.add_column("API Key")
|
||||
|
||||
for profile in profiles:
|
||||
table.add_row(profile.name, profile.base_url, _mask_key(profile.api_key))
|
||||
|
||||
console.print(table)
|
||||
console.print(f"[dim]Profile store: {_profile_store_dir()}[/dim]")
|
||||
|
||||
|
||||
@mcp_app.command("switch")
|
||||
def switch_command(
|
||||
profile_name: str | None = typer.Option(None, "--profile", help="Saved profile name to use without prompting."),
|
||||
dry_run: bool = typer.Option(False, "--dry-run", help="Preview updates without writing files."),
|
||||
) -> None:
|
||||
console.print(
|
||||
Panel(
|
||||
"[bold]Skyvern MCP Switcher[/bold]\n\n"
|
||||
"This command finds supported local MCP configs, lets you choose which apps to update, "
|
||||
"then swaps in a Skyvern API key/base URL from env, saved profiles, existing configs, or manual entry.",
|
||||
border_style="blue",
|
||||
)
|
||||
)
|
||||
|
||||
discovered, missing = _discover_switch_targets()
|
||||
if not discovered:
|
||||
console.print(
|
||||
"[red]No supported MCP client config files were found for Claude Code, Claude Desktop, Cursor, "
|
||||
"Windsurf, or Codex.[/red]"
|
||||
)
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
_print_discovery_results(discovered, missing)
|
||||
|
||||
selected_targets = _select_targets(discovered)
|
||||
profile = _select_profile(profile_name, discovered)
|
||||
|
||||
console.print(
|
||||
f"\n[bold]Selected source:[/bold] {profile.name}\n"
|
||||
f"[dim]API key:[/dim] {_mask_key(profile.api_key)}\n"
|
||||
f"[dim]Base URL:[/dim] {profile.base_url}\n"
|
||||
f"[dim]Remote MCP URL:[/dim] {_profile_to_mcp_url(profile.base_url)}\n"
|
||||
"[dim]Local stdio configs keep the base URL in SKYVERN_BASE_URL.[/dim]"
|
||||
)
|
||||
|
||||
target_names = ", ".join(target.name for target in selected_targets)
|
||||
if not dry_run and not Confirm.ask(f"Update {target_names} to use '{profile.name}'?", default=True):
|
||||
raise typer.Abort()
|
||||
|
||||
updated: list[str] = []
|
||||
unchanged: list[str] = []
|
||||
backups: list[Path] = []
|
||||
for target in selected_targets:
|
||||
changed, backup_path = _apply_profile_to_target(target, profile, dry_run=dry_run)
|
||||
if changed:
|
||||
updated.append(target.name)
|
||||
else:
|
||||
unchanged.append(target.name)
|
||||
if backup_path is not None:
|
||||
backups.append(backup_path)
|
||||
|
||||
if dry_run:
|
||||
console.print("\n[yellow]Dry run only. No config files were modified.[/yellow]")
|
||||
return
|
||||
|
||||
if updated:
|
||||
console.print(f"\n[green]Updated:[/green] {', '.join(updated)}")
|
||||
if unchanged:
|
||||
console.print(f"[dim]Already using that profile: {', '.join(unchanged)}[/dim]")
|
||||
if backups:
|
||||
console.print("[dim]Backups created:[/dim]")
|
||||
for backup in backups:
|
||||
console.print(f"[dim]- {backup}[/dim]")
|
||||
|
||||
console.print(
|
||||
"\n[bold yellow]Restart Claude Code, Claude Desktop, Cursor, Windsurf, or Codex. "
|
||||
"If you updated a project `.mcp.json`, reopen that project too.[/bold yellow]"
|
||||
)
|
||||
|
|
@ -9,6 +9,7 @@ import platform
|
|||
import shutil
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
from urllib.parse import urlparse
|
||||
|
|
@ -23,12 +24,14 @@ from skyvern.analytics import capture_setup_event
|
|||
from skyvern.cli.auth_command import run_signup
|
||||
from skyvern.cli.console import console
|
||||
from skyvern.cli.skill_commands import get_skill_dirs
|
||||
from skyvern.utils import detect_os, get_windows_appdata_roaming
|
||||
from skyvern.utils.env_paths import resolve_backend_env_path
|
||||
|
||||
# NOTE: These helpers back both `skyvern setup ...` commands and the
|
||||
# interactive MCP step used by `skyvern init` / `skyvern quickstart`.
|
||||
# Keep local stdio setup and Claude Code skill installation behavior here so
|
||||
# the standalone and wizard flows stay aligned.
|
||||
# interactive MCP step used by `skyvern init` / `skyvern quickstart`, plus
|
||||
# the MCP switcher in `skyvern/cli/mcp_commands.py`.
|
||||
# Keep shared config parsing/writes, local stdio setup, and Claude Code skill
|
||||
# installation behavior here so the standalone and wizard flows stay aligned.
|
||||
setup_app = typer.Typer(
|
||||
help="Register Skyvern MCP with AI coding tools.",
|
||||
invoke_without_command=True,
|
||||
|
|
@ -111,6 +114,7 @@ def _build_local_mcp_entry(
|
|||
api_key: str,
|
||||
base_url: str,
|
||||
use_python_path: bool = False,
|
||||
command: str | None = None,
|
||||
) -> dict:
|
||||
"""Build a stdio MCP entry for local self-hosted mode.
|
||||
|
||||
|
|
@ -125,8 +129,16 @@ def _build_local_mcp_entry(
|
|||
|
||||
_ = use_python_path
|
||||
|
||||
command_name = command or sys.executable
|
||||
if command_name == "skyvern":
|
||||
return {
|
||||
"command": command_name,
|
||||
"args": ["run", "mcp"],
|
||||
"env": env_block,
|
||||
}
|
||||
|
||||
return {
|
||||
"command": sys.executable,
|
||||
"command": command_name,
|
||||
"args": ["-m", "skyvern", "run", "mcp"],
|
||||
"env": env_block,
|
||||
}
|
||||
|
|
@ -138,6 +150,8 @@ def _has_api_key(entry: dict | None) -> bool:
|
|||
return False
|
||||
if entry.get("headers", {}).get("x-api-key"):
|
||||
return True
|
||||
if entry.get("http_headers", {}).get("x-api-key"):
|
||||
return True
|
||||
if entry.get("env", {}).get("SKYVERN_API_KEY"):
|
||||
return True
|
||||
# mcp-remote bridge: API key is in args as "--header", "x-api-key:..."
|
||||
|
|
@ -163,6 +177,10 @@ def _mask_secrets(entry: dict) -> dict:
|
|||
key = masked["headers"]["x-api-key"]
|
||||
masked["headers"]["x-api-key"] = _mask_key(key)
|
||||
|
||||
if "http_headers" in masked and "x-api-key" in masked["http_headers"]:
|
||||
key = masked["http_headers"]["x-api-key"]
|
||||
masked["http_headers"]["x-api-key"] = _mask_key(key)
|
||||
|
||||
# Local stdio format: env.SKYVERN_API_KEY
|
||||
if "env" in masked and "SKYVERN_API_KEY" in masked["env"]:
|
||||
key = masked["env"]["SKYVERN_API_KEY"]
|
||||
|
|
@ -182,6 +200,60 @@ def _mask_secrets(entry: dict) -> dict:
|
|||
return masked
|
||||
|
||||
|
||||
def _load_mcp_config(config_path: Path) -> tuple[dict | None, str | None]:
|
||||
"""Load and validate an MCP config file, returning (config, error)."""
|
||||
if not config_path.exists():
|
||||
return {}, None
|
||||
|
||||
try:
|
||||
existing = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError:
|
||||
return None, f"Cannot parse {config_path}. Fix the JSON and re-run."
|
||||
|
||||
if not isinstance(existing, dict):
|
||||
return None, f"{config_path} must contain a top-level JSON object."
|
||||
|
||||
servers = existing.get("mcpServers")
|
||||
if servers is not None and not isinstance(servers, dict):
|
||||
return None, f"{config_path} has invalid `mcpServers`; expected a JSON object."
|
||||
|
||||
return existing, None
|
||||
|
||||
|
||||
def _read_mcp_config(config_path: Path) -> dict:
|
||||
"""Load an MCP config or exit with a user-friendly message."""
|
||||
existing, error = _load_mcp_config(config_path)
|
||||
if error:
|
||||
console.print(f"[red]{error}[/red]")
|
||||
raise typer.Exit(code=1)
|
||||
return existing or {}
|
||||
|
||||
|
||||
def _find_server_key(servers: dict[str, object], preferred: str = "skyvern") -> str | None:
|
||||
"""Find an existing server key case-insensitively."""
|
||||
for key in servers:
|
||||
if key.lower() == preferred.lower():
|
||||
return key
|
||||
return None
|
||||
|
||||
|
||||
def _backup_config_path(config_path: Path) -> Path:
|
||||
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
|
||||
return config_path.with_name(f"{config_path.name}.bak-{timestamp}")
|
||||
|
||||
|
||||
def _write_mcp_config(config_path: Path, config: dict, create_backup: bool = True) -> Path | None:
|
||||
"""Write an MCP config, creating a backup of the prior file when overwriting."""
|
||||
backup_path: Path | None = None
|
||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if create_backup and config_path.exists():
|
||||
backup_path = _backup_config_path(config_path)
|
||||
shutil.copy2(config_path, backup_path)
|
||||
|
||||
config_path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")
|
||||
return backup_path
|
||||
|
||||
|
||||
def _upsert_mcp_config(
|
||||
config_path: Path,
|
||||
tool_name: str,
|
||||
|
|
@ -191,17 +263,10 @@ def _upsert_mcp_config(
|
|||
yes: bool = False,
|
||||
) -> None:
|
||||
"""Read config, diff, prompt, and write. Idempotent."""
|
||||
if config_path.exists():
|
||||
try:
|
||||
existing = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError:
|
||||
console.print(f"[red]Cannot parse {config_path}. Fix the JSON and re-run.[/red]")
|
||||
raise typer.Exit(code=1)
|
||||
else:
|
||||
existing = {}
|
||||
|
||||
existing = _read_mcp_config(config_path)
|
||||
servers = existing.setdefault("mcpServers", {})
|
||||
current = servers.get(server_key)
|
||||
resolved_server_key = _find_server_key(servers, preferred=server_key) or server_key
|
||||
current = servers.get(resolved_server_key)
|
||||
|
||||
if current == skyvern_entry:
|
||||
console.print(f"[green]Already configured for {tool_name} (no changes)[/green]")
|
||||
|
|
@ -233,10 +298,11 @@ def _upsert_mcp_config(
|
|||
if not typer.confirm("\nApply changes?"):
|
||||
raise typer.Abort()
|
||||
|
||||
servers[server_key] = skyvern_entry
|
||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
|
||||
servers[resolved_server_key] = skyvern_entry
|
||||
backup_path = _write_mcp_config(config_path, existing, create_backup=True)
|
||||
console.print(f"[green]Configured {tool_name} at {config_path}[/green]")
|
||||
if backup_path is not None:
|
||||
console.print(f"[dim]Backup saved to {backup_path}[/dim]")
|
||||
|
||||
|
||||
def _build_entry(
|
||||
|
|
@ -280,30 +346,31 @@ def _is_claude_code_installed() -> bool:
|
|||
|
||||
|
||||
def _is_cursor_installed() -> bool:
|
||||
return (Path.home() / ".cursor").is_dir()
|
||||
try:
|
||||
return _cursor_config_path().parent.is_dir()
|
||||
except typer.Exit:
|
||||
return False
|
||||
|
||||
|
||||
def _is_windsurf_installed() -> bool:
|
||||
return (Path.home() / ".codeium" / "windsurf").is_dir()
|
||||
try:
|
||||
return _windsurf_config_path().parent.is_dir()
|
||||
except typer.Exit:
|
||||
return False
|
||||
|
||||
|
||||
def _is_claude_desktop_installed() -> bool:
|
||||
system = platform.system()
|
||||
if system == "Darwin":
|
||||
return (Path.home() / "Library" / "Application Support" / "Claude").is_dir()
|
||||
if system == "Linux":
|
||||
return (Path.home() / ".config" / "Claude").is_dir()
|
||||
if system == "Windows":
|
||||
appdata = os.environ.get("APPDATA", "")
|
||||
return bool(appdata) and (Path(appdata) / "Claude").is_dir()
|
||||
return False
|
||||
try:
|
||||
return _claude_desktop_config_path().parent.is_dir()
|
||||
except typer.Exit:
|
||||
return False
|
||||
|
||||
|
||||
def _get_known_tools() -> list[DetectedTool]:
|
||||
"""Return all known AI coding tools in detection order.
|
||||
|
||||
Note: Codex is not included — it has no stable local config path to detect.
|
||||
Users can configure it manually via `skyvern setup codex` if/when that subcommand is added.
|
||||
Note: Codex is not included in the guided setup flow yet.
|
||||
Its config is TOML rather than the JSON shape used by the current setup commands.
|
||||
"""
|
||||
return [
|
||||
DetectedTool(
|
||||
|
|
@ -399,12 +466,26 @@ def _resolve_setup_credentials(*, api_key_flag: str | None, yes: bool, local: bo
|
|||
|
||||
|
||||
def _claude_desktop_config_path() -> Path:
|
||||
system = platform.system()
|
||||
if system == "Darwin":
|
||||
system = detect_os()
|
||||
if system == "wsl":
|
||||
roaming_path = get_windows_appdata_roaming()
|
||||
if roaming_path is None:
|
||||
console.print("[red]Could not locate Windows AppData\\\\Roaming from WSL.[/red]")
|
||||
raise typer.Exit(code=1)
|
||||
return Path(roaming_path) / "Claude" / "claude_desktop_config.json"
|
||||
if system == "darwin":
|
||||
return Path.home() / "Library" / "Application Support" / "Claude" / "claude_desktop_config.json"
|
||||
if system == "Linux":
|
||||
return Path.home() / ".config" / "Claude" / "claude_desktop_config.json"
|
||||
if system == "Windows":
|
||||
if system == "linux":
|
||||
candidates = [
|
||||
Path.home() / ".config" / "Claude",
|
||||
Path.home() / ".local" / "share" / "Claude",
|
||||
Path.home() / "Claude",
|
||||
]
|
||||
for candidate in candidates:
|
||||
if candidate.exists():
|
||||
return candidate / "claude_desktop_config.json"
|
||||
return candidates[0] / "claude_desktop_config.json"
|
||||
if system == "windows":
|
||||
appdata = os.environ.get("APPDATA")
|
||||
if not appdata:
|
||||
console.print("[red]APPDATA environment variable not set on Windows.[/red]")
|
||||
|
|
@ -414,11 +495,23 @@ def _claude_desktop_config_path() -> Path:
|
|||
raise typer.Exit(code=1)
|
||||
|
||||
|
||||
def _wsl_windows_user_home() -> Path:
|
||||
roaming_path = get_windows_appdata_roaming()
|
||||
if roaming_path is None:
|
||||
console.print("[red]Could not locate Windows AppData\\\\Roaming from WSL.[/red]")
|
||||
raise typer.Exit(code=1)
|
||||
return roaming_path.parent.parent
|
||||
|
||||
|
||||
def _cursor_config_path() -> Path:
|
||||
if detect_os() == "wsl":
|
||||
return _wsl_windows_user_home() / ".cursor" / "mcp.json"
|
||||
return Path.home() / ".cursor" / "mcp.json"
|
||||
|
||||
|
||||
def _windsurf_config_path() -> Path:
|
||||
if detect_os() == "wsl":
|
||||
return _wsl_windows_user_home() / ".codeium" / "windsurf" / "mcp_config.json"
|
||||
return Path.home() / ".codeium" / "windsurf" / "mcp_config.json"
|
||||
|
||||
|
||||
|
|
@ -426,6 +519,14 @@ def _claude_code_global_config_path() -> Path:
|
|||
return Path.home() / ".claude.json"
|
||||
|
||||
|
||||
def _claude_code_project_config_path() -> Path:
|
||||
return Path.cwd() / ".mcp.json"
|
||||
|
||||
|
||||
def _codex_config_path() -> Path:
|
||||
return Path.home() / ".codex" / "config.toml"
|
||||
|
||||
|
||||
_PROJECT_MARKERS = (
|
||||
".git",
|
||||
".mcp.json",
|
||||
|
|
|
|||
512
tests/unit/test_mcp_commands.py
Normal file
512
tests/unit/test_mcp_commands.py
Normal file
|
|
@ -0,0 +1,512 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import toml
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from skyvern.cli.mcp_commands import (
|
||||
MCPProfile,
|
||||
SwitchTarget,
|
||||
_apply_profile_to_target,
|
||||
_build_profile,
|
||||
_collect_profile_choices,
|
||||
_discover_switch_targets,
|
||||
_entry_kind,
|
||||
_list_profiles,
|
||||
_load_profile_from_path,
|
||||
_patch_entry_with_profile,
|
||||
_profile_to_mcp_url,
|
||||
_sanitize_prompt_response,
|
||||
_save_profile,
|
||||
mcp_app,
|
||||
)
|
||||
|
||||
|
||||
def test_save_profile_and_list_round_trip(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._profile_store_dir", lambda: tmp_path)
|
||||
|
||||
profile = _build_profile("Work Prod", "sk-test-1234567890", "https://api.skyvern.com/")
|
||||
saved_path = _save_profile(profile)
|
||||
|
||||
assert saved_path.exists()
|
||||
assert _list_profiles() == [
|
||||
MCPProfile(name="Work Prod", api_key="sk-test-1234567890", base_url="https://api.skyvern.com")
|
||||
]
|
||||
|
||||
|
||||
def test_save_profile_command_sanitizes_prompted_api_key_and_warns(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._profile_store_dir", lambda: tmp_path)
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._get_env_credentials", lambda: ("", ""))
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands.Prompt.ask", lambda *args, **kwargs: "\x1b[Cprompt-key-1234567890")
|
||||
|
||||
result = CliRunner().invoke(mcp_app, ["profile", "save", "Work Prod"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
saved = json.loads((tmp_path / "work-prod.json").read_text(encoding="utf-8"))
|
||||
assert saved["api_key"] == "prompt-key-1234567890"
|
||||
assert "plaintext JSON" in result.output
|
||||
|
||||
|
||||
def test_save_profile_restricts_permissions_on_posix(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
if os.name == "nt":
|
||||
pytest.skip("POSIX permissions are not enforced on Windows in the same way.")
|
||||
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._profile_store_dir", lambda: tmp_path / "profiles")
|
||||
|
||||
saved_path = _save_profile(_build_profile("Work Prod", "sk-test-1234567890", "https://api.skyvern.com"))
|
||||
|
||||
assert stat.S_IMODE(saved_path.stat().st_mode) == 0o600
|
||||
assert stat.S_IMODE(saved_path.parent.stat().st_mode) == 0o700
|
||||
|
||||
|
||||
def test_load_profile_from_path_rejects_non_string_fields(tmp_path: Path) -> None:
|
||||
profile_path = tmp_path / "invalid.json"
|
||||
profile_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"name": "Work Prod",
|
||||
"api_key": {"secret": "bad"},
|
||||
"base_url": "https://api.skyvern.com",
|
||||
}
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="field 'api_key' must be a string"):
|
||||
_load_profile_from_path(profile_path)
|
||||
|
||||
|
||||
def test_apply_profile_to_target_updates_local_entry_and_creates_backup(tmp_path: Path) -> None:
|
||||
config_path = tmp_path / "mcp.json"
|
||||
original_config = {
|
||||
"mcpServers": {
|
||||
"Skyvern": {
|
||||
"command": "/opt/homebrew/bin/python3.11",
|
||||
"args": ["-m", "skyvern", "run", "mcp"],
|
||||
"env": {
|
||||
"SKYVERN_BASE_URL": "http://localhost:8000",
|
||||
"SKYVERN_API_KEY": "old-key",
|
||||
"OTHER": "keep-me",
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
config_path.write_text(json.dumps(original_config, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
target = SwitchTarget(
|
||||
name="Cursor",
|
||||
config_path=config_path,
|
||||
entry_key="Skyvern",
|
||||
entry=original_config["mcpServers"]["Skyvern"],
|
||||
)
|
||||
profile = _build_profile("Prod", "new-key-1234567890", "https://api.skyvern.com")
|
||||
|
||||
changed, backup_path = _apply_profile_to_target(target, profile)
|
||||
|
||||
assert changed is True
|
||||
assert backup_path is not None
|
||||
assert backup_path.exists()
|
||||
|
||||
written = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
written_entry = written["mcpServers"]["Skyvern"]
|
||||
assert written_entry["command"] == "/opt/homebrew/bin/python3.11"
|
||||
assert written_entry["args"] == ["-m", "skyvern", "run", "mcp"]
|
||||
assert written_entry["env"]["SKYVERN_API_KEY"] == "new-key-1234567890"
|
||||
assert written_entry["env"]["SKYVERN_BASE_URL"] == "https://api.skyvern.com"
|
||||
assert written_entry["env"]["OTHER"] == "keep-me"
|
||||
|
||||
backup = json.loads(backup_path.read_text(encoding="utf-8"))
|
||||
assert backup["mcpServers"]["Skyvern"]["env"]["SKYVERN_API_KEY"] == "old-key"
|
||||
|
||||
|
||||
def test_patch_entry_with_profile_updates_mcp_remote_bridge() -> None:
|
||||
profile = _build_profile("Cloud Alt", "new-key-1234567890", "https://alt.skyvern.example")
|
||||
entry = {
|
||||
"command": "npx",
|
||||
"args": [
|
||||
"mcp-remote",
|
||||
"https://api.skyvern.com/mcp/",
|
||||
"--header",
|
||||
"x-api-key:old-key",
|
||||
"--transport",
|
||||
"stdio",
|
||||
],
|
||||
}
|
||||
|
||||
patched = _patch_entry_with_profile(entry, profile)
|
||||
|
||||
assert patched["args"][0] == "mcp-remote"
|
||||
assert patched["args"][1] == "https://alt.skyvern.example/mcp/"
|
||||
assert patched["args"].count("--header") == 1
|
||||
assert "x-api-key:new-key-1234567890" in patched["args"]
|
||||
assert "x-api-key:old-key" not in patched["args"]
|
||||
|
||||
|
||||
def test_patch_entry_with_profile_updates_mcp_remote_bridge_even_with_env() -> None:
|
||||
profile = _build_profile("Cloud Alt", "new-key-1234567890", "https://alt.skyvern.example")
|
||||
entry = {
|
||||
"command": "npx",
|
||||
"args": [
|
||||
"mcp-remote",
|
||||
"https://api.skyvern.com/mcp/",
|
||||
"--header",
|
||||
"x-api-key:old-key",
|
||||
],
|
||||
"env": {
|
||||
"DEBUG": "1",
|
||||
"SKYVERN_API_KEY": "do-not-touch",
|
||||
},
|
||||
}
|
||||
|
||||
patched = _patch_entry_with_profile(entry, profile)
|
||||
|
||||
assert _entry_kind(entry) == "mcp-remote bridge"
|
||||
assert "x-api-key:new-key-1234567890" in patched["args"]
|
||||
assert patched["env"]["SKYVERN_API_KEY"] == "do-not-touch"
|
||||
|
||||
|
||||
def test_entry_kind_requires_exact_npx_for_mcp_remote_bridge() -> None:
|
||||
entry = {
|
||||
"command": "npx-wrapper",
|
||||
"args": ["mcp-remote", "https://api.skyvern.com/mcp/"],
|
||||
}
|
||||
|
||||
assert _entry_kind(entry) == "unsupported"
|
||||
|
||||
|
||||
def test_sanitize_prompt_response_strips_arrow_escape_noise() -> None:
|
||||
assert _sanitize_prompt_response("\x1b[C\x1b[C\x1b[Call") == "all"
|
||||
|
||||
|
||||
def test_profile_to_mcp_url_normalizes_user_base_url() -> None:
|
||||
assert _profile_to_mcp_url("https://api.skyvern.com") == "https://api.skyvern.com/mcp/"
|
||||
assert _profile_to_mcp_url("https://api.skyvern.com/") == "https://api.skyvern.com/mcp/"
|
||||
assert _profile_to_mcp_url("https://api.skyvern.com/mcp/") == "https://api.skyvern.com/mcp/"
|
||||
|
||||
|
||||
def test_apply_profile_to_target_updates_codex_entry_and_creates_backup(tmp_path: Path) -> None:
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text(
|
||||
toml.dumps(
|
||||
{
|
||||
"model": "gpt-5.4",
|
||||
"mcp_servers": {
|
||||
"skyvern": {
|
||||
"url": "https://api.skyvern.com/mcp/",
|
||||
"http_headers": {"x-api-key": "old-key"},
|
||||
"startup_timeout_sec": 30,
|
||||
"tool_timeout_sec": 120,
|
||||
}
|
||||
},
|
||||
}
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
target = SwitchTarget(
|
||||
name="Codex",
|
||||
config_path=config_path,
|
||||
entry_key="skyvern",
|
||||
entry={
|
||||
"url": "https://api.skyvern.com/mcp/",
|
||||
"http_headers": {"x-api-key": "old-key"},
|
||||
"startup_timeout_sec": 30,
|
||||
"tool_timeout_sec": 120,
|
||||
},
|
||||
config_format="codex_toml",
|
||||
)
|
||||
profile = _build_profile("Prod", "new-key-1234567890", "https://alt.skyvern.example")
|
||||
|
||||
changed, backup_path = _apply_profile_to_target(target, profile)
|
||||
|
||||
assert changed is True
|
||||
assert backup_path is not None
|
||||
assert backup_path.exists()
|
||||
|
||||
written = toml.loads(config_path.read_text(encoding="utf-8"))
|
||||
written_entry = written["mcp_servers"]["skyvern"]
|
||||
assert written_entry["url"] == "https://alt.skyvern.example/mcp/"
|
||||
assert written_entry["http_headers"]["x-api-key"] == "new-key-1234567890"
|
||||
assert written_entry["startup_timeout_sec"] == 30
|
||||
assert written_entry["tool_timeout_sec"] == 120
|
||||
|
||||
backup = toml.loads(backup_path.read_text(encoding="utf-8"))
|
||||
assert backup["mcp_servers"]["skyvern"]["http_headers"]["x-api-key"] == "old-key"
|
||||
|
||||
|
||||
def test_collect_profile_choices_includes_env_and_existing_config(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._profile_store_dir", lambda: tmp_path / "profiles")
|
||||
monkeypatch.setattr(
|
||||
"skyvern.cli.mcp_commands._get_env_credentials",
|
||||
lambda: ("env-key-1234567890", "https://api.skyvern.com"),
|
||||
)
|
||||
|
||||
target = SwitchTarget(
|
||||
name="Cursor",
|
||||
config_path=tmp_path / "mcp.json",
|
||||
entry_key="Skyvern",
|
||||
entry={
|
||||
"command": "npx",
|
||||
"args": ["mcp-remote", "https://staging.skyvern.example/mcp/", "--header", "x-api-key:staging-key"],
|
||||
},
|
||||
)
|
||||
|
||||
choices = _collect_profile_choices([target])
|
||||
|
||||
assert len(choices) == 2
|
||||
assert any(choice.label == "Current environment" for choice in choices)
|
||||
assert any(choice.label == "Cursor current config" for choice in choices)
|
||||
|
||||
|
||||
def test_switch_uses_env_candidate_without_saved_profile(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "mcp.json"
|
||||
config_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"mcpServers": {
|
||||
"Skyvern": {
|
||||
"command": "skyvern",
|
||||
"args": ["run", "mcp"],
|
||||
"env": {
|
||||
"SKYVERN_BASE_URL": "http://localhost:8000",
|
||||
"SKYVERN_API_KEY": "old-key",
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
target = SwitchTarget(
|
||||
name="Cursor",
|
||||
config_path=config_path,
|
||||
entry_key="Skyvern",
|
||||
entry={
|
||||
"command": "skyvern",
|
||||
"args": ["run", "mcp"],
|
||||
"env": {
|
||||
"SKYVERN_BASE_URL": "http://localhost:8000",
|
||||
"SKYVERN_API_KEY": "old-key",
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._discover_switch_targets", lambda: ([target], []))
|
||||
monkeypatch.setattr(
|
||||
"skyvern.cli.mcp_commands._get_env_credentials",
|
||||
lambda: ("env-key-1234567890", "https://api.skyvern.com"),
|
||||
)
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._profile_store_dir", lambda: tmp_path / "profiles")
|
||||
|
||||
result = CliRunner().invoke(mcp_app, ["switch"], input="1\ny\n")
|
||||
|
||||
assert result.exit_code == 0
|
||||
written = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
assert written["mcpServers"]["Skyvern"]["env"]["SKYVERN_API_KEY"] == "env-key-1234567890"
|
||||
assert written["mcpServers"]["Skyvern"]["env"]["SKYVERN_BASE_URL"] == "https://api.skyvern.com"
|
||||
|
||||
|
||||
def test_switch_manual_entry_does_not_prompt_for_profile_name_and_normalizes_remote_base_url(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_path = tmp_path / "claude.json"
|
||||
config_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"mcpServers": {
|
||||
"Skyvern": {
|
||||
"type": "http",
|
||||
"url": "https://old.skyvern.example/mcp/",
|
||||
"headers": {
|
||||
"x-api-key": "old-key",
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
target = SwitchTarget(
|
||||
name="Claude Code (global)",
|
||||
config_path=config_path,
|
||||
entry_key="Skyvern",
|
||||
entry={
|
||||
"type": "http",
|
||||
"url": "https://old.skyvern.example/mcp/",
|
||||
"headers": {
|
||||
"x-api-key": "old-key",
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._discover_switch_targets", lambda: ([target], []))
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._get_env_credentials", lambda: ("", "https://api.skyvern.com"))
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._profile_store_dir", lambda: tmp_path / "profiles")
|
||||
prompt_calls: list[str] = []
|
||||
prompt_values = iter(
|
||||
[
|
||||
"2",
|
||||
"manual-key-1234567890",
|
||||
"https://alt.skyvern.example/mcp/",
|
||||
]
|
||||
)
|
||||
|
||||
def fake_prompt(prompt: str, *, default: str | None = None, password: bool = False) -> str:
|
||||
prompt_calls.append(prompt)
|
||||
return next(prompt_values)
|
||||
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._prompt_text", fake_prompt)
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands.Confirm.ask", lambda *args, **kwargs: True)
|
||||
|
||||
result = CliRunner().invoke(mcp_app, ["switch"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Available switch sources" in result.output
|
||||
assert "Available switch profiles" not in result.output
|
||||
assert "Selected source:" in result.output
|
||||
assert "Remote MCP URL:" in result.output
|
||||
assert "Profile name" not in prompt_calls
|
||||
|
||||
written = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
entry = written["mcpServers"]["Skyvern"]
|
||||
assert entry["headers"]["x-api-key"] == "manual-key-1234567890"
|
||||
assert entry["url"] == "https://alt.skyvern.example/mcp/"
|
||||
|
||||
|
||||
def test_switch_accepts_arrow_key_noise_in_target_prompt(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
config_a = tmp_path / "claude.json"
|
||||
config_b = tmp_path / "codex.json"
|
||||
for config_path in (config_a, config_b):
|
||||
config_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"mcpServers": {
|
||||
"Skyvern": {
|
||||
"command": "skyvern",
|
||||
"args": ["run", "mcp"],
|
||||
"env": {
|
||||
"SKYVERN_BASE_URL": "http://localhost:8000",
|
||||
"SKYVERN_API_KEY": "old-key",
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
targets = [
|
||||
SwitchTarget(
|
||||
name="Claude Code (global)",
|
||||
config_path=config_a,
|
||||
entry_key="Skyvern",
|
||||
entry={
|
||||
"command": "skyvern",
|
||||
"args": ["run", "mcp"],
|
||||
"env": {
|
||||
"SKYVERN_BASE_URL": "http://localhost:8000",
|
||||
"SKYVERN_API_KEY": "old-key",
|
||||
},
|
||||
},
|
||||
),
|
||||
SwitchTarget(
|
||||
name="Codex",
|
||||
config_path=config_b,
|
||||
entry_key="Skyvern",
|
||||
entry={
|
||||
"command": "skyvern",
|
||||
"args": ["run", "mcp"],
|
||||
"env": {
|
||||
"SKYVERN_BASE_URL": "http://localhost:8000",
|
||||
"SKYVERN_API_KEY": "old-key",
|
||||
},
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._discover_switch_targets", lambda: (targets, []))
|
||||
monkeypatch.setattr(
|
||||
"skyvern.cli.mcp_commands._select_profile",
|
||||
lambda profile_name, discovered: MCPProfile(
|
||||
name="Env",
|
||||
api_key="env-key-1234567890",
|
||||
base_url="https://api.skyvern.com",
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._profile_store_dir", lambda: tmp_path / "profiles")
|
||||
|
||||
result = CliRunner().invoke(mcp_app, ["switch"], input="\x1b[C\x1b[C\x1b[Call\ny\n")
|
||||
|
||||
assert result.exit_code == 0
|
||||
for config_path in (config_a, config_b):
|
||||
written = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
assert written["mcpServers"]["Skyvern"]["env"]["SKYVERN_API_KEY"] == "env-key-1234567890"
|
||||
assert written["mcpServers"]["Skyvern"]["env"]["SKYVERN_BASE_URL"] == "https://api.skyvern.com"
|
||||
|
||||
|
||||
def test_discover_switch_targets_finds_claude_code_and_codex(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
global_config = tmp_path / ".claude.json"
|
||||
global_config.write_text(
|
||||
json.dumps({"mcpServers": {"skyvern": {"type": "http", "url": "https://api.skyvern.com/mcp/"}}}) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
project_config = tmp_path / ".mcp.json"
|
||||
project_config.write_text(
|
||||
json.dumps({"mcpServers": {"Skyvern": {"command": "skyvern", "args": ["run", "mcp"], "env": {}}}}) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
codex_config = tmp_path / "config.toml"
|
||||
codex_config.write_text(
|
||||
toml.dumps({"mcp_servers": {"skyvern": {"url": "https://api.skyvern.com/mcp/"}}}) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._claude_code_global_config_path", lambda: global_config)
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._claude_code_project_config_path", lambda: project_config)
|
||||
monkeypatch.setattr(
|
||||
"skyvern.cli.mcp_commands._claude_desktop_config_path", lambda: tmp_path / "missing-claude.json"
|
||||
)
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._cursor_config_path", lambda: tmp_path / "missing-cursor.json")
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._windsurf_config_path", lambda: tmp_path / "missing-windsurf.json")
|
||||
monkeypatch.setattr("skyvern.cli.mcp_commands._codex_config_path", lambda: codex_config)
|
||||
|
||||
discovered, missing = _discover_switch_targets()
|
||||
|
||||
discovered_by_name = {target.name: target for target in discovered}
|
||||
assert "Claude Code (global)" in discovered_by_name
|
||||
assert "Claude Code (project)" in discovered_by_name
|
||||
assert "Codex" in discovered_by_name
|
||||
assert discovered_by_name["Codex"].config_format == "codex_toml"
|
||||
assert discovered_by_name["Codex"].entry_key == "skyvern"
|
||||
assert {name for name, _ in missing} == {"Claude Desktop", "Cursor", "Windsurf"}
|
||||
|
|
@ -9,7 +9,7 @@ from pathlib import Path
|
|||
import pytest
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from skyvern.cli.setup_commands import setup_app
|
||||
from skyvern.cli.setup_commands import _cursor_config_path, _windsurf_config_path, setup_app
|
||||
|
||||
_FAKE_ENV = ("test-key", "https://api.skyvern.com")
|
||||
|
||||
|
|
@ -81,3 +81,19 @@ def test_guided_setup_skips_claude_desktop_without_node_and_prints_bundle_hint(
|
|||
assert "one-click Skyvern bundle" in result.output
|
||||
assert ".mcpb" in result.output
|
||||
assert not config.exists()
|
||||
|
||||
|
||||
def test_cursor_config_path_uses_windows_home_on_wsl(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
roaming_path = Path("/mnt/c/Users/alice/AppData/Roaming")
|
||||
monkeypatch.setattr("skyvern.cli.setup_commands.detect_os", lambda: "wsl")
|
||||
monkeypatch.setattr("skyvern.cli.setup_commands.get_windows_appdata_roaming", lambda: roaming_path)
|
||||
|
||||
assert _cursor_config_path() == Path("/mnt/c/Users/alice/.cursor/mcp.json")
|
||||
|
||||
|
||||
def test_windsurf_config_path_uses_windows_home_on_wsl(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
roaming_path = Path("/mnt/c/Users/alice/AppData/Roaming")
|
||||
monkeypatch.setattr("skyvern.cli.setup_commands.detect_os", lambda: "wsl")
|
||||
monkeypatch.setattr("skyvern.cli.setup_commands.get_windows_appdata_roaming", lambda: roaming_path)
|
||||
|
||||
assert _windsurf_config_path() == Path("/mnt/c/Users/alice/.codeium/windsurf/mcp_config.json")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue