Merge branch 'main' into claude/issue-3049-20260131-2232

tests/cli/test_client_commands.py

@@ -0,0 +1,557 @@
+"""Tests for fastmcp list and fastmcp call CLI commands."""
+
+import json
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+import mcp.types
+import pytest
+
+from fastmcp import FastMCP
+from fastmcp.cli import client as client_module
+from fastmcp.cli.client import (
+    Client,
+    _build_client,
+    _build_stdio_from_command,
+    _format_call_result_text,
+    _is_http_target,
+    call_command,
+    coerce_value,
+    format_tool_signature,
+    list_command,
+    parse_tool_arguments,
+    resolve_server_spec,
+)
+from fastmcp.client.client import CallToolResult
+from fastmcp.client.transports.stdio import StdioTransport
+
+# ---------------------------------------------------------------------------
+# coerce_value
+# ---------------------------------------------------------------------------
+
+
+class TestCoerceValue:
+    def test_integer(self):
+        assert coerce_value("42", {"type": "integer"}) == 42
+
+    def test_integer_negative(self):
+        assert coerce_value("-7", {"type": "integer"}) == -7
+
+    def test_integer_invalid(self):
+        with pytest.raises(ValueError, match="Expected integer"):
+            coerce_value("abc", {"type": "integer"})
+
+    def test_number(self):
+        assert coerce_value("3.14", {"type": "number"}) == 3.14
+
+    def test_number_integer_value(self):
+        assert coerce_value("5", {"type": "number"}) == 5.0
+
+    def test_number_invalid(self):
+        with pytest.raises(ValueError, match="Expected number"):
+            coerce_value("xyz", {"type": "number"})
+
+    def test_boolean_true_variants(self):
+        for val in ("true", "True", "TRUE", "1", "yes"):
+            assert coerce_value(val, {"type": "boolean"}) is True
+
+    def test_boolean_false_variants(self):
+        for val in ("false", "False", "FALSE", "0", "no"):
+            assert coerce_value(val, {"type": "boolean"}) is False
+
+    def test_boolean_invalid(self):
+        with pytest.raises(ValueError, match="Expected boolean"):
+            coerce_value("maybe", {"type": "boolean"})
+
+    def test_array(self):
+        assert coerce_value("[1, 2, 3]", {"type": "array"}) == [1, 2, 3]
+
+    def test_array_invalid(self):
+        with pytest.raises(ValueError, match="Expected JSON array"):
+            coerce_value("not-json", {"type": "array"})
+
+    def test_object(self):
+        assert coerce_value('{"a": 1}', {"type": "object"}) == {"a": 1}
+
+    def test_string(self):
+        assert coerce_value("hello", {"type": "string"}) == "hello"
+
+    def test_string_default(self):
+        """Unknown or missing type treats value as string."""
+        assert coerce_value("hello", {}) == "hello"
+
+    def test_string_preserves_numeric_looking_values(self):
+        assert coerce_value("42", {"type": "string"}) == "42"
+
+
+# ---------------------------------------------------------------------------
+# parse_tool_arguments
+# ---------------------------------------------------------------------------
+
+
+class TestParseToolArguments:
+    SCHEMA: dict[str, Any] = {
+        "type": "object",
+        "properties": {
+            "query": {"type": "string"},
+            "limit": {"type": "integer"},
+            "verbose": {"type": "boolean"},
+        },
+        "required": ["query"],
+    }
+
+    def test_basic_key_value(self):
+        result = parse_tool_arguments(("query=hello", "limit=10"), None, self.SCHEMA)
+        assert result == {"query": "hello", "limit": 10}
+
+    def test_input_json_only(self):
+        result = parse_tool_arguments((), '{"query": "hello", "limit": 5}', self.SCHEMA)
+        assert result == {"query": "hello", "limit": 5}
+
+    def test_key_value_overrides_input_json(self):
+        result = parse_tool_arguments(
+            ("limit=20",), '{"query": "hello", "limit": 5}', self.SCHEMA
+        )
+        assert result == {"query": "hello", "limit": 20}
+
+    def test_value_containing_equals(self):
+        result = parse_tool_arguments(("query=a=b=c",), None, self.SCHEMA)
+        assert result == {"query": "a=b=c"}
+
+    def test_invalid_arg_format_exits(self):
+        with pytest.raises(SystemExit):
+            parse_tool_arguments(("noequalssign",), None, self.SCHEMA)
+
+    def test_invalid_input_json_exits(self):
+        with pytest.raises(SystemExit):
+            parse_tool_arguments((), "not-valid-json", self.SCHEMA)
+
+    def test_input_json_non_object_exits(self):
+        with pytest.raises(SystemExit):
+            parse_tool_arguments((), "[1,2,3]", self.SCHEMA)
+
+    def test_single_json_object_as_positional(self):
+        result = parse_tool_arguments(
+            ('{"query": "hello", "limit": 5}',), None, self.SCHEMA
+        )
+        assert result == {"query": "hello", "limit": 5}
+
+    def test_json_positional_ignored_when_input_json_set(self):
+        """When --input-json is already provided, a JSON positional arg is not special."""
+        with pytest.raises(SystemExit):
+            parse_tool_arguments(('{"limit": 99}',), '{"query": "hello"}', self.SCHEMA)
+
+    def test_coercion_error_exits(self):
+        with pytest.raises(SystemExit):
+            parse_tool_arguments(("limit=abc",), None, self.SCHEMA)
+
+
+# ---------------------------------------------------------------------------
+# format_tool_signature
+# ---------------------------------------------------------------------------
+
+
+class TestFormatToolSignature:
+    def _make_tool(
+        self,
+        name: str = "my_tool",
+        properties: dict[str, Any] | None = None,
+        required: list[str] | None = None,
+        output_schema: dict[str, Any] | None = None,
+        description: str | None = None,
+    ) -> mcp.types.Tool:
+        input_schema: dict[str, Any] = {"type": "object"}
+        if properties is not None:
+            input_schema["properties"] = properties
+        if required is not None:
+            input_schema["required"] = required
+        return mcp.types.Tool(
+            name=name,
+            description=description,
+            inputSchema=input_schema,
+            outputSchema=output_schema,
+        )
+
+    def test_no_params(self):
+        tool = self._make_tool()
+        assert format_tool_signature(tool) == "my_tool()"
+
+    def test_required_param(self):
+        tool = self._make_tool(
+            properties={"query": {"type": "string"}},
+            required=["query"],
+        )
+        assert format_tool_signature(tool) == "my_tool(query: str)"
+
+    def test_optional_param_with_default(self):
+        tool = self._make_tool(
+            properties={"limit": {"type": "integer", "default": 10}},
+        )
+        assert format_tool_signature(tool) == "my_tool(limit: int = 10)"
+
+    def test_optional_param_without_default(self):
+        tool = self._make_tool(
+            properties={"limit": {"type": "integer"}},
+        )
+        assert format_tool_signature(tool) == "my_tool(limit: int = ...)"
+
+    def test_mixed_required_and_optional(self):
+        tool = self._make_tool(
+            properties={
+                "query": {"type": "string"},
+                "limit": {"type": "integer", "default": 10},
+            },
+            required=["query"],
+        )
+        sig = format_tool_signature(tool)
+        assert sig == "my_tool(query: str, limit: int = 10)"
+
+    def test_with_output_schema(self):
+        tool = self._make_tool(
+            properties={"q": {"type": "string"}},
+            required=["q"],
+            output_schema={"type": "object"},
+        )
+        assert format_tool_signature(tool) == "my_tool(q: str) -> dict"
+
+    def test_anyof_type(self):
+        tool = self._make_tool(
+            properties={"value": {"anyOf": [{"type": "string"}, {"type": "integer"}]}},
+            required=["value"],
+        )
+        assert format_tool_signature(tool) == "my_tool(value: str | int)"
+
+
+# ---------------------------------------------------------------------------
+# resolve_server_spec
+# ---------------------------------------------------------------------------
+
+
+class TestResolveServerSpec:
+    def test_http_url(self):
+        assert (
+            resolve_server_spec("http://localhost:8000/mcp")
+            == "http://localhost:8000/mcp"
+        )
+
+    def test_https_url(self):
+        assert (
+            resolve_server_spec("https://example.com/mcp") == "https://example.com/mcp"
+        )
+
+    def test_python_file_existing(self, tmp_path: Path):
+        py_file = tmp_path / "server.py"
+        py_file.write_text("# empty")
+        result = resolve_server_spec(str(py_file))
+        assert isinstance(result, StdioTransport)
+        assert result.command == "fastmcp"
+        assert result.args == ["run", str(py_file.resolve()), "--no-banner"]
+
+    def test_json_mcp_config(self, tmp_path: Path):
+        config_file = tmp_path / "mcp.json"
+        config = {"mcpServers": {"test": {"url": "http://localhost:8000"}}}
+        config_file.write_text(json.dumps(config))
+        result = resolve_server_spec(str(config_file))
+        assert isinstance(result, dict)
+        assert "mcpServers" in result
+
+    def test_json_fastmcp_config_exits(self, tmp_path: Path):
+        config_file = tmp_path / "fastmcp.json"
+        config_file.write_text(json.dumps({"source": {"type": "file"}}))
+        with pytest.raises(SystemExit):
+            resolve_server_spec(str(config_file))
+
+    def test_json_not_found_exits(self, tmp_path: Path):
+        with pytest.raises(SystemExit):
+            resolve_server_spec(str(tmp_path / "nonexistent.json"))
+
+    def test_directory_exits(self, tmp_path: Path):
+        """Directories should not be treated as file paths."""
+        with pytest.raises(SystemExit):
+            resolve_server_spec(str(tmp_path))
+
+    def test_unrecognised_exits(self):
+        with pytest.raises(SystemExit):
+            resolve_server_spec("some_random_thing")
+
+    def test_command_returns_stdio_transport(self):
+        result = resolve_server_spec(None, command="npx -y @mcp/server")
+        assert isinstance(result, StdioTransport)
+        assert result.command == "npx"
+        assert result.args == ["-y", "@mcp/server"]
+
+    def test_command_single_word(self):
+        result = resolve_server_spec(None, command="myserver")
+        assert isinstance(result, StdioTransport)
+        assert result.command == "myserver"
+        assert result.args == []
+
+    def test_server_spec_and_command_exits(self):
+        with pytest.raises(SystemExit):
+            resolve_server_spec("http://localhost:8000", command="npx server")
+
+    def test_neither_server_spec_nor_command_exits(self):
+        with pytest.raises(SystemExit):
+            resolve_server_spec(None)
+
+    def test_transport_sse_rewrites_url(self):
+        result = resolve_server_spec("http://localhost:8000/mcp", transport="sse")
+        assert result == "http://localhost:8000/mcp/sse"
+
+    def test_transport_sse_no_duplicate_suffix(self):
+        result = resolve_server_spec("http://localhost:8000/sse", transport="sse")
+        assert result == "http://localhost:8000/sse"
+
+    def test_transport_sse_trailing_slash(self):
+        result = resolve_server_spec("http://localhost:8000/mcp/", transport="sse")
+        assert result == "http://localhost:8000/mcp/sse"
+
+    def test_transport_http_leaves_url_unchanged(self):
+        result = resolve_server_spec("http://localhost:8000/mcp", transport="http")
+        assert result == "http://localhost:8000/mcp"
+
+
+# ---------------------------------------------------------------------------
+# _build_stdio_from_command
+# ---------------------------------------------------------------------------
+
+
+class TestBuildStdioFromCommand:
+    def test_simple_command(self):
+        transport = _build_stdio_from_command("uvx my-server")
+        assert transport.command == "uvx"
+        assert transport.args == ["my-server"]
+
+    def test_quoted_args(self):
+        transport = _build_stdio_from_command("npx -y '@scope/server'")
+        assert transport.command == "npx"
+        assert transport.args == ["-y", "@scope/server"]
+
+    def test_empty_command_exits(self):
+        with pytest.raises(SystemExit):
+            _build_stdio_from_command("")
+
+    def test_invalid_shell_syntax_exits(self):
+        with pytest.raises(SystemExit):
+            _build_stdio_from_command("npx 'unterminated")
+
+
+# ---------------------------------------------------------------------------
+# _is_http_target
+# ---------------------------------------------------------------------------
+
+
+class TestIsHttpTarget:
+    def test_http_url(self):
+        assert _is_http_target("http://localhost:8000") is True
+
+    def test_https_url(self):
+        assert _is_http_target("https://example.com/mcp") is True
+
+    def test_file_path(self):
+        assert _is_http_target("/path/to/server.py") is False
+
+    def test_stdio_transport(self):
+        assert _is_http_target(StdioTransport(command="npx", args=[])) is False
+
+    def test_mcp_config_dict(self):
+        """MCPConfig dicts are not HTTP targets — auth is per-server internally."""
+        assert _is_http_target({"mcpServers": {}}) is False
+
+
+# ---------------------------------------------------------------------------
+# _build_client
+# ---------------------------------------------------------------------------
+
+
+class TestBuildClient:
+    def test_http_target_gets_oauth_by_default(self):
+        client = _build_client("http://localhost:8000/mcp")
+        # OAuth is applied during Client init via _set_auth
+        assert client.transport.auth is not None
+
+    def test_stdio_target_no_auth(self):
+        transport = StdioTransport(command="npx", args=["-y", "@mcp/server"])
+        client = _build_client(transport)
+        # Stdio transports don't support auth — no auth should be set
+        assert not hasattr(client.transport, "auth") or client.transport.auth is None
+
+    def test_explicit_auth_none_disables_oauth(self):
+        client = _build_client("http://localhost:8000/mcp", auth="none")
+        # "none" explicitly disables auth, even for HTTP targets
+        assert client.transport.auth is None
+
+    def test_mcp_config_no_auth(self):
+        """MCPConfig dicts handle auth per-server; no top-level auth applied."""
+        client = _build_client({"mcpServers": {"test": {"url": "http://localhost"}}})
+        # MCPConfigTransport doesn't support _set_auth — no crash means success
+        assert client.transport is not None
+
+
+# ---------------------------------------------------------------------------
+# Integration tests — invoke actual CLI commands via monkeypatched _build_client
+# ---------------------------------------------------------------------------
+
+
+def _build_test_server() -> FastMCP:
+    """Create a minimal FastMCP server for integration tests."""
+    server = FastMCP("TestServer")
+
+    @server.tool
+    def greet(name: str) -> str:
+        """Say hello to someone."""
+        return f"Hello, {name}!"
+
+    @server.tool
+    def add(a: int, b: int) -> int:
+        """Add two numbers."""
+        return a + b
+
+    @server.resource("test://greeting")
+    def greeting_resource() -> str:
+        """A static greeting resource."""
+        return "Hello from resource!"
+
+    @server.prompt
+    def ask(topic: str) -> str:
+        """Ask about a topic."""
+        return f"Tell me about {topic}"
+
+    return server
+
+
+@pytest.fixture()
+def _patch_client():
+    """Patch resolve_server_spec and _build_client so CLI commands use the
+    in-process test server without needing a real transport."""
+    server = _build_test_server()
+
+    def fake_resolve(server_spec: Any, **kwargs: Any) -> str:
+        return "fake"
+
+    def fake_build_client(resolved: Any, **kwargs: Any) -> Client:
+        return Client(server)
+
+    with (
+        patch.object(client_module, "resolve_server_spec", side_effect=fake_resolve),
+        patch.object(client_module, "_build_client", side_effect=fake_build_client),
+    ):
+        yield
+
+
+class TestListCommandCLI:
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_list_tools(self, capsys: pytest.CaptureFixture[str]):
+        await list_command("fake://server")
+        captured = capsys.readouterr()
+        assert "greet" in captured.out
+        assert "add" in captured.out
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_list_json(self, capsys: pytest.CaptureFixture[str]):
+        await list_command("fake://server", json_output=True)
+        captured = capsys.readouterr()
+        data = json.loads(captured.out)
+        names = {t["name"] for t in data["tools"]}
+        assert "greet" in names
+        assert "add" in names
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_list_resources(self, capsys: pytest.CaptureFixture[str]):
+        await list_command("fake://server", resources=True)
+        captured = capsys.readouterr()
+        assert "test://greeting" in captured.out
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_list_prompts(self, capsys: pytest.CaptureFixture[str]):
+        await list_command("fake://server", prompts=True)
+        captured = capsys.readouterr()
+        assert "ask" in captured.out
+
+
+class TestCallCommandCLI:
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_tool(self, capsys: pytest.CaptureFixture[str]):
+        await call_command("fake://server", "greet", "name=World")
+        captured = capsys.readouterr()
+        assert "Hello, World!" in captured.out
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_tool_json(self, capsys: pytest.CaptureFixture[str]):
+        await call_command("fake://server", "greet", "name=World", json_output=True)
+        captured = capsys.readouterr()
+        data = json.loads(captured.out)
+        assert data["is_error"] is False
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_tool_not_found(self):
+        with pytest.raises(SystemExit):
+            await call_command("fake://server", "nonexistent")
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_tool_missing_args(self):
+        with pytest.raises(SystemExit):
+            await call_command("fake://server", "greet")
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_resource_by_uri(self, capsys: pytest.CaptureFixture[str]):
+        await call_command("fake://server", "test://greeting")
+        captured = capsys.readouterr()
+        assert "Hello from resource!" in captured.out
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_resource_json(self, capsys: pytest.CaptureFixture[str]):
+        await call_command("fake://server", "test://greeting", json_output=True)
+        captured = capsys.readouterr()
+        data = json.loads(captured.out)
+        assert isinstance(data, list)
+        assert data[0]["text"] == "Hello from resource!"
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_prompt(self, capsys: pytest.CaptureFixture[str]):
+        await call_command("fake://server", "ask", "topic=Python", prompt=True)
+        captured = capsys.readouterr()
+        assert "Python" in captured.out
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_prompt_json(self, capsys: pytest.CaptureFixture[str]):
+        await call_command(
+            "fake://server", "ask", "topic=Python", prompt=True, json_output=True
+        )
+        captured = capsys.readouterr()
+        data = json.loads(captured.out)
+        assert "messages" in data
+
+    @pytest.mark.usefixtures("_patch_client")
+    async def test_call_prompt_not_found(self):
+        with pytest.raises(SystemExit):
+            await call_command("fake://server", "nonexistent", prompt=True)
+
+    async def test_call_missing_target(self):
+        with pytest.raises(SystemExit):
+            await call_command("fake://server", "")
+
+
+# ---------------------------------------------------------------------------
+# Structured content serialization
+# ---------------------------------------------------------------------------
+
+
+class TestFormatCallResult:
+    def test_structured_content_uses_dict_not_data(
+        self, capsys: pytest.CaptureFixture[str]
+    ):
+        """structured_content (raw dict) is used for display, not data (which may
+        be a non-serializable dataclass)."""
+        result = CallToolResult(
+            content=[mcp.types.TextContent(type="text", text="ok")],
+            structured_content={"key": "value"},
+            meta=None,
+            data=object(),  # non-serializable on purpose
+            is_error=False,
+        )
+        # Should not raise — uses structured_content, not data
+        _format_call_result_text(result)
+        captured = capsys.readouterr()
+        assert "value" in captured.out

