[Draft][WIP] studio: import MCP servers from a config file

studio/backend/core/inference/mcp_client.py

@@ -41,6 +41,19 @@ def parse_stdio_command(address: str) -> list[str]:
     return parts
 
 
+def join_stdio_command(parts: list[str]) -> str:
+    """Inverse of parse_stdio_command: join argv into a single command string
+    that parse_stdio_command() splits back into ``parts`` on this platform.
+    Config files (issue #5936) carry structured command + args; storage holds
+    one string in the url field. Windows uses list2cmdline so spaced/backslash
+    paths round-trip through the posix=False quote-strip; posix uses shlex."""
+    if sys.platform == "win32":
+        import subprocess
+
+        return subprocess.list2cmdline(parts)
+    return shlex.join(parts)
+
+
 def stdio_mcp_enabled() -> bool:
     """stdio MCP servers spawn local processes as the backend user (and bypass
     the python/terminal sandbox), so they are only allowed when the backend

studio/backend/core/inference/mcp_config_import.py

@@ -0,0 +1,97 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
+
+"""Parse a standard ``mcpServers`` JSON config (Claude Desktop / Cursor / Cline
+/ VS Code) into entries the existing MCP storage understands. See issue #5936.
+
+A stdio entry (``command`` + ``args`` + ``env``) is joined into the single
+command string the ``url`` field already stores; a remote entry (``url`` +
+``headers``) maps straight through. Parsing never raises on a single bad entry:
+it returns ``(entries, errors)`` so one malformed server can't sink the import.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+from core.inference.mcp_client import join_stdio_command
+
+_SCALAR = (str, int, float, bool)
+
+
+@dataclass
+class ParsedMcpEntry:
+    display_name: str
+    url: str  # joined command (stdio) or http(s) url (remote)
+    headers: Optional[dict[str, str]]  # env vars (stdio) or http headers (remote)
+    is_stdio: bool
+
+
+def _coerce_str_dict(value: dict) -> dict[str, str]:
+    return {str(k): str(v) for k, v in value.items()}
+
+
+def _parse_entry(
+    name: str, spec: object
+) -> tuple[Optional[ParsedMcpEntry], Optional[str]]:
+    label = str(name).strip()
+    if not label:
+        return None, "Server entry has an empty name."
+    if not isinstance(spec, dict):
+        return None, f"{label}: entry must be an object."
+
+    has_command = bool(spec.get("command"))
+    has_url = bool(spec.get("url"))
+    if has_command and has_url:
+        return None, f"{label}: entry has both 'command' and 'url'; use one."
+    if not has_command and not has_url:
+        return None, f"{label}: entry needs a 'command' (stdio) or 'url' (remote)."
+
+    if has_command:
+        command = spec["command"]
+        if not isinstance(command, str):
+            return None, f"{label}: 'command' must be a string."
+        args = spec.get("args") or []
+        if not isinstance(args, list) or not all(isinstance(a, _SCALAR) for a in args):
+            return None, f"{label}: 'args' must be a list of strings."
+        env = spec.get("env")
+        if env is not None and not isinstance(env, dict):
+            return None, f"{label}: 'env' must be an object."
+        url = join_stdio_command([command, *(str(a) for a in args)])
+        headers = _coerce_str_dict(env) if env else None
+        return ParsedMcpEntry(label, url, headers, True), None
+
+    url = spec["url"]
+    if not isinstance(url, str):
+        return None, f"{label}: 'url' must be a string."
+    headers_raw = spec.get("headers")
+    if headers_raw is not None and not isinstance(headers_raw, dict):
+        return None, f"{label}: 'headers' must be an object."
+    headers = _coerce_str_dict(headers_raw) if headers_raw else None
+    return ParsedMcpEntry(label, url.strip(), headers, False), None
+
+
+def parse_mcp_config(config: object) -> tuple[list[ParsedMcpEntry], list[str]]:
+    """Parse a Claude-Desktop/Cursor/Cline/VS Code config. Accepts the
+    ``mcpServers`` key (primary) or ``servers`` (VS Code alias). Returns
+    ``(entries, errors)``; a bad entry adds an error rather than raising."""
+    if not isinstance(config, dict):
+        return [], ["Config must be a JSON object."]
+    servers = config.get("mcpServers")
+    if servers is None:
+        servers = config.get("servers")
+    if servers is None:
+        return [], ["Config has no 'mcpServers' (or 'servers') object."]
+    if not isinstance(servers, dict):
+        return [], ["'mcpServers' must be an object mapping name -> server."]
+
+    entries: list[ParsedMcpEntry] = []
+    errors: list[str] = []
+    for name, spec in servers.items():
+        entry, error = _parse_entry(name, spec)
+        if error:
+            errors.append(error)
+        elif entry:
+            entries.append(entry)
+    return entries, errors

