Add Mason MCP service discovery

integrations/mason/README.md

@@ -86,16 +86,15 @@ mason [-p <profile>] [-o text|json]
   tracing
     setup      --catalog C --schema S [--experiment E]
     list | get | instrument
+  mcp
+    list             [--schema CATALOG.SCHEMA]
   init          [--framework openai|langgraph] [--profile P] [DIRECTORY]
   tools
     add sandbox      --scope SCOPE [--scope SCOPE ...] [--source PATH]
     add mcp          SERVICE [--name NAME] [--source PATH]
     add uc-function  FUNCTION [--name NAME] [--source PATH]
     add python       NAME [--source PATH]
     list             [--source PATH]
-  add-sandbox  --scope SCOPE [--scope SCOPE ...]
-               [--permission read_only|read_write] [--source PATH]
-               [--framework openai|langgraph]
   deploy       <name> --source PATH [--with-memory-store N]
                [--with-session-store N] [--actor-id ID]
                [--with-traces C.S] [--create-stores]
@@ -121,6 +120,16 @@ mason tools add python lookup-ticket
 mason tools list
 ```
 
+Discover the MCP Services available to your user before adding one. By default Mason lists the
+Databricks-managed services in `system.ai`; pass `--schema catalog.schema` for another Unity Catalog
+schema. Text output includes a copyable add command, while `--output json` returns normalized service
+records for scripts:
+
+```sh
+mason mcp list
+mason mcp list --schema main.tools
+```
+
 The Python command additionally creates user-owned `agent/tools/<name>.py` and
 `tests/tools/test_<name>.py` files using the LangGraph-native `@tool` decorator. `mason dev` and
 `mason deploy` preserve `agent.toml`; they do not generate or patch agent source.
@@ -130,10 +139,6 @@ Sandbox scopes default to read-only access. Repeat `--scope` to allow more than
 agent needs writes. Every sandbox call carries this fixed downscope in MCP `_meta`, outside the tool
 arguments controlled by the model.
 
-`mason add-sandbox` remains as a compatibility alias. For manifest-backed projects it follows the
-same LangGraph-only behavior as `mason tools add sandbox`; its older source-editing path remains for
-legacy projects that do not yet contain `agent.toml`.
-
 ## Initialize the chat app demo
 
 The chat app is a LangGraph-specific init overlay, not a command that mutates an existing project:

integrations/mason/src/databricks_mason/cli.py

@@ -15,8 +15,8 @@
 from databricks_mason.deploy import deploy, deployments
 from databricks_mason.dev import dev
 from databricks_mason.init import init
+from databricks_mason.mcp import mcp
 from databricks_mason.memory import memory
-from databricks_mason.sandbox import add_sandbox
 from databricks_mason.sessions import sessions
 from databricks_mason.tools import tools
 from databricks_mason.tracing import tracing
@@ -64,11 +64,11 @@ def mason(ctx: click.Context, profile: Optional[str], output: str) -> None:
 mason.add_command(init)
 mason.add_command(dev)
 mason.add_command(memory)
+mason.add_command(mcp)
 mason.add_command(sessions)
 mason.add_command(tracing)
 mason.add_command(deploy)
 mason.add_command(deployments)
-mason.add_command(add_sandbox)
 mason.add_command(tools)
 
 

integrations/mason/src/databricks_mason/client.py

@@ -22,6 +22,7 @@
 from databricks_mason.errors import AgentCliError, wrap_api_error
 
 _BASE = "/api/agents/v1"
+_MCP_SERVICES_PATH = "/api/2.1/unity-catalog/mcp-services"
 
 
 def _query(**kwargs: Any) -> dict[str, Any]:
@@ -122,6 +123,18 @@ def _do(
         except Exception as exc:  # noqa: BLE001 - normalized to AgentCliError
             raise wrap_api_error(exc) from exc
 
+    # --- Unity Catalog MCP Services -----------------------------------------
+
+    def list_mcp_services(
+        self, schema: str = "system.ai", page_token: Optional[str] = None
+    ) -> dict:
+        """List MCP Services visible to the user in a Unity Catalog schema."""
+        return self._do(
+            "GET",
+            _MCP_SERVICES_PATH,
+            query=_query(parent=f"schemas/{schema}", page_token=page_token),
+        )
+
     # --- memory stores -------------------------------------------------------
 
     def create_memory_store(

integrations/mason/src/databricks_mason/mcp.py

@@ -0,0 +1,89 @@
+"""Discover Unity Catalog MCP Services that can be added to an agent."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import click
+
+from databricks_mason import render
+from databricks_mason.errors import AgentCliError
+
+_RESOURCE_PREFIX = "mcp-services/"
+
+
+def _validate_schema(schema: str) -> str:
+    parts = schema.strip().split(".")
+    if (
+        len(parts) != 2
+        or any(not part for part in parts)
+        or any(character.isspace() for character in schema)
+    ):
+        raise AgentCliError(
+            f"Invalid schema {schema!r}.",
+            hint="Use a two-part Unity Catalog schema name: catalog.schema.",
+        )
+    return schema.strip()
+
+
+def _service_record(service: Any) -> dict[str, str] | None:
+    if not isinstance(service, dict):
+        return None
+    raw_name = service.get("name")
+    if not isinstance(raw_name, str) or not raw_name:
+        return None
+    name = raw_name.removeprefix(_RESOURCE_PREFIX)
+    record = {"name": name}
+    for field in ("id", "comment"):
+        value = service.get(field)
+        if isinstance(value, str) and value:
+            record[field] = value
+    return record
+
+
+def _list_services(client: Any, schema: str) -> list[dict[str, str]]:
+    by_name: dict[str, dict[str, str]] = {}
+    page_token = None
+    while True:
+        response = client.list_mcp_services(schema, page_token=page_token)
+        if not isinstance(response, dict):
+            raise AgentCliError("The MCP Services API returned an invalid response.")
+        services = response.get("mcp_services", [])
+        if not isinstance(services, list):
+            raise AgentCliError("The MCP Services API returned an invalid response.")
+        for service in services:
+            record = _service_record(service)
+            if record is not None and record["name"] not in by_name:
+                by_name[record["name"]] = record
+        page_token = response.get("next_page_token")
+        if not isinstance(page_token, str) or not page_token:
+            break
+    return [by_name[name] for name in sorted(by_name)]
+
+
+@click.group()
+def mcp() -> None:
+    """Discover managed MCP Services available through Unity Catalog."""
+
+
+@mcp.command("list")
+@click.option(
+    "--schema",
+    default="system.ai",
+    show_default=True,
+    help="Two-part Unity Catalog schema containing MCP Services.",
+)
+@click.pass_obj
+def list_mcp(obj: Any, schema: str) -> None:
+    """List MCP Services that can be added with ``mason tools add mcp``."""
+    schema = _validate_schema(schema)
+    services = _list_services(obj.client(), schema)
+    if getattr(obj, "output", "text") == "json":
+        render.emit_json({"schema_version": 1, "mcp_services": services})
+        return
+    render.resource_table(
+        "MCP Services",
+        [("Service", "left"), ("Add command", "left")],
+        [(service["name"], f"mason tools add mcp {service['name']}") for service in services],
+        subtitle=f"Available in {schema}",
+    )

integrations/mason/tests/unit_tests/cli_test.py

@@ -14,7 +14,7 @@ def test_sessions_verbs_are_flat_no_redundant_subgroup():
     assert {"stores", "items"} <= names
 
 
-def test_root_registers_login_and_logout():
+def test_root_registers_supported_commands():
     names = set(cli.mason.commands)
     assert {
         "login",
@@ -24,6 +24,7 @@ def test_root_registers_login_and_logout():
         "tracing",
         "deploy",
         "deployments",
-        "add-sandbox",
+        "mcp",
         "tools",
     } <= names
+    assert "add-sandbox" not in names

integrations/mason/tests/unit_tests/client_test.py

@@ -41,6 +41,20 @@ def test_list_memory_stores_query(workspace_client):
     )
 
 
+@mock.patch("databricks_mason.client.WorkspaceClient")
+def test_list_mcp_services_query(workspace_client):
+    c, do = _client(workspace_client)
+
+    c.list_mcp_services("system.ai", page_token="next")
+
+    do.assert_called_once_with(
+        "GET",
+        "/api/2.1/unity-catalog/mcp-services",
+        query={"parent": "schemas/system.ai", "page_token": "next"},
+        body=None,
+    )
+
+
 @mock.patch("databricks_mason.client.WorkspaceClient")
 def test_get_memory_store_normalizes_id(workspace_client):
     c, do = _client(workspace_client)

integrations/mason/tests/unit_tests/mcp_test.py

@@ -0,0 +1,97 @@
+"""Unit tests for ``mason mcp list`` discovery."""
+
+from __future__ import annotations
+
+import json
+
+from click.testing import CliRunner
+
+from databricks_mason.mcp import mcp
+
+
+class _Client:
+    def __init__(self, pages):
+        self.pages = iter(pages)
+        self.calls = []
+
+    def list_mcp_services(self, schema, page_token=None):
+        self.calls.append((schema, page_token))
+        return next(self.pages)
+
+
+class _Ctx:
+    def __init__(self, client, output="text"):
+        self._client = client
+        self.output = output
+
+    def client(self):
+        return self._client
+
+
+def test_list_json_normalizes_sorts_deduplicates_and_paginates():
+    client = _Client(
+        [
+            {
+                "mcp_services": [
+                    {
+                        "name": "mcp-services/system.ai.web_search",
+                        "id": "web-id",
+                        "comment": "Search the web",
+                    },
+                    {"name": "mcp-services/system.ai.slack", "id": "slack-id"},
+                    {"id": "missing-name"},
+                ],
+                "next_page_token": "page-2",
+            },
+            {
+                "mcp_services": [
+                    {"name": "mcp-services/system.ai.slack", "id": "duplicate"},
+                    {"name": "mcp-services/system.ai.github", "id": "github-id"},
+                ]
+            },
+        ]
+    )
+
+    result = CliRunner().invoke(mcp, ["list"], obj=_Ctx(client, output="json"))
+
+    assert result.exit_code == 0, result.output
+    assert client.calls == [("system.ai", None), ("system.ai", "page-2")]
+    assert json.loads(result.output) == {
+        "schema_version": 1,
+        "mcp_services": [
+            {"name": "system.ai.github", "id": "github-id"},
+            {"name": "system.ai.slack", "id": "slack-id"},
+            {
+                "name": "system.ai.web_search",
+                "id": "web-id",
+                "comment": "Search the web",
+            },
+        ],
+    }
+
+
+def test_list_text_shows_copyable_add_command_and_schema_override():
+    client = _Client(
+        [{"mcp_services": [{"name": "mcp-services/main.tools.ticket_search", "id": "ticket-id"}]}]
+    )
+
+    result = CliRunner().invoke(
+        mcp,
+        ["list", "--schema", "main.tools"],
+        obj=_Ctx(client),
+    )
+
+    assert result.exit_code == 0, result.output
+    assert client.calls == [("main.tools", None)]
+    assert "main.tools.ticket_search" in result.output
+    assert "mason tools add mcp main.tools.ticket_search" in result.output
+
+
+def test_list_rejects_invalid_schema_before_api_call():
+    client = _Client([])
+
+    result = CliRunner().invoke(mcp, ["list", "--schema", "system"], obj=_Ctx(client))
+
+    assert result.exit_code != 0
+    assert "catalog.schema" in result.output
+    assert client.calls == []

integrations/mason/tests/unit_tests/tools_test.py

@@ -11,7 +11,6 @@
 
 from databricks_mason.agent_project import AgentProject
 from databricks_mason.project_config import write_project_metadata
-from databricks_mason.sandbox import add_sandbox as legacy_add_sandbox
 from databricks_mason.tools import tools
 
 
@@ -46,20 +45,6 @@ def test_add_sandbox_only_updates_manifest(tmp_path: pathlib.Path):
     assert (project / "agent" / "mcps.py").read_text(encoding="utf-8") == "ORIGINAL = True\n"
 
 
-def test_top_level_add_sandbox_alias_delegates_for_manifest_projects(tmp_path: pathlib.Path):
-    project = _project(tmp_path)
-
-    result = CliRunner().invoke(
-        legacy_add_sandbox,
-        ["--scope", "table:samples.nyctaxi.trips", "--source", str(project)],
-        obj=_Ctx(),
-    )
-
-    assert result.exit_code == 0, result.output
-    assert AgentProject.load(project).tools[0].source.kind == "sandbox"
-    assert (project / "agent" / "mcps.py").read_text(encoding="utf-8") == "ORIGINAL = True\n"
-
-
 def test_generic_mcp_rejects_sandbox_scope(tmp_path: pathlib.Path):
     project = _project(tmp_path)