tests/cli/test_discovery.py

@@ -0,0 +1,668 @@
+"""Tests for MCP server discovery and name-based resolution."""
+
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+import yaml
+
+from fastmcp.cli.client import _is_http_target, resolve_server_spec
+from fastmcp.cli.discovery import (
+    DiscoveredServer,
+    _normalize_server_entry,
+    _parse_mcp_config,
+    _scan_claude_code,
+    _scan_claude_desktop,
+    _scan_cursor_workspace,
+    _scan_gemini,
+    _scan_goose,
+    _scan_project_mcp_json,
+    discover_servers,
+    resolve_name,
+)
+from fastmcp.client.transports.http import StreamableHttpTransport
+from fastmcp.client.transports.sse import SSETransport
+from fastmcp.client.transports.stdio import StdioTransport
+from fastmcp.mcp_config import RemoteMCPServer, StdioMCPServer
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_STDIO_CONFIG: dict[str, Any] = {
+    "mcpServers": {
+        "weather": {
+            "command": "npx",
+            "args": ["-y", "@mcp/weather"],
+        },
+        "github": {
+            "command": "npx",
+            "args": ["-y", "@mcp/github"],
+            "env": {"GITHUB_TOKEN": "xxx"},
+        },
+    }
+}
+
+_REMOTE_CONFIG: dict[str, Any] = {
+    "mcpServers": {
+        "api": {
+            "url": "http://localhost:8000/mcp",
+        },
+    }
+}
+
+
+def _write_config(path: Path, data: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data))
+
+
+# ---------------------------------------------------------------------------
+# DiscoveredServer properties
+# ---------------------------------------------------------------------------
+
+
+class TestDiscoveredServer:
+    def test_qualified_name(self):
+        server = DiscoveredServer(
+            name="weather",
+            source="claude-desktop",
+            config=StdioMCPServer(command="npx", args=["-y", "@mcp/weather"]),
+            config_path=Path("/fake/config.json"),
+        )
+        assert server.qualified_name == "claude-desktop:weather"
+
+    def test_transport_summary_stdio(self):
+        server = DiscoveredServer(
+            name="weather",
+            source="cursor",
+            config=StdioMCPServer(command="npx", args=["-y", "@mcp/weather"]),
+            config_path=Path("/fake/config.json"),
+        )
+        assert server.transport_summary == "stdio: npx -y @mcp/weather"
+
+    def test_transport_summary_remote(self):
+        server = DiscoveredServer(
+            name="api",
+            source="project",
+            config=RemoteMCPServer(url="http://localhost:8000/mcp"),
+            config_path=Path("/fake/config.json"),
+        )
+        assert server.transport_summary == "http: http://localhost:8000/mcp"
+
+    def test_transport_summary_remote_sse(self):
+        server = DiscoveredServer(
+            name="api",
+            source="project",
+            config=RemoteMCPServer(url="http://localhost:8000/sse", transport="sse"),
+            config_path=Path("/fake/config.json"),
+        )
+        assert server.transport_summary == "sse: http://localhost:8000/sse"
+
+
+# ---------------------------------------------------------------------------
+# _parse_mcp_config
+# ---------------------------------------------------------------------------
+
+
+class TestParseMcpConfig:
+    def test_valid_config(self, tmp_path: Path):
+        path = tmp_path / "config.json"
+        _write_config(path, _STDIO_CONFIG)
+        servers = _parse_mcp_config(path, "test-source")
+        assert len(servers) == 2
+        names = {s.name for s in servers}
+        assert names == {"weather", "github"}
+        assert all(s.source == "test-source" for s in servers)
+        assert all(s.config_path == path for s in servers)
+
+    def test_missing_file(self, tmp_path: Path):
+        path = tmp_path / "nonexistent.json"
+        servers = _parse_mcp_config(path, "test")
+        assert servers == []
+
+    def test_invalid_json(self, tmp_path: Path):
+        path = tmp_path / "bad.json"
+        path.write_text("{not json")
+        servers = _parse_mcp_config(path, "test")
+        assert servers == []
+
+    def test_no_mcp_servers_key(self, tmp_path: Path):
+        path = tmp_path / "config.json"
+        _write_config(path, {"something": "else"})
+        servers = _parse_mcp_config(path, "test")
+        assert servers == []
+
+    def test_empty_mcp_servers(self, tmp_path: Path):
+        path = tmp_path / "config.json"
+        _write_config(path, {"mcpServers": {}})
+        servers = _parse_mcp_config(path, "test")
+        assert servers == []
+
+    def test_remote_server(self, tmp_path: Path):
+        path = tmp_path / "config.json"
+        _write_config(path, _REMOTE_CONFIG)
+        servers = _parse_mcp_config(path, "test")
+        assert len(servers) == 1
+        assert isinstance(servers[0].config, RemoteMCPServer)
+        assert servers[0].config.url == "http://localhost:8000/mcp"
+
+
+# ---------------------------------------------------------------------------
+# Scanner: Claude Desktop
+# ---------------------------------------------------------------------------
+
+
+class TestScanClaudeDesktop:
+    def test_finds_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        config_dir = tmp_path / "Claude"
+        config_path = config_dir / "claude_desktop_config.json"
+        _write_config(config_path, _STDIO_CONFIG)
+
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        # Force darwin for deterministic path
+        monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "darwin")
+
+        # We need to override the path construction. On macOS it's
+        # ~/Library/Application Support/Claude — create that.
+        mac_dir = tmp_path / "Library" / "Application Support" / "Claude"
+        mac_path = mac_dir / "claude_desktop_config.json"
+        _write_config(mac_path, _STDIO_CONFIG)
+
+        servers = _scan_claude_desktop()
+        assert len(servers) == 2
+        assert all(s.source == "claude-desktop" for s in servers)
+
+    def test_missing_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "darwin")
+        servers = _scan_claude_desktop()
+        assert servers == []
+
+
+# ---------------------------------------------------------------------------
+# Normalize server entry
+# ---------------------------------------------------------------------------
+
+
+class TestNormalizeServerEntry:
+    def test_remote_type_becomes_transport(self):
+        entry = {"url": "http://localhost:8000/sse", "type": "sse"}
+        result = _normalize_server_entry(entry)
+        assert result["transport"] == "sse"
+        assert "type" not in result
+
+    def test_remote_with_transport_unchanged(self):
+        entry = {"url": "http://localhost:8000/mcp", "transport": "http"}
+        result = _normalize_server_entry(entry)
+        assert result["transport"] == "http"
+
+    def test_stdio_type_unchanged(self):
+        """Stdio entries have ``type`` as a proper field — leave it alone."""
+        entry = {"command": "npx", "args": [], "type": "stdio"}
+        result = _normalize_server_entry(entry)
+        assert result["type"] == "stdio"
+
+    def test_gemini_http_url_becomes_url(self):
+        entry = {"httpUrl": "https://api.example.com/mcp/"}
+        result = _normalize_server_entry(entry)
+        assert result["url"] == "https://api.example.com/mcp/"
+        assert "httpUrl" not in result
+
+    def test_gemini_http_url_does_not_override_url(self):
+        entry = {"url": "http://real.com", "httpUrl": "http://other.com"}
+        result = _normalize_server_entry(entry)
+        assert result["url"] == "http://real.com"
+
+
+# ---------------------------------------------------------------------------
+# Scanner: Claude Code
+# ---------------------------------------------------------------------------
+
+
+def _claude_code_config(
+    *,
+    global_servers: dict[str, Any] | None = None,
+    project_path: str | None = None,
+    project_servers: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build a minimal ~/.claude.json structure."""
+    data: dict[str, Any] = {}
+    if global_servers is not None:
+        data["mcpServers"] = global_servers
+    if project_path and project_servers is not None:
+        data["projects"] = {project_path: {"mcpServers": project_servers}}
+    return data
+
+
+class TestScanClaudeCode:
+    def test_global_servers(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        config_path = tmp_path / ".claude.json"
+        _write_config(
+            config_path,
+            _claude_code_config(global_servers=_STDIO_CONFIG["mcpServers"]),
+        )
+        servers = _scan_claude_code(tmp_path)
+        assert len(servers) == 2
+        assert all(s.source == "claude-code" for s in servers)
+
+    def test_project_servers(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        project_dir = tmp_path / "my-project"
+        project_dir.mkdir()
+        config_path = tmp_path / ".claude.json"
+        _write_config(
+            config_path,
+            _claude_code_config(
+                project_path=str(project_dir),
+                project_servers={"api": {"url": "http://localhost:8000/mcp"}},
+            ),
+        )
+        servers = _scan_claude_code(project_dir)
+        assert len(servers) == 1
+        assert servers[0].name == "api"
+
+    def test_global_and_project_combined(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        project_dir = tmp_path / "proj"
+        project_dir.mkdir()
+        config_path = tmp_path / ".claude.json"
+        _write_config(
+            config_path,
+            _claude_code_config(
+                global_servers={"global-tool": {"command": "echo", "args": ["hi"]}},
+                project_path=str(project_dir),
+                project_servers={"local-tool": {"command": "cat", "args": []}},
+            ),
+        )
+        servers = _scan_claude_code(project_dir)
+        names = {s.name for s in servers}
+        assert names == {"global-tool", "local-tool"}
+
+    def test_type_normalized_to_transport(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Claude Code uses ``type: sse`` — verify it becomes ``transport``."""
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        config_path = tmp_path / ".claude.json"
+        _write_config(
+            config_path,
+            _claude_code_config(
+                global_servers={
+                    "sse-server": {
+                        "type": "sse",
+                        "url": "http://localhost:8000/sse",
+                    }
+                }
+            ),
+        )
+        servers = _scan_claude_code(tmp_path)
+        assert len(servers) == 1
+        assert isinstance(servers[0].config, RemoteMCPServer)
+        assert servers[0].config.transport == "sse"
+
+    def test_missing_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        servers = _scan_claude_code(tmp_path)
+        assert servers == []
+
+    def test_no_matching_project(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        config_path = tmp_path / ".claude.json"
+        _write_config(
+            config_path,
+            _claude_code_config(
+                project_path="/some/other/project",
+                project_servers={"tool": {"command": "echo", "args": []}},
+            ),
+        )
+        servers = _scan_claude_code(tmp_path)
+        assert servers == []
+
+
+# ---------------------------------------------------------------------------
+# Scanner: Cursor workspace
+# ---------------------------------------------------------------------------
+
+
+class TestScanCursorWorkspace:
+    def test_finds_config_in_cwd(self, tmp_path: Path):
+        cursor_path = tmp_path / ".cursor" / "mcp.json"
+        _write_config(cursor_path, _STDIO_CONFIG)
+        servers = _scan_cursor_workspace(tmp_path)
+        assert len(servers) == 2
+        assert all(s.source == "cursor" for s in servers)
+
+    def test_finds_config_in_parent(self, tmp_path: Path):
+        cursor_path = tmp_path / ".cursor" / "mcp.json"
+        _write_config(cursor_path, _STDIO_CONFIG)
+        child = tmp_path / "src" / "deep"
+        child.mkdir(parents=True)
+        servers = _scan_cursor_workspace(child)
+        assert len(servers) == 2
+
+    def test_stops_at_home(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        # Place config above home — should not be found
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        above_home = tmp_path.parent / ".cursor" / "mcp.json"
+        _write_config(above_home, _STDIO_CONFIG)
+        child = tmp_path / "project"
+        child.mkdir()
+        servers = _scan_cursor_workspace(child)
+        assert servers == []
+
+    def test_no_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        # Confine walk to tmp_path so it doesn't find sibling test dirs
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        servers = _scan_cursor_workspace(tmp_path)
+        assert servers == []
+
+
+# ---------------------------------------------------------------------------
+# Scanner: project mcp.json
+# ---------------------------------------------------------------------------
+
+
+class TestScanProjectMcpJson:
+    def test_finds_config(self, tmp_path: Path):
+        config_path = tmp_path / "mcp.json"
+        _write_config(config_path, _STDIO_CONFIG)
+        servers = _scan_project_mcp_json(tmp_path)
+        assert len(servers) == 2
+        assert all(s.source == "project" for s in servers)
+
+    def test_no_config(self, tmp_path: Path):
+        servers = _scan_project_mcp_json(tmp_path)
+        assert servers == []
+
+
+# ---------------------------------------------------------------------------
+# Scanner: Gemini CLI
+# ---------------------------------------------------------------------------
+
+
+class TestScanGemini:
+    def test_user_level_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        config_path = tmp_path / ".gemini" / "settings.json"
+        _write_config(config_path, _STDIO_CONFIG)
+        servers = _scan_gemini(tmp_path)
+        assert len(servers) == 2
+        assert all(s.source == "gemini" for s in servers)
+
+    def test_project_level_config(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        project_dir = tmp_path / "my-project"
+        project_dir.mkdir()
+        config_path = project_dir / ".gemini" / "settings.json"
+        _write_config(config_path, _STDIO_CONFIG)
+        servers = _scan_gemini(project_dir)
+        assert len(servers) == 2
+
+    def test_http_url_normalized(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        """Gemini uses ``httpUrl`` — verify it becomes ``url``."""
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        config_path = tmp_path / ".gemini" / "settings.json"
+        _write_config(
+            config_path,
+            {
+                "mcpServers": {
+                    "api": {"httpUrl": "https://api.example.com/mcp/"},
+                }
+            },
+        )
+        servers = _scan_gemini(tmp_path)
+        assert len(servers) == 1
+        assert isinstance(servers[0].config, RemoteMCPServer)
+        assert servers[0].config.url == "https://api.example.com/mcp/"
+
+    def test_missing_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        servers = _scan_gemini(tmp_path)
+        assert servers == []
+
+
+# ---------------------------------------------------------------------------
+# Scanner: Goose
+# ---------------------------------------------------------------------------
+
+_GOOSE_CONFIG = {
+    "extensions": {
+        "developer": {
+            "enabled": True,
+            "name": "developer",
+            "type": "builtin",
+        },
+        "tavily": {
+            "cmd": "npx",
+            "args": ["-y", "mcp-tavily-search"],
+            "enabled": True,
+            "envs": {"TAVILY_API_KEY": "xxx"},
+            "type": "stdio",
+        },
+        "disabled-tool": {
+            "cmd": "echo",
+            "args": ["hi"],
+            "enabled": False,
+            "type": "stdio",
+        },
+    }
+}
+
+
+class TestScanGoose:
+    def test_finds_stdio_extensions(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
+        config_dir = tmp_path / ".config" / "goose"
+        config_path = config_dir / "config.yaml"
+        config_path.parent.mkdir(parents=True)
+        config_path.write_text(yaml.dump(_GOOSE_CONFIG))
+        # Force non-windows platform for path logic
+        monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "linux")
+        servers = _scan_goose()
+        assert len(servers) == 1
+        assert servers[0].name == "tavily"
+        assert servers[0].source == "goose"
+        assert isinstance(servers[0].config, StdioMCPServer)
+        assert servers[0].config.command == "npx"
+
+    def test_skips_builtin_and_disabled(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
+        config_dir = tmp_path / ".config" / "goose"
+        config_path = config_dir / "config.yaml"
+        config_path.parent.mkdir(parents=True)
+        config_path.write_text(yaml.dump(_GOOSE_CONFIG))
+        monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "linux")
+        servers = _scan_goose()
+        names = {s.name for s in servers}
+        assert "developer" not in names
+        assert "disabled-tool" not in names
+
+    def test_missing_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+        monkeypatch.setattr("fastmcp.cli.discovery.sys.platform", "linux")
+        servers = _scan_goose()
+        assert servers == []
+
+
+# ---------------------------------------------------------------------------
+# discover_servers
+# ---------------------------------------------------------------------------
+
+
+def _suppress_user_scanners(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Suppress all scanners that read real user config files."""
+    monkeypatch.setattr("fastmcp.cli.discovery._scan_claude_desktop", lambda: [])
+    monkeypatch.setattr("fastmcp.cli.discovery._scan_claude_code", lambda start_dir: [])
+    monkeypatch.setattr("fastmcp.cli.discovery._scan_gemini", lambda start_dir: [])
+    monkeypatch.setattr("fastmcp.cli.discovery._scan_goose", lambda: [])
+
+
+class TestDiscoverServers:
+    def test_combines_sources(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        # Set up project mcp.json
+        project_config = tmp_path / "mcp.json"
+        _write_config(project_config, _STDIO_CONFIG)
+
+        # Set up cursor config
+        cursor_config = tmp_path / ".cursor" / "mcp.json"
+        _write_config(cursor_config, _REMOTE_CONFIG)
+
+        _suppress_user_scanners(monkeypatch)
+
+        servers = discover_servers(start_dir=tmp_path)
+        sources = {s.source for s in servers}
+        assert "project" in sources
+        assert "cursor" in sources
+        assert len(servers) == 3  # 2 from project + 1 from cursor
+
+    def test_preserves_duplicates(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Same server name in multiple sources should appear multiple times."""
+        project_config = tmp_path / "mcp.json"
+        _write_config(project_config, _STDIO_CONFIG)
+
+        cursor_config = tmp_path / ".cursor" / "mcp.json"
+        _write_config(cursor_config, _STDIO_CONFIG)
+
+        _suppress_user_scanners(monkeypatch)
+
+        servers = discover_servers(start_dir=tmp_path)
+        weather_servers = [s for s in servers if s.name == "weather"]
+        assert len(weather_servers) == 2
+        assert {s.source for s in weather_servers} == {"cursor", "project"}
+
+
+# ---------------------------------------------------------------------------
+# resolve_name
+# ---------------------------------------------------------------------------
+
+
+class TestResolveName:
+    @pytest.fixture(autouse=True)
+    def _isolate_scanners(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        """Suppress scanners that read real user configs and confine walks to tmp_path."""
+        _suppress_user_scanners(monkeypatch)
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+
+    def test_unique_match(self, tmp_path: Path):
+        config_path = tmp_path / "mcp.json"
+        _write_config(config_path, _STDIO_CONFIG)
+        transport = resolve_name("weather", start_dir=tmp_path)
+        assert isinstance(transport, StdioTransport)
+
+    def test_qualified_match(self, tmp_path: Path):
+        config_path = tmp_path / "mcp.json"
+        _write_config(config_path, _STDIO_CONFIG)
+        transport = resolve_name("project:weather", start_dir=tmp_path)
+        assert isinstance(transport, StdioTransport)
+
+    def test_not_found_with_servers(self, tmp_path: Path):
+        config_path = tmp_path / "mcp.json"
+        _write_config(config_path, _STDIO_CONFIG)
+        with pytest.raises(ValueError, match="No server named 'nope'.*Available"):
+            resolve_name("nope", start_dir=tmp_path)
+
+    def test_not_found_no_servers(self, tmp_path: Path):
+        with pytest.raises(ValueError, match="No server named 'nope'.*Searched"):
+            resolve_name("nope", start_dir=tmp_path)
+
+    def test_ambiguous_name(self, tmp_path: Path):
+        project_config = tmp_path / "mcp.json"
+        _write_config(project_config, _STDIO_CONFIG)
+        cursor_config = tmp_path / ".cursor" / "mcp.json"
+        _write_config(cursor_config, _STDIO_CONFIG)
+        with pytest.raises(ValueError, match="Ambiguous server name 'weather'"):
+            resolve_name("weather", start_dir=tmp_path)
+
+    def test_ambiguous_resolved_by_qualified(self, tmp_path: Path):
+        project_config = tmp_path / "mcp.json"
+        _write_config(project_config, _STDIO_CONFIG)
+        cursor_config = tmp_path / ".cursor" / "mcp.json"
+        _write_config(cursor_config, _STDIO_CONFIG)
+        transport = resolve_name("cursor:weather", start_dir=tmp_path)
+        assert isinstance(transport, StdioTransport)
+
+    def test_qualified_not_found(self, tmp_path: Path):
+        config_path = tmp_path / "mcp.json"
+        _write_config(config_path, _STDIO_CONFIG)
+        with pytest.raises(
+            ValueError, match="No server named 'nope' found in source 'project'"
+        ):
+            resolve_name("project:nope", start_dir=tmp_path)
+
+    def test_remote_server_resolves_to_http_transport(self, tmp_path: Path):
+        config_path = tmp_path / "mcp.json"
+        _write_config(config_path, _REMOTE_CONFIG)
+        transport = resolve_name("api", start_dir=tmp_path)
+        assert isinstance(transport, StreamableHttpTransport)
+
+
+# ---------------------------------------------------------------------------
+# Integration: resolve_server_spec falls through to name resolution
+# ---------------------------------------------------------------------------
+
+
+class TestResolveServerSpecNameFallback:
+    def test_bare_name_resolves(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        config_path = tmp_path / "mcp.json"
+        _write_config(config_path, _STDIO_CONFIG)
+        _suppress_user_scanners(monkeypatch)
+        monkeypatch.setattr("fastmcp.cli.discovery.Path.home", lambda: tmp_path)
+
+        # Monkeypatch resolve_name in client module to use our tmp_path
+        original_resolve = resolve_name
+
+        def patched_resolve(name: str, start_dir: Path | None = None) -> Any:
+            return original_resolve(name, start_dir=tmp_path)
+
+        monkeypatch.setattr("fastmcp.cli.client.resolve_name", patched_resolve)
+
+        result = resolve_server_spec("weather")
+        assert isinstance(result, StdioTransport)
+
+    def test_url_takes_priority_over_name(self):
+        """URLs should be resolved before name lookup."""
+        result = resolve_server_spec("http://localhost:8000/mcp")
+        assert result == "http://localhost:8000/mcp"
+
+
+# ---------------------------------------------------------------------------
+# Integration: _is_http_target detects transport objects
+# ---------------------------------------------------------------------------
+
+
+class TestIsHttpTargetTransports:
+    def test_streamable_http_transport(self):
+        transport = StreamableHttpTransport("http://localhost:8000/mcp")
+        assert _is_http_target(transport) is True
+
+    def test_sse_transport(self):
+        transport = SSETransport("http://localhost:8000/sse")
+        assert _is_http_target(transport) is True
+
+    def test_stdio_transport(self):
+        transport = StdioTransport(command="echo", args=["hello"])
+        assert _is_http_target(transport) is False
+
+    def test_string_url(self):
+        assert _is_http_target("http://localhost:8000") is True
+
+    def test_string_non_url(self):
+        assert _is_http_target("server.py") is False
+
+    def test_dict_config(self):
+        assert _is_http_target({"mcpServers": {}}) is False