[#12953] feat(mcp-server): select the metalake per tool call

docs/gravitino-mcp-server.md

@@ -71,6 +71,7 @@ Gravitino MCP server supports the following tools, and you could export tool by
 
 | Tool name                           | Description                                                                    | Tag          |
 |-------------------------------------|--------------------------------------------------------------------------------|--------------|
+| `list_metalakes`                    | Retrieve the metalakes the caller can access.                                  | `metalake`   |
 | `get_list_of_catalogs`              | Retrieve a list of all catalogs in the system.                                 | `catalog`    |
 | `create_catalog`                    | Create a new catalog.                                                          | `catalog`    |
 | `alter_catalog`                     | Alter an existing catalog.                                                     | `catalog`    |
@@ -144,7 +145,7 @@ You could config Gravitino MCP server by arguments, `uv run mcp_server -h` shows
 
 | Argument                         | Description                                                                                                                     | Default value               | Required |
 |----------------------------------|---------------------------------------------------------------------------------------------------------------------------------|-----------------------------|----------|
-| `--metalake`                     | The Gravitino metalake name.                                                                                                    | none                        | Yes      |
+| `--metalake`                     | Default metalake, used by any tool call that does not name one. See Selecting a metalake.                                       | none                        | No       |
 | `--gravitino-uri`                | The URI of Gravitino server.                                                                                                    | `http://127.0.0.1:8090`     | No       |
 | `--transport`                    | Transport protocol: stdio (local), http / streamable-http (Streamable HTTP).                                                    | `stdio`                     | No       |
 | `--mcp-url`                      | The URL of MCP server if using HTTP transport.                                                                                  | `http://127.0.0.1:8000/mcp` | No       |
@@ -235,6 +236,48 @@ uv run mcp_server --metalake test --gravitino-uri http://127.0.0.1:8090 \
   --tls-cert /path/to/cert.pem --tls-key /path/to/key.pem
 ```
 
+## Selecting a metalake
+
+A metalake is Gravitino's top-level tenant boundary, and every tool operates inside one. `--metalake` sets the **default**: the metalake used by any tool call that does not name one itself. It is optional on every transport.
+
+Any tool call may name a different metalake with a `metalake` argument, which takes priority over the default. The argument is optional on every tool, so a server configured with `--metalake` behaves exactly as it always has for callers that ignore it.
+
+The metalake for a call is resolved in this order:
+
+1. The call's own `metalake` argument, when it passes one.
+2. The `--metalake` startup default, when it is configured.
+3. Otherwise the call fails, telling the agent to call `list_metalakes` and retry.
+
+Because each call carries its own metalake, one server instance can serve several metalakes at once: nothing is remembered between calls, so concurrent callers never see each other's metalake and the server stays correct however many replicas it runs as. This works identically over stdio and HTTP.
+
+Use the `list_metalakes` tool to discover which metalakes a caller may use. It is the one tool that does not need a metalake, so it works on a server started with no `--metalake` at all.
+
+The statistic tools (`list_statistics_for_metadata`, `list_statistics_for_partition`) shipped their own `metalake_name` argument before metalake selection was unified. It is still accepted as a deprecated alias for `metalake`, so existing callers keep working; passing both with different values is rejected. New callers should use `metalake`.
+
+Authorization is unchanged — the caller's identity (see above) determines what it may see in the named metalake exactly as it would through the REST API. Note that a caller can now reach any metalake its credentials permit, so scope the credentials accordingly when that matters.
+
+### Examples
+
+Single metalake, agents never think about it — the common case, and unchanged:
+
+```bash
+uv run mcp_server --metalake test --gravitino-uri http://127.0.0.1:8090
+```
+
+Several metalakes behind one server, with `prod` as the default:
+
+```bash
+uv run mcp_server --metalake prod --transport http --mcp-url http://0.0.0.0:8000/mcp
+```
+
+An agent then works in `prod` by default and switches per request when asked — "which catalogs are in the staging metalake?" sends `metalake=staging` on that call alone, without restarting or reconfiguring anything.
+
+No default at all, every call chooses:
+
+```bash
+uv run mcp_server --transport http --mcp-url http://0.0.0.0:8000/mcp
+```
+
 ## Audit Logging
 
 Every tool invocation is recorded as one structured JSON line in `gravitino-mcp-audit.log` (written to the server's working directory). Each record is attributed to the incoming HTTP `Authorization` header when present; otherwise to the configured service identity (`--token` or OAuth client id).

mcp-server/mcp_server/client/__init__.py

@@ -18,6 +18,7 @@
 from mcp_server.client.catalog_operation import CatalogOperation
 from mcp_server.client.gravitino_operation import GravitinoOperation
 from mcp_server.client.job_operation import JobOperation
+from mcp_server.client.metalake_operation import MetalakeOperation
 from mcp_server.client.model_operation import ModelOperation
 from mcp_server.client.policy_operation import PolicyOperation
 from mcp_server.client.schema_operation import SchemaOperation

mcp-server/mcp_server/client/gravitino_operation.py

@@ -20,6 +20,7 @@
 from mcp_server.client.catalog_operation import CatalogOperation
 from mcp_server.client.fileset_operation import FilesetOperation
 from mcp_server.client.job_operation import JobOperation
+from mcp_server.client.metalake_operation import MetalakeOperation
 from mcp_server.client.model_operation import ModelOperation
 from mcp_server.client.partition_operation import PartitionOperation
 from mcp_server.client.policy_operation import PolicyOperation
@@ -154,3 +155,13 @@ def as_view_operation(self) -> ViewOperation:
             ViewOperation: Interface for performing view-level operations
         """
         pass
+
+    @abstractmethod
+    def as_metalake_operation(self) -> MetalakeOperation:
+        """
+        Access the metalake operation interface of this Gravitino operation.
+
+        Returns:
+            MetalakeOperation: Interface for metalake-level operations
+        """
+        pass

mcp-server/mcp_server/client/metalake_operation.py

@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from abc import ABC, abstractmethod
+
+
+class MetalakeOperation(ABC):
+    """
+    Abstract base class for Gravitino metalake operations.
+
+    Unlike every other operation, these are not scoped to a single metalake:
+    they address the server's top-level ``/api/metalakes`` endpoint, so an
+    agent can discover which metalakes it may operate on before naming one.
+    """
+
+    @abstractmethod
+    async def get_list_of_metalakes(self) -> str:
+        """
+        Retrieve the list of metalakes the caller is allowed to see.
+
+        Returns:
+            str: JSON-formatted string containing metalake information.
+        """
+        pass

mcp-server/mcp_server/client/plain/plain_rest_client_metalake_operation.py

@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from httpx import AsyncClient
+
+from mcp_server.client.metalake_operation import MetalakeOperation
+from mcp_server.client.plain.utils import extract_content_from_response
+
+
+class PlainRESTClientMetalakeOperation(MetalakeOperation):
+    """Metalake operations against the server's top-level endpoint.
+
+    Takes no metalake name: this is the one operation that must work before a
+    metalake has been chosen.
+    """
+
+    def __init__(self, rest_client: AsyncClient):
+        self.rest_client = rest_client
+
+    async def get_list_of_metalakes(self) -> str:
+        response = await self.rest_client.get("/api/metalakes")
+        return extract_content_from_response(response, "metalakes", [])

mcp-server/mcp_server/client/plain/plain_rest_client_operation.py

@@ -22,6 +22,7 @@
 from mcp_server.client import (
     CatalogOperation,
     GravitinoOperation,
+    MetalakeOperation,
     ModelOperation,
     PolicyOperation,
     SchemaOperation,
@@ -38,6 +39,9 @@
 from mcp_server.client.plain.plain_rest_client_job_operation import (
     PlainRESTClientJobOperation,
 )
+from mcp_server.client.plain.plain_rest_client_metalake_operation import (
+    PlainRESTClientMetalakeOperation,
+)
 from mcp_server.client.plain.plain_rest_client_model_operation import (
     PlainRESTClientModelOperation,
 )
@@ -134,11 +138,18 @@ def __init__(
         self._view_operation = PlainRESTClientViewOperation(
             metalake_name, _rest_client
         )
+        # Not metalake-scoped: addresses the server's top-level endpoint.
+        self._metalake_operation = PlainRESTClientMetalakeOperation(
+            _rest_client
+        )
 
     async def close(self) -> None:
         """Close the shared httpx client and release its connection pool."""
         await self._rest_client.aclose()
 
+    def as_metalake_operation(self) -> MetalakeOperation:
+        return self._metalake_operation
+
     def as_catalog_operation(self) -> CatalogOperation:
         return self._catalog_operation
 

mcp-server/mcp_server/client/plain/plain_rest_client_statistic_operation.py

@@ -28,10 +28,10 @@ def __init__(self, metalake_name: str, rest_client):
         self.rest_client = rest_client
 
     async def list_of_statistics(
-        self, metalake_name: str, metadata_type: str, metadata_fullname: str
+        self, metadata_type: str, metadata_fullname: str
     ) -> str:
         response = await self.rest_client.get(
-            f"/api/metalakes/{encode_path_segment(metalake_name)}"
+            f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
             f"/objects/{encode_path_segment(metadata_type)}"
             f"/{encode_path_segment(metadata_fullname)}/statistics"
         )
@@ -40,7 +40,6 @@ async def list_of_statistics(
     # pylint: disable=R0917
     async def list_statistic_for_partition(
         self,
-        metalake_name: str,
         metadata_type: str,
         metadata_fullname: str,
         from_partition_name: str,
@@ -49,7 +48,7 @@ async def list_statistic_for_partition(
         to_inclusive: bool = False,
     ) -> str:
         response = await self.rest_client.get(
-            f"/api/metalakes/{encode_path_segment(metalake_name)}"
+            f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
             f"/objects/{encode_path_segment(metadata_type)}"
             f"/{encode_path_segment(metadata_fullname)}/statistics/partitions",
             params={

mcp-server/mcp_server/client/statistic_operation.py

@@ -25,12 +25,11 @@ class StatisticOperation(ABC):
 
     @abstractmethod
     async def list_of_statistics(
-        self, metalake_name: str, metadata_type: str, metadata_fullname: str
+        self, metadata_type: str, metadata_fullname: str
     ) -> str:
         """
         Retrieve the list of statistics for a specific metadata type and fullname within a metalake.
         Args:
-            metalake_name: Name of the metalake
             metadata_type: Type of metadata (e.g., table, column)
             metadata_fullname: Full name of the metadata item
 
@@ -43,7 +42,6 @@ async def list_of_statistics(
     @abstractmethod
     async def list_statistic_for_partition(
         self,
-        metalake_name: str,
         metadata_type: str,
         metadata_fullname: str,
         from_partition_name: str,
@@ -57,7 +55,6 @@ async def list_statistic_for_partition(
             So `metadata_type` should always be "table".
 
         Args:
-            metalake_name: Name of the metalake
             metadata_type: Type of metadata, should be "table" for partition statistics
             metadata_fullname: Full name of the metadata item, the format should be
                 "{catalog}.{schema}.{table}".

mcp-server/mcp_server/core/audit.py

@@ -63,6 +63,7 @@ def emit(
     tool: str,
     outcome: str,
     error_type: str = "",
+    metalake: str = "",
 ) -> None:
     """Write one structured JSON audit record to the audit logger.
 
@@ -74,13 +75,22 @@ def emit(
                    authorization denial being the common case), not only
                    authorization failures; inspect error_type to disambiguate.
         error_type: Exception class name when outcome is "deny", empty otherwise.
+        metalake:  Metalake the call operated on, resolved - so a call that
+                   relied on the server default records that default. Empty
+                   only for tools that are not metalake-scoped, such as the
+                   metalake listing, which spans every tenant the caller can
+                   see. Recorded because one server can now serve several
+                   metalakes, so "which tenant did this touch" is no longer
+                   answerable from the server config alone.
     """
     record = {
         "timestamp": datetime.now(timezone.utc).isoformat(),
         "principal": principal,
         "tool": tool,
         "outcome": outcome,
     }
+    if metalake:
+        record["metalake"] = metalake
     if error_type:
         record["error_type"] = error_type