admin-valentin/arcade-mcp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: added connect cli command (#819)

Browse source 
Summary

- New arcade connect command that logs in, creates/reuses an Arcade
Cloud gateway, and configures your MCP client in one step
- Supports 5 clients: Claude Desktop, Cursor, VS Code, Windsurf, Amazon
Q
- Selection modes: --toolkit, --tool, --preset, --gateway, --all, or
interactive picker
  - Reuses existing gateways when one already covers the requested tools
- Resolves gateway names to slugs (--gateway opencode finds slug
pascal_opencode)
- OAuth auth by default, --api-key fallback with auto-created project
key
  - --slug option to set a custom gateway slug on creation
- Tool catalog cached to ~/.arcade/cache/tools.json (5min TTL, scoped to
org/project)
- Fills in the three previously placeholder configure_*_arcade()
functions
  
  
  ```bash
❯ uv run arcade connect cursor --toolkit x
Fetching tool catalog...

Setting up gateway for toolkits: x

Checking existing gateways...
Found existing gateway: quickstart-x (slug:
gw_3CHqdAlQXSSQ28soevSheOJvXzs)

Configuring cursor to connect to gateway: gw_3CHqdAlQXSSQ28soevSheOJvXzs

Configured Cursor with Arcade gateway 'x'
Gateway URL: https://api.arcade.dev/mcp/gw_3CHqdAlQXSSQ28soevSheOJvXzs
   Config file: /Users/pascal/.cursor/mcp.json
   Restart Cursor for changes to take effect.

Setup complete!
Gateway URL: https://api.arcade.dev/mcp/gw_3CHqdAlQXSSQ28soevSheOJvXzs
   Auth: OAuth (handled by your MCP client)

Try asking your AI assistant:
   - Post a tweet saying 'Hello from Arcade!'
   - Search recent tweets about AI tools
  ```

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Adds a new end-to-end flow that performs OAuth login, calls Arcade
Engine/Coordinator APIs (gateway + API key creation), and writes MCP
client config files, so failures could affect remote resource creation
and local client configuration.
> 
> **Overview**
> Adds a new `arcade connect` CLI command that logs in (if needed),
fetches/caches the user’s tool catalog, creates or reuses an Arcade
Cloud gateway (optionally with a custom `--slug`), and writes the
appropriate MCP client config to point at the gateway.
> 
> Implements real Arcade Cloud gateway configuration for `claude`,
`cursor`, and `vscode` (replacing prior placeholders) and extends
support to **Windsurf** and **Amazon Q**, including optional `--api-key`
mode that auto-creates a project API key and writes it as a `Bearer`
header.
> 
> Refocuses `arcade configure` on *local filesystem* servers (and nudges
remote usage to `connect`), adds toolkit config helpers, expands test
coverage for gateway/toolkit configuration and the new connect flow, and
bumps the package version to `1.14.0`.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
d9357c144a8bddd05dfb39f9f922f577bdbb8bf0. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->
This commit is contained in:
Pascal Matthiesen 2026-04-15 13:16:50 -07:00 committed by GitHub
parent 1492c80fc5
commit 8f4fb1ad77
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 2425 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

435
libs/arcade-cli/arcade_cli/configure.py
View file

@ -199,6 +199,16 @@ def get_vscode_config_path() -> Path:
return Path.home() / ".config" / "Code" / "User" / "mcp.json"
def get_windsurf_config_path() -> Path:
"""Get the Windsurf (Codeium) configuration file path."""
return Path.home() / ".codeium" / "windsurf" / "mcp_config.json"
def get_amazonq_config_path() -> Path:
"""Get the Amazon Q Developer configuration file path."""
return Path.home() / ".aws" / "amazonq" / "mcp.json"
def is_uv_installed() -> bool:
"""Check if uv is installed and available in PATH."""
return shutil.which("uv") is not None
@ -328,13 +338,61 @@ def configure_claude_local(
console.print(" Restart Claude Desktop for changes to take effect.", style="yellow")
def configure_claude_arcade(
server_name: str, transport: str, config_path: Path | None = None
def _configure_mcpservers_arcade(
server_name: str,
gateway_url: str,
auth_token: str | None,
config_path: Path,
display_name: str,
) -> None:
"""Configure Claude Desktop to add an Arcade Cloud MCP server to the configuration."""
# This would connect to the Arcade Cloud to get the server URL
# For now, this is a placeholder
console.print("[red]Connecting to Arcade Cloud servers not yet implemented[/red]")
"""Shared helper for clients that use the ``mcpServers`` JSON key.
Used by Claude Desktop, Windsurf, and Amazon Q which all share
the same config format — only the file path and display name differ.
"""
if not config_path.is_absolute():
config_path = Path.cwd() / config_path
config_path.parent.mkdir(parents=True, exist_ok=True)
config: dict = {}
if config_path.exists():
with open(config_path, encoding="utf-8") as f:
config = json.load(f)
if "mcpServers" not in config:
config["mcpServers"] = {}
_warn_overwrite(config, "mcpServers", server_name, config_path)
entry: dict = {"url": gateway_url}
if auth_token:
entry["headers"] = {"Authorization": f"Bearer {auth_token}"}
config["mcpServers"][server_name] = entry
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
console.print(f"[green]Configured {display_name} with Arcade gateway '{server_name}'[/green]")
console.print(f" Gateway URL: {gateway_url}", style="dim")
console.print(f" Config file: {_format_path_for_display(config_path)}", style="dim")
console.print(f" Restart {display_name} for changes to take effect.", style="yellow")
def configure_claude_arcade(
server_name: str,
gateway_url: str,
auth_token: str | None = None,
config_path: Path | None = None,
) -> None:
"""Configure Claude Desktop to connect to an Arcade Cloud MCP gateway."""
_configure_mcpservers_arcade(
server_name,
gateway_url,
auth_token,
config_path or get_claude_config_path(),
"Claude Desktop",
)
def configure_cursor_local(
@ -422,10 +480,55 @@ def configure_cursor_local(
def configure_cursor_arcade(
server_name: str, transport: str, config_path: Path | None = None
server_name: str,
gateway_url: str,
auth_token: str | None = None,
config_path: Path | None = None,
) -> None:
"""Configure Cursor to add an Arcade Cloud MCP server to the configuration."""
console.print("[red]Connecting to Arcade Cloud servers not yet implemented[/red]")
"""Configure Cursor to connect to an Arcade Cloud MCP gateway."""
if config_path is not None:
target_paths = [config_path]
elif platform.system() == "Windows":
primary_path = get_cursor_config_path()
target_paths = _dedupe_paths([primary_path, *_get_windows_cursor_config_paths()])
else:
target_paths = [get_cursor_config_path()]
resolved_target_paths: list[Path] = []
for path in target_paths:
resolved_target_paths.append(path if path.is_absolute() else Path.cwd() / path)
server_config: dict = {"type": "sse", "url": gateway_url}
if auth_token:
server_config["headers"] = {"Authorization": f"Bearer {auth_token}"}
for idx, target in enumerate(resolved_target_paths):
target.parent.mkdir(parents=True, exist_ok=True)
config: dict = {}
if target.exists():
with open(target, encoding="utf-8") as f:
config = json.load(f)
if "mcpServers" not in config:
config["mcpServers"] = {}
if idx == 0:
_warn_overwrite(config, "mcpServers", server_name, target)
config["mcpServers"][server_name] = server_config
with open(target, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
primary_config_path = resolved_target_paths[0]
console.print(f"[green]Configured Cursor with Arcade gateway '{server_name}'[/green]")
console.print(f" Gateway URL: {gateway_url}", style="dim")
console.print(
f" Config file: {_format_path_for_display(primary_config_path)}",
style="dim",
)
console.print(" Restart Cursor for changes to take effect.", style="yellow")
def configure_vscode_local(
@ -495,9 +598,294 @@ def configure_vscode_local(
console.print(" Restart VS Code for changes to take effect.", style="yellow")
def configure_vscode_arcade(server_name: str, transport: str, path: Path | None = None) -> None:
"""Configure VS Code to add an Arcade Cloud MCP server to the configuration."""
console.print("[red]Connecting to Arcade Cloud servers not yet implemented[/red]")
def configure_vscode_arcade(
server_name: str,
gateway_url: str,
auth_token: str | None = None,
config_path: Path | None = None,
) -> None:
"""Configure VS Code to connect to an Arcade Cloud MCP gateway."""
config_path = config_path or get_vscode_config_path()
if config_path and not config_path.is_absolute():
config_path = Path.cwd() / config_path
config_path.parent.mkdir(parents=True, exist_ok=True)
config: dict = {}
if config_path.exists():
with open(config_path, encoding="utf-8") as f:
try:
config = json.load(f)
except json.JSONDecodeError as e:
raise ValueError(
f"\n\tFailed to load MCP configuration file at {_format_path_for_display(config_path)} "
f"\n\tThe file contains invalid JSON: {e}. "
"\n\tPlease check the file format or delete it to create a new configuration."
)
if "servers" not in config:
config["servers"] = {}
_warn_overwrite(config, "servers", server_name, config_path)
entry: dict = {"type": "http", "url": gateway_url}
if auth_token:
entry["headers"] = {"Authorization": f"Bearer {auth_token}"}
config["servers"][server_name] = entry
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
console.print(f"[green]Configured VS Code with Arcade gateway '{server_name}'[/green]")
console.print(f" Gateway URL: {gateway_url}", style="dim")
console.print(f" Config file: {_format_path_for_display(config_path)}", style="dim")
console.print(" Restart VS Code for changes to take effect.", style="yellow")
def configure_windsurf_arcade(
server_name: str,
gateway_url: str,
auth_token: str | None = None,
config_path: Path | None = None,
) -> None:
"""Configure Windsurf to connect to an Arcade Cloud MCP gateway."""
_configure_mcpservers_arcade(
server_name, gateway_url, auth_token, config_path or get_windsurf_config_path(), "Windsurf"
)
def configure_amazonq_arcade(
server_name: str,
gateway_url: str,
auth_token: str | None = None,
config_path: Path | None = None,
) -> None:
"""Configure Amazon Q Developer to connect to an Arcade Cloud MCP gateway."""
_configure_mcpservers_arcade(
server_name, gateway_url, auth_token, config_path or get_amazonq_config_path(), "Amazon Q"
)
def get_toolkit_stdio_config(tool_packages: list[str], server_name: str) -> dict:
"""Build a stdio config that runs ``arcade mcp stdio`` with ``--tool-package`` flags.
This configuration is used by MCP clients (Claude Desktop, Cursor, VS Code) to
launch an Arcade MCP server via ``uv tool run`` (or direct Python) with one or more
toolkit packages loaded.
"""
uv_executable = shutil.which("uv")
if uv_executable:
args = ["tool", "run", "arcade-mcp", "mcp", "stdio"]
for pkg in tool_packages:
args.extend(["--tool-package", pkg])
return {
"command": uv_executable,
"args": args,
"env": get_tool_secrets(),
}
else:
import sys
args = ["-m", "arcade_mcp_server", "stdio"]
for pkg in tool_packages:
args.extend(["--tool-package", pkg])
return {
"command": sys.executable,
"args": args,
"env": get_tool_secrets(),
}
def get_toolkit_http_config(client: str, tool_packages: list[str], port: int = 8000) -> dict:
"""Build an HTTP/SSE config entry pointing at a local ``arcade mcp http`` server.
The server must be started separately, e.g.::
arcade mcp http --tool-package github --port 8000
Each MCP client uses a slightly different JSON shape:
- Claude Desktop / Cursor: ``url`` (+ optional ``type`` for Cursor)
- VS Code: ``type: "http"`` + ``url``
"""
url = f"http://localhost:{port}/mcp"
client_lower = client.lower()
if client_lower == "cursor":
return {"type": "sse", "url": url}
elif client_lower == "vscode":
return {"type": "http", "url": url}
else:
# Claude Desktop and anything else: just url
return {"url": url}
def configure_client_gateway(
client: str,
server_name: str,
gateway_url: str,
auth_token: str | None = None,
config_path: Path | None = None,
) -> None:
"""Configure an MCP client to connect to an Arcade Cloud gateway.
If *auth_token* is ``None`` the config contains only the URL and the MCP
client handles OAuth natively. If an API key is provided it is written as
a ``Bearer`` header.
"""
client_lower = client.lower()
dispatch = {
"claude": configure_claude_arcade,
"cursor": configure_cursor_arcade,
"vscode": configure_vscode_arcade,
"windsurf": configure_windsurf_arcade,
"amazonq": configure_amazonq_arcade,
}
func = dispatch.get(client_lower)
if not func:
supported = ", ".join(sorted(dispatch))
raise typer.BadParameter(f"Unknown client: {client}. Supported clients: {supported}.")
func(server_name, gateway_url, auth_token, config_path)
def configure_client_toolkit(
client: str,
server_name: str,
tool_packages: list[str],
config_path: Path | None = None,
transport: str = "stdio",
port: int = 8000,
) -> None:
"""Configure an MCP client for Arcade toolkits.
When *transport* is ``"stdio"`` (default), writes a config that launches
``arcade mcp stdio --tool-package <pkg>`` via the MCP client.
When *transport* is ``"http"``, writes a config pointing the client at
``http://localhost:{port}/mcp``. The user must start the server separately::
arcade mcp http --tool-package <pkg> --port <port>
"""
client_lower = client.lower()
if transport == "http":
server_config = get_toolkit_http_config(client, tool_packages, port)
else:
server_config = get_toolkit_stdio_config(tool_packages, server_name)
if client_lower == "claude":
_config_path = config_path or get_claude_config_path()
if _config_path and not _config_path.is_absolute():
_config_path = Path.cwd() / _config_path
_config_path.parent.mkdir(parents=True, exist_ok=True)
config: dict = {}
if _config_path.exists():
with open(_config_path, encoding="utf-8") as f:
config = json.load(f)
if "mcpServers" not in config:
config["mcpServers"] = {}
_warn_overwrite(config, "mcpServers", server_name, _config_path)
config["mcpServers"][server_name] = server_config
with open(_config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
console.print(
f"[green]Configured Claude Desktop with Arcade toolkits: {', '.join(tool_packages)}[/green]"
)
console.print(f" Config file: {_format_path_for_display(_config_path)}", style="dim")
console.print(" Restart Claude Desktop for changes to take effect.", style="yellow")
elif client_lower == "cursor":
if config_path is not None:
target_paths = [config_path]
elif platform.system() == "Windows":
primary_path = get_cursor_config_path()
target_paths = _dedupe_paths([primary_path, *_get_windows_cursor_config_paths()])
else:
target_paths = [get_cursor_config_path()]
resolved_paths: list[Path] = []
for path in target_paths:
resolved_paths.append(path if path.is_absolute() else Path.cwd() / path)
for idx, target in enumerate(resolved_paths):
target.parent.mkdir(parents=True, exist_ok=True)
config = {}
if target.exists():
with open(target, encoding="utf-8") as f:
config = json.load(f)
if "mcpServers" not in config:
config["mcpServers"] = {}
if idx == 0:
_warn_overwrite(config, "mcpServers", server_name, target)
config["mcpServers"][server_name] = server_config
with open(target, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
console.print(
f"[green]Configured Cursor with Arcade toolkits: {', '.join(tool_packages)}[/green]"
)
console.print(f" Config file: {_format_path_for_display(resolved_paths[0])}", style="dim")
console.print(" Restart Cursor for changes to take effect.", style="yellow")
elif client_lower == "vscode":
_config_path = config_path or get_vscode_config_path()
if _config_path and not _config_path.is_absolute():
_config_path = Path.cwd() / _config_path
_config_path.parent.mkdir(parents=True, exist_ok=True)
config = {}
if _config_path.exists():
with open(_config_path, encoding="utf-8") as f:
try:
config = json.load(f)
except json.JSONDecodeError as e:
raise ValueError(
f"\n\tFailed to load MCP configuration file at {_format_path_for_display(_config_path)} "
f"\n\tThe file contains invalid JSON: {e}. "
"\n\tPlease check the file format or delete it to create a new configuration."
)
if "servers" not in config:
config["servers"] = {}
_warn_overwrite(config, "servers", server_name, _config_path)
config["servers"][server_name] = server_config
with open(_config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
console.print(
f"[green]Configured VS Code with Arcade toolkits: {', '.join(tool_packages)}[/green]"
)
console.print(f" Config file: {_format_path_for_display(_config_path)}", style="dim")
console.print(" Restart VS Code for changes to take effect.", style="yellow")
elif client_lower in ("windsurf", "amazonq"):
path_fn = (
get_windsurf_config_path if client_lower == "windsurf" else get_amazonq_config_path
)
display = "Windsurf" if client_lower == "windsurf" else "Amazon Q"
_config_path = config_path or path_fn()
if _config_path and not _config_path.is_absolute():
_config_path = Path.cwd() / _config_path
_config_path.parent.mkdir(parents=True, exist_ok=True)
config = {}
if _config_path.exists():
with open(_config_path, encoding="utf-8") as f:
config = json.load(f)
if "mcpServers" not in config:
config["mcpServers"] = {}
_warn_overwrite(config, "mcpServers", server_name, _config_path)
config["mcpServers"][server_name] = server_config
with open(_config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
console.print(
f"[green]Configured {display} with Arcade toolkits: {', '.join(tool_packages)}[/green]"
)
console.print(f" Config file: {_format_path_for_display(_config_path)}", style="dim")
console.print(f" Restart {display} for changes to take effect.", style="yellow")
else:
supported = "claude, cursor, vscode, windsurf, amazonq"
raise typer.BadParameter(f"Unknown client: {client}. Supported clients: {supported}.")
def configure_client(
@ -540,23 +928,22 @@ def configure_client(
client_lower = client.lower()
if host == "arcade":
console.print(
"Use [bold]arcade connect[/bold] to connect to Arcade Cloud gateways.\n"
"Example: [bold]arcade connect claude --gateway my-gateway[/bold]",
style="yellow",
)
return
if client_lower == "claude":
if transport != "stdio":
raise ValueError("Claude Desktop only supports stdio transport via configuration file")
if host == "local":
configure_claude_local(entrypoint_file, server_name, port, config_path)
else:
configure_claude_arcade(server_name, transport, config_path)
configure_claude_local(entrypoint_file, server_name, port, config_path)
elif client_lower == "cursor":
if host == "local":
configure_cursor_local(entrypoint_file, server_name, transport, port, config_path)
else:
configure_cursor_arcade(server_name, transport, config_path)
configure_cursor_local(entrypoint_file, server_name, transport, port, config_path)
elif client_lower == "vscode":
if host == "local":
configure_vscode_local(entrypoint_file, server_name, transport, port, config_path)
else:
configure_vscode_arcade(server_name, transport, config_path)
configure_vscode_local(entrypoint_file, server_name, transport, port, config_path)
else:
raise typer.BadParameter(
f"Unknown client: {client}. Supported clients: claude, cursor, vscode."

798
libs/arcade-cli/arcade_cli/connect.py Normal file
View file

@ -0,0 +1,798 @@
"""Connect command — one-command toolkit + gateway setup for Arcade MCP."""
from __future__ import annotations
import json as _json
import logging
import time
from pathlib import Path
import httpx
from arcade_core.constants import PROD_COORDINATOR_HOST, PROD_ENGINE_HOST
from arcade_cli.console import console
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Tool catalog cache
# ---------------------------------------------------------------------------
_CACHE_DIR = Path.home() / ".arcade" / "cache"
_CACHE_FILE = _CACHE_DIR / "tools.json"
_CACHE_TTL_SECONDS = 300 # 5 minutes
def _get_context_key() -> str:
"""Return a string identifying the active org+project (for cache scoping)."""
try:
from arcade_cli.utils import get_org_project_context
org_id, project_id = get_org_project_context()
except Exception:
return "unknown"
else:
return f"{org_id}:{project_id}"
def _read_cache(debug: bool = False) -> dict[str, list[str]] | None:
"""Return cached toolkit map if the cache file exists, is fresh, and matches the active context."""
try:
if not _CACHE_FILE.exists():
return None
data = _json.loads(_CACHE_FILE.read_text(encoding="utf-8"))
age = time.time() - data.get("ts", 0)
if age > _CACHE_TTL_SECONDS:
if debug:
console.print(f" [dim]Cache expired ({age:.0f}s old)[/dim]")
return None
# Invalidate if org/project changed
cached_ctx = data.get("context")
current_ctx = _get_context_key()
if cached_ctx and cached_ctx != current_ctx:
if debug:
console.print(" [dim]Cache stale (different project context)[/dim]")
return None
if debug:
console.print(f" [dim]Using cached tool catalog ({age:.0f}s old)[/dim]")
return data.get("toolkits", {})
except Exception:
return None
def _write_cache(toolkits: dict[str, list[str]]) -> None:
"""Persist the toolkit map to disk, scoped to the active org/project."""
try:
_CACHE_DIR.mkdir(parents=True, exist_ok=True)
_CACHE_FILE.write_text(
_json.dumps({
"ts": time.time(),
"context": _get_context_key(),
"toolkits": toolkits,
}),
encoding="utf-8",
)
except Exception:
logger.debug("Failed to write tool cache", exc_info=True)
# ---------------------------------------------------------------------------
# Well-known toolkit metadata
# ---------------------------------------------------------------------------
TOOLKIT_EXAMPLES: dict[str, list[str]] = {
"github": [
"List my open pull requests",
"Show recent issues in my repo",
"Create a new issue titled 'Bug: login fails'",
],
"slack": [
"Send a message to #general saying hello",
"List my unread Slack messages",
"Search Slack for messages about deployment",
],
"google": [
"List my upcoming Google Calendar events",
"Search my Gmail for emails from Alice",
"Create a new Google Doc titled 'Meeting Notes'",
],
"linear": [
"Show my assigned Linear issues",
"Create a new Linear issue for the API refactor",
],
"notion": [
"Search my Notion workspace for project plans",
"Create a new Notion page in my workspace",
],
"jira": [
"List my open Jira tickets",
"Create a Jira issue for the backend migration",
],
"spotify": [
"Play my Discover Weekly playlist",
"What song is currently playing?",
],
"x": [
"Post a tweet saying 'Hello from Arcade!'",
"Search recent tweets about AI tools",
],
"reddit": [
"Search Reddit for posts about MCP tools",
"Get the top posts from r/programming",
],
"figma": [
"List my recent Figma files",
"Get comments on my latest Figma design",
],
"atlassian": [
"List my Confluence pages",
"Search Jira for open bugs",
],
"dropbox": [
"List files in my Dropbox root folder",
"Search Dropbox for 'project plan'",
],
"asana": [
"List my Asana tasks",
"Create a new Asana task for the launch",
],
"hubspot": [
"List my recent HubSpot contacts",
"Search HubSpot deals closing this month",
],
"discord": [
"Send a message to my Discord server",
"List channels in my Discord server",
],
"zoom": [
"List my upcoming Zoom meetings",
"Create a Zoom meeting for tomorrow at 2pm",
],
"microsoft": [
"List my recent Outlook emails",
"Search OneDrive for 'quarterly report'",
],
"pagerduty": [
"List my on-call schedules",
"Show recent PagerDuty incidents",
],
}
PRESET_BUNDLES: dict[str, list[str]] = {
"Productivity": ["google", "slack", "notion"],
"Development": ["github", "linear", "jira"],
"Communication": ["slack", "google", "x"],
"Project Management": ["linear", "jira", "notion"],
"DevOps": ["github", "slack", "linear"],
"Social": ["x", "slack", "reddit"],
"Creative": ["spotify", "figma", "notion"],
}
def get_toolkit_examples(toolkits: list[str]) -> list[str]:
"""Return example prompts for the given toolkit names."""
examples: list[str] = []
for tk in toolkits:
tk_lower = tk.lower().replace("arcade-", "").replace("arcade_", "")
if tk_lower in TOOLKIT_EXAMPLES:
examples.extend(TOOLKIT_EXAMPLES[tk_lower][:2])
if not examples:
examples.append("Ask your AI assistant to use one of the configured tools!")
return examples
# ---------------------------------------------------------------------------
# Login helper
# ---------------------------------------------------------------------------
def ensure_login(coordinator_url: str | None = None) -> str:
"""Ensure the user is logged in, triggering OAuth if needed.
Returns the valid access token.
"""
from arcade_cli.authn import (
OAuthLoginError,
check_existing_login,
get_valid_access_token,
perform_oauth_login,
save_credentials_from_whoami,
)
resolved_url = coordinator_url or f"https://{PROD_COORDINATOR_HOST}"
if check_existing_login(suppress_message=True):
return get_valid_access_token(resolved_url)
console.print("You need to log in to Arcade first.\n", style="yellow")
try:
result = perform_oauth_login(
resolved_url,
on_status=lambda msg: console.print(msg, style="dim"),
)
save_credentials_from_whoami(result.tokens, result.whoami, resolved_url)
console.print(f"\nLogged in as {result.email}.", style="bold green")
return get_valid_access_token(resolved_url)
except OAuthLoginError as e:
raise SystemExit(f"Login failed: {e}") from e
# ---------------------------------------------------------------------------
# Arcade API helpers
# ---------------------------------------------------------------------------
def fetch_available_toolkits(
base_url: str | None = None,
debug: bool = False,
skip_cache: bool = False,
) -> dict[str, list[str]]:
"""Fetch tools from the Arcade Engine and group them by toolkit name.
Results are cached to ``~/.arcade/cache/tools.json`` for 5 minutes so
repeated invocations (e.g. interactive → allow-list) are instant.
Returns a dict mapping toolkit names to lists of tool qualified names
(e.g. ``"Github.ListPRs"``).
"""
if not skip_cache:
cached = _read_cache(debug=debug)
if cached is not None:
return cached
from arcadepy import NOT_GIVEN, APIConnectionError
from arcade_cli.utils import compute_base_url, get_arcade_client
url = base_url or compute_base_url(False, False, PROD_ENGINE_HOST, None, default_port=None)
if debug:
console.print(f" [dim]Connecting to Arcade Engine at {url}[/dim]")
client = get_arcade_client(url)
toolkits: dict[str, list[str]] = {}
tool_count = 0
try:
# limit= is the page size, not a cap — the iterator auto-paginates
for tool in client.tools.list(toolkit=NOT_GIVEN, limit=1000):
toolkit_name = getattr(tool.toolkit, "name", None) or "unknown"
tool_name = tool.name or "unknown"
# Gateway API requires qualified names: "ToolkitName.ToolName"
qualified = f"{toolkit_name}.{tool_name}"
toolkits.setdefault(toolkit_name, []).append(qualified)
tool_count += 1
if debug:
console.print(f" [dim] Found tool: {qualified}[/dim]")
except APIConnectionError:
console.print(f"Could not connect to Arcade Engine at {url}.", style="bold red")
except Exception as e:
if debug:
console.print(f" [dim]Error fetching toolkits: {e}[/dim]")
else:
logger.debug("Failed to fetch toolkits: %s", e)
console.print(
"Could not fetch available toolkits from your account.",
style="bold red",
)
if debug:
console.print(
f" [dim]Fetched {tool_count} tools across {len(toolkits)} toolkits: "
f"{list(toolkits.keys())}[/dim]"
)
if toolkits:
_write_cache(toolkits)
return toolkits
def list_gateways(
access_token: str,
base_url: str | None = None,
debug: bool = False,
) -> list[dict]:
"""List existing MCP gateways from the user's project.
Returns a list of gateway dicts (each with ``id``, ``slug``, ``name``,
``tool_filter``, etc.).
"""
from arcade_cli.utils import compute_base_url, get_org_project_context
url = base_url or compute_base_url(False, False, PROD_ENGINE_HOST, None, default_port=None)
org_id, project_id = get_org_project_context()
endpoint = f"{url}/v1/orgs/{org_id}/projects/{project_id}/gateways"
if debug:
console.print(f" [dim]GET {endpoint}[/dim]")
resp = httpx.get(
endpoint,
headers={"Authorization": f"Bearer {access_token}"},
timeout=30,
)
if resp.status_code != 200:
if debug:
console.print(f" [dim]Failed to list gateways: {resp.status_code}[/dim]")
return []
data = resp.json()
return data.get("items", [])
def find_matching_gateway(
gateways: list[dict],
tool_allow_list: list[str],
auth_type: str = "arcade",
debug: bool = False,
) -> dict | None:
"""Find an existing gateway whose allow-list is a superset of *tool_allow_list*
and whose ``auth_type`` matches."""
needed = set(tool_allow_list)
for gw in gateways:
if gw.get("auth_type", "arcade") != auth_type:
continue
existing = set(gw.get("tool_filter", {}).get("allowed_tools", []))
if needed <= existing:
if debug:
console.print(
f" [dim]Found existing gateway '{gw.get('slug')}' "
f"with {len(existing)} tools (covers all {len(needed)} needed)[/dim]"
)
return gw
return None
def create_gateway(
access_token: str,
name: str,
tool_allow_list: list[str],
auth_type: str = "arcade",
slug: str | None = None,
base_url: str | None = None,
debug: bool = False,
) -> dict:
"""Create a new MCP gateway on Arcade Cloud.
Args:
access_token: OAuth access token for the Engine API.
name: Human-readable gateway name.
tool_allow_list: Qualified tool names (e.g. ``"Github.CreateIssue"``).
auth_type: ``"arcade"`` (OAuth, default) or ``"arcade_header"`` (API key).
slug: Custom slug for the gateway URL. Auto-generated if not provided.
base_url: Engine API base URL override.
debug: Print debug output.
Returns the gateway response dict (with ``slug``, ``id``, ``name``, etc.).
"""
from arcade_cli.utils import compute_base_url, get_org_project_context
url = base_url or compute_base_url(False, False, PROD_ENGINE_HOST, None, default_port=None)
org_id, project_id = get_org_project_context()
endpoint = f"{url}/v1/orgs/{org_id}/projects/{project_id}/gateways"
body: dict = {
"name": name,
"auth_type": auth_type,
"tool_filter": {"allowed_tools": tool_allow_list},
}
if slug:
body["slug"] = slug
if debug:
console.print(f" [dim]POST {endpoint}[/dim]")
console.print(f" [dim]Body: {body}[/dim]")
resp = httpx.post(
endpoint,
json=body,
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
},
timeout=30,
)
if debug:
console.print(f" [dim]Response: {resp.status_code}[/dim]")
console.print(f" [dim]{resp.text[:500]}[/dim]")
if resp.status_code not in (200, 201):
raise RuntimeError(f"Failed to create gateway ({resp.status_code}): {resp.text}")
data = resp.json()
# The API may return the gateway directly or wrapped in a list/items envelope
if "slug" in data:
return data
if data.get("items"):
return data["items"][0]
if "id" in data:
return data
if debug:
console.print(f" [dim]Unexpected response shape: {list(data.keys())}[/dim]")
return data
def create_project_api_key(
access_token: str,
label: str = "quickstart",
debug: bool = False,
) -> str:
"""Create a project API key via the Coordinator.
Calls ``POST /v1/projects/{project_id}/api_keys``.
Returns the raw API key string (``arc_...``).
"""
from arcade_cli.utils import get_org_project_context
_, project_id = get_org_project_context()
coordinator_url = f"https://{PROD_COORDINATOR_HOST}"
endpoint = f"{coordinator_url}/v1/projects/{project_id}/api_keys"
if debug:
console.print(f" [dim]POST {endpoint}[/dim]")
resp = httpx.post(
endpoint,
json={"label": label},
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
},
timeout=30,
)
if debug:
console.print(f" [dim]Response: {resp.status_code}[/dim]")
console.print(f" [dim]{resp.text[:300]}[/dim]")
if resp.status_code not in (200, 201):
raise RuntimeError(f"Failed to create API key ({resp.status_code}): {resp.text}")
return resp.json()["api_key"]
# ---------------------------------------------------------------------------
# Interactive selection
# ---------------------------------------------------------------------------
def prompt_toolkit_selection(available: dict[str, list[str]]) -> list[str]:
"""Interactively prompt the user to select toolkits.
Returns a list of selected toolkit names.
"""
if not available:
console.print("No toolkits available in your account.", style="bold red")
raise SystemExit(1)
console.print("\n[bold]Available toolkits:[/bold]\n")
# Case-insensitive lookup: preset says "github", API returns "Github"
avail_lower: dict[str, str] = {k.lower(): k for k in available}
# Show preset bundles first
bundle_choices: list[tuple[str, list[str]]] = []
for bundle_name, bundle_tks in PRESET_BUNDLES.items():
# Resolve each preset toolkit to its actual API key
matching = [avail_lower[t] for t in bundle_tks if t in avail_lower]
if matching:
bundle_choices.append((bundle_name, matching))
sorted_toolkits = sorted(available.keys())
# Number the options
idx = 1
option_map: dict[int, list[str]] = {}
for bundle_name, bundle_tks in bundle_choices:
tool_count = sum(len(available.get(t, [])) for t in bundle_tks)
display_names = ", ".join(t.lower() for t in bundle_tks)
console.print(
f" [bold cyan]{idx}.[/bold cyan] {bundle_name} bundle "
f"({display_names}) — {tool_count} tools"
)
option_map[idx] = bundle_tks
idx += 1
if bundle_choices:
console.print()
for tk_name in sorted_toolkits:
tools = available[tk_name]
console.print(f" [bold cyan]{idx}.[/bold cyan] {tk_name} — {len(tools)} tools")
option_map[idx] = [tk_name]
idx += 1
console.print(f"\n [bold cyan]{idx}.[/bold cyan] All available toolkits")
option_map[idx] = sorted_toolkits
console.print()
try:
raw = input("Select toolkits (comma-separated numbers, e.g. 1,3): ").strip()
except (EOFError, KeyboardInterrupt):
raise SystemExit("\nCancelled.")
selected: list[str] = []
for part in raw.split(","):
part = part.strip()
if not part:
continue
try:
choice = int(part)
except ValueError:
console.print(f" Skipping invalid choice: {part}", style="yellow")
continue
if choice in option_map:
for tk in option_map[choice]:
if tk not in selected:
selected.append(tk)
else:
console.print(f" Skipping unknown option: {choice}", style="yellow")
if not selected:
console.print("No toolkits selected.", style="bold red")
raise SystemExit(1)
return selected
# ---------------------------------------------------------------------------
# Main orchestrator
# ---------------------------------------------------------------------------
def run_connect(
client: str,
toolkits: list[str] | None = None,
tools: list[str] | None = None,
gateway: str | None = None,
all_tools: bool = False,
use_api_key: bool = False,
gateway_slug: str | None = None,
config_path: Path | None = None,
debug: bool = False,
) -> None:
"""Run the quickstart flow: login → determine mode → configure client.
Everything is configured as a cloud gateway — no local server required.
Args:
toolkits: Whole toolkit names (e.g. ``["github"]``) — adds all tools.
tools: Individual qualified tool names (e.g. ``["Github.CreateIssue"]``).
gateway: Existing gateway slug to connect to directly.
"""
# Step 1: Ensure login
access_token = ensure_login()
# Step 2: Determine mode
if gateway:
# --- Direct gateway mode (existing gateway) ---
# Resolve the input: user may pass a name ("opencode") or slug ("pascal_opencode")
slug = _resolve_gateway_slug(gateway, access_token, debug=debug)
api_key: str | None = None
if use_api_key:
console.print("Creating project API key...", style="dim")
api_key = create_project_api_key(access_token, label=f"connect-{slug}", debug=debug)
console.print(" API key created.", style="green")
_configure_gateway(client, slug, config_path, api_key=api_key, name=gateway)
return
# --- Toolkit / tool → gateway mode ---
# If only --tool is given (no --toolkit, --all, or interactive), skip catalog fetch
if tools and not toolkits and not all_tools:
tool_allow_list = list(tools)
selected_toolkits = sorted({t.split(".")[0] for t in tools})
console.print(
f"\nSetting up gateway with {len(tool_allow_list)} individual tool(s): "
f"[bold]{', '.join(tool_allow_list)}[/bold]\n"
)
else:
selected_toolkits_list: list[str]
# Fetch the tool catalog once — used for selection and allow-list
console.print("Fetching tool catalog...", style="dim")
available = fetch_available_toolkits(debug=debug)
if toolkits:
selected_toolkits_list = toolkits
elif all_tools:
if not available:
console.print(
"No toolkits found. Make sure you have tools deployed in your Arcade account.",
style="bold red",
)
raise SystemExit(1)
selected_toolkits_list = sorted(available.keys())
else:
# Interactive mode
if not available:
console.print(
"No toolkits found in your account. You can specify toolkits manually:\n"
" [bold]arcade connect claude --toolkit github[/bold]",
style="yellow",
)
raise SystemExit(1)
selected_toolkits_list = prompt_toolkit_selection(available)
selected_toolkits = selected_toolkits_list
console.print(
f"\nSetting up gateway for toolkits: [bold]{', '.join(selected_toolkits)}[/bold]\n"
)
# Build a case-insensitive lookup: "github" -> "Github", etc.
tk_lower_map: dict[str, str] = {k.lower(): k for k in available}
if debug:
console.print(f" [dim]Available toolkit keys: {list(available.keys())}[/dim]")
console.print(f" [dim]Looking for: {selected_toolkits}[/dim]")
tool_allow_list = []
for tk in selected_toolkits:
actual_key = tk_lower_map.get(tk.lower())
if actual_key:
tk_tools = available[actual_key]
tool_allow_list.extend(tk_tools)
if debug:
console.print(
f" [dim]Matched '{tk}' -> '{actual_key}' ({len(tk_tools)} tools)[/dim]"
)
else:
console.print(f" [yellow]Warning: No tools found for toolkit '{tk}'.[/yellow]")
if available:
console.print(
f" [yellow]Available toolkits: "
f"{', '.join(sorted(available.keys()))}[/yellow]"
)
# Append any individual --tool names
if tools:
for t in tools:
if t not in tool_allow_list:
tool_allow_list.append(t)
if debug:
console.print(f" [dim]Added {len(tools)} individual tool(s)[/dim]")
if not tool_allow_list:
console.print(
"\nNo tools to add to the gateway. Deploy toolkits first with "
"[bold]arcade deploy[/bold].",
style="bold red",
)
raise SystemExit(1)
# Check if an existing gateway already covers these tools
auth_type = "arcade_header" if use_api_key else "arcade"
console.print("Checking existing gateways...", style="dim")
existing_gateways = list_gateways(access_token, debug=debug)
existing = find_matching_gateway(
existing_gateways, tool_allow_list, auth_type=auth_type, debug=debug
)
if existing:
slug = existing["slug"]
console.print(
f" Found existing gateway: [bold]{existing.get('name', slug)}[/bold] (slug: {slug})\n",
style="green",
)
else:
# Create a new gateway
if len(selected_toolkits) == 1:
gateway_name = selected_toolkits[0].lower()
else:
gateway_name = "-".join(sorted({tk.lower() for tk in selected_toolkits}))
console.print(
f"Creating gateway '{gateway_name}' with {len(tool_allow_list)} tools "
f"(auth: {auth_type})...",
style="dim",
)
gw = create_gateway(
access_token=access_token,
name=gateway_name,
tool_allow_list=tool_allow_list,
auth_type=auth_type,
slug=gateway_slug,
debug=debug,
)
slug = gw.get("slug", gateway_name)
if debug:
console.print(f" [dim]Gateway response: id={gw.get('id')}, slug={slug}[/dim]")
console.print(f" Gateway created: [bold]{slug}[/bold]\n", style="green")
# For API-key auth, create a project key and include it in the config
api_key: str | None = None
if use_api_key:
console.print("Creating project API key...", style="dim")
api_key = create_project_api_key(access_token, label=f"connect-{slug}", debug=debug)
console.print(" API key created.", style="green")
# Config key: prefer --slug if given, otherwise derive from toolkit names
if gateway_slug:
display_name = gateway_slug
elif len(selected_toolkits) == 1:
display_name = selected_toolkits[0].lower()
else:
display_name = "-".join(sorted({tk.lower() for tk in selected_toolkits}))
_configure_gateway(client, slug, config_path, api_key=api_key, name=display_name)
# Print examples
examples = get_toolkit_examples(selected_toolkits)
console.print("\nTry asking your AI assistant:", style="bold")
for ex in examples[:3]:
console.print(f" - {ex}", style="dim")
def _resolve_gateway_slug(
user_input: str,
access_token: str,
debug: bool = False,
) -> str:
"""Resolve a gateway name or slug to the actual slug.
The user may pass a name (``opencode``) or a slug (``pascal_opencode``).
We look up existing gateways and match by slug first, then by name.
Falls back to the original input if no match is found.
"""
gateways = list_gateways(access_token, debug=debug)
input_lower = user_input.lower()
for gw in gateways:
if gw.get("slug", "").lower() == input_lower:
if debug:
console.print(f" [dim]Matched by slug: {gw['slug']}[/dim]")
return gw["slug"]
for gw in gateways:
if gw.get("name", "").lower() == input_lower:
slug = gw["slug"]
if debug:
console.print(f" [dim]Matched by name '{gw['name']}' -> slug: {slug}[/dim]")
return slug
if debug:
available = [f"{g.get('name')} ({g.get('slug')})" for g in gateways]
console.print(f" [dim]No match for '{user_input}', available: {available}[/dim]")
return user_input
def _configure_gateway(
client: str,
slug: str,
config_path: Path | None,
api_key: str | None = None,
name: str | None = None,
) -> None:
"""Configure the MCP client to connect to a gateway by slug.
*name* is the human-readable label used as the config key (e.g. ``github``).
Defaults to *slug* if not provided.
"""
from arcade_cli.configure import configure_client_gateway
from arcade_cli.utils import compute_base_url
api_base = compute_base_url(False, False, PROD_ENGINE_HOST, None, default_port=None)
gateway_url = f"{api_base}/mcp/{slug}"
server_name = name or slug
console.print(f"Configuring [bold]{client}[/bold] to connect to gateway: [bold]{slug}[/bold]\n")
configure_client_gateway(
client=client,
server_name=server_name,
gateway_url=gateway_url,
auth_token=api_key,
config_path=config_path,
)
console.print("\n[bold green]Setup complete![/bold green]")
console.print(f" Gateway URL: {gateway_url}", style="dim")
if api_key:
console.print(" Auth: API key (included in config)", style="dim")
else:
console.print(" Auth: OAuth (handled by your MCP client)", style="dim")

133
libs/arcade-cli/arcade_cli/main.py
View file

@ -693,7 +693,10 @@ def evals(
handle_cli_error("Failed to run evaluations", e, debug)
@cli.command(help="Configure MCP clients to connect to your server", rich_help_panel="Manage")
@cli.command(
help="Configure an MCP client to use a local server on your filesystem",
rich_help_panel="Manage",
)
def configure(
client: str = typer.Argument(
...,
@ -727,7 +730,7 @@ def configure(
"local",
"--host",
"-h",
help="The host of the HTTP server to configure. Use 'local' to connect to a local MCP server or 'arcade' to connect to an Arcade Cloud MCP server.",
help="The host for HTTP transport. Use 'local' for a local server. ('arcade' is supported but 'arcade connect' is the recommended way to set up remote gateways.)",
click_type=click.Choice(["local", "arcade"], case_sensitive=False),
show_choices=True,
rich_help_panel="HTTP Options",
@ -750,16 +753,19 @@ def configure(
debug: bool = typer.Option(False, "--debug", "-d", help="Show debug information"),
) -> None:
"""
Configure MCP clients to connect to your server.
Configure an MCP client to use a local server on your filesystem.
The default behavior is to configure the specified client for a local stdio server that
runs when the server.py file in the current directory is invoked directly.
Points your MCP client at a server you are developing or running locally.
By default, configures a stdio transport that launches the server.py file
in the current directory. Use --transport http for a running local HTTP server.
To connect to remote Arcade Cloud gateways instead, use 'arcade connect'.
Examples:
arcade configure claude
arcade configure cursor --transport http --port 8080
arcade configure vscode --host arcade --entrypoint my_server.py --config .vscode/mcp.json
arcade configure claude --host local --name my_server_name
arcade configure vscode --entrypoint my_server.py --config .vscode/mcp.json
arcade configure claude --name my_server_name
"""
from arcade_cli.configure import configure_client
@ -777,6 +783,118 @@ def configure(
handle_cli_error(f"Failed to configure {client}", e, debug)
@cli.command(
name="connect",
help="Connect an MCP client to a remote Arcade Cloud gateway",
rich_help_panel="Run",
)
def connect(
client: str = typer.Argument(
...,
help="MCP client to connect to the remote gateway",
click_type=click.Choice(
["claude", "cursor", "vscode", "windsurf", "amazonq"],
case_sensitive=False,
),
show_choices=True,
),
server: Optional[list[str]] = typer.Option(
None,
"--server",
"-t",
help="Server(s) to set up — adds all tools from each server. Can be repeated.",
),
tool: Optional[list[str]] = typer.Option(
None,
"--tool",
help="Individual tool(s) by qualified name (e.g., Github.CreateIssue). Can be repeated.",
),
preset: Optional[str] = typer.Option(
None,
"--preset",
help="Use a preset bundle (productivity, development, communication, devops, social, creative, project-management).",
),
gateway: Optional[str] = typer.Option(
None,
"--gateway",
"-g",
help="Connect to an Arcade Cloud gateway by slug instead of local toolkits.",
),
all_tools: bool = typer.Option(
False,
"--all",
help="Set up all available toolkits from your account without prompting.",
),
slug: Optional[str] = typer.Option(
None,
"--slug",
"-s",
help="Custom slug for the created gateway (only with --server/--tool/--preset).",
),
api_key: bool = typer.Option(
False,
"--api-key",
help="Use API-key auth instead of OAuth. Creates a project API key and includes it in the client config.",
),
config_path: Optional[Path] = typer.Option(
None,
"--config",
"-c",
exists=False,
help="Custom path to the MCP client config file (overrides default).",
),
debug: bool = typer.Option(False, "--debug", "-d", help="Show debug information"),
) -> None:
"""
Connect an MCP client to a remote Arcade Cloud gateway.
No local server needed — tools run in the cloud. Logs you in (if needed),
creates an Arcade Cloud gateway for the selected toolkits, and writes your
MCP client config, all in one step.
By default gateways use OAuth (the MCP client handles the auth flow).
Pass --api-key to use API-key auth instead (creates a key automatically).
To configure a local server on your filesystem instead, use 'arcade configure'.
Examples:\n
arcade connect claude --server github\n
arcade connect cursor --preset productivity\n
arcade connect claude --tool Github.CreateIssue --tool Linear.UpdateIssue\n
arcade connect claude --gateway my-existing-gw\n
arcade connect vscode --all --api-key\n
"""
from arcade_cli.connect import PRESET_BUNDLES, run_connect
# Resolve --preset to toolkit list
resolved_toolkits = list(server) if server else None
if preset:
preset_lower = preset.lower().replace("-", " ")
match = {k.lower(): v for k, v in PRESET_BUNDLES.items()}.get(preset_lower)
if not match:
available = ", ".join(k.lower().replace(" ", "-") for k in PRESET_BUNDLES)
handle_cli_error(f"Unknown preset '{preset}'. Available presets: {available}")
return
resolved_toolkits = (resolved_toolkits or []) + match
try:
run_connect(
client=client,
toolkits=resolved_toolkits,
tools=list(tool) if tool else None,
gateway=gateway,
all_tools=all_tools,
use_api_key=api_key,
gateway_slug=slug,
config_path=config_path,
debug=debug,
)
except SystemExit:
raise
except Exception as e:
handle_cli_error("Quickstart failed", e, debug)
@cli.command(
name="deploy",
help="Deploy MCP servers to Arcade",
@ -1005,6 +1123,7 @@ def main_callback(
new.__name__,
show.__name__,
configure.__name__,
connect.__name__,
update.__name__,
upgrade.__name__,
}

328
libs/tests/cli/test_configure.py
View file

@ -1,4 +1,4 @@
"""Tests for get_tool_secrets() in arcade configure."""
"""Tests for get_tool_secrets() and gateway configuration in arcade configure."""
import json
import sys
@ -11,8 +11,17 @@ from arcade_cli.configure import (
_format_path_for_display,
_resolve_windows_appdata,
_warn_overwrite,
configure_amazonq_arcade,
configure_claude_arcade,
configure_client,
configure_client_gateway,
configure_client_toolkit,
configure_cursor_arcade,
configure_vscode_arcade,
configure_windsurf_arcade,
get_tool_secrets,
get_toolkit_http_config,
get_toolkit_stdio_config,
)
@ -89,8 +98,7 @@ def test_format_path_for_display_posix_escapes() -> None:
else:
path = Path("/tmp/with space/mcp.json")
assert (
_format_path_for_display(path, platform_system="Linux")
== "/tmp/with\\ space/mcp.json"
_format_path_for_display(path, platform_system="Linux") == "/tmp/with\\ space/mcp.json"
)
@ -108,9 +116,7 @@ def test_resolve_windows_appdata_delegates_to_platformdirs(
monkeypatch.delenv("USERPROFILE", raising=False)
fake_platformdirs = types.ModuleType("platformdirs")
fake_platformdirs.user_data_dir = (
lambda *args, **kwargs: r"C:\Users\Alice\AppData\Roaming"
)
fake_platformdirs.user_data_dir = lambda *args, **kwargs: r"C:\Users\Alice\AppData\Roaming"
monkeypatch.setitem(sys.modules, "platformdirs", fake_platformdirs)
assert _resolve_windows_appdata() == Path(r"C:\Users\Alice\AppData\Roaming")
@ -145,7 +151,9 @@ def test_resolve_windows_appdata_handles_older_platformdirs(
assert len(received_args) == 1, "Fallback must make exactly one positional call"
fallback_args = received_args[0]
# args: (None, False, None, True) — roaming is the 4th positional arg
assert len(fallback_args) == 4, f"Expected 4 positional args, got {len(fallback_args)}: {fallback_args}"
assert len(fallback_args) == 4, (
f"Expected 4 positional args, got {len(fallback_args)}: {fallback_args}"
)
assert fallback_args[3] is True, f"4th arg (roaming) must be True, got {fallback_args[3]}"
assert fallback_args[2] is None, f"3rd arg (version) must be None, got {fallback_args[2]}"
@ -273,7 +281,9 @@ def test_config_written_as_utf8(tmp_path: Path, monkeypatch: pytest.MonkeyPatch)
assert "demo" in data["mcpServers"]
def test_config_roundtrip_preserves_unicode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
def test_config_roundtrip_preserves_unicode(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Write a config with Unicode, then overwrite and verify it still decodes."""
monkeypatch.chdir(tmp_path)
_write_entrypoint(tmp_path)
@ -507,3 +517,305 @@ def test_claude_config_stdio_only(tmp_path: Path, monkeypatch: pytest.MonkeyPatc
port=8000,
config_path=config_path,
)
# ---------------------------------------------------------------------------
# configure_*_arcade() — gateway configuration
# ---------------------------------------------------------------------------
class TestConfigureClaudeArcade:
def test_writes_gateway_url_and_headers(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
configure_claude_arcade(
server_name="my-gw",
gateway_url="https://api.arcade.dev/mcp/my-gw",
auth_token="tok_abc",
config_path=config_path,
)
config = _load_json(config_path)
entry = config["mcpServers"]["my-gw"]
assert entry["url"] == "https://api.arcade.dev/mcp/my-gw"
assert entry["headers"]["Authorization"] == "Bearer tok_abc"
def test_preserves_existing_entries(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
config_path.write_text(
json.dumps({"mcpServers": {"existing": {"command": "old"}}}),
encoding="utf-8",
)
configure_claude_arcade(
server_name="new-gw",
gateway_url="https://api.arcade.dev/mcp/new-gw",
auth_token="tok",
config_path=config_path,
)
config = _load_json(config_path)
assert "existing" in config["mcpServers"]
assert "new-gw" in config["mcpServers"]
class TestConfigureCursorArcade:
def test_writes_sse_config(self, tmp_path: Path) -> None:
config_path = tmp_path / "cursor.json"
configure_cursor_arcade(
server_name="my-gw",
gateway_url="https://api.arcade.dev/mcp/my-gw",
auth_token="tok_abc",
config_path=config_path,
)
config = _load_json(config_path)
entry = config["mcpServers"]["my-gw"]
assert entry["type"] == "sse"
assert entry["url"] == "https://api.arcade.dev/mcp/my-gw"
assert entry["headers"]["Authorization"] == "Bearer tok_abc"
class TestConfigureVscodeArcade:
def test_writes_http_config(self, tmp_path: Path) -> None:
config_path = tmp_path / "vscode.json"
configure_vscode_arcade(
server_name="my-gw",
gateway_url="https://api.arcade.dev/mcp/my-gw",
auth_token="tok_abc",
config_path=config_path,
)
config = _load_json(config_path)
entry = config["servers"]["my-gw"]
assert entry["type"] == "http"
assert entry["url"] == "https://api.arcade.dev/mcp/my-gw"
assert entry["headers"]["Authorization"] == "Bearer tok_abc"
# ---------------------------------------------------------------------------
# configure_client_gateway() — dispatcher
# ---------------------------------------------------------------------------
class TestConfigureClientGateway:
@pytest.mark.parametrize(
"client,section",
[
("claude", "mcpServers"),
("cursor", "mcpServers"),
("vscode", "servers"),
("windsurf", "mcpServers"),
("amazonq", "mcpServers"),
],
)
def test_dispatches_to_correct_client(self, tmp_path: Path, client: str, section: str) -> None:
config_path = tmp_path / f"{client}.json"
configure_client_gateway(
client=client,
server_name="test-gw",
gateway_url="https://api.arcade.dev/mcp/test-gw",
auth_token="tok",
config_path=config_path,
)
config = _load_json(config_path)
assert "test-gw" in config[section]
# ---------------------------------------------------------------------------
# configure_client_toolkit() — toolkit stdio config
# ---------------------------------------------------------------------------
class TestConfigureClientToolkit:
def test_claude_toolkit_stdio(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
configure_client_toolkit(
client="claude",
server_name="arcade-github",
tool_packages=["github"],
config_path=config_path,
transport="stdio",
)
config = _load_json(config_path)
entry = config["mcpServers"]["arcade-github"]
assert "command" in entry
assert "--tool-package" in entry["args"]
assert "github" in entry["args"]
def test_claude_toolkit_http(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
configure_client_toolkit(
client="claude",
server_name="arcade-github",
tool_packages=["github"],
config_path=config_path,
transport="http",
port=8000,
)
config = _load_json(config_path)
entry = config["mcpServers"]["arcade-github"]
assert entry["url"] == "http://localhost:8000/mcp"
assert "command" not in entry
def test_cursor_toolkit_http(self, tmp_path: Path) -> None:
config_path = tmp_path / "cursor.json"
configure_client_toolkit(
client="cursor",
server_name="arcade-github",
tool_packages=["github"],
config_path=config_path,
transport="http",
port=9000,
)
config = _load_json(config_path)
entry = config["mcpServers"]["arcade-github"]
assert entry["type"] == "sse"
assert entry["url"] == "http://localhost:9000/mcp"
def test_vscode_toolkit_stdio(self, tmp_path: Path) -> None:
config_path = tmp_path / "vscode.json"
configure_client_toolkit(
client="vscode",
server_name="arcade-tools",
tool_packages=["github", "slack"],
config_path=config_path,
transport="stdio",
)
config = _load_json(config_path)
entry = config["servers"]["arcade-tools"]
assert "command" in entry
args_str = " ".join(str(a) for a in entry["args"])
assert "github" in args_str
assert "slack" in args_str
def test_vscode_toolkit_http(self, tmp_path: Path) -> None:
config_path = tmp_path / "vscode.json"
configure_client_toolkit(
client="vscode",
server_name="arcade-tools",
tool_packages=["github", "slack"],
config_path=config_path,
transport="http",
)
config = _load_json(config_path)
entry = config["servers"]["arcade-tools"]
assert entry["type"] == "http"
assert entry["url"] == "http://localhost:8000/mcp"
def test_windsurf_toolkit_stdio(self, tmp_path: Path) -> None:
config_path = tmp_path / "windsurf.json"
configure_client_toolkit(
client="windsurf",
server_name="arcade-github",
tool_packages=["github"],
config_path=config_path,
)
config = _load_json(config_path)
entry = config["mcpServers"]["arcade-github"]
assert "command" in entry
assert "--tool-package" in entry["args"]
def test_amazonq_toolkit_stdio(self, tmp_path: Path) -> None:
config_path = tmp_path / "amazonq.json"
configure_client_toolkit(
client="amazonq",
server_name="arcade-github",
tool_packages=["github"],
config_path=config_path,
)
config = _load_json(config_path)
entry = config["mcpServers"]["arcade-github"]
assert "command" in entry
assert "--tool-package" in entry["args"]
# ---------------------------------------------------------------------------
# get_toolkit_stdio_config()
# ---------------------------------------------------------------------------
class TestGetToolkitStdioConfig:
def test_uses_uv_when_available(self, monkeypatch: pytest.MonkeyPatch) -> None:
import arcade_cli.configure as configure_mod
monkeypatch.setattr(
configure_mod.shutil, "which", lambda exe: "/usr/bin/uv" if exe == "uv" else None
)
config = get_toolkit_stdio_config(["github"], "arcade-github")
assert config["command"] == "/usr/bin/uv"
assert "tool" in config["args"]
assert "run" in config["args"]
assert "--tool-package" in config["args"]
assert "github" in config["args"]
def test_falls_back_to_python(self, monkeypatch: pytest.MonkeyPatch) -> None:
import arcade_cli.configure as configure_mod
monkeypatch.setattr(configure_mod.shutil, "which", lambda exe: None)
config = get_toolkit_stdio_config(["github"], "arcade-github")
assert "python" in config["command"].lower() or config["command"].endswith("python3")
assert "--tool-package" in config["args"]
# ---------------------------------------------------------------------------
# get_toolkit_http_config()
# ---------------------------------------------------------------------------
class TestGetToolkitHttpConfig:
def test_claude_config(self) -> None:
config = get_toolkit_http_config("claude", ["github"])
assert config["url"] == "http://localhost:8000/mcp"
assert "type" not in config
def test_cursor_config(self) -> None:
config = get_toolkit_http_config("cursor", ["github"])
assert config["type"] == "sse"
assert config["url"] == "http://localhost:8000/mcp"
def test_vscode_config(self) -> None:
config = get_toolkit_http_config("vscode", ["github"])
assert config["type"] == "http"
assert config["url"] == "http://localhost:8000/mcp"
def test_custom_port(self) -> None:
config = get_toolkit_http_config("claude", ["github"], port=9000)
assert config["url"] == "http://localhost:9000/mcp"
# ---------------------------------------------------------------------------
# New clients: Windsurf, Amazon Q, Zed
# ---------------------------------------------------------------------------
class TestConfigureWindsurfArcade:
def test_writes_mcpservers_config(self, tmp_path: Path) -> None:
config_path = tmp_path / "windsurf.json"
configure_windsurf_arcade(
server_name="my-gw",
gateway_url="https://api.arcade.dev/mcp/my-gw",
config_path=config_path,
)
config = _load_json(config_path)
entry = config["mcpServers"]["my-gw"]
assert entry["url"] == "https://api.arcade.dev/mcp/my-gw"
assert "headers" not in entry
def test_with_api_key(self, tmp_path: Path) -> None:
config_path = tmp_path / "windsurf.json"
configure_windsurf_arcade(
server_name="my-gw",
gateway_url="https://api.arcade.dev/mcp/my-gw",
auth_token="arc_test",
config_path=config_path,
)
config = _load_json(config_path)
assert config["mcpServers"]["my-gw"]["headers"]["Authorization"] == "Bearer arc_test"
class TestConfigureAmazonqArcade:
def test_writes_mcpservers_config(self, tmp_path: Path) -> None:
config_path = tmp_path / "amazonq.json"
configure_amazonq_arcade(
server_name="my-gw",
gateway_url="https://api.arcade.dev/mcp/my-gw",
config_path=config_path,
)
config = _load_json(config_path)
entry = config["mcpServers"]["my-gw"]
assert entry["url"] == "https://api.arcade.dev/mcp/my-gw"

769
libs/tests/cli/test_connect.py Normal file
View file

@ -0,0 +1,769 @@
"""Tests for the arcade connect command."""
from __future__ import annotations
import json
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from arcade_cli.connect import (
_get_context_key,
_read_cache,
_write_cache,
create_gateway,
ensure_login,
fetch_available_toolkits,
find_matching_gateway,
get_toolkit_examples,
list_gateways,
run_connect,
)
# ---------------------------------------------------------------------------
# get_toolkit_examples
# ---------------------------------------------------------------------------
class TestGetToolkitExamples:
def test_known_toolkit_returns_examples(self) -> None:
examples = get_toolkit_examples(["github"])
assert len(examples) == 2
assert any("pull request" in e.lower() for e in examples)
def test_multiple_toolkits(self) -> None:
examples = get_toolkit_examples(["github", "slack"])
assert len(examples) == 4
def test_unknown_toolkit_returns_fallback(self) -> None:
examples = get_toolkit_examples(["nonexistent_toolkit_xyz"])
assert len(examples) == 1
assert "assistant" in examples[0].lower()
def test_strips_arcade_prefix(self) -> None:
examples = get_toolkit_examples(["arcade-github"])
assert len(examples) == 2
def test_empty_list_returns_fallback(self) -> None:
examples = get_toolkit_examples([])
assert len(examples) == 1
# ---------------------------------------------------------------------------
# ensure_login
# ---------------------------------------------------------------------------
class TestEnsureLogin:
@patch("arcade_cli.connect.console")
@patch("arcade_cli.authn.get_valid_access_token", return_value="tok_abc")
@patch("arcade_cli.authn.check_existing_login", return_value=True)
def test_already_logged_in_returns_token(
self, _check: MagicMock, _get_token: MagicMock, _console: MagicMock
) -> None:
token = ensure_login()
assert token == "tok_abc"
@patch("arcade_cli.connect.console")
@patch("arcade_cli.authn.get_valid_access_token", return_value="tok_new")
@patch("arcade_cli.authn.save_credentials_from_whoami")
@patch("arcade_cli.authn.check_existing_login", return_value=False)
def test_not_logged_in_triggers_oauth(
self,
_check: MagicMock,
_save: MagicMock,
_get_token: MagicMock,
_console: MagicMock,
) -> None:
mock_result = MagicMock()
mock_result.email = "user@example.com"
mock_result.tokens = MagicMock()
mock_result.whoami = MagicMock()
with patch(
"arcade_cli.authn.perform_oauth_login",
return_value=mock_result,
):
token = ensure_login()
assert token == "tok_new"
# ---------------------------------------------------------------------------
# fetch_available_toolkits
# ---------------------------------------------------------------------------
class TestFetchAvailableToolkits:
def test_groups_by_toolkit_name(self) -> None:
tool1 = SimpleNamespace(toolkit=SimpleNamespace(name="github"), name="GithubListPRs")
tool2 = SimpleNamespace(toolkit=SimpleNamespace(name="github"), name="GithubCreateIssue")
tool3 = SimpleNamespace(toolkit=SimpleNamespace(name="slack"), name="SlackSendMessage")
mock_client = MagicMock()
mock_client.tools.list.return_value = [tool1, tool2, tool3]
with patch("arcade_cli.utils.get_arcade_client", return_value=mock_client):
result = fetch_available_toolkits("https://api.example.com", skip_cache=True)
assert "github" in result
assert len(result["github"]) == 2
assert "slack" in result
assert len(result["slack"]) == 1
@patch("arcade_cli.connect.console")
def test_connection_error_returns_empty(self, _console: MagicMock) -> None:
from arcadepy import APIConnectionError
mock_client = MagicMock()
mock_client.tools.list.side_effect = APIConnectionError(request=MagicMock())
with patch("arcade_cli.utils.get_arcade_client", return_value=mock_client):
result = fetch_available_toolkits("https://api.example.com", skip_cache=True)
assert result == {}
# ---------------------------------------------------------------------------
# Cache functions
# ---------------------------------------------------------------------------
class TestCache:
def test_write_and_read_cache(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
import arcade_cli.connect as mod
cache_file = tmp_path / "tools.json"
monkeypatch.setattr(mod, "_CACHE_DIR", tmp_path)
monkeypatch.setattr(mod, "_CACHE_FILE", cache_file)
monkeypatch.setattr(mod, "_get_context_key", lambda: "org:proj")
toolkits = {"github": ["Github.CreateIssue"]}
_write_cache(toolkits)
assert cache_file.exists()
result = _read_cache()
assert result == toolkits
def test_read_cache_returns_none_when_missing(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
import arcade_cli.connect as mod
monkeypatch.setattr(mod, "_CACHE_FILE", tmp_path / "nonexistent.json")
assert _read_cache() is None
def test_read_cache_invalidates_on_context_change(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
import arcade_cli.connect as mod
cache_file = tmp_path / "tools.json"
monkeypatch.setattr(mod, "_CACHE_DIR", tmp_path)
monkeypatch.setattr(mod, "_CACHE_FILE", cache_file)
# Write with one context
monkeypatch.setattr(mod, "_get_context_key", lambda: "org1:proj1")
_write_cache({"github": ["Github.CreateIssue"]})
# Read with different context
monkeypatch.setattr(mod, "_get_context_key", lambda: "org2:proj2")
assert _read_cache() is None
def test_get_context_key_returns_unknown_without_credentials(self) -> None:
# On CI or without credentials, should return "unknown" not raise
with patch(
"arcade_cli.utils.get_org_project_context",
side_effect=Exception("no creds"),
):
assert _get_context_key() == "unknown"
# ---------------------------------------------------------------------------
# find_matching_gateway
# ---------------------------------------------------------------------------
class TestFindMatchingGateway:
def test_finds_superset_gateway(self) -> None:
gateways = [
{
"slug": "my-gw",
"tool_filter": {"allowed_tools": ["Github.CreateIssue", "Github.ListPRs"]},
}
]
result = find_matching_gateway(gateways, ["Github.CreateIssue"])
assert result is not None
assert result["slug"] == "my-gw"
def test_returns_none_when_no_match(self) -> None:
gateways = [{"slug": "my-gw", "tool_filter": {"allowed_tools": ["Slack.SendMessage"]}}]
result = find_matching_gateway(gateways, ["Github.CreateIssue"])
assert result is None
def test_returns_none_for_empty_gateways(self) -> None:
assert find_matching_gateway([], ["Github.CreateIssue"]) is None
def test_skips_gateway_with_wrong_auth_type(self) -> None:
gateways = [
{
"slug": "oauth-gw",
"auth_type": "arcade",
"tool_filter": {"allowed_tools": ["Github.CreateIssue"]},
}
]
# Looking for arcade_header auth — should not match the OAuth gateway
result = find_matching_gateway(gateways, ["Github.CreateIssue"], auth_type="arcade_header")
assert result is None
def test_matches_gateway_with_correct_auth_type(self) -> None:
gateways = [
{
"slug": "apikey-gw",
"auth_type": "arcade_header",
"tool_filter": {"allowed_tools": ["Github.CreateIssue"]},
}
]
result = find_matching_gateway(gateways, ["Github.CreateIssue"], auth_type="arcade_header")
assert result is not None
assert result["slug"] == "apikey-gw"
# ---------------------------------------------------------------------------
# list_gateways
# ---------------------------------------------------------------------------
class TestListGateways:
@patch("arcade_cli.connect.httpx.get")
@patch("arcade_cli.utils.get_org_project_context", return_value=("org1", "proj1"))
def test_returns_items(self, _ctx: MagicMock, mock_get: MagicMock) -> None:
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"items": [{"slug": "gw1"}]}
mock_get.return_value = mock_resp
result = list_gateways("tok")
assert len(result) == 1
assert result[0]["slug"] == "gw1"
@patch("arcade_cli.connect.httpx.get")
@patch("arcade_cli.utils.get_org_project_context", return_value=("org1", "proj1"))
def test_returns_empty_on_error(self, _ctx: MagicMock, mock_get: MagicMock) -> None:
mock_resp = MagicMock()
mock_resp.status_code = 401
mock_get.return_value = mock_resp
result = list_gateways("tok")
assert result == []
# ---------------------------------------------------------------------------
# create_gateway
# ---------------------------------------------------------------------------
class TestCreateGateway:
@patch("arcade_cli.connect.httpx.post")
@patch("arcade_cli.utils.get_org_project_context", return_value=("org1", "proj1"))
def test_returns_gateway_dict(self, _ctx: MagicMock, mock_post: MagicMock) -> None:
mock_resp = MagicMock()
mock_resp.status_code = 201
mock_resp.json.return_value = {"slug": "my-gw", "id": "gw-123"}
mock_post.return_value = mock_resp
result = create_gateway("tok", "my-gw", ["Github.CreateIssue"])
assert result["slug"] == "my-gw"
@patch("arcade_cli.connect.httpx.post")
@patch("arcade_cli.utils.get_org_project_context", return_value=("org1", "proj1"))
def test_unwraps_items_envelope(self, _ctx: MagicMock, mock_post: MagicMock) -> None:
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"items": [{"slug": "gw-abc", "id": "123"}]}
mock_post.return_value = mock_resp
result = create_gateway("tok", "test", ["Github.CreateIssue"])
assert result["slug"] == "gw-abc"
@patch("arcade_cli.connect.httpx.post")
@patch("arcade_cli.utils.get_org_project_context", return_value=("org1", "proj1"))
def test_raises_on_error(self, _ctx: MagicMock, mock_post: MagicMock) -> None:
mock_resp = MagicMock()
mock_resp.status_code = 400
mock_resp.text = "bad request"
mock_post.return_value = mock_resp
with pytest.raises(RuntimeError, match="400"):
create_gateway("tok", "test", ["Github.CreateIssue"])
@patch("arcade_cli.connect.httpx.post")
@patch("arcade_cli.utils.get_org_project_context", return_value=("org1", "proj1"))
def test_passes_slug_and_auth_type(self, _ctx: MagicMock, mock_post: MagicMock) -> None:
mock_resp = MagicMock()
mock_resp.status_code = 201
mock_resp.json.return_value = {"slug": "custom"}
mock_post.return_value = mock_resp
create_gateway("tok", "test", ["T.A"], auth_type="arcade_header", slug="custom")
call_body = mock_post.call_args[1]["json"]
assert call_body["auth_type"] == "arcade_header"
assert call_body["slug"] == "custom"
# ---------------------------------------------------------------------------
# _resolve_gateway_slug
# ---------------------------------------------------------------------------
class TestResolveGatewaySlug:
@patch("arcade_cli.connect.list_gateways")
def test_matches_by_slug(self, mock_list: MagicMock) -> None:
from arcade_cli.connect import _resolve_gateway_slug
mock_list.return_value = [{"slug": "pascal_opencode", "name": "opencode"}]
assert _resolve_gateway_slug("pascal_opencode", "tok") == "pascal_opencode"
@patch("arcade_cli.connect.list_gateways")
def test_matches_by_name(self, mock_list: MagicMock) -> None:
from arcade_cli.connect import _resolve_gateway_slug
mock_list.return_value = [{"slug": "pascal_opencode", "name": "opencode"}]
assert _resolve_gateway_slug("opencode", "tok") == "pascal_opencode"
@patch("arcade_cli.connect.list_gateways")
def test_falls_back_to_input(self, mock_list: MagicMock) -> None:
from arcade_cli.connect import _resolve_gateway_slug
mock_list.return_value = [{"slug": "other", "name": "other"}]
assert _resolve_gateway_slug("unknown-gw", "tok") == "unknown-gw"
# ---------------------------------------------------------------------------
# run_connect — tool-only mode
# ---------------------------------------------------------------------------
class TestRunConnectToolOnly:
def test_tool_only_creates_gateway(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
_mock_list_gw(),
patch(
"arcade_cli.connect.create_gateway",
return_value={"slug": "custom-tools", "id": "gw-999"},
),
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="claude",
tools=["Github.CreateIssue", "Slack.SendMessage"],
config_path=config_path,
)
config = json.loads(config_path.read_text(encoding="utf-8"))
assert "mcpServers" in config
# ---------------------------------------------------------------------------
# Helpers: fresh mocks per test (patch objects are single-use as context managers)
# ---------------------------------------------------------------------------
def _mock_list_gw(): # type: ignore[no-untyped-def]
return patch("arcade_cli.connect.list_gateways", return_value=[])
def _mock_resolve_slug(): # type: ignore[no-untyped-def]
return patch("arcade_cli.connect._resolve_gateway_slug", side_effect=lambda gw, *a, **kw: gw)
# ---------------------------------------------------------------------------
# run_connect — gateway mode (direct slug)
# ---------------------------------------------------------------------------
class TestRunConnectGateway:
def test_gateway_mode_configures_claude(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
_mock_resolve_slug(),
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="claude",
gateway="my-production-gw",
config_path=config_path,
)
config = json.loads(config_path.read_text(encoding="utf-8"))
entry = config["mcpServers"]["my-production-gw"]
assert entry["url"] == "https://api.arcade.dev/mcp/my-production-gw"
assert "headers" not in entry
def test_gateway_mode_configures_cursor(self, tmp_path: Path) -> None:
config_path = tmp_path / "cursor.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
_mock_resolve_slug(),
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="cursor",
gateway="test-gw",
config_path=config_path,
)
config = json.loads(config_path.read_text(encoding="utf-8"))
entry = config["mcpServers"]["test-gw"]
assert entry["type"] == "sse"
assert "api.arcade.dev/mcp/test-gw" in entry["url"]
def test_gateway_mode_configures_vscode(self, tmp_path: Path) -> None:
config_path = tmp_path / "vscode.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
_mock_resolve_slug(),
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="vscode",
gateway="test-gw",
config_path=config_path,
)
config = json.loads(config_path.read_text(encoding="utf-8"))
entry = config["servers"]["test-gw"]
assert entry["type"] == "http"
assert "api.arcade.dev/mcp/test-gw" in entry["url"]
# ---------------------------------------------------------------------------
# run_connect — toolkit mode (creates gateway)
# ---------------------------------------------------------------------------
class TestRunConnectToolkit:
def test_toolkit_creates_gateway_and_configures_client(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
patch(
"arcade_cli.connect.fetch_available_toolkits",
return_value={"github": ["Github.ListPRs", "Github.CreateIssue"]},
),
_mock_list_gw(),
patch(
"arcade_cli.connect.create_gateway",
return_value={"slug": "github", "id": "gw-123"},
) as mock_create,
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="claude",
toolkits=["github"],
config_path=config_path,
)
mock_create.assert_called_once()
call_kwargs = mock_create.call_args[1]
assert call_kwargs["name"] == "github"
assert "Github.ListPRs" in call_kwargs["tool_allow_list"]
assert "Github.CreateIssue" in call_kwargs["tool_allow_list"]
config = json.loads(config_path.read_text(encoding="utf-8"))
entry = config["mcpServers"]["github"]
assert entry["url"] == "https://api.arcade.dev/mcp/github"
assert "headers" not in entry
def test_multiple_toolkits_creates_combined_gateway(self, tmp_path: Path) -> None:
config_path = tmp_path / "cursor.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
patch(
"arcade_cli.connect.fetch_available_toolkits",
return_value={
"github": ["Github.ListPRs"],
"slack": ["Slack.SendMessage"],
},
),
_mock_list_gw(),
patch(
"arcade_cli.connect.create_gateway",
return_value={"slug": "github-slack", "id": "gw-456"},
),
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="cursor",
toolkits=["github", "slack"],
config_path=config_path,
)
config = json.loads(config_path.read_text(encoding="utf-8"))
entry = config["mcpServers"]["github-slack"]
assert entry["type"] == "sse"
assert "api.arcade.dev/mcp/github-slack" in entry["url"]
# ---------------------------------------------------------------------------
# run_connect — --all and interactive modes
# ---------------------------------------------------------------------------
class TestRunConnectInteractive:
def test_all_mode_creates_gateway_for_all_toolkits(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
patch(
"arcade_cli.connect.fetch_available_toolkits",
return_value={
"github": ["Github.ListPRs"],
"slack": ["Slack.SendMessage"],
},
),
_mock_list_gw(),
patch(
"arcade_cli.connect.create_gateway",
return_value={"slug": "github-slack", "id": "gw-789"},
) as mock_create,
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="claude",
all_tools=True,
config_path=config_path,
)
call_kwargs = mock_create.call_args[1]
assert len(call_kwargs["tool_allow_list"]) == 2
config = json.loads(config_path.read_text(encoding="utf-8"))
assert "mcpServers" in config
def test_all_mode_no_toolkits_exits(self) -> None:
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
patch("arcade_cli.connect.fetch_available_toolkits", return_value={}),
patch("arcade_cli.connect.console"),
pytest.raises(SystemExit),
):
run_connect(client="claude", all_tools=True)
def test_toolkit_not_found_exits(self) -> None:
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
patch("arcade_cli.connect.fetch_available_toolkits", return_value={}),
_mock_list_gw(),
patch("arcade_cli.connect.console"),
pytest.raises(SystemExit),
):
run_connect(client="claude", toolkits=["nonexistent"])
# ---------------------------------------------------------------------------
# create_project_api_key
# ---------------------------------------------------------------------------
class TestCreateProjectApiKey:
@patch("arcade_cli.connect.httpx.post")
@patch("arcade_cli.utils.get_org_project_context", return_value=("org1", "proj1"))
def test_returns_api_key(self, _ctx: MagicMock, mock_post: MagicMock) -> None:
from arcade_cli.connect import create_project_api_key
mock_resp = MagicMock()
mock_resp.status_code = 201
mock_resp.json.return_value = {"api_key": "arc_test123"}
mock_post.return_value = mock_resp
result = create_project_api_key("tok", label="test")
assert result == "arc_test123"
@patch("arcade_cli.connect.httpx.post")
@patch("arcade_cli.utils.get_org_project_context", return_value=("org1", "proj1"))
def test_raises_on_error(self, _ctx: MagicMock, mock_post: MagicMock) -> None:
from arcade_cli.connect import create_project_api_key
mock_resp = MagicMock()
mock_resp.status_code = 403
mock_resp.text = "forbidden"
mock_post.return_value = mock_resp
with pytest.raises(RuntimeError, match="403"):
create_project_api_key("tok")
# ---------------------------------------------------------------------------
# prompt_toolkit_selection
# ---------------------------------------------------------------------------
class TestPromptToolkitSelection:
from arcade_cli.connect import prompt_toolkit_selection
def test_selects_single_toolkit(self, monkeypatch: pytest.MonkeyPatch) -> None:
from arcade_cli.connect import prompt_toolkit_selection
monkeypatch.setattr("builtins.input", lambda _: "1")
with patch("arcade_cli.connect.console"):
result = prompt_toolkit_selection({"github": ["Github.CreateIssue"]})
assert result == ["github"]
def test_selects_multiple(self, monkeypatch: pytest.MonkeyPatch) -> None:
from arcade_cli.connect import prompt_toolkit_selection
# Bundles come first, then individual toolkits, then "all"
# With no matching bundles, "github" is option 1, "slack" is 2, "all" is 3
monkeypatch.setattr("builtins.input", lambda _: "1,2")
with patch("arcade_cli.connect.console"):
result = prompt_toolkit_selection({
"github": ["Github.CreateIssue"],
"slack": ["Slack.Send"],
})
assert "github" in result
assert "slack" in result
def test_empty_input_exits(self, monkeypatch: pytest.MonkeyPatch) -> None:
from arcade_cli.connect import prompt_toolkit_selection
monkeypatch.setattr("builtins.input", lambda _: "")
with patch("arcade_cli.connect.console"), pytest.raises(SystemExit):
prompt_toolkit_selection({"github": ["Github.CreateIssue"]})
def test_empty_available_exits(self) -> None:
from arcade_cli.connect import prompt_toolkit_selection
with patch("arcade_cli.connect.console"), pytest.raises(SystemExit):
prompt_toolkit_selection({})
# ---------------------------------------------------------------------------
# run_connect — gateway reuse and api-key paths
# ---------------------------------------------------------------------------
class TestRunConnectAdvanced:
def test_reuses_existing_gateway(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
existing_gw = {
"slug": "existing-gw",
"name": "existing",
"tool_filter": {"allowed_tools": ["Github.CreateIssue"]},
}
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
patch(
"arcade_cli.connect.fetch_available_toolkits",
return_value={"github": ["Github.CreateIssue"]},
),
patch("arcade_cli.connect.list_gateways", return_value=[existing_gw]),
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="claude",
toolkits=["github"],
config_path=config_path,
)
config = json.loads(config_path.read_text(encoding="utf-8"))
entry = config["mcpServers"]["github"]
assert "existing-gw" in entry["url"]
def test_gateway_with_api_key(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
_mock_resolve_slug(),
patch("arcade_cli.connect.create_project_api_key", return_value="arc_key123"),
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="claude",
gateway="my-gw",
use_api_key=True,
config_path=config_path,
)
config = json.loads(config_path.read_text(encoding="utf-8"))
entry = config["mcpServers"]["my-gw"]
assert entry["headers"]["Authorization"] == "Bearer arc_key123"
def test_toolkit_with_custom_slug(self, tmp_path: Path) -> None:
config_path = tmp_path / "claude.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
patch(
"arcade_cli.connect.fetch_available_toolkits",
return_value={"github": ["Github.CreateIssue"]},
),
_mock_list_gw(),
patch(
"arcade_cli.connect.create_gateway",
return_value={"slug": "my-custom", "id": "gw-1"},
),
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="claude",
toolkits=["github"],
gateway_slug="my-custom",
config_path=config_path,
)
config = json.loads(config_path.read_text(encoding="utf-8"))
# Display name should be the slug when --slug is given
assert "my-custom" in config["mcpServers"]
def test_tool_with_toolkit_combo(self, tmp_path: Path) -> None:
"""--server github --tool Slack.SendMessage merges both."""
config_path = tmp_path / "claude.json"
with (
patch("arcade_cli.connect.ensure_login", return_value="tok_abc"),
patch(
"arcade_cli.connect.fetch_available_toolkits",
return_value={"github": ["Github.CreateIssue"]},
),
_mock_list_gw(),
patch(
"arcade_cli.connect.create_gateway",
return_value={"slug": "combo", "id": "gw-2"},
) as mock_create,
patch("arcade_cli.connect.console"),
patch("arcade_cli.configure.console"),
):
run_connect(
client="claude",
toolkits=["github"],
tools=["Slack.SendMessage"],
config_path=config_path,
)
call_kwargs = mock_create.call_args[1]
assert "Github.CreateIssue" in call_kwargs["tool_allow_list"]
assert "Slack.SendMessage" in call_kwargs["tool_allow_list"]

2
pyproject.toml
View file

@ -1,6 +1,6 @@
[project]
name = "arcade-mcp"
version = "1.13.3"
version = "1.14.0"
description = "Arcade.dev - Tool Calling platform for Agents"
readme = "README.md"
license = { file = "LICENSE" }