studio/backend/models/mcp_servers.py

@@ -44,3 +44,16 @@ class McpServerProbeResult(BaseModel):
     ok: bool
     tool_count: int = 0
     error: Optional[str] = None
+
+
+class McpServerImportRequest(BaseModel):
+    # A standard mcpServers JSON config (Claude Desktop / Cursor / Cline / VS Code).
+    config: dict
+
+
+class McpServerImportResult(BaseModel):
+    created: list[McpServerResponse] = Field(default_factory = list)
+    skipped: list[str] = Field(
+        default_factory = list
+    )  # display names skipped as duplicates
+    errors: list[str] = Field(default_factory = list)

studio/backend/routes/mcp_servers.py

@@ -18,8 +18,11 @@
     probe_timeout,
     stdio_mcp_enabled,
 )
+from core.inference.mcp_config_import import parse_mcp_config
 from models.mcp_servers import (
     McpServerCreate,
+    McpServerImportRequest,
+    McpServerImportResult,
     McpServerProbeResult,
     McpServerResponse,
     McpServerTestRequest,
@@ -247,6 +250,47 @@ async def refresh_mcp_server_tools(
     return McpServerProbeResult(ok = True, tool_count = len(tools))
 
 
+@router.post("/import", response_model = McpServerImportResult)
+async def import_mcp_servers(
+    payload: McpServerImportRequest,
+    current_subject: str = Depends(get_current_subject),
+):
+    """Bulk-register servers from a standard mcpServers JSON config (issue
+    #5936). Each entry rides the existing create path: _validate_url applies
+    the same stdio gate (a stdio entry becomes a per-entry error when stdio is
+    off; http still imports), and entries whose url already exists are skipped
+    so re-importing the same file is idempotent. One bad entry never 400s the
+    whole batch -- failures are reported per entry."""
+    entries, errors = parse_mcp_config(payload.config)
+    created: list[McpServerResponse] = []
+    skipped: list[str] = []
+    seen_urls = {row["url"] for row in mcp_servers_db.list_servers()}
+
+    for entry in entries:
+        try:
+            url = _validate_url(entry.url)
+        except HTTPException as exc:
+            errors.append(f"{entry.display_name}: {exc.detail}")
+            continue
+        if url in seen_urls:
+            skipped.append(entry.display_name)
+            continue
+        headers = _normalize_headers(entry.headers)
+        server_id = uuid.uuid4().hex[:16]
+        mcp_servers_db.create_server(
+            id = server_id,
+            display_name = entry.display_name,
+            url = url,
+            headers_json = json.dumps(headers) if headers else None,
+            is_enabled = True,
+            use_oauth = False,
+        )
+        seen_urls.add(url)
+        created.append(_row_to_response(mcp_servers_db.get_server(server_id)))
+
+    return McpServerImportResult(created = created, skipped = skipped, errors = errors)
+
+
 @router.post("/test", response_model = McpServerProbeResult)
 async def test_mcp_server(
     payload: McpServerTestRequest,